dinislage/CouchPotatoServer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Include module for desktop build

Browse Source
This commit is contained in:
Ruud
2012-09-09 17:29:09 +02:00
parent 89426d75ab
commit 36f653f65b
88 changed files with 9967 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/pyutil/test/__init__.py
View File
libs/pyutil/test/current/__init__.py
View File
libs/pyutil/test/current/json_tests/__init__.py
View File
libs/pyutil/test/current/json_tests/test_decode.py
+15
View File
@@ -0,0 +1,15 @@
import decimal
from unittest import TestCase
from pyutil import jsonutil as json
class TestDecode(TestCase):
def test_decimal(self):
rval = json.loads('1.1', parse_float=decimal.Decimal)
self.assert_(isinstance(rval, decimal.Decimal))
self.assertEquals(rval, decimal.Decimal('1.1'))
def test_float(self):
rval = json.loads('1', parse_int=float)
self.assert_(isinstance(rval, float))
self.assertEquals(rval, 1.0)
libs/pyutil/test/current/json_tests/test_default.py
+9
View File
@@ -0,0 +1,9 @@
from unittest import TestCase
from pyutil import jsonutil as json
class TestDefault(TestCase):
def test_default(self):
self.assertEquals(
json.dumps(type, default=repr),
json.dumps(repr(type)))
libs/pyutil/test/current/json_tests/test_dump.py
+13
View File
@@ -0,0 +1,13 @@
from unittest import TestCase
from cStringIO import StringIO
from pyutil import jsonutil as json
class TestDump(TestCase):
def test_dump(self):
sio = StringIO()
json.dump({}, sio)
self.assertEquals(sio.getvalue(), '{}')
def test_dumps(self):
self.assertEquals(json.dumps({}), '{}')
libs/pyutil/test/current/json_tests/test_encode_basestring_ascii.py
+36
View File
@@ -0,0 +1,36 @@
from twisted.trial.unittest import SkipTest, TestCase
from pyutil.jsonutil import encoder
CASES = [
(u'/\\"\ucafe\ubabe\uab98\ufcde\ubcda\uef4a\x08\x0c\n\r\t`1~!@#$%^&*()_+-=[]{}|;:\',./<>?', '"/\\\\\\"\\ucafe\\ubabe\\uab98\\ufcde\\ubcda\\uef4a\\b\\f\\n\\r\\t`1~!@#$%^&*()_+-=[]{}|;:\',./<>?"'),
(u'\u0123\u4567\u89ab\ucdef\uabcd\uef4a', '"\\u0123\\u4567\\u89ab\\ucdef\\uabcd\\uef4a"'),
(u'controls', '"controls"'),
(u'\x08\x0c\n\r\t', '"\\b\\f\\n\\r\\t"'),
(u'{"object with 1 member":["array with 1 element"]}', '"{\\"object with 1 member\\":[\\"array with 1 element\\"]}"'),
(u' s p a c e d ', '" s p a c e d "'),
(u'\U0001d120', '"\\ud834\\udd20"'),
(u'\u03b1\u03a9', '"\\u03b1\\u03a9"'),
('\xce\xb1\xce\xa9', '"\\u03b1\\u03a9"'),
(u'\u03b1\u03a9', '"\\u03b1\\u03a9"'),
('\xce\xb1\xce\xa9', '"\\u03b1\\u03a9"'),
(u'\u03b1\u03a9', '"\\u03b1\\u03a9"'),
(u'\u03b1\u03a9', '"\\u03b1\\u03a9"'),
(u"`1~!@#$%^&*()_+-={':[,]}|;.</>?", '"`1~!@#$%^&*()_+-={\':[,]}|;.</>?"'),
(u'\x08\x0c\n\r\t', '"\\b\\f\\n\\r\\t"'),
(u'\u0123\u4567\u89ab\ucdef\uabcd\uef4a', '"\\u0123\\u4567\\u89ab\\ucdef\\uabcd\\uef4a"'),
]
class TestEncodeBaseStringAscii(TestCase):
def test_py_encode_basestring_ascii(self):
self._test_encode_basestring_ascii(encoder.py_encode_basestring_ascii)
def test_c_encode_basestring_ascii(self):
if not encoder.c_encode_basestring_ascii:
raise SkipTest("no C extension speedups available to test")
self._test_encode_basestring_ascii(encoder.c_encode_basestring_ascii)
def _test_encode_basestring_ascii(self, encode_basestring_ascii):
for input_string, expect in CASES:
result = encode_basestring_ascii(input_string)
self.assertEquals(result, expect)
libs/pyutil/test/current/json_tests/test_fail.py
+76
View File
@@ -0,0 +1,76 @@
from unittest import TestCase
from pyutil import jsonutil as json
# Fri Dec 30 18:57:26 2005
JSONDOCS = [
# http://json.org/JSON_checker/test/fail1.json
'"A JSON payload should be an object or array, not a string."',
# http://json.org/JSON_checker/test/fail2.json
'["Unclosed array"',
# http://json.org/JSON_checker/test/fail3.json
'{unquoted_key: "keys must be quoted}',
# http://json.org/JSON_checker/test/fail4.json
'["extra comma",]',
# http://json.org/JSON_checker/test/fail5.json
'["double extra comma",,]',
# http://json.org/JSON_checker/test/fail6.json
'[ , "<-- missing value"]',
# http://json.org/JSON_checker/test/fail7.json
'["Comma after the close"],',
# http://json.org/JSON_checker/test/fail8.json
'["Extra close"]]',
# http://json.org/JSON_checker/test/fail9.json
'{"Extra comma": true,}',
# http://json.org/JSON_checker/test/fail10.json
'{"Extra value after close": true} "misplaced quoted value"',
# http://json.org/JSON_checker/test/fail11.json
'{"Illegal expression": 1 + 2}',
# http://json.org/JSON_checker/test/fail12.json
'{"Illegal invocation": alert()}',
# http://json.org/JSON_checker/test/fail13.json
'{"Numbers cannot have leading zeroes": 013}',
# http://json.org/JSON_checker/test/fail14.json
'{"Numbers cannot be hex": 0x14}',
# http://json.org/JSON_checker/test/fail15.json
'["Illegal backslash escape: \\x15"]',
# http://json.org/JSON_checker/test/fail16.json
'["Illegal backslash escape: \\\'"]',
# http://json.org/JSON_checker/test/fail17.json
'["Illegal backslash escape: \\017"]',
# http://json.org/JSON_checker/test/fail18.json
'[[[[[[[[[[[[[[[[[[[["Too deep"]]]]]]]]]]]]]]]]]]]]',
# http://json.org/JSON_checker/test/fail19.json
'{"Missing colon" null}',
# http://json.org/JSON_checker/test/fail20.json
'{"Double colon":: null}',
# http://json.org/JSON_checker/test/fail21.json
'{"Comma instead of colon", null}',
# http://json.org/JSON_checker/test/fail22.json
'["Colon instead of comma": false]',
# http://json.org/JSON_checker/test/fail23.json
'["Bad value", truth]',
# http://json.org/JSON_checker/test/fail24.json
"['single quote']",
# http://code.google.com/p/simplejson/issues/detail?id=3
u'["A\u001FZ control characters in string"]',
]
SKIPS = {
1: "why not have a string payload?",
18: "spec doesn't specify any nesting limitations",
}
class TestFail(TestCase):
def test_failures(self):
for idx, doc in enumerate(JSONDOCS):
idx = idx + 1
if idx in SKIPS:
json.loads(doc)
continue
try:
json.loads(doc)
except ValueError:
pass
else:
self.fail("Expected failure for fail%d.json: %r" % (idx, doc))
libs/pyutil/test/current/json_tests/test_float.py
+9
View File
@@ -0,0 +1,9 @@
import math
from unittest import TestCase
from pyutil import jsonutil as json
class TestFloat(TestCase):
def test_floats(self):
for num in [1617161771.7650001, math.pi, math.pi**100, math.pi**-100]:
self.assertEquals(float(json.dumps(num)), num)
libs/pyutil/test/current/json_tests/test_indent.py
+41
View File
@@ -0,0 +1,41 @@
from unittest import TestCase
from pyutil import jsonutil as json
import textwrap
class TestIndent(TestCase):
def test_indent(self):
h = [['blorpie'], ['whoops'], [], 'd-shtaeou', 'd-nthiouh', 'i-vhbjkhnth',
{'nifty': 87}, {'field': 'yes', 'morefield': False} ]
expect = textwrap.dedent("""\
[
[
"blorpie"
],
[
"whoops"
],
[],
"d-shtaeou",
"d-nthiouh",
"i-vhbjkhnth",
{
"nifty": 87
},
{
"field": "yes",
"morefield": false
}
]""")
d1 = json.dumps(h)
d2 = json.dumps(h, indent=2, sort_keys=True, separators=(',', ': '))
h1 = json.loads(d1)
h2 = json.loads(d2)
self.assertEquals(h1, h)
self.assertEquals(h2, h)
self.assertEquals(d2, expect)
libs/pyutil/test/current/json_tests/test_pass1.py
+71
View File
@@ -0,0 +1,71 @@
from unittest import TestCase
from pyutil import jsonutil as json
# from http://json.org/JSON_checker/test/pass1.json
JSON = r'''
[
"JSON Test Pattern pass1",
{"object with 1 member":["array with 1 element"]},
{},
[],
-42,
true,
false,
null,
{
"integer": 1234567890,
"real": -9876.543210,
"e": 0.123456789e-12,
"E": 1.234567890E+34,
"": 23456789012E666,
"zero": 0,
"one": 1,
"space": " ",
"quote": "\"",
"backslash": "\\",
"controls": "\b\f\n\r\t",
"slash": "/ & \/",
"alpha": "abcdefghijklmnopqrstuvwyz",
"ALPHA": "ABCDEFGHIJKLMNOPQRSTUVWYZ",
"digit": "0123456789",
"special": "`1~!@#$%^&*()_+-={':[,]}|;.</>?",
"hex": "\u0123\u4567\u89AB\uCDEF\uabcd\uef4A",
"true": true,
"false": false,
"null": null,
"array":[ ],
"object":{ },
"address": "50 St. James Street",
"url": "http://www.JSON.org/",
"comment": "// /* <!-- --",
"# -- --> */": " ",
" s p a c e d " :[1,2 , 3
,
4 , 5 , 6 ,7 ],
"compact": [1,2,3,4,5,6,7],
"jsontext": "{\"object with 1 member\":[\"array with 1 element\"]}",
"quotes": "&#34; \u0022 %22 0x22 034 &#x22;",
"\/\\\"\uCAFE\uBABE\uAB98\uFCDE\ubcda\uef4A\b\f\n\r\t`1~!@#$%^&*()_+-=[]{}|;:',./<>?"
: "A key can be any string"
},
0.5 ,98.6
,
99.44
,
1066
,"rosebud"]
'''
class TestPass1(TestCase):
def test_parse(self):
# test in/out equivalence and parsing
res = json.loads(JSON)
out = json.dumps(res)
self.assertEquals(res, json.loads(out))
self.failUnless("2.3456789012E+676" in json.dumps(res, allow_nan=False))
libs/pyutil/test/current/json_tests/test_pass2.py
+14
View File
@@ -0,0 +1,14 @@
from unittest import TestCase
from pyutil import jsonutil as json
# from http://json.org/JSON_checker/test/pass2.json
JSON = r'''
[[[[[[[[[[[[[[[[[[["Not too deep"]]]]]]]]]]]]]]]]]]]
'''
class TestPass2(TestCase):
def test_parse(self):
# test in/out equivalence and parsing
res = json.loads(JSON)
out = json.dumps(res)
self.assertEquals(res, json.loads(out))
libs/pyutil/test/current/json_tests/test_pass3.py
+20
View File
@@ -0,0 +1,20 @@
from unittest import TestCase
from pyutil import jsonutil as json
# from http://json.org/JSON_checker/test/pass3.json
JSON = r'''
{
"JSON Test Pattern pass3": {
"The outermost value": "must be an object or array.",
"In this test": "It is an object."
}
}
'''
class TestPass3(TestCase):
def test_parse(self):
# test in/out equivalence and parsing
res = json.loads(JSON)
out = json.dumps(res)
self.assertEquals(res, json.loads(out))
libs/pyutil/test/current/json_tests/test_recursion.py
+67
View File
@@ -0,0 +1,67 @@
from unittest import TestCase
from pyutil import jsonutil as json
class JSONTestObject:
pass
class RecursiveJSONEncoder(json.JSONEncoder):
recurse = False
def default(self, o):
if o is JSONTestObject:
if self.recurse:
return [JSONTestObject]
else:
return 'JSONTestObject'
return json.JSONEncoder.default(o)
class TestRecursion(TestCase):
def test_listrecursion(self):
x = []
x.append(x)
try:
json.dumps(x)
except ValueError:
pass
else:
self.fail("didn't raise ValueError on list recursion")
x = []
y = [x]
x.append(y)
try:
json.dumps(x)
except ValueError:
pass
else:
self.fail("didn't raise ValueError on alternating list recursion")
y = []
x = [y, y]
# ensure that the marker is cleared
json.dumps(x)
def test_dictrecursion(self):
x = {}
x["test"] = x
try:
json.dumps(x)
except ValueError:
pass
else:
self.fail("didn't raise ValueError on dict recursion")
x = {}
{"a": x, "b": x}
# ensure that the marker is cleared
json.dumps(x)
def test_defaultrecursion(self):
enc = RecursiveJSONEncoder()
self.assertEquals(enc.encode(JSONTestObject), '"JSONTestObject"')
enc.recurse = True
try:
enc.encode(JSONTestObject)
except ValueError:
pass
else:
self.fail("didn't raise ValueError on default recursion")
libs/pyutil/test/current/json_tests/test_separators.py
+42
View File
@@ -0,0 +1,42 @@
import textwrap
from unittest import TestCase
from pyutil import jsonutil as json
class TestSeparators(TestCase):
def test_separators(self):
h = [['blorpie'], ['whoops'], [], 'd-shtaeou', 'd-nthiouh', 'i-vhbjkhnth',
{'nifty': 87}, {'field': 'yes', 'morefield': False} ]
expect = textwrap.dedent("""\
[
[
"blorpie"
] ,
[
"whoops"
] ,
[] ,
"d-shtaeou" ,
"d-nthiouh" ,
"i-vhbjkhnth" ,
{
"nifty" : 87
} ,
{
"field" : "yes" ,
"morefield" : false
}
]""")
d1 = json.dumps(h)
d2 = json.dumps(h, indent=2, sort_keys=True, separators=(' ,', ' : '))
h1 = json.loads(d1)
h2 = json.loads(d2)
self.assertEquals(h1, h)
self.assertEquals(h2, h)
self.assertEquals(d2, expect)
libs/pyutil/test/current/json_tests/test_speedups.py
+18
View File
@@ -0,0 +1,18 @@
from twisted.trial.unittest import SkipTest, TestCase
from pyutil.jsonutil import decoder
from pyutil.jsonutil import encoder
class TestSpeedups(TestCase):
def test_scanstring(self):
if not encoder.c_encode_basestring_ascii:
raise SkipTest("no C extension speedups available to test")
self.assertEquals(decoder.scanstring.__module__, "simplejson._speedups")
self.assert_(decoder.scanstring is decoder.c_scanstring)
def test_encode_basestring_ascii(self):
if not encoder.c_encode_basestring_ascii:
raise SkipTest("no C extension speedups available to test")
self.assertEquals(encoder.encode_basestring_ascii.__module__, "simplejson._speedups")
self.assert_(encoder.encode_basestring_ascii is
encoder.c_encode_basestring_ascii)
libs/pyutil/test/current/json_tests/test_unicode.py
+55
View File
@@ -0,0 +1,55 @@
from unittest import TestCase
from pyutil import jsonutil as json
class TestUnicode(TestCase):
def test_encoding1(self):
encoder = json.JSONEncoder(encoding='utf-8')
u = u'\N{GREEK SMALL LETTER ALPHA}\N{GREEK CAPITAL LETTER OMEGA}'
s = u.encode('utf-8')
ju = encoder.encode(u)
js = encoder.encode(s)
self.assertEquals(ju, js)
def test_encoding2(self):
u = u'\N{GREEK SMALL LETTER ALPHA}\N{GREEK CAPITAL LETTER OMEGA}'
s = u.encode('utf-8')
ju = json.dumps(u, encoding='utf-8')
js = json.dumps(s, encoding='utf-8')
self.assertEquals(ju, js)
def test_encoding3(self):
u = u'\N{GREEK SMALL LETTER ALPHA}\N{GREEK CAPITAL LETTER OMEGA}'
j = json.dumps(u)
self.assertEquals(j, '"\\u03b1\\u03a9"')
def test_encoding4(self):
u = u'\N{GREEK SMALL LETTER ALPHA}\N{GREEK CAPITAL LETTER OMEGA}'
j = json.dumps([u])
self.assertEquals(j, '["\\u03b1\\u03a9"]')
def test_encoding5(self):
u = u'\N{GREEK SMALL LETTER ALPHA}\N{GREEK CAPITAL LETTER OMEGA}'
j = json.dumps(u, ensure_ascii=False)
self.assertEquals(j, u'"%s"' % (u,))
def test_encoding6(self):
u = u'\N{GREEK SMALL LETTER ALPHA}\N{GREEK CAPITAL LETTER OMEGA}'
j = json.dumps([u], ensure_ascii=False)
self.assertEquals(j, u'["%s"]' % (u,))
def test_big_unicode_encode(self):
u = u'\U0001d120'
self.assertEquals(json.dumps(u), '"\\ud834\\udd20"')
self.assertEquals(json.dumps(u, ensure_ascii=False), u'"\U0001d120"')
def test_big_unicode_decode(self):
u = u'z\U0001d120x'
self.assertEquals(json.loads('"' + u + '"'), u)
self.assertEquals(json.loads('"z\\ud834\\udd20x"'), u)
def test_unicode_decode(self):
for i in range(0, 0xd7ff):
u = unichr(i)
js = '"\\u%04x"' % (i,)
self.assertEquals(json.loads(js), u)
libs/pyutil/test/current/test_assertutil.py
+18
View File
@@ -0,0 +1,18 @@
#!/usr/bin/env python
# Copyright (c) 2002-2009 Zooko Wilcox-O'Hearn
# portions Copyright (c) 2001 Autonomous Zone Industries
# This file is part of pyutil; see README.rst for licensing terms.
# Python Standard Library modules
import unittest
from pyutil import assertutil
class Testy(unittest.TestCase):
def test_bad_precond(self):
adict=23
try:
assertutil.precondition(isinstance(adict, dict), "adict is required to be a dict.", 23, adict=adict, foo=None)
except AssertionError, le:
self.failUnless(le.args[0] == "precondition: 'adict is required to be a dict.' <type 'str'>, 23 <type 'int'>, foo: None <type 'NoneType'>, 'adict': 23 <type 'int'>")
libs/pyutil/test/current/test_fileutil.py
+33
View File
@@ -0,0 +1,33 @@
import unittest
import os
from pyutil import fileutil
class FileUtil(unittest.TestCase):
def mkdir(self, basedir, path, mode=0777):
fn = os.path.join(basedir, path)
fileutil.make_dirs(fn, mode)
def touch(self, basedir, path, mode=None, data="touch\n"):
fn = os.path.join(basedir, path)
f = open(fn, "w")
f.write(data)
f.close()
if mode is not None:
os.chmod(fn, mode)
def test_du(self):
basedir = "util/FileUtil/test_du"
fileutil.make_dirs(basedir)
d = os.path.join(basedir, "space-consuming")
self.mkdir(d, "a/b")
self.touch(d, "a/b/1.txt", data="a"*10)
self.touch(d, "a/b/2.txt", data="b"*11)
self.mkdir(d, "a/c")
self.touch(d, "a/c/1.txt", data="c"*12)
self.touch(d, "a/c/2.txt", data="d"*13)
used = fileutil.du(basedir)
self.failUnlessEqual(10+11+12+13, used)
libs/pyutil/test/current/test_iputil.py
+33
View File
@@ -0,0 +1,33 @@
#!/usr/bin/env python
try:
from twisted.trial import unittest
unittest # http://divmod.org/trac/ticket/1499
except ImportError, le:
print "Skipping test_iputil since it requires Twisted and Twisted could not be imported: %s" % (le,)
else:
from pyutil import iputil, testutil
import re
DOTTED_QUAD_RE=re.compile("^[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+$")
class ListAddresses(testutil.SignalMixin):
def test_get_local_ip_for(self):
addr = iputil.get_local_ip_for('127.0.0.1')
self.failUnless(DOTTED_QUAD_RE.match(addr))
def test_list_async(self):
try:
from twisted.trial import unittest
unittest # http://divmod.org/trac/ticket/1499
from pyutil import iputil
except ImportError, le:
raise unittest.SkipTest("iputil could not be imported (probably because its dependency, Twisted, is not installed). %s" % (le,))
d = iputil.get_local_addresses_async()
def _check(addresses):
self.failUnless(len(addresses) >= 1) # always have localhost
self.failUnless("127.0.0.1" in addresses, addresses)
d.addCallbacks(_check)
return d
test_list_async.timeout=2
libs/pyutil/test/current/test_jsonutil.py
+18
View File
@@ -0,0 +1,18 @@
#!/usr/bin/env python
import unittest
from decimal import Decimal
from pyutil import jsonutil
zero_point_one = Decimal("0.1")
class TestDecimal(unittest.TestCase):
def test_encode(self):
self.failUnlessEqual(jsonutil.dumps(zero_point_one), "0.1")
def test_decode(self):
self.failUnlessEqual(jsonutil.loads("0.1"), zero_point_one)
def test_no_exception_on_convergent_parse_float(self):
self.failUnlessEqual(jsonutil.loads("0.1", parse_float=Decimal), zero_point_one)
libs/pyutil/test/current/test_mathutil.py
+135
View File
@@ -0,0 +1,135 @@
#!/usr/bin/env python
import unittest
from pyutil import mathutil
from pyutil.assertutil import _assert
class MathUtilTestCase(unittest.TestCase):
def _help_test_is_power_of_k(self, k):
for i in range(2, 40):
_assert(mathutil.is_power_of_k(k**i, k), k, i)
def test_is_power_of_k(self):
for i in range(2, 5):
self._help_test_is_power_of_k(i)
def test_log_ceil(self):
f = mathutil.log_ceil
self.failUnlessEqual(f(1, 2), 0)
self.failUnlessEqual(f(1, 3), 0)
self.failUnlessEqual(f(2, 2), 1)
self.failUnlessEqual(f(2, 3), 1)
self.failUnlessEqual(f(3, 2), 2)
def test_log_floor(self):
f = mathutil.log_floor
self.failUnlessEqual(f(1, 2), 0)
self.failUnlessEqual(f(1, 3), 0)
self.failUnlessEqual(f(2, 2), 1)
self.failUnlessEqual(f(2, 3), 0)
self.failUnlessEqual(f(3, 2), 1)
def test_div_ceil(self):
f = mathutil.div_ceil
self.failUnlessEqual(f(0, 1), 0)
self.failUnlessEqual(f(0, 2), 0)
self.failUnlessEqual(f(0, 3), 0)
self.failUnlessEqual(f(1, 3), 1)
self.failUnlessEqual(f(2, 3), 1)
self.failUnlessEqual(f(3, 3), 1)
self.failUnlessEqual(f(4, 3), 2)
self.failUnlessEqual(f(5, 3), 2)
self.failUnlessEqual(f(6, 3), 2)
self.failUnlessEqual(f(7, 3), 3)
def test_next_multiple(self):
f = mathutil.next_multiple
self.failUnlessEqual(f(5, 1), 5)
self.failUnlessEqual(f(5, 2), 6)
self.failUnlessEqual(f(5, 3), 6)
self.failUnlessEqual(f(5, 4), 8)
self.failUnlessEqual(f(5, 5), 5)
self.failUnlessEqual(f(5, 6), 6)
self.failUnlessEqual(f(32, 1), 32)
self.failUnlessEqual(f(32, 2), 32)
self.failUnlessEqual(f(32, 3), 33)
self.failUnlessEqual(f(32, 4), 32)
self.failUnlessEqual(f(32, 5), 35)
self.failUnlessEqual(f(32, 6), 36)
self.failUnlessEqual(f(32, 7), 35)
self.failUnlessEqual(f(32, 8), 32)
self.failUnlessEqual(f(32, 9), 36)
self.failUnlessEqual(f(32, 10), 40)
self.failUnlessEqual(f(32, 11), 33)
self.failUnlessEqual(f(32, 12), 36)
self.failUnlessEqual(f(32, 13), 39)
self.failUnlessEqual(f(32, 14), 42)
self.failUnlessEqual(f(32, 15), 45)
self.failUnlessEqual(f(32, 16), 32)
self.failUnlessEqual(f(32, 17), 34)
self.failUnlessEqual(f(32, 18), 36)
self.failUnlessEqual(f(32, 589), 589)
def test_pad_size(self):
f = mathutil.pad_size
self.failUnlessEqual(f(0, 4), 0)
self.failUnlessEqual(f(1, 4), 3)
self.failUnlessEqual(f(2, 4), 2)
self.failUnlessEqual(f(3, 4), 1)
self.failUnlessEqual(f(4, 4), 0)
self.failUnlessEqual(f(5, 4), 3)
def test_is_power_of_k_part_2(self):
f = mathutil.is_power_of_k
for i in range(1, 100):
if i in (1, 2, 4, 8, 16, 32, 64):
self.failUnless(f(i, 2), "but %d *is* a power of 2" % i)
else:
self.failIf(f(i, 2), "but %d is *not* a power of 2" % i)
for i in range(1, 100):
if i in (1, 3, 9, 27, 81):
self.failUnless(f(i, 3), "but %d *is* a power of 3" % i)
else:
self.failIf(f(i, 3), "but %d is *not* a power of 3" % i)
def test_next_power_of_k(self):
f = mathutil.next_power_of_k
self.failUnlessEqual(f(0,2), 1)
self.failUnlessEqual(f(1,2), 1)
self.failUnlessEqual(f(2,2), 2)
self.failUnlessEqual(f(3,2), 4)
self.failUnlessEqual(f(4,2), 4)
for i in range(5, 8): self.failUnlessEqual(f(i,2), 8, "%d" % i)
for i in range(9, 16): self.failUnlessEqual(f(i,2), 16, "%d" % i)
for i in range(17, 32): self.failUnlessEqual(f(i,2), 32, "%d" % i)
for i in range(33, 64): self.failUnlessEqual(f(i,2), 64, "%d" % i)
for i in range(65, 100): self.failUnlessEqual(f(i,2), 128, "%d" % i)
self.failUnlessEqual(f(0,3), 1)
self.failUnlessEqual(f(1,3), 1)
self.failUnlessEqual(f(2,3), 3)
self.failUnlessEqual(f(3,3), 3)
for i in range(4, 9): self.failUnlessEqual(f(i,3), 9, "%d" % i)
for i in range(10, 27): self.failUnlessEqual(f(i,3), 27, "%d" % i)
for i in range(28, 81): self.failUnlessEqual(f(i,3), 81, "%d" % i)
for i in range(82, 200): self.failUnlessEqual(f(i,3), 243, "%d" % i)
def test_ave(self):
f = mathutil.ave
self.failUnlessEqual(f([1,2,3]), 2)
self.failUnlessEqual(f([0,0,0,4]), 1)
self.failUnlessAlmostEqual(f([0.0, 1.0, 1.0]), .666666666666)
def failUnlessEqualContents(self, a, b):
self.failUnlessEqual(sorted(a), sorted(b))
def test_permute(self):
f = mathutil.permute
self.failUnlessEqualContents(f([]), [])
self.failUnlessEqualContents(f([1]), [[1]])
self.failUnlessEqualContents(f([1,2]), [[1,2], [2,1]])
self.failUnlessEqualContents(f([1,2,3]),
[[1,2,3], [1,3,2],
[2,1,3], [2,3,1],
[3,1,2], [3,2,1]])
libs/pyutil/test/current/test_time_format.py
+97
View File
@@ -0,0 +1,97 @@
#!/usr/bin/env python
"""\
Test time_format.py
"""
import os, time, unittest
from pyutil import time_format, increasing_timer
class TimeUtilTestCase(unittest.TestCase):
def setUp(self):
pass
def tearDown(self):
pass
def test_iso8601_utc_time(self, timer=increasing_timer.timer):
ts1 = time_format.iso_utc(timer.time() - 20)
ts2 = time_format.iso_utc()
assert ts1 < ts2, "failed: %s < %s" % (ts1, ts2)
ts3 = time_format.iso_utc(timer.time() + 20)
assert ts2 < ts3, "failed: %s < %s" % (ts2, ts3)
def test_iso_utc_time_to_localseconds(self, timer=increasing_timer.timer):
# test three times of the year so that a DST problem would hopefully be triggered
t1 = int(timer.time() - 365*3600/3)
iso_utc_t1 = time_format.iso_utc(t1)
t1_2 = time_format.iso_utc_time_to_seconds(iso_utc_t1)
assert t1 == t1_2, (t1, t1_2)
t1 = int(timer.time() - (365*3600*2/3))
iso_utc_t1 = time_format.iso_utc(t1)
t1_2 = time_format.iso_utc_time_to_seconds(iso_utc_t1)
self.failUnlessEqual(t1, t1_2)
t1 = int(timer.time())
iso_utc_t1 = time_format.iso_utc(t1)
t1_2 = time_format.iso_utc_time_to_seconds(iso_utc_t1)
self.failUnlessEqual(t1, t1_2)
def test_epoch(self):
return self._help_test_epoch()
def test_epoch_in_London(self):
# Europe/London is a particularly troublesome timezone. Nowadays, its
# offset from GMT is 0. But in 1970, its offset from GMT was 1.
# (Apparently in 1970 Britain had redefined standard time to be GMT+1
# and stayed in standard time all year round, whereas today
# Europe/London standard time is GMT and Europe/London Daylight
# Savings Time is GMT+1.) The current implementation of
# time_format.iso_utc_time_to_seconds() breaks if the timezone is
# Europe/London. (As soon as this unit test is done then I'll change
# that implementation to something that works even in this case...)
origtz = os.environ.get('TZ')
os.environ['TZ'] = "Europe/London"
if hasattr(time, 'tzset'):
time.tzset()
try:
return self._help_test_epoch()
finally:
if origtz is None:
del os.environ['TZ']
else:
os.environ['TZ'] = origtz
if hasattr(time, 'tzset'):
time.tzset()
def _help_test_epoch(self):
origtzname = time.tzname
s = time_format.iso_utc_time_to_seconds("1970-01-01T00:00:01Z")
self.failUnlessEqual(s, 1.0)
s = time_format.iso_utc_time_to_seconds("1970-01-01_00:00:01Z")
self.failUnlessEqual(s, 1.0)
s = time_format.iso_utc_time_to_seconds("1970-01-01 00:00:01Z")
self.failUnlessEqual(s, 1.0)
self.failUnlessEqual(time_format.iso_utc(1.0), "1970-01-01 00:00:01Z")
self.failUnlessEqual(time_format.iso_utc(1.0, sep="_"),
"1970-01-01_00:00:01Z")
now = time.time()
isostr = time_format.iso_utc(now)
timestamp = time_format.iso_utc_time_to_seconds(isostr)
self.failUnlessEqual(int(timestamp), int(now))
def my_time():
return 1.0
self.failUnlessEqual(time_format.iso_utc(t=my_time),
"1970-01-01 00:00:01Z")
self.failUnlessRaises(ValueError,
time_format.iso_utc_time_to_seconds,
"invalid timestring")
s = time_format.iso_utc_time_to_seconds("1970-01-01 00:00:01.500Z")
self.failUnlessEqual(s, 1.5)
# Look for daylight-savings-related errors.
thatmomentinmarch = time_format.iso_utc_time_to_seconds("2009-03-20 21:49:02.226536Z")
self.failUnlessEqual(thatmomentinmarch, 1237585742.226536)
self.failUnlessEqual(origtzname, time.tzname)
libs/pyutil/test/current/test_verlib.py
+124
View File
@@ -0,0 +1,124 @@
# -*- coding: utf-8 -*-
"""Tests for distutils.version."""
import unittest
import doctest
from pyutil.verlib import NormalizedVersion as V
from pyutil.verlib import IrrationalVersionError
from pyutil.verlib import suggest_normalized_version as suggest
class VersionTestCase(unittest.TestCase):
versions = ((V('1.0'), '1.0'),
(V('1.1'), '1.1'),
(V('1.2.3'), '1.2.3'),
(V('1.2'), '1.2'),
(V('1.2.3a4'), '1.2.3a4'),
(V('1.2c4'), '1.2c4'),
(V('1.2.3.4'), '1.2.3.4'),
(V('1.2.3.4.0b3'), '1.2.3.4b3'),
(V('1.2.0.0.0'), '1.2'),
(V('1.0.dev345'), '1.0.dev345'),
(V('1.0.post456.dev623'), '1.0.post456.dev623'))
def test_basic_versions(self):
for v, s in self.versions:
self.assertEquals(str(v), s)
def test_from_parts(self):
for v, s in self.versions:
v2 = V.from_parts(*v.parts)
self.assertEquals(v, v2)
self.assertEquals(str(v), str(v2))
def test_irrational_versions(self):
irrational = ('1', '1.2a', '1.2.3b', '1.02', '1.2a03',
'1.2a3.04', '1.2.dev.2', '1.2dev', '1.2.dev',
'1.2.dev2.post2', '1.2.post2.dev3.post4')
for s in irrational:
self.assertRaises(IrrationalVersionError, V, s)
def test_comparison(self):
r"""
>>> V('1.2.0') == '1.2'
Traceback (most recent call last):
...
TypeError: cannot compare NormalizedVersion and str
>>> V('1.2.0') == V('1.2')
True
>>> V('1.2.0') == V('1.2.3')
False
>>> V('1.2.0') < V('1.2.3')
True
>>> (V('1.0') > V('1.0b2'))
True
>>> (V('1.0') > V('1.0c2') > V('1.0c1') > V('1.0b2') > V('1.0b1')
... > V('1.0a2') > V('1.0a1'))
True
>>> (V('1.0.0') > V('1.0.0c2') > V('1.0.0c1') > V('1.0.0b2') > V('1.0.0b1')
... > V('1.0.0a2') > V('1.0.0a1'))
True
>>> V('1.0') < V('1.0.post456.dev623')
True
>>> V('1.0.post456.dev623') < V('1.0.post456') < V('1.0.post1234')
True
>>> (V('1.0a1')
... < V('1.0a2.dev456')
... < V('1.0a2')
... < V('1.0a2.1.dev456') # e.g. need to do a quick post release on 1.0a2
... < V('1.0a2.1')
... < V('1.0b1.dev456')
... < V('1.0b2')
... < V('1.0c1.dev456')
... < V('1.0c1')
... < V('1.0.dev7')
... < V('1.0.dev18')
... < V('1.0.dev456')
... < V('1.0.dev1234')
... < V('1.0')
... < V('1.0.post456.dev623') # development version of a post release
... < V('1.0.post456'))
True
"""
# must be a simpler way to call the docstrings
doctest.run_docstring_examples(self.test_comparison, globals(),
name='test_comparison')
def test_suggest_normalized_version(self):
self.assertEquals(suggest('1.0'), '1.0')
self.assertEquals(suggest('1.0-alpha1'), '1.0a1')
self.assertEquals(suggest('1.0c2'), '1.0c2')
self.assertEquals(suggest('walla walla washington'), None)
self.assertEquals(suggest('2.4c1'), '2.4c1')
# from setuptools
self.assertEquals(suggest('0.4a1.r10'), '0.4a1.post10')
self.assertEquals(suggest('0.7a1dev-r66608'), '0.7a1.dev66608')
self.assertEquals(suggest('0.6a9.dev-r41475'), '0.6a9.dev41475')
self.assertEquals(suggest('2.4preview1'), '2.4c1')
self.assertEquals(suggest('2.4pre1') , '2.4c1')
self.assertEquals(suggest('2.1-rc2'), '2.1c2')
# from pypi
self.assertEquals(suggest('0.1dev'), '0.1.dev0')
self.assertEquals(suggest('0.1.dev'), '0.1.dev0')
# we want to be able to parse Twisted
# development versions are like post releases in Twisted
self.assertEquals(suggest('9.0.0+r2363'), '9.0.0.post2363')
# pre-releases are using markers like "pre1"
self.assertEquals(suggest('9.0.0pre1'), '9.0.0c1')
# we want to be able to parse Tcl-TK
# they us "p1" "p2" for post releases
self.assertEquals(suggest('1.4p1'), '1.4.post1')
libs/pyutil/test/current/test_version_class.py
+23
View File
@@ -0,0 +1,23 @@
import unittest
from pyutil import version_class
V = version_class.Version
class T(unittest.TestCase):
def test_rc_regex_rejects_rc_suffix(self):
self.failUnlessRaises(ValueError, V, '9.9.9rc9')
def test_rc_regex_rejects_trailing_garbage(self):
self.failUnlessRaises(ValueError, V, '9.9.9c9HEYTHISISNTRIGHT')
def test_comparisons(self):
self.failUnless(V('1.0') < V('1.1'))
self.failUnless(V('1.0a1') < V('1.0'))
self.failUnless(V('1.0a1') < V('1.0b1'))
self.failUnless(V('1.0b1') < V('1.0c1'))
self.failUnless(V('1.0a1') < V('1.0a1-r99'))
self.failUnlessEqual(V('1.0a1.post987'), V('1.0a1-r987'))
self.failUnlessEqual(str(V('1.0a1.post999')), '1.0.0a1-r999')
self.failUnlessEqual(str(V('1.0a1-r999')), '1.0.0a1-r999')
self.failIfEqual(V('1.0a1'), V('1.0a1-r987'))
libs/pyutil/test/deprecated/__init__.py
View File
libs/pyutil/test/deprecated/test_dictutil.py
+115
View File
@@ -0,0 +1,115 @@
#!/usr/bin/env python
# Copyright (c) 2002-2009 Zooko "Zooko" Wilcox-O'Hearn
# This file is part of pyutil; see README.rst for licensing terms.
import random, sys, traceback, unittest
from pyutil.assertutil import _assert
from pyutil import dictutil
class EqButNotIs:
def __init__(self, x):
self.x = x
self.hash = int(random.randrange(0, 2**31))
def __repr__(self):
return "<%s %s>" % (self.__class__.__name__, self.x,)
def __hash__(self):
return self.hash
def __le__(self, other):
return self.x <= other
def __lt__(self, other):
return self.x < other
def __ge__(self, other):
return self.x >= other
def __gt__(self, other):
return self.x > other
def __ne__(self, other):
return self.x != other
def __eq__(self, other):
return self.x == other
class Testy(unittest.TestCase):
def _help_test_empty_dict(self, klass):
d1 = klass()
d2 = klass({})
self.failUnless(d1 == d2, "d1: %r, d2: %r" % (d1, d2,))
self.failUnless(len(d1) == 0)
self.failUnless(len(d2) == 0)
def _help_test_nonempty_dict(self, klass):
d1 = klass({'a': 1, 'b': "eggs", 3: "spam",})
d2 = klass({'a': 1, 'b': "eggs", 3: "spam",})
self.failUnless(d1 == d2)
self.failUnless(len(d1) == 3, "%s, %s" % (len(d1), d1,))
self.failUnless(len(d2) == 3)
def _help_test_eq_but_notis(self, klass):
d = klass({'a': 3, 'b': EqButNotIs(3), 'c': 3})
d.pop('b')
d.clear()
d['a'] = 3
d['b'] = EqButNotIs(3)
d['c'] = 3
d.pop('b')
d.clear()
d['b'] = EqButNotIs(3)
d['a'] = 3
d['c'] = 3
d.pop('b')
d.clear()
d['a'] = EqButNotIs(3)
d['c'] = 3
d['a'] = 3
d.clear()
fake3 = EqButNotIs(3)
fake7 = EqButNotIs(7)
d[fake3] = fake7
d[3] = 7
d[3] = 8
_assert(filter(lambda x: x is 8, d.itervalues()))
_assert(filter(lambda x: x is fake7, d.itervalues()))
_assert(not filter(lambda x: x is 7, d.itervalues())) # The real 7 should have been ejected by the d[3] = 8.
_assert(filter(lambda x: x is fake3, d.iterkeys()))
_assert(filter(lambda x: x is 3, d.iterkeys()))
d[fake3] = 8
d.clear()
d[3] = 7
fake3 = EqButNotIs(3)
fake7 = EqButNotIs(7)
d[fake3] = fake7
d[3] = 8
_assert(filter(lambda x: x is 8, d.itervalues()))
_assert(filter(lambda x: x is fake7, d.itervalues()))
_assert(not filter(lambda x: x is 7, d.itervalues())) # The real 7 should have been ejected by the d[3] = 8.
_assert(filter(lambda x: x is fake3, d.iterkeys()))
_assert(filter(lambda x: x is 3, d.iterkeys()))
d[fake3] = 8
def test_em(self):
for klass in (dictutil.UtilDict, dictutil.NumDict, dictutil.ValueOrderedDict,):
# print "name of class: ", klass
for helper in (self._help_test_empty_dict, self._help_test_nonempty_dict, self._help_test_eq_but_notis,):
# print "name of test func: ", helper
try:
helper(klass)
except:
(etype, evalue, realtb) = sys.exc_info()
traceback.print_exception(etype, evalue, realtb)
self.fail(evalue)
del realtb
def suite():
suite = unittest.makeSuite(Testy, 'test')
return suite
if __name__ == '__main__':
unittest.main()
libs/pyutil/test/deprecated/test_picklesaver.py
+36
View File
@@ -0,0 +1,36 @@
#!/usr/bin/env python
# Copyright (c) 2002 Luke 'Artimage' Nelson
# Copyright (c) 2005-2010 Zooko Wilcox-O'Hearn
# This file is part of pyutil; see README.rst for licensing terms.
import os
try:
from twisted.trial import unittest
except ImportError, le:
print "Skipping %s since it requires Twisted and Twisted could not be imported: %s" % (__name__, le,)
else:
from pyutil import PickleSaver, fileutil
class Thingie(PickleSaver.PickleSaver):
def __init__(self, fname, delay=30):
PickleSaver.PickleSaver.__init__(self, fname=fname, attrs={'tmp_store':'False'}, DELAY=delay)
class PickleSaverTest(unittest.TestCase):
def _test_save_now(self, fname):
thingie = Thingie(fname, delay=0)
thingie.tmp_store = 'True'
thingie.lazy_save() # Note: it was constructed with default save delay of 0.
def test_save_now(self):
"""
This test should create a lazy save object, save it with no delay and check if the file exists.
"""
tempdir = fileutil.NamedTemporaryDirectory()
fname = os.path.join(tempdir.name, "picklesavertest")
self._test_save_now(fname)
self.failUnless(os.path.isfile(fname), "The file [%s] does not exist." %(fname,))
tempdir.shutdown()
libs/pyutil/test/deprecated/test_xor.py
+24
View File
@@ -0,0 +1,24 @@
#!/usr/bin/env python
# Copyright (c) 2002-2009 Zooko Wilcox-O'Hearn
# portions Copyright (c) 2001 Autonomous Zone Industries
# This file is part of pyutil; see README.rst for licensing terms.
#
import unittest
from pyutil.xor import xor
# unit tests
def _help_test(xf):
assert xf('\000', '\000') == '\000'
assert xf('\001', '\000') == '\001'
assert xf('\001', '\001') == '\000'
assert xf('\000\001', '\000\001') == '\000\000'
assert xf('\100\101', '\000\101') == '\100\000'
class Testy(unittest.TestCase):
def test_em(self):
for xorfunc in (xor.py_xor, xor.py_xor_simple, xor.xor,):
if callable(xorfunc):
# print "testing xorfunc ", xorfunc
_help_test(xorfunc)
libs/pyutil/test/out_of_shape/__init__.py
View File
libs/pyutil/test/out_of_shape/test_cache.py
+454
View File
@@ -0,0 +1,454 @@
#!/usr/bin/env python
# Copyright (c) 2002-2010 Zooko "Zooko" Wilcox-O'Hearn
# This file is part of pyutil; see README.rst for licensing terms.
import random, unittest
from pyutil.assertutil import _assert
from pyutil.humanreadable import hr
from pyutil import memutil
from pyutil import cache
class Bencher:
def __init__(self, klass, MAXREPS=2**8, MAXTIME=5):
print klass
self.klass = klass
self.MAXREPS = MAXREPS
self.MAXTIME = MAXTIME
self.d = {}
self.lrun = None
def _generic_benchmarking_init(self, n):
self.d.clear()
global lrun
self.lrun = self.klass(maxsize=n)
for i in range(n):
self.d[i] = i
self.lrun[n+i] = n+i
def _benchmark_init(self, n):
MAXSIZE=n/2
d2 = self.klass(initialdata=self.d, maxsize=MAXSIZE)
assert len(d2) == min(len(self.d), MAXSIZE)
return True
def _benchmark_update(self, n):
MAXSIZE=n/2
d2 = self.klass(maxsize=MAXSIZE)
assert len(d2) == 0
d2.update(self.d)
assert len(d2) == min(len(self.d), MAXSIZE)
return True
def _benchmark_insert(self, n):
MAXSIZE=n/2
d2 = self.klass(maxsize=MAXSIZE)
assert len(d2) == 0
for k, v, in self.d.iteritems():
d2[k] = v
assert len(d2) == min(len(self.d), MAXSIZE)
return True
def _benchmark_init_and_popitem(self, n):
MAXSIZE=n/2
d2 = self.klass(initialdata=self.d, maxsize=MAXSIZE)
assert len(d2) == min(len(self.d), MAXSIZE)
for i in range(len(d2), 0, -1):
assert len(d2) == i
d2.popitem()
return True
def _benchmark_init_and_has_key_and_del(self, n):
MAXSIZE=n/2
d2 = self.klass(initialdata=self.d, maxsize=MAXSIZE)
assert len(d2) == min(len(self.d), MAXSIZE)
for k in self.d.iterkeys():
if d2.has_key(k):
del d2[k]
return True
def _benchmark_init_and_remove(self, n):
MAXSIZE=n/2
d2 = self.klass(initialdata=self.d, maxsize=MAXSIZE)
assert len(d2) == min(len(self.d), MAXSIZE)
for k in self.d.iterkeys():
d2.remove(k, strictkey=False)
return True
def bench(self, BSIZES=(128, 250, 2048, 5000, 2**13, 2**20,)):
from pyutil import benchutil
funcs = ("_benchmark_insert", "_benchmark_init_and_has_key_and_del", "_benchmark_init_and_remove", "_benchmark_init_and_popitem", "_benchmark_update", "_benchmark_init",)
max = 0
for func in funcs:
if len(func) > max:
max = len(func)
for func in funcs:
print func + " " * (max + 1 - len(func))
for BSIZE in BSIZES:
f = getattr(self, func)
benchutil.rep_bench(f, BSIZE, self._generic_benchmarking_init, MAXREPS=self.MAXREPS, MAXTIME=self.MAXTIME)
def quick_bench():
Bencher(cache.LRUCache, MAXTIME=2).bench(BSIZES=(2**7, 2**12, 2**14, 2**15, 2**16,))
Bencher(cache.LinkedListLRUCache, MAXTIME=2).bench(BSIZES=(2**7, 2**12, 2**14, 2**15,))
Bencher(cache.SmallLRUCache, MAXTIME=2).bench(BSIZES=(2**7, 2**12, 2**14, 2**15,))
def slow_bench():
Bencher(cache.LRUCache, MAXTIME=5).bench(BSIZES=[2**x for x in range(7, 21)])
Bencher(cache.LinkedListLRUCache, MAXTIME=5).bench(BSIZES=[2**x for x in range(7, 21)])
Bencher(cache.SmallLRUCache, MAXTIME=5).bench(BSIZES=[2**x for x in range(7, 17)])
MUCHADDINGSIZE=2**4
MUCHADDINGNUM = 2**4
# The following parameters are for testing for memory leakage.
MIN_SLOPE = 512.0 # If it leaks less than 512.0 bytes per iteration, then it's probably just some kind of noise from the interpreter or something...
SAMPLES = 2**5
# MIN_SLOPE is high because samples is low, which is because taking a statistically useful numbers of samples takes too long.
# For a *good* test, turn samples up as high as you can stand (maybe 2**10) and set MIN_SLOPE to about 1.0.
# For a *really* good test, add a variance measure to memutil.measure_mem_leakage(), and only consider it to be leaking if the slope is > 0.1 *and* is a "pretty good" fit for the data.
# MIN_SLOPE = 1.0
# SAMPLES = 2**10
class Testy(unittest.TestCase):
def _test_empty_lookup(self, d) :
self.failUnless(d.get('spam') is None)
def _test_key_error(self, C) :
d = C()
try:
d['spam']
self.fail(d)
except KeyError :
pass
def _test_insert_and_get(self, d) :
d.insert("spam", "eggs")
d["spam2"] = "eggs2"
self.failUnless(d.get("spam") == "eggs", str(d))
self.failUnless(d.get("spam2") == "eggs2")
self.failUnless(d["spam"] == "eggs")
self.failUnless(d["spam2"] == "eggs2")
def _test_insert_and_remove(self, d):
d.insert('spam', "eggs")
self.failUnless(d.has_key('spam'))
self.failUnless(d.get('spam') == "eggs")
self.failUnless(d['spam'] == "eggs")
x = d.remove('spam')
self.failUnless(x == "eggs", "x: %s" % `x`)
self.failUnless(not d.has_key('spam'))
d['spam'] = "eggs"
self.failUnless(d.has_key('spam'))
self.failUnless(d.get('spam') == "eggs")
self.failUnless(d['spam'] == "eggs")
del d['spam']
self.failUnless(not d.has_key('spam'))
def _test_setdefault(self, d):
d.setdefault('spam', "eggs")
self.failUnless(d.has_key('spam'))
self.failUnless(d.get('spam') == "eggs")
self.failUnless(d['spam'] == "eggs")
x = d.remove('spam')
self.failUnless(x == "eggs", "x: %s" % `x`)
self.failUnless(not d.has_key('spam'))
def _test_extracted_bound_method(self, d):
insmeth = d.insert
insmeth('spammy', "eggsy")
self.failUnless(d.get('spammy') == "eggsy")
def _test_extracted_unbound_method(self, d):
insumeth = d.__class__.insert
insumeth(d, 'spammy', "eggsy")
self.failUnless(d.get('spammy') == "eggsy")
def _test_unbound_method(self, C, d):
umeth = C.insert
umeth(d, 'spammy', "eggsy")
self.failUnless(d.get('spammy') == "eggsy")
def _test_clear(self, d):
d[11] = 11
d._assert_invariants()
self.failUnless(len(d) == 1)
d.clear()
d._assert_invariants()
self.failUnless(len(d) == 0)
def _test_update(self, d):
self.failUnless(d._assert_invariants())
d['b'] = 99
self.failUnless(d._assert_invariants())
d2={ 'a': 0, 'b': 1, 'c': 2,}
d.update(d2)
self.failUnless(d._assert_invariants())
self.failUnless(d.get('a') == 0, "d.get('a'): %s" % d.get('a'))
self.failUnless(d._assert_invariants())
self.failUnless(d.get('b') == 1, "d.get('b'): %s" % d.get('b'))
self.failUnless(d._assert_invariants())
self.failUnless(d.get('c') == 2)
self.failUnless(d._assert_invariants())
def _test_popitem(self, C):
c = C({"a": 1})
res = c.popitem()
_assert(res == ("a", 1,), C, c, res)
self.failUnless(res == ("a", 1,))
def _test_iterate_items(self, C):
c = C({"a": 1})
i = c.iteritems()
x = i.next()
self.failUnless(x == ("a", 1,))
try:
i.next()
self.fail() # Should have gotten StopIteration exception
except StopIteration:
pass
def _test_iterate_keys(self, C):
c = C({"a": 1})
i = c.iterkeys()
x = i.next()
self.failUnless(x == "a")
try:
i.next()
self.fail() # Should have gotten StopIteration exception
except StopIteration:
pass
def _test_iterate_values(self, C):
c = C({"a": 1})
i = c.itervalues()
x = i.next()
self.failUnless(x == 1)
try:
i.next()
self.fail() # Should have gotten StopIteration exception
except StopIteration:
pass
def _test_LRU_much_adding_some_removing(self, C):
c = C(maxsize=MUCHADDINGSIZE)
for i in range(MUCHADDINGNUM):
c[i] = i
if (i % 400) == 0:
k = random.choice(c.keys())
del c[k]
for i in range(MUCHADDINGSIZE):
c[i] = i
self.failUnless(len(c) == MUCHADDINGSIZE)
def _test_LRU_1(self, C):
c = C(maxsize=10)
c[11] = 11
c._assert_invariants()
c[11] = 11
c._assert_invariants()
c[11] = 1001
c._assert_invariants()
c[11] = 11
c._assert_invariants()
c[11] = 1001
c._assert_invariants()
c[11] = 1001
c._assert_invariants()
c[11] = 1001
c._assert_invariants()
def _test_LRU_2(self, C):
c = C(maxsize=10)
c[11] = 11
c._assert_invariants()
del c[11]
c._assert_invariants()
c[11] = 11
c._assert_invariants()
c[11] = 11
c._assert_invariants()
def _test_LRU_3(self, C):
c = C(maxsize=10)
c[11] = 11
c._assert_invariants()
c[11] = 12
c._assert_invariants()
c[11] = 13
c._assert_invariants()
del c[11]
c._assert_invariants()
c[11] = 14
c._assert_invariants()
c[11] = 15
c._assert_invariants()
c[11] = 16
c._assert_invariants()
def _test_LRU_full(self, C):
c = C(maxsize=10)
c._assert_invariants()
for i in xrange(11):
c._assert_invariants()
c[i] = i
c._assert_invariants()
self.failUnless(len(c) == 10)
self.failUnless(10 in c.values(), c.values())
self.failUnless(0 not in c.values())
del c[1]
c._assert_invariants()
self.failUnless(1 not in c.values())
self.failUnless(len(c) == 9)
c[11] = 11
c._assert_invariants()
self.failUnless(len(c) == 10)
self.failUnless(1 not in c.values())
self.failUnless(11 in c.values())
del c[11]
c._assert_invariants()
c[11] = 11
c._assert_invariants()
self.failUnless(len(c) == 10)
self.failUnless(1 not in c.values())
self.failUnless(11 in c.values())
c[11] = 11
c._assert_invariants()
self.failUnless(len(c) == 10)
self.failUnless(1 not in c.values())
self.failUnless(11 in c.values())
for i in xrange(200):
c[i] = i
c._assert_invariants()
self.failUnless(199 in c.values())
self.failUnless(190 in c.values())
def _test_LRU_has_key(self, C):
c = C(maxsize=10)
c._assert_invariants()
for i in xrange(11):
c._assert_invariants()
c[i] = i
c._assert_invariants()
self.failUnless(len(c) == 10)
self.failUnless(10 in c.values())
self.failUnless(0 not in c.values())
# c.has_key(1) # this touches `1' and makes it fresher so that it will live and `2' will die next time we overfill.
c[1] = 1 # this touches `1' and makes it fresher so that it will live and `2' will die next time we overfill.
c._assert_invariants()
c[99] = 99
c._assert_invariants()
self.failUnless(len(c) == 10)
self.failUnless(1 in c.values(), "C: %s, c.values(): %s" % (hr(C), hr(c.values(),),))
self.failUnless(not 2 in c.values())
self.failUnless(99 in c.values())
def _test_LRU_not_overfull_on_idempotent_add(self, C):
c = C(maxsize=10)
for i in xrange(11):
c[i] = i
c[1] = "spam"
# Now 1 is the freshest, so 2 is the next one that would be removed *if* we went over limit.
c[3] = "eggs"
self.failUnless(c.has_key(2))
self.failUnless(len(c) == 10)
c._assert_invariants()
def _test_LRU_overflow_on_update(self, C):
d = C(maxsize=10)
self.failUnless(d._assert_invariants())
d2 = {}
for i in range(12):
d2[i] = i
d.update(d2)
self.failUnless(d._assert_invariants())
self.failUnless(len(d) == 10)
def _test_LRU_overflow_on_init(self, C):
d2 = {}
for i in range(12):
d2[i] = i
d = C(d2, maxsize=10)
self.failUnless(d._assert_invariants())
self.failUnless(len(d) == 10)
def _test_em(self):
for klass in (cache.LRUCache, cache.SmallLRUCache,):
for testfunc in (self._test_empty_lookup, self._test_insert_and_get, self._test_insert_and_remove, self._test_extracted_bound_method, self._test_extracted_unbound_method, self._test_clear, self._test_update, self._test_setdefault,):
testfunc(klass())
for testfunc in (self._test_popitem, self._test_iterate_items, self._test_iterate_keys, self._test_iterate_values, self._test_key_error, ):
testfunc(klass)
self._test_unbound_method(klass, klass())
for klass in (cache.LRUCache, cache.SmallLRUCache,):
for testfunc in (self._test_LRU_1, self._test_LRU_2, self._test_LRU_3, self._test_LRU_full, self._test_LRU_has_key, self._test_LRU_not_overfull_on_idempotent_add, self._test_LRU_overflow_on_update, self._test_LRU_overflow_on_init,):
testfunc(klass)
def test_em(self):
self._test_em()
def _mem_test_LRU_much_adding_some_removing(self):
for klass in (cache.LRUCache, cache.SmallLRUCache,):
return self._test_LRU_much_adding_some_removing(klass)
def test_mem_leakage(self):
try:
self._test_mem_leakage()
except memutil.NotSupportedException:
print "Skipping memory leak test since measurement of current mem usage isn't implemented on this platform."
pass
del test_mem_leakage # This test takes too long.
def _test_mem_leakage(self):
# measure one and throw it away, in order to reach a "steady state" in terms of initialization of memory state.
memutil.measure_mem_leakage(self.test_em, max(2**3, SAMPLES/2**3), iterspersample=2**0)
slope = memutil.measure_mem_leakage(self.test_em, max(2**3, SAMPLES/2**3), iterspersample=2**0)
self.failUnless(slope <= MIN_SLOPE, "%s leaks memory at a rate of approximately %s system bytes per invocation" % (self.test_em, "%0.3f" % slope,))
def test_mem_leakage_much_adding_some_removing(self):
try:
self._test_mem_leakage_much_adding_some_removing()
except memutil.NotSupportedException:
print "Skipping memory leak test since measurement of current mem usage isn't implemented on this platform."
pass
del test_mem_leakage_much_adding_some_removing # This test takes too long.
def _test_mem_leakage_much_adding_some_removing(self):
# measure one and throw it away, in order to reach a "steady state" in terms of initialization of memory state.
memutil.measure_mem_leakage(self._mem_test_LRU_much_adding_some_removing, SAMPLES, iterspersample=2**0)
slope = memutil.measure_mem_leakage(self._mem_test_LRU_much_adding_some_removing, SAMPLES, iterspersample=2**0)
self.failUnless(slope <= MIN_SLOPE, "%s leaks memory at a rate of approximately %s system bytes per invocation" % (self._mem_test_LRU_much_adding_some_removing, "%0.3f" % slope,))
def test_obj_leakage(self):
self._test_obj_leakage()
del test_obj_leakage # This test takes too long.
def _test_obj_leakage(self):
# measure one and throw it away, in order to reach a "steady state" in terms of initialization of objects state.
memutil.measure_obj_leakage(self.test_em, max(2**3, SAMPLES/2**3), iterspersample=2**0)
slope = memutil.measure_obj_leakage(self.test_em, max(2**3, SAMPLES/2**3), iterspersample=2**0)
self.failUnless(slope <= MIN_SLOPE, "%s leaks objects at a rate of approximately %s system bytes per invocation" % (self.test_em, "%0.3f" % slope,))
def test_obj_leakage_much_adding_some_removing(self):
self._test_obj_leakage_much_adding_some_removing()
del test_obj_leakage_much_adding_some_removing # This test takes too long.
def _test_obj_leakage_much_adding_some_removing(self):
# measure one and throw it away, in order to reach a "steady state" in terms of initialization of objects state.
memutil.measure_obj_leakage(self._mem_test_LRU_much_adding_some_removing, SAMPLES, iterspersample=2**0)
slope = memutil.measure_obj_leakage(self._mem_test_LRU_much_adding_some_removing, SAMPLES, iterspersample=2**0)
self.failUnless(slope <= MIN_SLOPE, "%s leaks objects at a rate of approximately %s system bytes per invocation" % (self._mem_test_LRU_much_adding_some_removing, "%0.3f" % slope,))
libs/pyutil/test/out_of_shape/test_odict.py
+441
View File
@@ -0,0 +1,441 @@
#!/usr/bin/env python
# Copyright (c) 2002-2010 Zooko Wilcox-O'Hearn
# This file is part of pyutil; see README.rst for licensing terms.
import random, unittest
from pyutil.humanreadable import hr
from pyutil import memutil
from pyutil import odict
class Bencher:
def __init__(self, klass, MAXREPS=2**8, MAXTIME=5):
print klass
self.klass = klass
self.MAXREPS = MAXREPS
self.MAXTIME = MAXTIME
self.d = {}
self.lrun = None
def _generic_benchmarking_init(self, n):
self.d.clear()
self.lrun = self.klass()
for i in range(n):
self.d[i] = i
self.lrun[n+i] = n+i
def _benchmark_init(self, n):
d2 = self.klass(initialdata=self.d)
assert len(d2) == len(self.d)
return True
def _benchmark_update(self, n):
d2 = self.klass()
assert len(d2) == 0
d2.update(self.d)
assert len(d2) == len(self.d)
return True
def _benchmark_insert(self, n):
d2 = self.klass()
assert len(d2) == 0
for k, v, in self.d.iteritems():
d2[k] = v
assert len(d2) == len(self.d)
return True
def _benchmark_init_and_popitem(self, n):
d2 = self.klass(initialdata=self.d)
assert len(d2) == len(self.d)
for i in range(len(d2), 0, -1):
assert len(d2) == i
d2.popitem()
return True
def _benchmark_init_and_has_key_and_del(self, n):
d2 = self.klass(initialdata=self.d)
assert len(d2) == len(self.d)
for k in self.d.iterkeys():
if d2.has_key(k):
del d2[k]
return True
def _benchmark_init_and_remove(self, n):
d2 = self.klass(initialdata=self.d)
assert len(d2) == len(self.d)
for k in self.d.iterkeys():
d2.remove(k, strictkey=False)
return True
def bench(self, BSIZES=(128, 250, 2048, 5000, 2**13, 2**20,)):
from pyutil import benchutil
funcs = ("_benchmark_insert", "_benchmark_init_and_has_key_and_del", "_benchmark_init_and_remove", "_benchmark_init_and_popitem", "_benchmark_update", "_benchmark_init",)
max = 0
for func in funcs:
if len(func) > max:
max = len(func)
for func in funcs:
print func + " " * (max + 1 - len(func))
for BSIZE in BSIZES:
f = getattr(self, func)
benchutil.rep_bench(f, BSIZE, self._generic_benchmarking_init, MAXREPS=self.MAXREPS, MAXTIME=self.MAXTIME)
def quick_bench():
Bencher(odict.LRUCache, MAXTIME=2).bench(BSIZES=(2**7, 2**12, 2**14, 2**15, 2**16,))
Bencher(odict.LinkedListLRUCache, MAXTIME=2).bench(BSIZES=(2**7, 2**12, 2**14, 2**15,))
Bencher(odict.SmallLRUCache, MAXTIME=2).bench(BSIZES=(2**7, 2**12, 2**14, 2**15,))
def slow_bench():
Bencher(odict.LRUCache, MAXTIME=5).bench(BSIZES=[2**x for x in range(7, 21)])
Bencher(odict.LinkedListLRUCache, MAXTIME=5).bench(BSIZES=[2**x for x in range(7, 21)])
Bencher(odict.SmallLRUCache, MAXTIME=5).bench(BSIZES=[2**x for x in range(7, 17)])
MUCHADDINGSIZE=2**4
# The following parameters are for testing for memory leakage.
MIN_SLOPE = 512.0 # If it leaks less than 512.0 bytes per iteration, then it's probably just some kind of noise from the interpreter or something...
SAMPLES = 2**5
# MIN_SLOPE is high because samples is low, which is because taking a statistically useful numbers of samples takes too long.
# For a *good* test, turn samples up as high as you can stand (maybe 2**10) and set MIN_SLOPE to about 1.0.
# For a *really* good test, add a variance measure to memutil.measure_mem_leakage(), and only consider it to be leaking if the slope is > 0.1 *and* is a "pretty good" fit for the data.
# MIN_SLOPE = 1.0
# SAMPLES = 2**10
class Testy(unittest.TestCase):
def _test_empty_lookup(self, d) :
self.failUnless(d.get('spam') is None)
def _test_key_error(self, C) :
d = C()
try:
d['spam']
self.fail(d)
except KeyError :
pass
def _test_insert_and_get_and_items(self, d) :
d.insert("spam", "eggs")
d["spam2"] = "eggs2"
self.failUnless(d.get("spam") == "eggs", str(d))
self.failUnless(d.get("spam2") == "eggs2")
self.failUnless(d["spam"] == "eggs")
self.failUnless(d["spam2"] == "eggs2")
self.failUnlessEqual(d.items(), [("spam", "eggs"), ("spam2", "eggs2")], d)
def _test_move_to_most_recent(self, d) :
d.insert("spam", "eggs")
d["spam2"] = "eggs2"
self.failUnless(d.get("spam") == "eggs", str(d))
self.failUnless(d.get("spam2") == "eggs2")
self.failUnless(d["spam"] == "eggs")
self.failUnless(d["spam2"] == "eggs2")
self.failUnlessEqual(d.items(), [("spam", "eggs"), ("spam2", "eggs2")])
d.move_to_most_recent("spam")
self.failUnlessEqual(d.items(), [("spam2", "eggs2"), ("spam", "eggs")])
def _test_insert_and_remove(self, d):
d.insert('spam', "eggs")
self.failUnless(d.has_key('spam'))
self.failUnless(d.get('spam') == "eggs")
self.failUnless(d['spam'] == "eggs")
self.failUnlessEqual(d.items(), [("spam", "eggs")])
x = d.remove('spam')
self.failUnless(x == "eggs", "x: %s" % `x`)
self.failUnless(not d.has_key('spam'))
self.failUnlessEqual(d.items(), [])
d['spam'] = "eggsy"
self.failUnless(d.has_key('spam'))
self.failUnless(d.get('spam') == "eggsy")
self.failUnless(d['spam'] == "eggsy")
self.failUnlessEqual(d.items(), [("spam", "eggsy")])
del d['spam']
self.failUnless(not d.has_key('spam'))
self.failUnlessEqual(d.items(), [])
def _test_setdefault(self, d):
d.setdefault('spam', "eggs")
self.failUnless(d.has_key('spam'))
self.failUnless(d.get('spam') == "eggs")
self.failUnless(d['spam'] == "eggs")
self.failUnlessEqual(d.items(), [("spam", "eggs")])
x = d.remove('spam')
self.failUnless(x == "eggs", "x: %s" % `x`)
self.failUnless(not d.has_key('spam'))
self.failUnlessEqual(d.items(), [])
def _test_extracted_bound_method(self, d):
insmeth = d.insert
insmeth('spammy', "eggsy")
self.failUnless(d.get('spammy') == "eggsy")
def _test_extracted_unbound_method(self, d):
insumeth = d.__class__.insert
insumeth(d, 'spammy', "eggsy")
self.failUnless(d.get('spammy') == "eggsy")
def _test_unbound_method(self, C, d):
umeth = C.insert
umeth(d, 'spammy', "eggsy")
self.failUnless(d.get('spammy') == "eggsy")
def _test_clear(self, d):
d[11] = 11
d._assert_invariants()
self.failUnless(len(d) == 1)
d.clear()
d._assert_invariants()
self.failUnless(len(d) == 0)
self.failUnlessEqual(d.items(), [])
def _test_update_from_dict(self, d):
self.failUnless(d._assert_invariants())
d['b'] = 99
self.failUnless(d._assert_invariants())
d2={ 'a': 0, 'b': 1, 'c': 2,}
d.update(d2)
self.failUnless(d._assert_invariants())
self.failUnless(d.get('a') == 0, "d.get('a'): %s" % d.get('a'))
self.failUnless(d._assert_invariants())
self.failUnless(d.get('b') == 1, "d.get('b'): %s" % d.get('b'))
self.failUnless(d._assert_invariants())
self.failUnless(d.get('c') == 2)
self.failUnless(d._assert_invariants())
def _test_update_from_odict(self, d):
self.failUnless(d._assert_invariants())
d['b'] = 99
self.failUnless(d._assert_invariants())
d2 = odict.OrderedDict()
d2['a'] = 0
d2['b'] = 1
d2['c'] = 2
d.update(d2)
self.failUnless(d._assert_invariants())
self.failUnless(d.get('a') == 0, "d.get('a'): %s" % d.get('a'))
self.failUnless(d._assert_invariants())
self.failUnless(d.get('b') == 1, "d.get('b'): %s" % d.get('b'))
self.failUnless(d._assert_invariants())
self.failUnless(d.get('c') == 2)
self.failUnless(d._assert_invariants())
self.failUnlessEqual(d.items(), [("b", 1), ("a", 0), ("c", 2)])
def _test_popitem(self, C):
c = C({"a": 1})
res = c.popitem()
self.failUnlessEqual(res, ("a", 1,))
c["a"] = 1
c["b"] = 2
res = c.popitem()
self.failUnlessEqual(res, ("b", 2,))
def _test_pop(self, C):
c = C({"a": 1})
res = c.pop()
self.failUnlessEqual(res, "a")
c["a"] = 1
c["b"] = 2
res = c.pop()
self.failUnlessEqual(res, "b")
def _test_iterate_items(self, C):
c = C({"a": 1})
c["b"] = 2
i = c.iteritems()
x = i.next()
self.failUnlessEqual(x, ("a", 1,))
x = i.next()
self.failUnlessEqual(x, ("b", 2,))
try:
i.next()
self.fail() # Should have gotten StopIteration exception
except StopIteration:
pass
def _test_iterate_keys(self, C):
c = C({"a": 1})
c["b"] = 2
i = c.iterkeys()
x = i.next()
self.failUnlessEqual(x, "a")
x = i.next()
self.failUnlessEqual(x, "b")
try:
i.next()
self.fail() # Should have gotten StopIteration exception
except StopIteration:
pass
def _test_iterate_values(self, C):
c = C({"a": 1})
c["b"] = 2
i = c.itervalues()
x = i.next()
self.failUnless(x == 1)
x = i.next()
self.failUnless(x == 2)
try:
i.next()
self.fail() # Should have gotten StopIteration exception
except StopIteration:
pass
def _test_much_adding_some_removing(self, C):
c = C()
for i in range(MUCHADDINGSIZE):
c[i] = i
if (i % 4) == 0:
k = random.choice(c.keys())
del c[k]
for i in range(MUCHADDINGSIZE):
c[i] = i
self.failUnlessEqual(len(c), MUCHADDINGSIZE)
def _test_1(self, C):
c = C()
c[11] = 11
c._assert_invariants()
c[11] = 11
c._assert_invariants()
c[11] = 1001
c._assert_invariants()
c[11] = 11
c._assert_invariants()
c[11] = 1001
c._assert_invariants()
c[11] = 1001
c._assert_invariants()
c[11] = 1001
c._assert_invariants()
def _test_2(self, C):
c = C()
c[11] = 11
c._assert_invariants()
del c[11]
c._assert_invariants()
c[11] = 11
c._assert_invariants()
c[11] = 11
c._assert_invariants()
def _test_3(self, C):
c = C()
c[11] = 11
c._assert_invariants()
c[11] = 12
c._assert_invariants()
c[11] = 13
c._assert_invariants()
del c[11]
c._assert_invariants()
c[11] = 14
c._assert_invariants()
c[11] = 15
c._assert_invariants()
c[11] = 16
c._assert_invariants()
def _test_has_key(self, C):
c = C()
c._assert_invariants()
for i in xrange(11):
c._assert_invariants()
c[i] = i
c._assert_invariants()
del c[0]
self.failUnless(len(c) == 10)
self.failUnless(10 in c.values())
self.failUnless(0 not in c.values())
c.has_key(1) # this touches `1' but does not make it fresher so that it will get popped next time we pop.
c[1] = 1 # this touches `1' but does not make it fresher so that it will get popped.
c._assert_invariants()
x = c.pop()
self.failUnlessEqual(x, 10)
c[99] = 99
c._assert_invariants()
self.failUnless(len(c) == 10)
self.failUnless(1 in c.values(), "C: %s, c.values(): %s" % (hr(C), hr(c.values(),),))
self.failUnless(2 in c.values(), "C: %s, c.values(): %s" % (hr(C), hr(c.values(),),))
self.failIf(10 in c.values(), "C: %s, c.values(): %s" % (hr(C), hr(c.values(),),))
self.failUnless(99 in c.values())
def _test_em(self):
for klass in (odict.OrderedDict,):
for testfunc in (self._test_empty_lookup, self._test_insert_and_get_and_items, self._test_insert_and_remove, self._test_extracted_bound_method, self._test_extracted_unbound_method, self._test_clear, self._test_update_from_dict, self._test_update_from_odict, self._test_setdefault,):
testfunc(klass())
for testfunc in (self._test_pop, self._test_popitem, self._test_iterate_items, self._test_iterate_keys, self._test_iterate_values, self._test_key_error, ):
testfunc(klass)
self._test_unbound_method(klass, klass())
for klass in (odict.OrderedDict,):
for testfunc in (self._test_1, self._test_2, self._test_3, self._test_has_key,):
testfunc(klass)
def test_em(self):
self._test_em()
def _mem_test_much_adding_some_removing(self):
for klass in (odict.LRUCache, odict.SmallLRUCache,):
return self._test_much_adding_some_removing(klass)
def test_mem_leakage(self):
try:
self._test_mem_leakage()
except memutil.NotSupportedException:
print "Skipping memory leak test since measurement of current mem usage isn't implemented on this platform."
pass
del test_mem_leakage # This test takes too long.
def _test_mem_leakage(self):
# measure one and throw it away, in order to reach a "steady state" in terms of initialization of memory state.
memutil.measure_mem_leakage(self.test_em, max(2**3, SAMPLES/2**3), iterspersample=2**0)
slope = memutil.measure_mem_leakage(self.test_em, max(2**3, SAMPLES/2**3), iterspersample=2**0)
self.failUnless(slope <= MIN_SLOPE, "%s leaks memory at a rate of approximately %s system bytes per invocation" % (self.test_em, "%0.3f" % slope,))
def test_mem_leakage_much_adding_some_removing(self):
try:
self._test_mem_leakage_much_adding_some_removing()
except memutil.NotSupportedException:
print "Skipping memory leak test since measurement of current mem usage isn't implemented on this platform."
pass
del test_mem_leakage_much_adding_some_removing # This test takes too long.
def _test_mem_leakage_much_adding_some_removing(self):
# measure one and throw it away, in order to reach a "steady state" in terms of initialization of memory state.
memutil.measure_mem_leakage(self._mem_test_much_adding_some_removing, SAMPLES, iterspersample=2**0)
slope = memutil.measure_mem_leakage(self._mem_test_much_adding_some_removing, SAMPLES, iterspersample=2**0)
self.failUnless(slope <= MIN_SLOPE, "%s leaks memory at a rate of approximately %s system bytes per invocation" % (self._mem_test_much_adding_some_removing, "%0.3f" % slope,))
def test_obj_leakage(self):
self._test_obj_leakage()
del test_obj_leakage # This test takes too long.
def _test_obj_leakage(self):
# measure one and throw it away, in order to reach a "steady state" in terms of initialization of objects state.
memutil.measure_obj_leakage(self.test_em, max(2**3, SAMPLES/2**3), iterspersample=2**0)
slope = memutil.measure_obj_leakage(self.test_em, max(2**3, SAMPLES/2**3), iterspersample=2**0)
self.failUnless(slope <= MIN_SLOPE, "%s leaks objects at a rate of approximately %s system bytes per invocation" % (self.test_em, "%0.3f" % slope,))
def test_obj_leakage_much_adding_some_removing(self):
self._test_obj_leakage_much_adding_some_removing()
del test_obj_leakage_much_adding_some_removing # This test takes too long.
def _test_obj_leakage_much_adding_some_removing(self):
# measure one and throw it away, in order to reach a "steady state" in terms of initialization of objects state.
memutil.measure_obj_leakage(self._mem_test_much_adding_some_removing, SAMPLES, iterspersample=2**0)
slope = memutil.measure_obj_leakage(self._mem_test_much_adding_some_removing, SAMPLES, iterspersample=2**0)
self.failUnless(slope <= MIN_SLOPE, "%s leaks objects at a rate of approximately %s system bytes per invocation" % (self._mem_test_much_adding_some_removing, "%0.3f" % slope,))
libs/pyutil/test/out_of_shape/test_strutil.py
+30
View File
@@ -0,0 +1,30 @@
#!/usr/bin/env python
# Copyright (c) 2004-2009 Zooko "Zooko" Wilcox-O'Hearn
# This file is part of pyutil; see README.rst for licensing terms.
import unittest
from pyutil.assertutil import _assert
from pyutil import strutil
class Teststrutil(unittest.TestCase):
def test_short_input(self):
self.failUnless(strutil.pop_trailing_newlines("\r\n") == "")
self.failUnless(strutil.pop_trailing_newlines("\r") == "")
self.failUnless(strutil.pop_trailing_newlines("x\r\n") == "x")
self.failUnless(strutil.pop_trailing_newlines("x\r") == "x")
def test_split(self):
_assert(strutil.split_on_newlines("x\r\ny") == ["x", "y",], strutil.split_on_newlines("x\r\ny"))
_assert(strutil.split_on_newlines("x\r\ny\r\n") == ["x", "y", '',], strutil.split_on_newlines("x\r\ny\r\n"))
_assert(strutil.split_on_newlines("x\n\ny\n\n") == ["x", '', "y", '', '',], strutil.split_on_newlines("x\n\ny\n\n"))
def test_commonprefix(self):
_assert(strutil.commonprefix(["foo","foobarooo", "foosplat",]) == 'foo', strutil.commonprefix(["foo","foobarooo", "foosplat",]))
_assert(strutil.commonprefix(["foo","afoobarooo", "foosplat",]) == '', strutil.commonprefix(["foo","afoobarooo", "foosplat",]))
def test_commonsuffix(self):
_assert(strutil.commonsuffix(["foo","foobarooo", "foosplat",]) == '', strutil.commonsuffix(["foo","foobarooo", "foosplat",]))
_assert(strutil.commonsuffix(["foo","foobarooo", "foosplato",]) == 'o', strutil.commonsuffix(["foo","foobarooo", "foosplato",]))
_assert(strutil.commonsuffix(["foo","foobarooofoo", "foosplatofoo",]) == 'foo', strutil.commonsuffix(["foo","foobarooofoo", "foosplatofoo",]))
libs/pyutil/test/out_of_shape/test_zlibutil.py
+77
View File
@@ -0,0 +1,77 @@
#!/usr/bin/env python
import unittest
from pyutil import randutil
from pyutil import zlibutil
class Accumulator:
def __init__(self):
self.buf = ''
def write(self, str):
self.buf += str
def make_decomp(realdecomp):
def decomp(str, maxlen, maxmem):
d = Accumulator()
realdecomp(str, d, maxlen, maxmem)
return d.buf
return decomp
def genrandstr(strlen):
return randutil.insecurerandstr(strlen)
def genbombstr(strlen):
return '0' * strlen
MAXMEM=65*2**20
class ZlibTestCase(unittest.TestCase):
def _help_test(self, genstring, decomp, strlen):
s = genstring(strlen)
cs = zlibutil.zlib.compress(s)
s2 = decomp(cs, maxlen=strlen, maxmem=strlen*2**3 + zlibutil.MINMAXMEM)
self.failUnless(s == s2)
s2 = decomp(cs, maxlen=strlen, maxmem=strlen*2**6 + zlibutil.MINMAXMEM)
self.failUnless(s == s2)
self.failUnlessRaises(zlibutil.TooBigError, decomp, cs, maxlen=strlen-1, maxmem=strlen*2**3 + zlibutil.MINMAXMEM)
def _help_test_inplace_minmaxmem(self, genstring, decomp, strlen):
s = genstring(strlen)
cs = zlibutil.zlib.compress(s)
s2 = decomp(cs, maxlen=strlen, maxmem=zlibutil.MINMAXMEM)
self.failUnless(s == s2)
self.failUnlessRaises(zlibutil.TooBigError, decomp, cs, maxlen=strlen-1, maxmem=zlibutil.MINMAXMEM)
def _help_test_inplace(self, genstring, decomp, strlen):
# ### XXX self.failUnlessRaises(UnsafeDecompressError, decomp, zlib.compress(genstring(strlen)), maxlen=strlen, maxmem=strlen-1)
s = genstring(strlen)
cs = zlibutil.zlib.compress(s)
s2 = decomp(cs, maxlen=strlen, maxmem=max(strlen*2**3, zlibutil.MINMAXMEM))
self.failUnless(s == s2)
s2 = decomp(cs, maxlen=strlen, maxmem=max(strlen*2**6, zlibutil.MINMAXMEM))
self.failUnless(s == s2)
s2 = decomp(cs, maxlen=strlen, maxmem=max(strlen-1, zlibutil.MINMAXMEM))
self.failUnless(s == s2)
s2 = decomp(cs, maxlen=strlen, maxmem=max(strlen/2, zlibutil.MINMAXMEM))
self.failUnless(s == s2)
self.failUnlessRaises(zlibutil.TooBigError, decomp, cs, maxlen=strlen-1, maxmem=max(strlen*2**3, zlibutil.MINMAXMEM))
def testem(self):
# for strlen in [2**1, 2**2, 2**10, 2**14, 2**21]: # a *real* test ought to include 2**21, which exercises different cases re: maxmem. But it takes too long.
for strlen in [2, 3, 4, 99,]:
# print "strlen: %s\n" % (strlen,)
for decomp in [zlibutil.decompress, make_decomp(zlibutil.decompress_to_fileobj), make_decomp(zlibutil.decompress_to_spool),]:
# print "decomp: %s\n" % (decomp,)
for genstring in [genrandstr, genbombstr,]:
# print "genstring: %s\n" % (genstring,)
self._help_test(genstring, decomp, strlen)
for decomp in [make_decomp(zlibutil.decompress_to_spool),]:
# print "decomp: %s\n" % (decomp,)
for genstring in [genrandstr, genbombstr,]:
# print "genstring: %s\n" % (genstring,)
self._help_test_inplace(genstring, decomp, strlen)
self._help_test_inplace_minmaxmem(genstring, decomp, strlen)