From ab85ae76b468b99250defe86e07bb85c323cbe79 Mon Sep 17 00:00:00 2001 From: "brian m. carlson" Date: Sun, 13 Jul 2014 16:19:19 +0000 Subject: [PATCH] testdatabase: bring up to PEP 8 standards. Signed-off-by: brian m. carlson --- test/testdatabase.py | 50 +++++++++++++++++++++++++++++++++++++------- 1 file changed, 42 insertions(+), 8 deletions(-) diff --git a/test/testdatabase.py b/test/testdatabase.py index 7eaddca..19c5618 100755 --- a/test/testdatabase.py +++ b/test/testdatabase.py @@ -9,46 +9,59 @@ import socket import tempfile import unittest + class TestDatabaseVersion(unittest.TestCase): def test_default_serialization(self): self.assertEqual(DatabaseVersion().serialization(), "csv") + def test_preferred_serialization(self): self.assertEqual(DatabaseVersion.preferred().serialization(), "json") + def test_preferred_serialization_version(self): - self.assertEqual(DatabaseVersion.preferred().serialization_version(), 2) + self.assertEqual(DatabaseVersion.preferred().serialization_version(), + 2) + def test_preferred_compression(self): self.assertEqual(DatabaseVersion.preferred().compression(), None) + def test_preferred_compression_version(self): self.assertEqual(DatabaseVersion.preferred().compression_version(), 0) + def test_preferred_record_version(self): self.assertEqual(DatabaseVersion.preferred().record_version(), 3) + def test_preferred_version_version(self): self.assertEqual(DatabaseVersion.preferred().version_version(), 0) + class TestDatabaseAccessors(unittest.TestCase): def test_location(self): for location in ("/tmp/foo", "/bar", "/home/quux/.newfol"): obj = Database(location, 2, Schema(), []) self.assertEqual(obj.location(), location) + def test_records(self): obj = Database("/tmp/foo", 2, Schema(), [Record([1, 2, 3])]) records = obj.records() self.assertEqual(len(records), 1) self.assertEqual(len(records[0].fields), 3) + def test_version(self): obj = Database("/tmp/foo", DatabaseVersion(), Schema(), []) self.assertEqual(obj.version(), 0) obj = Database("/tmp/foo", DatabaseVersion.preferred(), Schema(), []) self.assertEqual(obj.version(), DatabaseVersion.preferred()) + def test_serialization(self): obj = Database("/tmp/foo", DatabaseVersion(), Schema(), []) self.assertEqual(obj.serialization(), - DatabaseVersion(0).serialization()) + DatabaseVersion(0).serialization()) obj = Database("/tmp/foo", DatabaseVersion.preferred(), Schema(), []) self.assertEqual(obj.serialization(), - DatabaseVersion.preferred().serialization()) + DatabaseVersion.preferred().serialization()) + class TestDatabaseIntegrity(unittest.TestCase): def create_temp_db(self): @@ -58,37 +71,43 @@ class TestDatabaseIntegrity(unittest.TestCase): fp.close() db = Database.load(ddir.name) return (ddir, db) + def test_version(self): ddir, db = self.create_temp_db() self.assertEqual(Database.read_version(ddir.name), - DatabaseVersion()) + DatabaseVersion()) db.store() db.upgrade() self.assertEqual(Database.read_version(ddir.name), - DatabaseVersion.preferred()) + DatabaseVersion.preferred()) ddir.cleanup() + def test_validate(self): ddir, db = self.create_temp_db() db.store() db.validate() ddir.cleanup() + def test_validate_strict(self): ddir, db = self.create_temp_db() db.store() db.validate(strict=True) ddir.cleanup() + def test_repair_doesnt_raise(self): ddir, db = self.create_temp_db() db.store() db.repair() db.validate(strict=True) ddir.cleanup() + def test_upgrade_records(self): ddir, db = self.create_temp_db() db.store() db.upgrade_records() ddir.cleanup() + class TestDatabaseUpgrades(unittest.TestCase): def create_temp_db(self): ddir = tempfile.TemporaryDirectory() @@ -97,31 +116,36 @@ class TestDatabaseUpgrades(unittest.TestCase): fp.close() db = Database.load(ddir.name) return (ddir, db) + def do_upgrade_test(self, version, pattern): ddir, db = self.create_temp_db() if not isinstance(version, DatabaseVersion): version = DatabaseVersion(version) self.assertEqual(Database.read_version(ddir.name), - DatabaseVersion()) + DatabaseVersion()) db.store() db.upgrade(version=version) self.assertEqual(Database.read_version(ddir.name), - version) + version) fp = open(ddir.name + "/dtb", "rb") data = fp.read(len(pattern)) self.assertEqual(data, pattern) fp.close() ddir.cleanup() + def test_upgrade_to_xz(self): version = (DatabaseVersion.preferred() & ~0x00ff0000) | 0x00010000 self.do_upgrade_test(version, b"\xfd7zXZ\x00") + def test_upgrade_to_json(self): version = (DatabaseVersion.preferred() & ~0x0000ff00) | 0x00000200 self.do_upgrade_test(version, b"[") + def test_upgrade_to_pickle(self): version = (DatabaseVersion.preferred() & ~0x0000ff00) | 0x00000100 self.do_upgrade_test(version, b"(lp0\n") + class TestGitTransactions(unittest.TestCase): def setUp(self): self.ddir = tempfile.TemporaryDirectory() @@ -129,9 +153,11 @@ class TestGitTransactions(unittest.TestCase): fp.write("fmt:0:newfol schema file:\ntxn:git\n") fp.close() self.db = Database.load(self.ddir.name) + def test_has_git_dir(self): self.db.store() self.assertTrue(os.path.isdir(self.ddir.name + "/.git")) + def get_format_data(self, fmt): cwd = os.getcwd() data = None @@ -143,15 +169,18 @@ class TestGitTransactions(unittest.TestCase): finally: os.chdir(cwd) return data + def test_correct_committer_name(self): self.db.store() self.assertEqual(self.get_format_data("%cn"), "newfol") + def test_correct_committer_email(self): self.db.store() user = pwd.getpwuid(os.getuid())[0] fqdn = socket.getfqdn() self.assertEqual(self.get_format_data("%ce"), user + "@" + fqdn) + class TestMultipleTransactions(unittest.TestCase): def test_multiple_types(self): ddir = tempfile.TemporaryDirectory() @@ -162,9 +191,10 @@ class TestMultipleTransactions(unittest.TestCase): db.records()[:] = [Record([1, 2, 3])] db.store() self.assertEqual(set(db.schema().transaction_types()), - set(["git", "hash"])) + set(["git", "hash"])) ddir.cleanup() + class TestDatabaseFiltering(unittest.TestCase): def create_temp_db(self): ddir = tempfile.TemporaryDirectory() @@ -173,6 +203,7 @@ class TestDatabaseFiltering(unittest.TestCase): fp.close() db = Database.load(ddir.name) return (ddir, db) + def test_filtering(self): ddir, db = self.create_temp_db() records = [ @@ -181,6 +212,7 @@ class TestDatabaseFiltering(unittest.TestCase): Record([7, 8, 9]) ] db.records().extend(records) + def has_only_numbers(rec): for field in rec.fields: try: @@ -193,6 +225,7 @@ class TestDatabaseFiltering(unittest.TestCase): self.assertEqual(set(selected), set(records[1:])) ddir.cleanup() + class TestSingleton(unittest.TestCase): def test_singleton(self): def TestClass(metaclass=Singleton): @@ -201,5 +234,6 @@ class TestSingleton(unittest.TestCase): b = TestClass() self.assertIs(a, b) + if __name__ == '__main__': unittest.main()