Skip to content

Commit

Permalink
Test cleanup. Added basic threaded test.
Browse files Browse the repository at this point in the history
  • Loading branch information
Daniel Lundin committed Jul 20, 2010
1 parent 1989586 commit 4c85c1e
Show file tree
Hide file tree
Showing 3 changed files with 35 additions and 51 deletions.
6 changes: 1 addition & 5 deletions libuuid/__init__.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,5 @@
from libuuid._uuid import *

from uuid import UUID as _UUID
from uuid import (NAMESPACE_DNS, NAMESPACE_OID, NAMESPACE_URL, NAMESPACE_X500,
RESERVED_FUTURE, RESERVED_MICROSOFT, RESERVED_NCS, RFC_4122)



from libuuid._uuid import *

3 changes: 1 addition & 2 deletions setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,14 +6,13 @@
__version__ = ".".join(map(str, __version_info__))
__author__ = "Daniel Lundin"
__contact__ = "[email protected]"
__homepage__ = "http://github.com/dln/pylibuuid/"
__homepage__ = "http://github.com/dln/python-libuuid/"
__docformat__ = "restructuredtext"

import codecs
import os
from glob import glob


try:
import setuptools
except ImportError:
Expand Down
77 changes: 33 additions & 44 deletions tests/test_uuid.py
Original file line number Diff line number Diff line change
@@ -1,71 +1,50 @@
# -*- coding: utf-8 -*-
from random import choice, randrange
import os
import shutil
import string
import sys
import tempfile
from threading import Thread
from Queue import Queue
import time
import unittest
import uuid

import libuuid


def test_properties():
def test_property():
_PROPERTIES = [
'bytes', 'bytes_le', 'clock_seq', 'clock_seq_hi_variant',
'clock_seq_low', 'fields', 'hex', 'node', 'time', 'time_hi_version',
'time_low', 'time_mid', 'urn', 'variant', 'version'
]

def _check_property(factory_func, prop):
u = factory_func()
u2 = uuid.UUID(bytes=u.bytes)
a1 = getattr(u, prop)
a2 = getattr(u2, prop)
assert a1 == a2

'time_low', 'time_mid', 'urn', 'variant', 'version']
def _check_property(func_name, prop):
u = getattr(libuuid, func_name)()
c = uuid.UUID(bytes=u.bytes)
assert getattr(u, prop) == getattr(c, prop)
for prop in _PROPERTIES:
test_properties.__doc__ = "Property: " + prop
yield _check_property, libuuid.uuid1, prop
yield _check_property, libuuid.uuid4, prop
yield _check_property, 'uuid1', prop
yield _check_property, 'uuid4', prop


def test_methods():
def test_method():
_METHODS = [
'__hash__', '__int__', '__repr__', '__str__', 'get_bytes',
'get_bytes_le', 'get_clock_seq', 'get_clock_seq_hi_variant',
'get_clock_seq_low', 'get_fields', 'get_hex', 'get_node', 'get_time',
'get_time_hi_version', 'get_time_low', 'get_time_mid', 'get_urn',
'get_variant', 'get_version'
]

def _check_method(factory_func, method, *args, **kwargs):
u = factory_func()
u2 = uuid.UUID(bytes=u.bytes)
m1 = getattr(u, method)(*args, **kwargs)
m2 = getattr(u2, method)(*args, **kwargs)
assert m1 == m2

'get_variant', 'get_version']
def _check_method(func_name, method):
u = getattr(libuuid, func_name)()
c = uuid.UUID(bytes=u.bytes)
assert getattr(u, method)() == getattr(c, method)()
for method in _METHODS:
test_methods.__doc__ = "Method: " + method
yield _check_method, libuuid.uuid1, method
yield _check_method, libuuid.uuid4, method
yield _check_method, 'uuid1', method
yield _check_method, 'uuid4', method


def test_constants():

_CONSTANTS = ['NAMESPACE_DNS', 'NAMESPACE_OID', 'NAMESPACE_URL',
'NAMESPACE_X500', 'RESERVED_FUTURE', 'RESERVED_MICROSOFT',
'RESERVED_NCS', 'RFC_4122']

def _check_constant(const):
m1 = getattr(libuuid, const)
m2 = getattr(uuid, const)
assert m1 == m2

assert getattr(libuuid, const) == getattr(uuid, const)
for constant in _CONSTANTS:
test_methods.__doc__ = "Constant: " + constant
yield _check_constant, constant


Expand Down Expand Up @@ -122,7 +101,17 @@ def test_basic_sanity_uuid1(self):
diff = abs(t - clocks[-1])
self.assert_(diff < 10000, "Timestamp reasonable")




def test_multiple_threads(self):
q = Queue()
def _runsome():
for _ in xrange(200):
q.put(libuuid.uuid4().hex)
q.put(libuuid.uuid1().hex)
threads = [Thread(target=_runsome) for _ in xrange(50)]
for t in threads:
t.start()
for t in threads:
t.join()
result = list(q.queue)
self.assertEquals(len(result), len(set(result)))

0 comments on commit 4c85c1e

Please sign in to comment.