# -*- coding: utf-8 -*- from tests.helper.Stubs import Core from tests.helper.BenchmarkTest import BenchmarkTest from module.Api import DownloadState, PackageInfo, FileInfo from module.database import DatabaseBackend # disable asyncronous queries DatabaseBackend.async = DatabaseBackend.queue from random import choice class TestDatabase(BenchmarkTest): bench = ["insert", "insert_links", "insert_many", "get_packages", "get_files", "get_files_queued", "get_package_childs", "get_package_files", "get_package_data", "get_file_data", "find_files", "collector", "purge"] pids = None fids = None owner = 123 pstatus = 0 @classmethod def setUpClass(cls): cls.pids = [-1] cls.fids = [] cls.db = DatabaseBackend(Core()) cls.db.manager = cls.db.core cls.db.setup() cls.db.purgeAll() @classmethod def tearDownClass(cls): cls.db.purgeAll() cls.db.shutdown() # benchmarker ignore setup def setUp(self): self.db.purgeAll() self.pids = [-1] self.fids = [] self.test_insert(20) self.test_insert_many() self.fids = self.db.getAllFiles().keys() def test_insert(self, n=200): for i in range(n): pid = self.db.addPackage("name", "folder", choice(self.pids), "password", "site", "comment", self.pstatus, self.owner) self.pids.append(pid) def test_insert_links(self): for i in range(10000): fid = self.db.addLink("url %s" % i, "name", "plugin", choice(self.pids), self.owner) self.fids.append(fid) def test_insert_many(self): for pid in self.pids: self.db.addLinks([("url %s" % i, "plugin") for i in range(50)], pid, self.owner) def test_get_packages(self): packs = self.db.getAllPackages() n = len(packs) assert n == len(self.pids) - 1 print "Fetched %d packages" % n self.assert_pack(choice(packs.values())) def test_get_files(self): files = self.db.getAllFiles() n = len(files) assert n >= len(self.pids) print "Fetched %d files" % n self.assert_file(choice(files.values())) def test_get_files_queued(self): files = self.db.getAllFiles(state=DownloadState.Unfinished) print "Fetched %d files queued" % len(files) def test_delete(self): pid = choice(self.pids) self.db.deletePackage(pid) self.pids.remove(pid) def test_get_package_childs(self): pid = choice(self.pids) packs = self.db.getAllPackages(root=pid) print "Package %d has %d packages" % (pid, len(packs)) self.assert_pack(choice(packs.values())) def test_get_package_files(self): pid = choice(self.pids) files = self.db.getAllFiles(package=pid) print "Package %d has %d files" % (pid, len(files)) self.assert_file(choice(files.values())) def test_get_package_data(self, stats=False): pid = choice(self.pids) p = self.db.getPackageInfo(pid, stats) self.assert_pack(p) # test again with stat if not stats: self.test_get_package_data(True) def test_get_file_data(self): fid = choice(self.fids) f = self.db.getFileInfo(fid) self.assert_file(f) def test_find_files(self): files = self.db.getAllFiles(search="1") print "Found %s files" % len(files) f = choice(files.values()) assert "1" in f.name names = self.db.getMatchingFilenames("1") for name in names: assert "1" in name def test_collector(self): self.db.saveCollector(0, "data") assert self.db.retrieveCollector(0) == "data" self.db.deleteCollector(0) def test_purge(self): self.db.purgeLinks() def test_user_context(self): self.db.purgeAll() p1 = self.db.addPackage("name", "folder", 0, "password", "site", "comment", self.pstatus, 0) self.db.addLink("url", "name", "plugin", p1, 0) p2 = self.db.addPackage("name", "folder", 0, "password", "site", "comment", self.pstatus, 1) self.db.addLink("url", "name", "plugin", p2, 1) assert len(self.db.getAllPackages(owner=0)) == 1 == len(self.db.getAllFiles(owner=0)) assert len(self.db.getAllPackages(root=0, owner=0)) == 1 == len(self.db.getAllFiles(package=p1, owner=0)) assert len(self.db.getAllPackages(owner=1)) == 1 == len(self.db.getAllFiles(owner=1)) assert len(self.db.getAllPackages(root=0, owner=1)) == 1 == len(self.db.getAllFiles(package=p2, owner=1)) assert len(self.db.getAllPackages()) == 2 == len(self.db.getAllFiles()) self.db.deletePackage(p1, 1) assert len(self.db.getAllPackages(owner=0)) == 1 == len(self.db.getAllFiles(owner=0)) self.db.deletePackage(p1, 0) assert len(self.db.getAllPackages(owner=1)) == 1 == len(self.db.getAllFiles(owner=1)) self.db.deletePackage(p2) assert len(self.db.getAllPackages()) == 0 def test_count(self): self.db.purgeAll() assert self.db.downloadstats() == (0,0) assert self.db.queuestats() == (0,0) assert self.db.processcount() == 0 def test_update(self): p1 = self.db.addPackage("name", "folder", 0, "password", "site", "comment", self.pstatus, 0) pack = self.db.getPackageInfo(p1) assert isinstance(pack, PackageInfo) pack.folder = "new folder" pack.comment = "lol" pack.tags.append("video") self.db.updatePackage(pack) pack = self.db.getPackageInfo(p1) assert pack.folder == "new folder" assert pack.comment == "lol" assert "video" in pack.tags def assert_file(self, f): try: assert f is not None assert isinstance(f, FileInfo) self.assert_int(f, ("fid", "status", "size", "media", "fileorder", "added", "package", "owner")) assert f.status in range(5) assert f.owner == self.owner assert f.media in range(1024) assert f.package in self.pids assert f.added > 10 ** 6 # date is usually big integer except: print f raise def assert_pack(self, p): try: assert p is not None assert isinstance(p, PackageInfo) self.assert_int(p, ("pid", "root", "added", "status", "packageorder", "owner")) assert p.pid in self.pids assert p.owner == self.owner assert p.status in range(5) assert p.root in self.pids assert p.added > 10 ** 6 assert isinstance(p.tags, list) assert p.shared in (0, 1) except: print p raise def assert_int(self, obj, list): for attr in list: assert type(getattr(obj, attr)) == int if __name__ == "__main__": TestDatabase.benchmark()