diff options
author | Sascha Silbe <sascha-pgp@silbe.org> | 2011-01-19 19:08:17 (GMT) |
---|---|---|
committer | Sascha Silbe <sascha-pgp@silbe.org> | 2011-01-19 19:08:17 (GMT) |
commit | 111e96dc6ec6ce26eba406e7aed85713bc069e4a (patch) | |
tree | 490276167c71a148388728c0a899f3e1f21b2045 | |
parent | be1f2707c8e2fe02dff3b0c43483e1aa8b473f65 (diff) |
pep8 / pylint / style fixes
-rw-r--r-- | tests/basic_api_v2.txt | 6 | ||||
-rwxr-xr-x | tests/runalltests.py | 69 | ||||
-rw-r--r-- | tests/test_massops.py | 58 | ||||
-rw-r--r-- | tests/test_migration_v1_v2.py | 43 |
4 files changed, 95 insertions, 81 deletions
diff --git a/tests/basic_api_v2.txt b/tests/basic_api_v2.txt index b56e629..b0c4a88 100644 --- a/tests/basic_api_v2.txt +++ b/tests/basic_api_v2.txt @@ -19,9 +19,9 @@ Define some helper functions Connect to datastore using DBus and wait for it to get ready: >>> import dbus ->>> DS_DBUS_SERVICE = "org.laptop.sugar.DataStore" ->>> DS_DBUS_INTERFACE = "org.laptop.sugar.DataStore" ->>> DS_DBUS_PATH = "/org/laptop/sugar/DataStore" +>>> DS_DBUS_SERVICE = 'org.laptop.sugar.DataStore' +>>> DS_DBUS_INTERFACE = 'org.laptop.sugar.DataStore' +>>> DS_DBUS_PATH = '/org/laptop/sugar/DataStore' >>> bus = dbus.SessionBus() >>> ds = dbus.Interface(bus.get_object(DS_DBUS_SERVICE, DS_DBUS_PATH), DS_DBUS_INTERFACE) diff --git a/tests/runalltests.py b/tests/runalltests.py index f59200f..9a7c8f8 100755 --- a/tests/runalltests.py +++ b/tests/runalltests.py @@ -27,19 +27,20 @@ import gobject logging.basicConfig(level=logging.WARN, - format="%(asctime)-15s %(name)s %(levelname)s: %(message)s", - stream=sys.stderr) + format='%(asctime)-15s %(name)s %(levelname)s:' + ' %(message)s', + stream=sys.stderr) DOCTESTS = [ - "basic_api_v2.txt", + 'basic_api_v2.txt', ] DOCTEST_OPTIONS = doctest.ELLIPSIS DOCTEST_OPTIONS |= doctest.REPORT_ONLY_FIRST_FAILURE -DS_DBUS_SERVICE = "org.laptop.sugar.DataStore" -DS_DBUS_INTERFACE = "org.laptop.sugar.DataStore" -DS_DBUS_PATH = "/org/laptop/sugar/DataStore" +DS_DBUS_SERVICE = 'org.laptop.sugar.DataStore' +DS_DBUS_INTERFACE = 'org.laptop.sugar.DataStore' +DS_DBUS_PATH = '/org/laptop/sugar/DataStore' ENVIRONMENT_WHITELIST = [ 'LD_LIBRARY_PATH', @@ -48,6 +49,13 @@ ENVIRONMENT_WHITELIST = [ 'SUGAR_LOGGER_LEVEL', ] +SERVICE_TEMPLATE = """ +[D-BUS Service] +Name = org.laptop.sugar.DataStore +Exec = %s/bin/datastore-service +""" + + def setup(): """Prepare for testing and return environment. @@ -69,18 +77,16 @@ def setup(): basedir = os.path.join(os.path.dirname(os.path.abspath(sys.argv[0])), '..') python_path = [os.path.join(basedir, 'src')] + python_path environment['PYTHONPATH'] = ':'.join(python_path) - environment['PATH'] = os.path.join(basedir, 'bin')+':'+os.environ['PATH'] - - servicedir = os.path.join(environment['HOME'], 'dbus-1', 'services') - servicepath = os.path.join(servicedir, 'org.laptop.sugar.DataStore.service') - os.makedirs(servicedir) - servicefile = file(servicepath, 'w') - servicefile.write(""" - [D-BUS Service] - Name = org.laptop.sugar.DataStore - Exec = %s/bin/datastore-service - """.replace(' ', '') % (basedir, )) - servicefile.close() + environment['PATH'] = ':'.join([os.path.join(basedir, 'bin'), + os.environ['PATH']]) + + service_dir = os.path.join(environment['HOME'], 'dbus-1', 'services') + service_path = os.path.join(service_dir, + 'org.laptop.sugar.DataStore.service') + os.makedirs(service_dir) + service_file = file(service_path, 'w') + service_file.write(SERVICE_TEMPLATE % (basedir, )) + service_file.close() environment['XDG_DATA_DIRS'] = environment['HOME'] os.setpgid(0, 0) @@ -172,7 +178,7 @@ class TestSuiteWrapper(unittest.TestCase): def shortDescription(self): doc = self._wrapped_suite.__doc__ - return doc and doc.split("\n")[0].strip() or None + return doc and doc.split('\n')[0].strip() or None def tearDown(self): self._kill_data_store() @@ -180,13 +186,13 @@ class TestSuiteWrapper(unittest.TestCase): def _kill_data_store(self): pgrep = subprocess.Popen(['pgrep', '-g', os.environ['DBUS_PID'], - '-f', 'datastore-service'], - close_fds=True, stdout=subprocess.PIPE) + '-f', 'datastore-service'], + close_fds=True, stdout=subprocess.PIPE) stdout, stderr_ = pgrep.communicate() pids = stdout.strip().split('\n') if len(pids) != 1 or not pids[0]: raise ValueError("Can't find (a single) data store process " - "(pgrep output %r)" % (stdout, )) + "(pgrep output %r)" % (stdout, )) pid = int(pids[0]) self._loop = gobject.MainLoop() @@ -212,7 +218,7 @@ class TimedTestResult(unittest._TextTestResult): # Depending on a private class is bad style, but the only alternative is # copying it verbatim. - # pylint: disable-msg=W0212 + # pylint: disable=W0212 def __init__(self, stream, descriptions, verbosity): unittest._TextTestResult.__init__(self, stream, descriptions, @@ -250,7 +256,7 @@ class TimedTestResult(unittest._TextTestResult): return if self.showAll: - self.stream.writeln("ok (%.3fs)" % (run_time, )) + self.stream.writeln('ok (%.3fs)' % (run_time, )) elif self.dots: self.stream.write('.') @@ -297,7 +303,8 @@ def run_tests(tests): def _start_dbus(environment): pipe = subprocess.Popen(['dbus-launch'], stdout=subprocess.PIPE, - close_fds=True, env=environment, cwd=environment['HOME']) + close_fds=True, env=environment, + cwd=environment['HOME']) stdout, stderr_ = pipe.communicate() pid = None address = None @@ -319,11 +326,11 @@ def _parse_options(): """Parse command line arguments.""" parser = OptionParser() parser.add_option('-k', '--keep', dest='keep', - action='store_true', default=False, - help='Keep temporary files') + action='store_true', default=False, + help='Keep temporary files') parser.add_option('', '--stage2', dest='stage2', - action='store_true', default=False, - help='For internal use only') + action='store_true', default=False, + help='For internal use only') return parser.parse_args() @@ -339,8 +346,8 @@ def main(my_name, arguments): environment['DBUS_PID'] = str(dbus_pid) pipe = subprocess.Popen([os.path.abspath(my_name), - '--stage2']+arguments, - cwd=environment['HOME'], env=environment) + '--stage2'] + arguments, + cwd=environment['HOME'], env=environment) return pipe.wait() finally: diff --git a/tests/test_massops.py b/tests/test_massops.py index 8638bc1..edce0c3 100644 --- a/tests/test_massops.py +++ b/tests/test_massops.py @@ -9,9 +9,9 @@ import time import unittest -DS_DBUS_SERVICE = "org.laptop.sugar.DataStore" -DS_DBUS_INTERFACE = "org.laptop.sugar.DataStore" -DS_DBUS_PATH = "/org/laptop/sugar/DataStore" +DS_DBUS_SERVICE = 'org.laptop.sugar.DataStore' +DS_DBUS_INTERFACE = 'org.laptop.sugar.DataStore' +DS_DBUS_PATH = '/org/laptop/sugar/DataStore' NUM_RUNS = int(os.environ.get('MASSOPS_RUNS', '100')) IGNORE_PROPERTIES = [ 'checksum', @@ -33,17 +33,17 @@ class MassOpsTestCase(unittest.TestCase): """Large number of operations intended for measuring performance.""" def setUp(self): - # pylint: disable-msg=C0103 + # pylint: disable=C0103 self._bus = dbus.SessionBus() self._datastore = dbus.Interface(self._bus.get_object(DS_DBUS_SERVICE, - DS_DBUS_PATH), DS_DBUS_INTERFACE) + DS_DBUS_PATH), DS_DBUS_INTERFACE) _create_properties = { 'title': 'DS test object', 'mime_type': 'text/plain', 'activity': 'org.sugarlabs.DataStoreTest1', } - _create_content = 'Foo bar\n'*1000 + _create_content = 'Foo bar\n' * 1000 def test_create(self): """Run create() lots of times to create new objects.""" @@ -61,17 +61,18 @@ class MassOpsTestCase(unittest.TestCase): def test_find_all(self): """Run find() to list all entries.""" entries, total_count = self._datastore.find({}, ['number'], - byte_arrays=True) + byte_arrays=True) self.assertEquals(total_count, NUM_RUNS) self.assertEquals(total_count, len(entries)) for position, entry in enumerate(entries): - self.assertEquals(int(entry['number']), NUM_RUNS-position-1) + self.assertEquals(int(entry['number']), NUM_RUNS - position - 1) @repeat def test_find_all_reverse_time(self): """Run find() to list all entries in reverse chronological order.""" - entries, total_count = self._datastore.find({'order_by': - ['-timestamp']}, ['number'], byte_arrays=True) + entries, total_count = \ + self._datastore.find({'order_by': ['-timestamp']}, ['number'], + byte_arrays=True) self.assertEquals(total_count, NUM_RUNS) self.assertEquals(total_count, len(entries)) for position, entry in enumerate(entries): @@ -80,16 +81,18 @@ class MassOpsTestCase(unittest.TestCase): @repeat def test_find_all_title(self): """Run find() to list all entries ordered by title.""" - entries, total_count = self._datastore.find({'order_by': - ['+title']}, ['tree_id'], byte_arrays=True) + entries, total_count = \ + self._datastore.find({'order_by': ['+title']}, ['tree_id'], + byte_arrays=True) self.assertEquals(total_count, NUM_RUNS) self.assertEquals(total_count, len(entries)) @repeat def test_find_all_reverse_title(self): """Run find() to list all entries ordered by title (reversed).""" - entries, total_count = self._datastore.find({'order_by': - ['-title']}, ['tree_id'], byte_arrays=True) + entries, total_count = \ + self._datastore.find({'order_by': ['-title']}, ['tree_id'], + byte_arrays=True) self.assertEquals(total_count, NUM_RUNS) self.assertEquals(total_count, len(entries)) @@ -98,21 +101,22 @@ class MassOpsTestCase(unittest.TestCase): """Run find() to list all entries in small chunks.""" chunk_size = 30 for chunk_start in range(0, NUM_RUNS, 30): - entries, total_count = self._datastore.find({ - 'offset': chunk_start, 'limit': chunk_size}, - ['number'], byte_arrays=True) + entries, total_count = \ + self._datastore.find({'offset': chunk_start, + 'limit': chunk_size}, ['number'], + byte_arrays=True) self.assertEquals(len(entries), - min(chunk_size, NUM_RUNS-chunk_start)) + min(chunk_size, NUM_RUNS - chunk_start)) self.assertEquals(total_count, NUM_RUNS) for position, entry in enumerate(entries): self.assertEquals(int(entry['number']), - NUM_RUNS-(chunk_start+position)-1) + NUM_RUNS - (chunk_start + position) - 1) def test_get_properties(self): """Run get_properties() on all entries and verify result.""" for entry in self._datastore.find({}, ['uid'], byte_arrays=True)[0]: - properties = self._datastore.get_properties(entry['uid'], - byte_arrays=True) + properties = \ + self._datastore.get_properties(entry['uid'], byte_arrays=True) self.assertEquals(properties.pop('filesize'), str(len(self._create_content))) self._filter_properties(properties) @@ -122,7 +126,7 @@ class MassOpsTestCase(unittest.TestCase): """Run get_filename() on all entries and verify content.""" for entry in self._datastore.find({}, ['uid'], byte_arrays=True)[0]: filename = self._datastore.get_filename(entry['uid'], - byte_arrays=True) + byte_arrays=True) try: self.assertEquals(file(filename).read(), self._create_content) finally: @@ -133,7 +137,7 @@ class MassOpsTestCase(unittest.TestCase): 'mime_type': 'text/plain', 'activity': 'org.sugarlabs.DataStoreTest1', } - _update_content = 'Foo bar baz\n'*1000 + _update_content = 'Foo bar baz\n' * 1000 def test_update(self): """Update the content of all existing entries""" @@ -142,13 +146,15 @@ class MassOpsTestCase(unittest.TestCase): content_file.flush() for entry in self._datastore.find({}, ['uid'], byte_arrays=True)[0]: self._datastore.update(entry['uid'], self._update_properties, - content_file.name, False) + content_file.name, False) def test_update_verify(self): - """Verify test_update() has changed content and metadata of all entries.""" + """ + Verify test_update() has changed content and metadata of all entries. + """ for entry in self._datastore.find({}, [], byte_arrays=True)[0]: filename = self._datastore.get_filename(entry['uid'], - byte_arrays=True) + byte_arrays=True) self.assertEquals(entry.pop('filesize'), str(len(self._update_content))) self._filter_properties(entry) diff --git a/tests/test_migration_v1_v2.py b/tests/test_migration_v1_v2.py index fd7010d..bad1009 100644 --- a/tests/test_migration_v1_v2.py +++ b/tests/test_migration_v1_v2.py @@ -2,18 +2,16 @@ """Test datastore migration from version 1 to version 2.""" import dbus -import decorator import hashlib import os -import tempfile import time import unittest import uuid -DS_DBUS_SERVICE = "org.laptop.sugar.DataStore" -DS_DBUS_INTERFACE = "org.laptop.sugar.DataStore" -DS_DBUS_PATH = "/org/laptop/sugar/DataStore" +DS_DBUS_SERVICE = 'org.laptop.sugar.DataStore' +DS_DBUS_INTERFACE = 'org.laptop.sugar.DataStore' +DS_DBUS_PATH = '/org/laptop/sugar/DataStore' IGNORE_PROPERTIES = [ 'activity_id', 'checksum', @@ -31,10 +29,10 @@ class MigrationV1V2TestCase(unittest.TestCase): def __init__(self, *args, **kwargs): unittest.TestCase.__init__(self, *args, **kwargs) - self._templates = self._v1_properties*10 + self._templates = self._v1_properties * 10 def setUp(self): - # pylint: disable-msg=C0103 + # pylint: disable=C0103 profile = os.environ.get('SUGAR_PROFILE', 'default') base_dir = os.path.join(os.path.expanduser('~'), '.sugar', profile) self._root_path = os.path.join(base_dir, 'datastore') @@ -43,7 +41,8 @@ class MigrationV1V2TestCase(unittest.TestCase): self._bus = dbus.SessionBus() self._datastore = dbus.Interface(self._bus.get_object(DS_DBUS_SERVICE, - DS_DBUS_PATH), DS_DBUS_INTERFACE) + DS_DBUS_PATH), + DS_DBUS_INTERFACE) _v1_properties = [ { @@ -60,8 +59,8 @@ class MigrationV1V2TestCase(unittest.TestCase): 'keep': '1', 'mime_type': 'text/html', 'activity': 'org.sugarlabs.DataStoreTest2', - 'activity_id': lambda *args: str(uuid.uuid4()), - 'timestamp': lambda *args: time.time(), + 'activity_id': lambda number_: str(uuid.uuid4()), + 'timestamp': lambda number_: time.time(), 'icon-color': '#00ff00,#0000ff', 'buddies': '{}', 'description': 'DS migration test object', @@ -71,19 +70,20 @@ class MigrationV1V2TestCase(unittest.TestCase): { 'title': lambda number: 'DS test object %d' % (number, ), 'activity': 'org.sugarlabs.DataStoreTest3', - 'activity_id': lambda *args: str(uuid.uuid4()), - 'ctime': lambda *args: time.strftime('%Y-%m-%dT%H:%M:%S'), + 'activity_id': lambda number_: str(uuid.uuid4()), + 'ctime': lambda number_: time.strftime('%Y-%m-%dT%H:%M:%S'), }, { 'title': lambda number: 'DS test object %d' % (number, ), 'activity': 'org.sugarlabs.DataStoreTest4', - 'activity_id': lambda *args: str(uuid.uuid4()), - 'mtime': lambda *args: time.strftime('%Y-%m-%dT%H:%M:%S'), + 'activity_id': lambda number_: str(uuid.uuid4()), + 'mtime': lambda number_: time.strftime('%Y-%m-%dT%H:%M:%S'), }, {}, ] + def _v1_content(self, num): - return ('Foo bar %d\n' % (num, ))*1000 + return ('Foo bar %d\n' % (num, )) * 1000 def _create_v1_datastore(self): """Create a version 1 datastore on disk.""" @@ -126,17 +126,17 @@ class MigrationV1V2TestCase(unittest.TestCase): def test_find_all(self): """Run find() to list all migrated entries.""" - entries, count = self._find({}, ['uid']) + entries_, count = self._find({}, ['uid']) self.assertEquals(count, len(self._templates)) def test_get_properties(self): """Run get_properties() on all entries and verify result.""" for entry in self._find({}, ['uid'])[0]: properties = self._datastore.get_properties(entry['uid'], - byte_arrays=True) + byte_arrays=True) number = int(properties['number']) expected = self._fill_template(self._templates[number], - number) + number) self.assertEquals(properties.pop('filesize'), str(len(self._v1_content(number)))) self._filter_properties(properties) @@ -147,7 +147,7 @@ class MigrationV1V2TestCase(unittest.TestCase): """Run get_filename() on all entries and verify content.""" for entry in self._find({}, ['number', 'uid'])[0]: filename = self._datastore.get_filename(entry['uid'], - byte_arrays=True) + byte_arrays=True) content = file(filename).read() os.remove(filename) number = int(entry['number']) @@ -156,7 +156,7 @@ class MigrationV1V2TestCase(unittest.TestCase): def _find(self, query, properties): return self._datastore.find(dbus.Dictionary(query, signature='sv'), - properties, byte_arrays=True) + properties, byte_arrays=True) def _filter_properties(self, properties): for key in IGNORE_PROPERTIES: @@ -164,6 +164,7 @@ class MigrationV1V2TestCase(unittest.TestCase): def suite(): - test_suite = unittest.TestLoader().loadTestsFromTestCase(MigrationV1V2TestCase) + test_loader = unittest.TestLoader() + test_suite = test_loader.loadTestsFromTestCase(MigrationV1V2TestCase) test_suite.__doc__ = MigrationV1V2TestCase.__doc__ return test_suite |