e9b211eea0
It looks like Portage now stores unicode paths correctly in its metadata as opposed to what it used to be. We need to make sure that we parse those "CONTENTS" file and content metadata in general using the correct encoding. This will allow us to store and retrieve such metadata from the sqlite3 database correctly and also match the stored paths with the filesystem paths exactly. This commit may need a bit more real-life testing. Backward compat wrt old Entropy and Portage tbz2 files should be as expected. Unit tests attached.
1197 lines
45 KiB
Python
1197 lines
45 KiB
Python
# -*- coding: utf-8 -*-
|
|
import sys
|
|
sys.path.insert(0, '.')
|
|
sys.path.insert(0, '../')
|
|
import unittest
|
|
import os
|
|
import time
|
|
import threading
|
|
|
|
from entropy.client.interfaces import Client
|
|
from entropy.const import etpConst, const_convert_to_unicode, \
|
|
const_convert_to_rawstring, const_mkstemp
|
|
from entropy.output import set_mute
|
|
from entropy.core.settings.base import SystemSettings
|
|
from entropy.misc import ParallelTask
|
|
from entropy.db import EntropyRepository
|
|
import tests._misc as _misc
|
|
|
|
import entropy.dep
|
|
import entropy.tools
|
|
|
|
|
|
class EntropyRepositoryTest(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.Client = Client(installed_repo = -1, indexing = False,
|
|
xcache = False, repo_validation = False)
|
|
self.Spm = self.Client.Spm()
|
|
self.test_db_name = "test_suite"
|
|
self.test_db = self.__open_test_db(":memory:")
|
|
# GenericRepository supports package masking if this property is set
|
|
self.test_db.enable_mask_filter = True
|
|
self.test_db2 = self.__open_test_db(":memory:")
|
|
# GenericRepository supports package masking if this property is set
|
|
self.test_db2.enable_mask_filter = True
|
|
self._settings = SystemSettings()
|
|
# since package files have been produced on amd64, add the same
|
|
# arch to etpConst['keywords'] to avoid atomMatch failures on x86
|
|
# and arm/other arches.
|
|
self._original_keywords = etpConst['keywords'].copy()
|
|
etpConst['keywords'].add("~amd64")
|
|
etpConst['keywords'].add("amd64")
|
|
|
|
def tearDown(self):
|
|
"""
|
|
tearDown is run after each test
|
|
"""
|
|
self.test_db.close()
|
|
self.test_db2.close()
|
|
# calling destroy() and shutdown()
|
|
# need to call destroy() directly to remove all the SystemSettings
|
|
# plugins because shutdown() doesn't, since it's meant to be called
|
|
# right before terminating the process
|
|
self.Client.destroy()
|
|
self.Client.shutdown()
|
|
etpConst['keywords'] = self._original_keywords.copy()
|
|
|
|
def __open_test_db(self, tmp_path):
|
|
return self.Client.open_temp_repository(name = self.test_db_name,
|
|
temp_file = tmp_path)
|
|
|
|
def test_db_clearcache(self):
|
|
self.test_db.clearCache()
|
|
|
|
def test_treeupdates_config_files_update(self):
|
|
files = _misc.get_config_files_updates_test_files()
|
|
actions = [
|
|
"move app-admin/anaconda app-admin/fuckaconda",
|
|
"slotmove app-admin/anaconda 0 2", # unsupported
|
|
"move media-sound/xmms2 media-sound/deadashell",
|
|
"move media-sound/pulseaudio media-sound/idiotaudio",
|
|
"move sys-auth/pambase sys-auth/fuckbase",
|
|
"move sys-devel/gcc sys-devel/fuckcc"
|
|
]
|
|
config_map = {
|
|
'._cfg0000_packages.db.critical': 'faa50df927223bb6de967e33179803b7',
|
|
'._cfg0000_packages.db.system_mask': 'b7f536785e315f7c104c7185b0bfe608',
|
|
'._cfg0000_packages.server.dep_blacklist.test': '8180f9e89d57f788e5b4bab05e30d447',
|
|
'._cfg0000_packages.server.dep_rewrite.test': 'c31d66b7f03c725e586a6e22941b8082',
|
|
}
|
|
for file_path in files:
|
|
updated_files = self.test_db._runConfigurationFilesUpdate(actions,
|
|
[file_path])
|
|
self.assertTrue(len(updated_files) == 1)
|
|
updated_file = list(updated_files)[0]
|
|
md5sum = entropy.tools.md5sum(updated_file)
|
|
os.remove(updated_file)
|
|
updated_name = os.path.basename(updated_file)
|
|
self.assertEqual(config_map[updated_name], md5sum)
|
|
|
|
def test_treeupdates_actions(self):
|
|
self.assertEqual(self.test_db.listAllTreeUpdatesActions(), tuple())
|
|
|
|
updates = (
|
|
('move media-libs/x264-svn media-libs/x264', '2020', '1210199116.46'),
|
|
('slotmove x11-libs/lesstif 2.1 0', '2020', '1210753863.16')
|
|
)
|
|
updates_out = (
|
|
(1, 'test_suite', 'move media-libs/x264-svn media-libs/x264', '2020', '1210199116.46'),
|
|
(2, 'test_suite', 'slotmove x11-libs/lesstif 2.1 0', '2020', '1210753863.16')
|
|
)
|
|
actions = tuple(sorted(['move media-libs/x264-svn media-libs/x264',
|
|
'slotmove x11-libs/lesstif 2.1 0']))
|
|
|
|
self.test_db.insertTreeUpdatesActions(updates, self.test_db_name)
|
|
db_actions = self.test_db.retrieveTreeUpdatesActions(self.test_db_name)
|
|
self.assertEqual(actions, db_actions)
|
|
self.assertEqual(updates_out, self.test_db.listAllTreeUpdatesActions())
|
|
|
|
self.test_db.removeTreeUpdatesActions(self.test_db_name)
|
|
db_actions = self.test_db.retrieveTreeUpdatesActions(self.test_db_name)
|
|
self.assertEqual(tuple(), db_actions)
|
|
|
|
def test_contentsafety(self):
|
|
test_pkg = _misc.get_test_package()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
_misc.clean_pkg_metadata(db_data)
|
|
_misc.clean_pkg_metadata(data)
|
|
|
|
self.assertEqual(data, db_data)
|
|
path = "/usr/include/zconf.h"
|
|
content_safety = self.test_db.searchContentSafety(path)
|
|
self.assertEqual(content_safety, (
|
|
{'package_id': 1,
|
|
'sha256': data['content_safety'][path]['sha256'],
|
|
'path': '/usr/include/zconf.h',
|
|
'mtime': data['content_safety'][path]['mtime']},)
|
|
)
|
|
|
|
def test_needed(self):
|
|
test_pkg1 = _misc.get_test_package()
|
|
test_pkg2 = _misc.get_test_package4()
|
|
for test_pkg in (test_pkg1, test_pkg2):
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
|
|
_misc.clean_pkg_metadata(db_data)
|
|
_misc.clean_pkg_metadata(data)
|
|
self.assertEqual(data, db_data)
|
|
|
|
db_needed = self.test_db.retrieveNeededLibraries(
|
|
idpackage)
|
|
self.assertEqual(db_needed, data['needed_libs'])
|
|
|
|
db_needed = self.test_db.retrieveNeededLibraries(idpackage)
|
|
self.assertEqual(db_needed, data['needed_libs'])
|
|
|
|
def test_dependencies(self):
|
|
test_pkg = _misc.get_test_package3()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
_misc.clean_pkg_metadata(db_data)
|
|
_misc.clean_pkg_metadata(data)
|
|
self.assertEqual(data, db_data)
|
|
|
|
pkg_deps = sorted(
|
|
self.test_db.retrieveDependencies(
|
|
idpackage, extended = True))
|
|
orig_pkg_deps = sorted([
|
|
('=dev-libs/apr-1*', 0),
|
|
('dev-libs/openssl', 0),
|
|
('dev-libs/libpcre', 0),
|
|
('=dev-libs/apr-util-1*', 0),
|
|
('=dev-libs/apr-1*', 3),
|
|
('dev-libs/openssl', 3),
|
|
('dev-libs/libpcre', 3),
|
|
('=dev-libs/apr-util-1*', 3),
|
|
])
|
|
|
|
self.assertEqual(pkg_deps, orig_pkg_deps)
|
|
|
|
def test_use_dependencies(self):
|
|
test_pkg = _misc.get_test_entropy_package6()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
|
|
_misc.clean_pkg_metadata(db_data)
|
|
_misc.clean_pkg_metadata(data)
|
|
self.assertEqual(data, db_data)
|
|
|
|
useflags = self.test_db.retrieveUseflags(idpackage)
|
|
self.assertTrue("gtk" not in useflags)
|
|
self.assertTrue("-gtk" in useflags)
|
|
self.assertTrue("-kde" in useflags)
|
|
self.assertTrue("-debug" in useflags)
|
|
self.assertTrue("-examples" in useflags)
|
|
|
|
def test_content(self):
|
|
test_pkg = _misc.get_test_package3()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
|
|
_misc.clean_pkg_metadata(db_data)
|
|
_misc.clean_pkg_metadata(data)
|
|
|
|
self.assertEqual(data, db_data)
|
|
content = self.test_db.retrieveContent(
|
|
idpackage, extended = True, order_by="file")
|
|
orig_content = (('/usr/sbin/ab2-ssl', 'sym'),
|
|
('/usr/sbin/logresolve2', 'sym'),
|
|
('/usr/sbin/log_server_status', 'obj'),
|
|
('/usr/sbin/checkgid2', 'sym'),
|
|
('/usr/sbin/htdbm', 'obj'),
|
|
('/usr/sbin/rotatelogs2', 'sym'),
|
|
('/usr/share/man/man1/htpasswd.1.bz2', 'obj'),
|
|
('/usr/sbin/ab-ssl', 'sym'),
|
|
('/usr/sbin/htcacheclean2', 'sym'),
|
|
('/usr/sbin/split-logfile2', 'sym'),
|
|
('/usr/share/man/man8', 'dir'),
|
|
('/usr/sbin/htcacheclean', 'obj'),
|
|
('/usr/sbin', 'dir'), ('/usr/sbin/ab', 'obj'),
|
|
('/usr/share/doc/apache-tools-2.2.11/CHANGES.bz2', 'obj'),
|
|
('/usr/sbin/htpasswd', 'obj'), ('/usr', 'dir'),
|
|
('/usr/bin/htpasswd', 'sym'),
|
|
('/usr/share/man/man1/htdigest.1.bz2', 'obj'),
|
|
('/usr/sbin/dbmmanage', 'obj'), ('/usr/share', 'dir'),
|
|
('/usr/share/man/man1', 'dir'), ('/usr/sbin/htdbm2', 'sym'),
|
|
('/usr/sbin/log_server_status2', 'sym'),
|
|
('/usr/share/man/man1/dbmmanage.1.bz2', 'obj'),
|
|
('/usr/share/man', 'dir'), ('/usr/sbin/htpasswd2', 'sym'),
|
|
('/usr/sbin/htdigest2', 'sym'), ('/usr/sbin/httxt2dbm2', 'sym'),
|
|
('/usr/bin', 'dir'), ('/usr/sbin/logresolve', 'obj'),
|
|
('/usr/share/doc', 'dir'), ('/usr/share/man/man8/ab.8.bz2', 'obj'),
|
|
('/usr/share/man/man8/logresolve.8.bz2', 'obj'),
|
|
('/usr/share/man/man8/htcacheclean.8.bz2', 'obj'),
|
|
('/usr/sbin/rotatelogs', 'obj'), ('/usr/sbin/checkgid', 'obj'),
|
|
('/usr/share/man/man1/htdbm.1.bz2', 'obj'),
|
|
('/usr/sbin/dbmmanage2', 'sym'), ('/usr/sbin/httxt2dbm', 'obj'),
|
|
('/usr/sbin/split-logfile', 'obj'),
|
|
('/usr/sbin/htdigest', 'obj'),
|
|
('/usr/share/doc/apache-tools-2.2.11', 'dir'),
|
|
('/usr/sbin/ab2', 'sym'),
|
|
('/usr/share/man/man8/rotatelogs.8.bz2', 'obj')
|
|
)
|
|
self.assertEqual(
|
|
content,
|
|
tuple(sorted(orig_content, key = lambda x: x[0])))
|
|
|
|
def test_db_creation(self):
|
|
self.assertTrue(isinstance(self.test_db, EntropyRepository))
|
|
self.assertEqual(self.test_db_name, self.test_db.repository_id())
|
|
self.assertTrue(self.test_db._doesTableExist('baseinfo'))
|
|
self.assertTrue(self.test_db._doesTableExist('extrainfo'))
|
|
|
|
def test_db_metadata_handling(self):
|
|
|
|
test_entry = {
|
|
const_convert_to_unicode("/path/to/foo", "utf-8"): \
|
|
const_convert_to_unicode("dir", "utf-8"),
|
|
const_convert_to_unicode("/path/to/foo/foo", "utf-8"): \
|
|
const_convert_to_unicode("obj", "utf-8"),
|
|
}
|
|
|
|
test_pkg = _misc.get_test_package()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
data['content'].update(test_entry.copy())
|
|
idpackage = self.test_db.addPackage(data)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
|
|
test_pkg2 = _misc.get_test_package2()
|
|
data2 = self.Spm.extract_package_metadata(test_pkg2)
|
|
data2['content'].update(test_entry.copy())
|
|
idpackage2 = self.test_db2.addPackage(data2)
|
|
db_data2 = self.test_db2.getPackageData(idpackage2)
|
|
|
|
cont_diff = self.test_db.contentDiff(idpackage, self.test_db2,
|
|
idpackage2)
|
|
|
|
for key in test_entry:
|
|
try:
|
|
self.assertTrue(key not in cont_diff)
|
|
except AssertionError:
|
|
print(key)
|
|
raise
|
|
|
|
py_diff = sorted([x for x in db_data['content'] if x not in \
|
|
db_data2['content']])
|
|
|
|
self.assertEqual(sorted(cont_diff), py_diff)
|
|
|
|
orig_diff = ['/lib64', '/lib64/libz.so', '/lib64/libz.so.1',
|
|
'/lib64/libz.so.1.2.3', '/usr/include', '/usr/include/zconf.h',
|
|
'/usr/include/zlib.h', '/usr/lib64/libz.a',
|
|
'/usr/lib64/libz.so', '/usr/share/doc/zlib-1.2.3-r1',
|
|
'/usr/share/doc/zlib-1.2.3-r1/ChangeLog.bz2',
|
|
'/usr/share/doc/zlib-1.2.3-r1/FAQ.bz2',
|
|
'/usr/share/doc/zlib-1.2.3-r1/README.bz2',
|
|
'/usr/share/doc/zlib-1.2.3-r1/algorithm.txt.bz2',
|
|
'/usr/share/man', '/usr/share/man/man3',
|
|
'/usr/share/man/man3/zlib.3.bz2'
|
|
]
|
|
orig_diff = [const_convert_to_unicode(x, 'utf-8') for x in orig_diff]
|
|
self.assertEqual(orig_diff, py_diff)
|
|
|
|
versioning_data = self.test_db.getVersioningData(idpackage)
|
|
dbverdata = (self.test_db.retrieveVersion(idpackage),
|
|
self.test_db.retrieveTag(idpackage),
|
|
self.test_db.retrieveRevision(idpackage),)
|
|
self.assertEqual(versioning_data, dbverdata)
|
|
|
|
strict_scope = self.test_db.getStrictScopeData(idpackage)
|
|
dbverdata = (self.test_db.retrieveAtom(idpackage),
|
|
self.test_db.retrieveSlot(idpackage),
|
|
self.test_db.retrieveRevision(idpackage),)
|
|
self.assertEqual(strict_scope, dbverdata)
|
|
|
|
scope_data = self.test_db.getScopeData(idpackage)
|
|
dbverdata = (
|
|
self.test_db.retrieveAtom(idpackage),
|
|
self.test_db.retrieveCategory(idpackage),
|
|
self.test_db.retrieveName(idpackage),
|
|
self.test_db.retrieveVersion(idpackage),
|
|
self.test_db.retrieveSlot(idpackage),
|
|
self.test_db.retrieveTag(idpackage),
|
|
self.test_db.retrieveRevision(idpackage),
|
|
self.test_db.retrieveBranch(idpackage),
|
|
self.test_db.retrieveApi(idpackage),
|
|
)
|
|
self.assertEqual(scope_data, dbverdata)
|
|
|
|
trigger_info = self.test_db.getTriggerData(idpackage)
|
|
trigger_keys = ['version', 'etpapi', 'slot', 'cxxflags', 'cflags',
|
|
'chost', 'atom', 'category', 'name', 'versiontag', 'content',
|
|
'trigger', 'branch', 'spm_phases', 'revision']
|
|
self.assertEqual(sorted(trigger_keys), sorted(trigger_info.keys()))
|
|
|
|
def test_db_insert_compare_match_provide(self):
|
|
test_pkg = _misc.get_test_entropy_package_provide()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
|
|
_misc.clean_pkg_metadata(db_data)
|
|
_misc.clean_pkg_metadata(data)
|
|
self.assertEqual(data, db_data)
|
|
|
|
def test_db_cache(self):
|
|
test_pkg = _misc.get_test_entropy_package_provide()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
|
|
# enable cache
|
|
self.test_db._caching = True
|
|
key = data['category'] + "/" + data['name']
|
|
|
|
from entropy.cache import EntropyCacher
|
|
cacher = EntropyCacher()
|
|
started = cacher.is_started()
|
|
cacher.start()
|
|
# avoid race conditions, unittest bug
|
|
time.sleep(2)
|
|
|
|
cached = self.test_db._EntropyRepositoryBase__atomMatchFetchCache(
|
|
key, True, False, False, None, None, False, False, True)
|
|
self.assertTrue(cached is None)
|
|
|
|
# now store
|
|
self.test_db._EntropyRepositoryBase__atomMatchStoreCache(
|
|
key, True, False, False, None, None, False, False, True,
|
|
result = (123, 0)
|
|
)
|
|
cacher.sync()
|
|
|
|
cached = self.test_db._EntropyRepositoryBase__atomMatchFetchCache(
|
|
key, True, False, False, None, None, False, False, True)
|
|
self.assertEqual(cached, (123, 0))
|
|
if not started:
|
|
cacher.stop()
|
|
|
|
def test_db_insert_compare_match(self):
|
|
|
|
# insert/compare
|
|
test_pkg = _misc.get_test_package()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
|
|
_misc.clean_pkg_metadata(db_data)
|
|
_misc.clean_pkg_metadata(data)
|
|
self.assertEqual(data, db_data)
|
|
|
|
# match
|
|
nf_match = (-1, 1)
|
|
f_match = (1, 0)
|
|
pkg_atom = _misc.get_test_package_atom()
|
|
pkg_name = _misc.get_test_package_name()
|
|
self.assertEqual(nf_match, self.test_db.atomMatch("slib"))
|
|
self.assertEqual(f_match,
|
|
self.test_db.atomMatch(pkg_name))
|
|
self.assertEqual(f_match,
|
|
self.test_db.atomMatch(pkg_atom))
|
|
|
|
# test package masking
|
|
masking_validation = \
|
|
self.Client.ClientSettings()['masking_validation']['cache']
|
|
f_match_mask = (1, self.test_db_name,)
|
|
|
|
self._settings['live_packagemasking']['mask_matches'].add(
|
|
f_match_mask)
|
|
masking_validation.clear()
|
|
self.assertEqual((-1, 1), self.test_db.atomMatch(pkg_atom))
|
|
|
|
self._settings['live_packagemasking']['mask_matches'].discard(
|
|
f_match_mask)
|
|
masking_validation.clear()
|
|
self.assertNotEqual((-1, 1), self.test_db.atomMatch(pkg_atom))
|
|
|
|
# now test multimatch
|
|
idpackage = self.test_db.addPackage(db_data)
|
|
results, rc = self.test_db.atomMatch(pkg_name, multiMatch = True)
|
|
self.assertEqual(2, len(results))
|
|
self.assertTrue(isinstance(results, set))
|
|
self.assertTrue(rc == 0)
|
|
|
|
results, rc = self.test_db.atomMatch(pkg_name+"foo", multiMatch = True)
|
|
self.assertEqual(0, len(results))
|
|
self.assertTrue(isinstance(results, set))
|
|
self.assertTrue(rc == 1)
|
|
|
|
def test_db_handle_unicode_puppet_agent(self):
|
|
test_pkg = _misc.get_test_package_puppet_agent()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
|
|
idpackage = self.test_db.addPackage(data)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
|
|
_misc.clean_pkg_metadata(db_data)
|
|
_misc.clean_pkg_metadata(data)
|
|
self.assertEqual(data, db_data)
|
|
|
|
self.test_db.removePackage(idpackage)
|
|
|
|
def test_db_handle_unicode_ca_certs(self):
|
|
test_pkg = _misc.get_test_package_ca_certs()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
|
|
idpackage = self.test_db.addPackage(data)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
|
|
_misc.clean_pkg_metadata(db_data)
|
|
_misc.clean_pkg_metadata(data)
|
|
self.assertEqual(data, db_data)
|
|
|
|
self.test_db.removePackage(idpackage)
|
|
|
|
def test_db_insert_compare_match_utf(self):
|
|
|
|
# insert/compare
|
|
test_pkg = _misc.get_test_package2()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
# Portage stores them this way
|
|
unicode_msg = const_convert_to_unicode(
|
|
"#248083).\n\n 06 Feb 2009; Ra\xc3\xbal Porcel")
|
|
data['changelog'] = unicode_msg
|
|
data['license'] = const_convert_to_unicode('GPL-2')
|
|
data['licensedata'] = {
|
|
const_convert_to_unicode('GPL-2'): unicode_msg,
|
|
}
|
|
data['content_safety'] = {
|
|
unicode_msg: {
|
|
'sha256': "abcdbbcdbcdbcdbcdbcdbcdbcbd",
|
|
'mtime': 1024.0,
|
|
}
|
|
}
|
|
idpackage = self.test_db.addPackage(data)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
|
|
_misc.clean_pkg_metadata(db_data)
|
|
_misc.clean_pkg_metadata(data)
|
|
self.assertEqual(data, db_data)
|
|
|
|
# match
|
|
nf_match = (-1, 1)
|
|
f_match = (1, 0)
|
|
pkg_atom = _misc.get_test_package_atom2()
|
|
pkg_name = _misc.get_test_package_name2()
|
|
self.assertEqual(nf_match, self.test_db.atomMatch("slib"))
|
|
self.assertEqual(f_match,
|
|
self.test_db.atomMatch(pkg_name))
|
|
self.assertEqual(f_match,
|
|
self.test_db.atomMatch(pkg_atom))
|
|
|
|
# test package masking
|
|
masking_validation = self.Client.ClientSettings(
|
|
)['masking_validation']['cache']
|
|
f_match_mask = (1, self.test_db_name,)
|
|
|
|
self._settings['live_packagemasking']['mask_matches'].add(
|
|
f_match_mask)
|
|
masking_validation.clear()
|
|
self.assertEqual((-1, 1), self.test_db.atomMatch(pkg_atom))
|
|
|
|
self._settings['live_packagemasking']['mask_matches'].discard(
|
|
f_match_mask)
|
|
masking_validation.clear()
|
|
self.assertNotEqual((-1, 1), self.test_db.atomMatch(pkg_atom))
|
|
|
|
def test_db_insert_compare_match_utf2(self):
|
|
|
|
# insert/compare
|
|
test_pkg = _misc.get_test_package3()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
|
|
_misc.clean_pkg_metadata(db_data)
|
|
_misc.clean_pkg_metadata(data)
|
|
self.assertEqual(data, db_data)
|
|
|
|
# match
|
|
nf_match = (-1, 1)
|
|
f_match = (1, 0)
|
|
pkg_atom = _misc.get_test_package_atom3()
|
|
pkg_name = _misc.get_test_package_name3()
|
|
self.assertEqual(nf_match, self.test_db.atomMatch("slib"))
|
|
self.assertEqual(f_match,
|
|
self.test_db.atomMatch(pkg_name))
|
|
self.assertEqual(f_match,
|
|
self.test_db.atomMatch(pkg_atom))
|
|
|
|
# test package masking
|
|
masking_validation = self.Client.ClientSettings(
|
|
)['masking_validation']['cache']
|
|
f_match_mask = (1, self.test_db_name,)
|
|
|
|
self._settings['live_packagemasking']['mask_matches'].add(
|
|
f_match_mask)
|
|
masking_validation.clear()
|
|
self.assertEqual((-1, 1), self.test_db.atomMatch(pkg_atom))
|
|
|
|
self._settings['live_packagemasking']['mask_matches'].discard(
|
|
f_match_mask)
|
|
masking_validation.clear()
|
|
self.assertNotEqual((-1, 1), self.test_db.atomMatch(pkg_atom))
|
|
|
|
def test_db_insert_compare_match_mime(self):
|
|
|
|
# insert/compare
|
|
test_pkg = _misc.get_test_package4()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
|
|
_misc.clean_pkg_metadata(db_data)
|
|
_misc.clean_pkg_metadata(data)
|
|
self.assertEqual(data, db_data)
|
|
|
|
known_mime = set(['application/ogg', 'audio/x-oggflac', 'audio/x-mp3',
|
|
'audio/x-pn-realaudio', 'audio/mpeg', 'application/x-ogm-audio',
|
|
'audio/vorbis', 'video/x-ms-asf', 'audio/x-speex', 'audio/x-scpls',
|
|
'audio/x-vorbis', 'audio/mpegurl', 'audio/aac', 'audio/x-ms-wma',
|
|
'audio/ogg', 'audio/x-mpegurl', 'audio/mp4',
|
|
'audio/vnd.rn-realaudio', 'audio/x-vorbis+ogg', 'audio/x-musepack',
|
|
'audio/x-flac', 'audio/x-wav'])
|
|
self.assertEqual(db_data['provided_mime'], known_mime)
|
|
|
|
# match
|
|
nf_match = (-1, 1)
|
|
f_match = (1, 0)
|
|
pkg_atom = _misc.get_test_package_atom4()
|
|
pkg_name = _misc.get_test_package_name4()
|
|
self.assertEqual(nf_match, self.test_db.atomMatch("slib"))
|
|
self.assertEqual(f_match,
|
|
self.test_db.atomMatch(pkg_name))
|
|
self.assertEqual(f_match,
|
|
self.test_db.atomMatch(pkg_atom))
|
|
|
|
# test package masking
|
|
masking_validation = self.Client.ClientSettings(
|
|
)['masking_validation']['cache']
|
|
f_match_mask = (1, self.test_db_name,)
|
|
|
|
self._settings['live_packagemasking']['mask_matches'].add(
|
|
f_match_mask)
|
|
masking_validation.clear()
|
|
self.assertEqual((-1, 1), self.test_db.atomMatch(pkg_atom))
|
|
|
|
self._settings['live_packagemasking']['mask_matches'].discard(
|
|
f_match_mask)
|
|
masking_validation.clear()
|
|
self.assertNotEqual((-1, 1), self.test_db.atomMatch(pkg_atom))
|
|
|
|
def test_db_insert_compare_match_tag(self):
|
|
|
|
# insert/compare
|
|
test_pkg = _misc.get_test_entropy_package_tag()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
|
|
_misc.clean_pkg_metadata(db_data)
|
|
_misc.clean_pkg_metadata(data)
|
|
self.assertEqual(data, db_data)
|
|
|
|
# match
|
|
f_match = (1, 0)
|
|
|
|
for atom, pkg_id, branch in self.test_db.listAllPackages():
|
|
pkg_key = entropy.dep.dep_getkey(atom)
|
|
self.assertEqual(f_match, self.test_db.atomMatch(pkg_key))
|
|
self.assertEqual(f_match, self.test_db.atomMatch(atom))
|
|
self.assertEqual(f_match, self.test_db.atomMatch("~"+atom))
|
|
|
|
def test_db_multithread(self):
|
|
|
|
# insert/compare
|
|
test_pkg = _misc.get_test_entropy_package_tag()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
_misc.clean_pkg_metadata(data)
|
|
|
|
def handle_pkg(xdata):
|
|
idpackage = self.test_db.addPackage(xdata)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
|
|
_misc.clean_pkg_metadata(db_data)
|
|
self.assertEqual(xdata, db_data)
|
|
|
|
t1 = ParallelTask(handle_pkg, data)
|
|
t2 = ParallelTask(handle_pkg, data)
|
|
t3 = ParallelTask(handle_pkg, data)
|
|
t4 = ParallelTask(handle_pkg, data)
|
|
t1.start()
|
|
t2.start()
|
|
t3.start()
|
|
t4.start()
|
|
|
|
t1.join()
|
|
t2.join()
|
|
t3.join()
|
|
t4.join()
|
|
|
|
cur_cache = self.test_db._cursor_pool().keys()
|
|
self.assertTrue(len(cur_cache) > 0)
|
|
self.test_db._cleanup_all()
|
|
cur_cache = self.test_db._cursor_pool().keys()
|
|
self.assertEqual(len(cur_cache), 0)
|
|
|
|
def test_db_close_all(self):
|
|
"""
|
|
This tests if EntropyRepository.close() really closes
|
|
all the resources, including those allocated by the
|
|
main thread.
|
|
"""
|
|
|
|
test_pkg = _misc.get_test_entropy_package_tag()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
_misc.clean_pkg_metadata(data)
|
|
|
|
_tmp_data = {
|
|
"path": None,
|
|
"db": None,
|
|
"T1": False,
|
|
"T2": False,
|
|
"T3": False,
|
|
}
|
|
|
|
def handle_pkg(_tmp_data, xdata):
|
|
idpackage = self.test_db.addPackage(xdata)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
|
|
_misc.clean_pkg_metadata(db_data)
|
|
self.assertEqual(xdata, db_data)
|
|
self.test_db.commit()
|
|
fd, buf_file = const_mkstemp()
|
|
os.close(fd)
|
|
buf = open(buf_file, "wb")
|
|
|
|
set_mute(True)
|
|
self.test_db.exportRepository(buf)
|
|
set_mute(False)
|
|
|
|
buf.flush()
|
|
buf.close()
|
|
|
|
fd, buf_file_db = const_mkstemp()
|
|
os.close(fd)
|
|
self.test_db.importRepository(buf_file, buf_file_db)
|
|
os.remove(buf_file)
|
|
db = self.Client.open_generic_repository(buf_file_db)
|
|
self.assertTrue(db is not None)
|
|
pkg_ids = db.listAllPackageIds()
|
|
self.assertTrue(1 in pkg_ids)
|
|
_tmp_data['path'] = buf_file_db
|
|
_tmp_data['db'] = db
|
|
_tmp_data['T1'] = True
|
|
|
|
def select_pkg(_tmp_data, t1):
|
|
t1.join() # wait for t1 to finish
|
|
pkg_ids = _tmp_data['db'].listAllPackageIds()
|
|
self.assertTrue(1 in pkg_ids)
|
|
self.assertTrue(len(pkg_ids) == 1)
|
|
_tmp_data['T2'] = True
|
|
|
|
def close_all(_tmp_data, t1, t2):
|
|
t1.join()
|
|
t2.join()
|
|
|
|
_tmp_data['db']._cleanup_all(_cleanup_main_thread=False)
|
|
with _tmp_data['db']._cursor_pool_mutex():
|
|
cur_cache = _tmp_data['db']._cursor_pool().keys()
|
|
self.assertEqual(len(cur_cache), 1) # just MainThread
|
|
|
|
_tmp_data['db'].close()
|
|
with _tmp_data['db']._cursor_pool_mutex():
|
|
cur_cache = _tmp_data['db']._cursor_pool().keys()
|
|
self.assertEqual(len(cur_cache), 0) # nothing left
|
|
_tmp_data['T3'] = True
|
|
|
|
t1 = ParallelTask(handle_pkg, _tmp_data, data)
|
|
t1.name = "T1"
|
|
t2 = ParallelTask(select_pkg, _tmp_data, t1)
|
|
t2.name = "T2"
|
|
t3 = ParallelTask(close_all, _tmp_data, t1, t2)
|
|
t3.name = "T3"
|
|
t1.start()
|
|
t2.start()
|
|
|
|
t1.join()
|
|
pkg_ids = _tmp_data['db'].listAllPackageIds()
|
|
|
|
t2.join()
|
|
t3.start()
|
|
t3.join()
|
|
|
|
self.assertTrue(_tmp_data['T1'] and _tmp_data['T2'] \
|
|
and _tmp_data['T3'])
|
|
|
|
self.assertTrue(1 in pkg_ids)
|
|
self.assertTrue(len(pkg_ids) == 1)
|
|
_tmp_data['db'].close()
|
|
with _tmp_data['db']._cursor_pool_mutex():
|
|
cur_cache = _tmp_data['db']._cursor_pool().keys()
|
|
self.assertEqual(len(cur_cache), 0) # nothing left
|
|
os.remove(_tmp_data['path'])
|
|
|
|
def test_db_reverse_deps(self):
|
|
|
|
test_pkg = _misc.get_test_package()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
test_pkg2 = _misc.get_test_package2()
|
|
data2 = self.Spm.extract_package_metadata(test_pkg2)
|
|
data['pkg_dependencies'] += ((
|
|
_misc.get_test_package_atom2(),
|
|
etpConst['dependency_type_ids']['rdepend_id']),)
|
|
data2['pkg_dependencies'] += ((
|
|
_misc.get_test_package_atom(),
|
|
etpConst['dependency_type_ids']['rdepend_id']),)
|
|
|
|
idpackage = self.test_db.addPackage(data)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
|
|
_misc.clean_pkg_metadata(data)
|
|
_misc.clean_pkg_metadata(db_data)
|
|
self.assertEqual(data, db_data)
|
|
|
|
idpackage2 = self.test_db.addPackage(data2)
|
|
db_data2 = self.test_db.getPackageData(idpackage2)
|
|
|
|
_misc.clean_pkg_metadata(data2)
|
|
_misc.clean_pkg_metadata(db_data2)
|
|
self.assertEqual(data2, db_data2)
|
|
|
|
rev_deps = self.test_db.retrieveReverseDependencies(idpackage)
|
|
rev_deps2 = self.test_db.retrieveReverseDependencies(idpackage2)
|
|
|
|
self.assertTrue(idpackage in rev_deps2)
|
|
self.assertTrue(idpackage2 in rev_deps)
|
|
rev_deps_t = self.test_db.retrieveReverseDependencies(idpackage,
|
|
key_slot = True)
|
|
self.assertEqual(rev_deps_t, (('app-dicts/aspell-es', '0'),))
|
|
|
|
pkg_data = self.test_db.retrieveUnusedPackageIds()
|
|
self.assertEqual(pkg_data, tuple())
|
|
|
|
def test_similar(self):
|
|
test_pkg = _misc.get_test_package()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
|
|
_misc.clean_pkg_metadata(data)
|
|
_misc.clean_pkg_metadata(db_data)
|
|
self.assertEqual(data, db_data)
|
|
out = self.test_db.searchSimilarPackages(_misc.get_test_package_name())
|
|
self.assertEqual(out, (1,))
|
|
|
|
def test_search(self):
|
|
test_pkg = _misc.get_test_package()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
|
|
_misc.clean_pkg_metadata(data)
|
|
_misc.clean_pkg_metadata(db_data)
|
|
self.assertEqual(data, db_data)
|
|
|
|
out = self.test_db.searchPackages(_misc.get_test_package_name())
|
|
self.assertEqual(out, (('sys-libs/zlib-1.2.3-r1', 1, '5'),))
|
|
out = self.test_db.searchPackages(_misc.get_test_package_name(),
|
|
slot = "0")
|
|
self.assertEqual(out, (('sys-libs/zlib-1.2.3-r1', 1, '5'),))
|
|
out = self.test_db.searchPackages(_misc.get_test_package_name(),
|
|
slot = "0", just_id = True)
|
|
self.assertEqual(out, (1,))
|
|
|
|
def test_list_packages(self):
|
|
test_pkg = _misc.get_test_package()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
out = self.test_db.listAllPackages()
|
|
self.assertEqual(out, (('sys-libs/zlib-1.2.3-r1', 1, '5'),))
|
|
|
|
def test_spmuids(self):
|
|
test_pkg = _misc.get_test_package()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
out = self.test_db.listAllSpmUids()
|
|
self.assertEqual(out, ((22331, 1),))
|
|
|
|
def test_list_pkg_ids(self):
|
|
test_pkg = _misc.get_test_package()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
out = self.test_db.listAllPackageIds(order_by="atom")
|
|
self.assertEqual(out, (1,))
|
|
|
|
def test_list_files(self):
|
|
test_pkg = _misc.get_test_package()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
out = sorted(self.test_db.listAllFiles())
|
|
self.assertEqual(out, sorted([
|
|
'/lib64/libz.so', '/usr/share/doc/zlib-1.2.3-r1',
|
|
'/usr/share/doc/zlib-1.2.3-r1/algorithm.txt.bz2',
|
|
'/usr/share/doc/zlib-1.2.3-r1/FAQ.bz2',
|
|
'/usr/share/doc/zlib-1.2.3-r1/ChangeLog.bz2',
|
|
'/usr', '/usr/include', '/usr/lib64',
|
|
'/usr/share/man/man3/zlib.3.bz2', '/usr/lib64/libz.a', '/lib64',
|
|
'/usr/share', '/usr/share/doc/zlib-1.2.3-r1/README.bz2',
|
|
'/usr/lib64/libz.so', '/usr/share/man', '/usr/include/zconf.h',
|
|
'/lib64/libz.so.1.2.3', '/usr/include/zlib.h', '/usr/share/doc',
|
|
'/usr/share/man/man3', '/lib64/libz.so.1'])
|
|
)
|
|
|
|
def test_list_categories(self):
|
|
test_pkg = _misc.get_test_package()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
out = self.test_db.listAllCategories()
|
|
self.assertEqual(out, frozenset(('sys-libs',)))
|
|
|
|
def test_list_downloads(self):
|
|
test_pkg = _misc.get_test_package()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
out = self.test_db.listAllDownloads()
|
|
self.assertEqual(out, ('sys-libs:zlib-1.2.3-r1.334f75538e6eddde753d9a247609dd8b1123a541.tbz2',))
|
|
|
|
def test_search_name(self):
|
|
test_pkg = _misc.get_test_package()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
|
|
_misc.clean_pkg_metadata(data)
|
|
_misc.clean_pkg_metadata(db_data)
|
|
self.assertEqual(db_data, data)
|
|
|
|
out = self.test_db.searchName(_misc.get_test_package_name())
|
|
self.assertEqual(out, frozenset([('sys-libs/zlib-1.2.3-r1', 1)]))
|
|
|
|
def test_db_indexes(self):
|
|
self.test_db.createAllIndexes()
|
|
|
|
def test_db_import_export(self):
|
|
|
|
test_pkg = _misc.get_test_package2()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
# Portage stores them this way
|
|
data['changelog'] = const_convert_to_unicode(
|
|
"#248083).\n\n 06 Feb 2009; Ra\xc3\xbal Porcel")
|
|
data['license'] = const_convert_to_unicode('GPL-2')
|
|
data['licensedata'] = {
|
|
const_convert_to_unicode('GPL-2'): \
|
|
const_convert_to_unicode(
|
|
"#248083).\n\n 06 Feb 2009; Ra\xc3\xbal Porcel"),
|
|
}
|
|
idpackage = self.test_db.addPackage(data)
|
|
db_data = self.test_db.getPackageData(idpackage)
|
|
|
|
_misc.clean_pkg_metadata(data)
|
|
_misc.clean_pkg_metadata(db_data)
|
|
assert set(data.keys()) == set(db_data.keys())
|
|
for k, v in data.items():
|
|
if v != db_data[k]:
|
|
print(k, v, "vs", db_data[k])
|
|
self.assertEqual(data, db_data)
|
|
|
|
set_mute(True)
|
|
|
|
# export
|
|
fd, buf_file = const_mkstemp()
|
|
os.close(fd)
|
|
buf = open(buf_file, "wb")
|
|
self.test_db.exportRepository(buf)
|
|
buf.flush()
|
|
buf.close()
|
|
|
|
fd, new_db_path = const_mkstemp()
|
|
os.close(fd)
|
|
self.test_db.importRepository(buf_file, new_db_path)
|
|
new_db = self.Client.open_generic_repository(new_db_path)
|
|
new_db_data = new_db.getPackageData(idpackage)
|
|
|
|
_misc.clean_pkg_metadata(new_db_data)
|
|
new_db.close()
|
|
set_mute(True)
|
|
self.assertEqual(new_db_data, db_data)
|
|
os.remove(buf_file)
|
|
os.remove(new_db_path)
|
|
|
|
def test_use_defaults(self):
|
|
test_pkg = _misc.get_test_package()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
key, slot = self.test_db.retrieveKeySlot(idpackage)
|
|
valid_test_deps = [
|
|
"%s[%s(+)]" % (key, "doesntexistforsure"),
|
|
"%s[-%s(-)]" % (key, "doesntexistforsure"),
|
|
"%s[%s(+)]" % (key, "kernel_linux"),
|
|
"%s[%s(-)]" % (key, "kernel_linux"),
|
|
]
|
|
invalid_test_deps = [
|
|
"%s[%s(-)]" % (key, "doesntexistforsure"),
|
|
"%s[-%s(+)]" % (key, "kernel_linux"),
|
|
"%s[-%s(+)]" % (key, "doesntexistforsure"),
|
|
"%s[-%s(-)]" % (key, "kernel_linux"),
|
|
]
|
|
for dep in valid_test_deps:
|
|
self.assertEqual((1, 0), self.test_db.atomMatch(dep))
|
|
for dep in invalid_test_deps:
|
|
self.assertEqual((-1, 1), self.test_db.atomMatch(dep))
|
|
|
|
def test_db_package_sets(self):
|
|
|
|
set_name = 'my_test_set'
|
|
set_deps = ["app-foo/foo", "app-pling/plong", "media-foo/ajez"]
|
|
set_name2 = 'my_test_set2'
|
|
set_deps2 = ["app-foo/foo2", "app-pling/plong2", "media-foo/ajez2"]
|
|
pkgsets = {
|
|
set_name: set(set_deps),
|
|
set_name2: set(set_deps2),
|
|
}
|
|
self.test_db.insertPackageSets(pkgsets)
|
|
self.assertEqual(self.test_db.retrievePackageSets(), pkgsets)
|
|
set_search = self.test_db.searchSets(set_name2)
|
|
self.assertEqual(set([set_name2]), set_search)
|
|
|
|
def test_db_license_data_str_insert(self):
|
|
lic_txt = const_convert_to_rawstring('[3]\xab foo\n\n', 'utf-8')
|
|
lic_name = const_convert_to_unicode('CCPL-Attribution-2.0')
|
|
lic_data = {lic_name: lic_txt}
|
|
self.test_db._insertLicenses(lic_data)
|
|
db_lic_txt = self.test_db.retrieveLicenseText(lic_name)
|
|
self.assertEqual(db_lic_txt, lic_txt)
|
|
|
|
def test_settings(self):
|
|
self.assertRaises(KeyError, self.test_db.getSetting, "fuck")
|
|
self.test_db._setSetting("something_cool", "abcdef\nabcdef")
|
|
self.assertEqual(self.test_db.getSetting("something_cool"),
|
|
"abcdef\nabcdef")
|
|
|
|
def test_new_entropyrepository_schema(self):
|
|
test_pkg = _misc.get_test_package2()
|
|
data = self.Spm.extract_package_metadata(test_pkg)
|
|
idpackage = self.test_db.addPackage(data)
|
|
old_data = self.test_db.getPackageData(idpackage)
|
|
old_base_data = self.test_db.getBaseData(idpackage)
|
|
old_cats = self.test_db.listAllCategories()
|
|
|
|
test_db = self.__open_test_db(":memory:")
|
|
idpackage = test_db.addPackage(data)
|
|
new_data = test_db.getPackageData(idpackage)
|
|
new_base_data = test_db.getBaseData(idpackage)
|
|
new_cats = test_db.listAllCategories()
|
|
|
|
self.assertTrue(test_db._isBaseinfoExtrainfo2010())
|
|
self.assertEqual(old_data, new_data)
|
|
self.assertEqual(old_base_data, new_base_data)
|
|
self.assertEqual(old_cats, new_cats)
|
|
|
|
test_db.close()
|
|
|
|
def test_preserved_libs(self):
|
|
data = self.test_db.listAllPreservedLibraries()
|
|
self.assertEqual(data, tuple())
|
|
|
|
atom = "app-foo/bar-1.2.3"
|
|
fake_data = [
|
|
("libfoo.so.3", 2, "/usr/lib64/libfoo.so.3", atom),
|
|
("libfoo.so.3", 2, "/usr/lib64/libfoo.so.3.0.0", atom),
|
|
("libfoo.so.3", 2, "/usr/lib64/libfoo.so", atom),
|
|
("libbar.so.10", 2, "/usr/lib64/libbar.so.10", atom),
|
|
("libbar.so.10", 2, "/usr/lib64/libbar.so", atom),
|
|
("libbar.so.10", 2, "/usr/lib64/libbar.so.10.0.0", atom),
|
|
]
|
|
for entry in fake_data:
|
|
self.test_db.insertPreservedLibrary(*entry)
|
|
|
|
# test that insert order equals list
|
|
data = self.test_db.listAllPreservedLibraries()
|
|
self.assertEqual(tuple(data), tuple(fake_data))
|
|
|
|
grouped = {}
|
|
for lib, elfclass, path, _atom in data:
|
|
obj = grouped.setdefault((lib, elfclass), [])
|
|
obj.append(path)
|
|
|
|
# make sure that retrieve works as intended
|
|
for (lib, elfclass), paths in grouped.items():
|
|
out_paths = self.test_db.retrievePreservedLibraries(
|
|
lib, elfclass)
|
|
self.assertEqual(sorted(out_paths), sorted(paths))
|
|
|
|
# test removal
|
|
count = len(data)
|
|
for lib, elfclass, path, _atom in data:
|
|
self.test_db.removePreservedLibrary(lib, elfclass, path)
|
|
count -= 1
|
|
current_preserved_libs = self.test_db.listAllPreservedLibraries()
|
|
self.assertEqual(len(current_preserved_libs), count)
|
|
self.assertTrue((lib, elfclass, path) not in current_preserved_libs)
|
|
|
|
data = self.test_db.listAllPreservedLibraries()
|
|
self.assertEqual(data, tuple())
|
|
|
|
def _test_repository_locking(self, test_db):
|
|
|
|
with test_db.shared():
|
|
self.assertRaises(RuntimeError, test_db.try_acquire_exclusive)
|
|
|
|
with test_db.exclusive():
|
|
opaque = test_db.try_acquire_shared()
|
|
self.assertNotEqual(opaque, None)
|
|
test_db.release_shared(opaque)
|
|
|
|
opaque_shared = test_db.try_acquire_shared()
|
|
self.assert_(opaque_shared is not None)
|
|
|
|
self.assertRaises(RuntimeError, test_db.try_acquire_exclusive)
|
|
|
|
test_db.release_shared(opaque_shared)
|
|
|
|
# test reentrancy
|
|
opaque_exclusive = test_db.try_acquire_exclusive()
|
|
self.assert_(opaque_exclusive is not None)
|
|
|
|
opaque_exclusive2 = test_db.try_acquire_exclusive()
|
|
self.assert_(opaque_exclusive2 is not None)
|
|
|
|
self.assert_(opaque_exclusive._file_lock_setup() is
|
|
opaque_exclusive2._file_lock_setup())
|
|
|
|
# test reference counting
|
|
self.assertEquals(
|
|
2,
|
|
opaque_exclusive._file_lock_setup()['count'])
|
|
|
|
test_db.release_exclusive(opaque_exclusive)
|
|
|
|
self.assertEquals(
|
|
1,
|
|
opaque_exclusive._file_lock_setup()['count'])
|
|
|
|
test_db.release_exclusive(opaque_exclusive2)
|
|
|
|
self.assertEquals(
|
|
0,
|
|
opaque_exclusive._file_lock_setup()['count'])
|
|
|
|
# test that refcount doesn't go below zero
|
|
self.assertRaises(
|
|
RuntimeError, test_db.release_exclusive, opaque_exclusive)
|
|
|
|
self.assertEquals(
|
|
0,
|
|
opaque_exclusive._file_lock_setup()['count'])
|
|
|
|
opaque_exclusive = test_db.try_acquire_exclusive()
|
|
self.assert_(opaque_exclusive is not None)
|
|
|
|
self.assertRaises(RuntimeError, test_db.release_shared,
|
|
opaque_exclusive)
|
|
|
|
test_db.release_exclusive(opaque_exclusive)
|
|
|
|
def test_locking_file(self):
|
|
|
|
fd, db_file = const_mkstemp()
|
|
os.close(fd)
|
|
test_db = None
|
|
|
|
try:
|
|
test_db = self.Client.open_generic_repository(db_file)
|
|
test_db.initializeRepository()
|
|
|
|
return self._test_repository_locking(test_db)
|
|
|
|
finally:
|
|
if test_db is not None:
|
|
test_db.close()
|
|
os.remove(db_file)
|
|
|
|
def test_locking_memory(self):
|
|
self.assert_(self.test_db._is_memory())
|
|
return self._test_repository_locking(self.test_db)
|
|
|
|
def test_direct_access(self):
|
|
local = self.test_db._tls
|
|
|
|
self.assertEquals(self.test_db.directed(), False)
|
|
|
|
counter = getattr(local, "_EntropyRepositoryCacheCounter", "foo")
|
|
self.assertEquals(counter, "foo")
|
|
|
|
with self.test_db.direct():
|
|
self.assertEquals(self.test_db.directed(), True)
|
|
|
|
counter = local._EntropyRepositoryCacheCounter
|
|
self.assertEquals(counter, 0)
|
|
|
|
self.assertEquals(self.test_db.directed(), False)
|
|
|
|
with self.test_db.direct():
|
|
|
|
counter = local._EntropyRepositoryCacheCounter
|
|
self.assertEquals(counter, 1)
|
|
|
|
with self.test_db.direct():
|
|
counter = local._EntropyRepositoryCacheCounter
|
|
self.assertEquals(counter, 2)
|
|
|
|
counter = local._EntropyRepositoryCacheCounter
|
|
self.assertEquals(counter, 1)
|
|
|
|
counter = local._EntropyRepositoryCacheCounter
|
|
self.assertEquals(counter, 0)
|
|
|
|
self.test_db._direct_enabled = True
|
|
self.assertEquals(self.test_db.directed(), True)
|
|
|
|
self.test_db._direct_enabled = False
|
|
self.assertEquals(self.test_db.directed(), False)
|
|
|
|
with self.test_db.direct():
|
|
opaque = self.test_db.try_acquire_exclusive()
|
|
self.assertTrue(opaque is not None)
|
|
self.assertEquals(opaque.directed(), True)
|
|
|
|
opaque_shared = self.test_db.try_acquire_shared()
|
|
self.assertTrue(opaque_shared is not None)
|
|
self.assertEquals(opaque_shared.directed(), True)
|
|
|
|
opaque_exclusive = self.test_db.try_acquire_exclusive()
|
|
self.assertTrue(opaque_exclusive is not None)
|
|
self.assertEquals(opaque_exclusive.directed(), True)
|
|
self.assertEquals(self.test_db.directed(), True)
|
|
|
|
self.assertRaises(
|
|
RuntimeError,
|
|
self.test_db.release_shared, opaque_exclusive)
|
|
self.test_db.release_exclusive(opaque_exclusive)
|
|
self.assertEquals(self.test_db.directed(), True)
|
|
|
|
self.assertRaises(
|
|
RuntimeError,
|
|
self.test_db.release_exclusive, opaque_shared)
|
|
self.test_db.release_shared(opaque_shared)
|
|
self.assertEquals(self.test_db.directed(), True)
|
|
|
|
self.assertRaises(
|
|
RuntimeError,
|
|
self.test_db.release_shared, opaque)
|
|
self.test_db.release_exclusive(opaque)
|
|
self.assertEquals(self.test_db.directed(), True)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|
|
raise SystemExit(0)
|