diff options
| -rw-r--r-- | g_sorcery/db_layout.py | 6 | ||||
| -rw-r--r-- | tests/server.py | 4 | ||||
| -rw-r--r-- | tests/test_PackageDB.py | 133 |
3 files changed, 80 insertions, 63 deletions
diff --git a/g_sorcery/db_layout.py b/g_sorcery/db_layout.py index 1f4514c..da779bd 100644 --- a/g_sorcery/db_layout.py +++ b/g_sorcery/db_layout.py @@ -40,7 +40,7 @@ SUPPORTED_FILE_FORMATS = {JSON_FILE_SUFFIX: CategoryJSON} # bson module is optional, we should check if it is installed try: - from g_sorcery.file_bson.file_bson import FileBSON + from .file_bson.file_bson import FileBSON class CategoryBSON(FileBSON): """ @@ -172,9 +172,13 @@ def get_layout(metadata): return (CategoryJSON, [file_name(CATEGORIES_FILE_NAME)]) elif layout_version == 1: category_format = metadata['category_format'] + wrong_fmt = True try: category_cls = SUPPORTED_FILE_FORMATS[category_format] + wrong_fmt = False except KeyError: + pass + if wrong_fmt: raise DBLayoutError("unsupported packages file format: " + category_format) return (category_cls, [file_name(CATEGORIES_FILE_NAME), file_name(METADATA_FILE_NAME)]) else: diff --git a/tests/server.py b/tests/server.py index 5cb813f..51d49b7 100644 --- a/tests/server.py +++ b/tests/server.py @@ -38,10 +38,10 @@ def HTTPRequestHandlerGenerator(direct): class Server(threading.Thread): - def __init__(self, directory): + def __init__(self, directory, port=8080): super(Server, self).__init__() HTTPServer.allow_reuse_address = True - server_address = ('127.0.0.1', 8080) + server_address = ('127.0.0.1', port) self.httpd = HTTPServer(server_address, HTTPRequestHandlerGenerator(directory)) def run(self): diff --git a/tests/test_PackageDB.py b/tests/test_PackageDB.py index 2a67385..221be00 100644 --- a/tests/test_PackageDB.py +++ b/tests/test_PackageDB.py @@ -16,6 +16,7 @@ import time import unittest from g_sorcery.compatibility import TemporaryDirectory +from g_sorcery.db_layout import JSON_FILE_SUFFIX, BSON_FILE_SUFFIX from g_sorcery.exceptions import IntegrityError, InvalidKeyError, SyncError from g_sorcery.g_collections import Package from g_sorcery.package_db import PackageDB @@ -23,6 +24,14 @@ from g_sorcery.package_db import PackageDB from tests.base import BaseTest from tests.server import Server +SUPPORTED_FILE_FORMATS = [JSON_FILE_SUFFIX] +# bson module is optional, we should check if it is installed +try: + from g_sorcery.file_bson.file_bson import FileBSON + SUPPORTED_FILE_FORMATS.append(BSON_FILE_SUFFIX) +except ImportError as e: + pass + class TestDB(PackageDB): def get_real_db_uri(self, db_uri): @@ -32,66 +41,70 @@ class TestDB(PackageDB): class TestPackageDB(BaseTest): def test_functionality(self): - orig_tempdir = TemporaryDirectory() - orig_path = os.path.join(orig_tempdir.name, "db") - os.makedirs(orig_path) - orig_db = PackageDB(orig_path) - orig_db.add_category("app-test1") - orig_db.add_category("app-test2") - ebuild_data = {"test1": "tst1", "test2": "tst2"} - common_data = {"common1": "cmn1", "common2": "cmn2"} - packages = [Package("app-test1", "test", "1"), Package("app-test1", "test", "2"), - Package("app-test1", "test1", "1"), Package("app-test2", "test2", "1")] - for package in packages: - orig_db.add_package(package, ebuild_data) - orig_db.set_common_data("app-test1", common_data) - full_data = dict(ebuild_data) - full_data.update(common_data) - - orig_db.write() - os.system("cd " + orig_tempdir.name + " && tar cvzf good.tar.gz db") - os.system("echo invalid >> " + orig_tempdir.name + "/db/app-test1/packages.json") - os.system("cd " + orig_tempdir.name + " && tar cvzf dummy.tar.gz db") - - test_db = TestDB(self.tempdir.name) - self.assertRaises(SyncError, test_db.sync, "127.0.0.1:8080") - - srv = Server(orig_tempdir.name) - srv.start() - self.assertRaises(IntegrityError, test_db.sync, "127.0.0.1:8080") - os.system("cd " + orig_tempdir.name + " && mv good.tar.gz dummy.tar.gz") - test_db.sync("127.0.0.1:8080") - srv.shutdown() - srv.join() - test_db.read() - self.assertEqual(orig_db.database, test_db.database) - self.assertEqual(orig_db.get_common_data("app-test1"), test_db.get_common_data("app-test1")) - self.assertEqual(orig_db.get_common_data("app-test2"), test_db.get_common_data("app-test2")) - self.assertEqual(set(test_db.list_categories()), set(["app-test1", "app-test2"])) - self.assertTrue(test_db.in_category("app-test1", "test")) - self.assertFalse(test_db.in_category("app-test2", "test")) - self.assertRaises(InvalidKeyError, test_db.in_category, "app-test3", "test") - self.assertEqual(set(test_db.list_package_names("app-test1")), set(['test', 'test1'])) - self.assertEqual(set(test_db.list_catpkg_names()),set(['app-test1/test', 'app-test1/test1', 'app-test2/test2'])) - self.assertRaises(InvalidKeyError, test_db.list_package_versions, "invalid", "test") - self.assertRaises(InvalidKeyError, test_db.list_package_versions, "app-test1", "invalid") - self.assertEqual(set(test_db.list_package_versions("app-test1", "test")), set(['1', '2'])) - self.assertEqual(set(test_db.list_all_packages()), set(packages)) - self.assertEqual(test_db.get_package_description(packages[0]), full_data) - self.assertRaises(KeyError, test_db.get_package_description, Package("invalid", "invalid", "1")) - self.assertEqual(test_db.get_max_version("app-test1", "test"), "2") - self.assertEqual(test_db.get_max_version("app-test1", "test1"), "1") - self.assertRaises(InvalidKeyError, test_db.get_max_version, "invalid", "invalid") - pkg_set = set(packages) - for package, data in test_db: - self.assertTrue(package in pkg_set) - if package.category == "app-test1": - self.assertEqual(data, full_data) - else: - self.assertEqual(data, ebuild_data) - pkg_set.remove(package) - self.assertTrue(not pkg_set) - self.assertEqual(orig_db.database, test_db.database) + port = 8080 + for fmt in SUPPORTED_FILE_FORMATS: + sync_address = "127.0.0.1:" + str(port) + orig_tempdir = TemporaryDirectory() + orig_path = os.path.join(orig_tempdir.name, "db") + os.makedirs(orig_path) + orig_db = PackageDB(orig_path, preferred_category_format=fmt) + orig_db.add_category("app-test1") + orig_db.add_category("app-test2") + ebuild_data = {"test1": "tst1", "test2": "tst2"} + common_data = {"common1": "cmn1", "common2": "cmn2"} + packages = [Package("app-test1", "test", "1"), Package("app-test1", "test", "2"), + Package("app-test1", "test1", "1"), Package("app-test2", "test2", "1")] + for package in packages: + orig_db.add_package(package, ebuild_data) + orig_db.set_common_data("app-test1", common_data) + full_data = dict(ebuild_data) + full_data.update(common_data) + + orig_db.write() + os.system("cd " + orig_tempdir.name + " && tar cvzf good.tar.gz db") + os.system("echo invalid >> " + orig_tempdir.name + "/db/app-test1/packages." + fmt) + os.system("cd " + orig_tempdir.name + " && tar cvzf dummy.tar.gz db") + + test_db = TestDB(self.tempdir.name) + self.assertRaises(SyncError, test_db.sync, sync_address) + + srv = Server(orig_tempdir.name, port=port) + srv.start() + self.assertRaises(IntegrityError, test_db.sync, sync_address) + os.system("cd " + orig_tempdir.name + " && mv good.tar.gz dummy.tar.gz") + test_db.sync(sync_address) + srv.shutdown() + srv.join() + test_db.read() + self.assertEqual(orig_db.database, test_db.database) + self.assertEqual(orig_db.get_common_data("app-test1"), test_db.get_common_data("app-test1")) + self.assertEqual(orig_db.get_common_data("app-test2"), test_db.get_common_data("app-test2")) + self.assertEqual(set(test_db.list_categories()), set(["app-test1", "app-test2"])) + self.assertTrue(test_db.in_category("app-test1", "test")) + self.assertFalse(test_db.in_category("app-test2", "test")) + self.assertRaises(InvalidKeyError, test_db.in_category, "app-test3", "test") + self.assertEqual(set(test_db.list_package_names("app-test1")), set(['test', 'test1'])) + self.assertEqual(set(test_db.list_catpkg_names()),set(['app-test1/test', 'app-test1/test1', 'app-test2/test2'])) + self.assertRaises(InvalidKeyError, test_db.list_package_versions, "invalid", "test") + self.assertRaises(InvalidKeyError, test_db.list_package_versions, "app-test1", "invalid") + self.assertEqual(set(test_db.list_package_versions("app-test1", "test")), set(['1', '2'])) + self.assertEqual(set(test_db.list_all_packages()), set(packages)) + self.assertEqual(test_db.get_package_description(packages[0]), full_data) + self.assertRaises(KeyError, test_db.get_package_description, Package("invalid", "invalid", "1")) + self.assertEqual(test_db.get_max_version("app-test1", "test"), "2") + self.assertEqual(test_db.get_max_version("app-test1", "test1"), "1") + self.assertRaises(InvalidKeyError, test_db.get_max_version, "invalid", "invalid") + pkg_set = set(packages) + for package, data in test_db: + self.assertTrue(package in pkg_set) + if package.category == "app-test1": + self.assertEqual(data, full_data) + else: + self.assertEqual(data, ebuild_data) + pkg_set.remove(package) + self.assertTrue(not pkg_set) + self.assertEqual(orig_db.database, test_db.database) + port = port + 1 def suite(): suite = unittest.TestSuite() |