本文整理汇总了Python中yum_importer.importer.YumImporter类的典型用法代码示例。如果您正苦于以下问题:Python YumImporter类的具体用法?Python YumImporter怎么用?Python YumImporter使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了YumImporter类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_resolve_deps
def test_resolve_deps(self):
repo = mock.Mock(spec=Repository)
repo.working_dir = "/tmp/test_resolve_deps"
repo.id = "test_resolve_deps"
unit_key_a = {'id' : '','name' :'pulp-server', 'version' :'0.0.309', 'release' : '1.fc17', 'epoch':'0', 'arch' : 'noarch', 'checksumtype' : 'sha256',
'checksum': 'ee5afa0aaf8bd2130b7f4a9b35f4178336c72e95358dd33bda8acaa5f28ea6e9', 'type_id' : 'rpm'}
unit_key_a_obj = Unit(RPM_TYPE_ID, unit_key_a, {}, '')
unit_key_a_obj.metadata = constants.PULP_SERVER_RPM_METADATA
unit_key_b = {'id' : '', 'name' :'pulp-rpm-server', 'version' :'0.0.309', 'release' :'1.fc17', 'epoch':'0','arch' : 'noarch', 'checksumtype' :'sha256',
'checksum': '1e6c3a3bae26423fe49d26930b986e5f5ee25523c13f875dfcd4bf80f770bf56', 'type_id' : 'rpm', }
unit_key_b_obj = Unit(RPM_TYPE_ID, unit_key_b, {}, '')
unit_key_b_obj.metadata = constants.PULP_RPM_SERVER_RPM_METADATA
existing_units = []
for unit in [unit_key_a_obj, unit_key_b_obj]:
existing_units.append(unit)
conduit = importer_mocks.get_dependency_conduit(type_id=RPM_TYPE_ID, existing_units=existing_units, pkg_dir=self.pkg_dir)
config = importer_mocks.get_basic_config()
importer = YumImporter()
units = [Unit(RPM_TYPE_ID, unit_key_b, {}, '')]
result = importer.resolve_dependencies(repo, units, conduit, config)
self.assertEqual(len(list(itertools.chain(*result['resolved'].values()))), 1)
self.assertEqual(len(list(itertools.chain(*result['unresolved'].values()))), 0)
开发者ID:jwmatthews,项目名称:pulp_rpm,代码行数:25,代码来源:test_resolve_dependencies.py
示例2: test_upload_rpm
def test_upload_rpm(self):
repo = mock.Mock(spec=Repository)
repo.working_dir = self.working_dir
repo.id = "test_upload_rpm"
upload_conduit = importer_mocks.get_upload_conduit(pkg_dir=self.pkg_dir)
config = importer_mocks.get_basic_config()
importer = YumImporter()
file_path = "%s/%s" % (self.data_dir, "incisura-7.1.4-1.elfake.noarch.rpm")
mdata = {'filename' : "incisura-7.1.4-1.elfake.noarch.rpm", 'checksum' : 'e0e98e76e4e06dad65a82b0111651d7aca5b00fe'}
unit_key = {'name' : 'incisura', 'version' : '7.1.4', 'release' : '1', 'arch' : 'noarch', 'checksum' : 'e0e98e76e4e06dad65a82b0111651d7aca5b00fe', 'checksumtype' : 'sha1'}
type_id = "rpm"
status, summary, details = importer._upload_unit(repo, type_id, unit_key, mdata, file_path, upload_conduit, config)
self.assertTrue(status)
self.assertTrue(summary is not None)
self.assertTrue(details is not None)
# validate if metadata was generated
self.assertTrue(mdata.has_key('repodata'))
self.assertTrue(mdata['repodata']['primary'] is not None)
self.assertTrue(mdata['repodata']['other'] is not None)
self.assertTrue(mdata['repodata']['filelists'] is not None)
primary_snippet = mdata['repodata']['primary']
location_start_index = primary_snippet.find("href=")
# verify the location matches the basename of the package
self.assertTrue(primary_snippet[location_start_index:location_start_index + len("href=") + len("incisura-7.1.4-1.elfake.noarch.rpm") + 2 ] ==
"href=\"incisura-7.1.4-1.elfake.noarch.rpm\"")
self.assertEquals(summary["num_units_saved"], 1)
self.assertEquals(summary["num_units_processed"], 1)
self.assertEquals(summary["filename"], "incisura-7.1.4-1.elfake.noarch.rpm")
self.assertEquals(details["errors"], [])
开发者ID:jessegonzalez,项目名称:pulp_rpm,代码行数:29,代码来源:test_upload_unit.py
示例3: test_repo_scratchpad_settings
def test_repo_scratchpad_settings(self):
global repo_scratchpad
repo_scratchpad = {}
def set_repo_scratchpad(data):
global repo_scratchpad
repo_scratchpad = data
def get_repo_scratchpad():
global repo_scratchpad
return repo_scratchpad
feed_url = "http://repos.fedorapeople.org/repos/pulp/pulp/demo_repos/test_drpm_repo/"
importer = YumImporter()
repo = mock.Mock(spec=Repository)
repo.working_dir = self.working_dir
repo.id = "test_repo_scratchpad"
sync_conduit = importer_mocks.get_sync_conduit(pkg_dir=self.pkg_dir)
sync_conduit.set_repo_scratchpad = mock.Mock()
sync_conduit.set_repo_scratchpad.side_effect = set_repo_scratchpad
sync_conduit.get_repo_scratchpad = mock.Mock()
sync_conduit.get_repo_scratchpad.side_effect = get_repo_scratchpad
config = importer_mocks.get_basic_config(feed_url=feed_url)
importer._sync_repo(repo, sync_conduit, config)
print "SCRATCHPAD %s" % repo_scratchpad
self.assertEquals(repo_scratchpad['checksum_type'], 'sha256')
self.assertTrue(repo_scratchpad.has_key("repodata"))
self.assertTrue(repo_scratchpad["repodata"].has_key("prestodelta"))
开发者ID:ehelms,项目名称:pulp_rpm,代码行数:28,代码来源:test_repo_scratchpad.py
示例4: test_skip_packagegroups
def test_skip_packagegroups(self):
global updated_progress
updated_progress = None
def set_progress(progress):
global updated_progress
updated_progress = progress
yi = YumImporter()
skip = ["packagegroup"]
repo_src_dir = os.path.join(self.data_dir, "pulp_unittest")
feed_url = "file://%s" % (repo_src_dir)
config = importer_mocks.get_basic_config(feed_url=feed_url,skip_content_types=skip)
repo = mock.Mock(spec=Repository)
repo.working_dir = self.working_dir
repo.id = "test_skip_packagegroup"
# Simulate a repo sync, copy the source contents to the repo.working_dir
self.simulate_sync(repo, repo_src_dir)
sync_conduit = importer_mocks.get_sync_conduit(pkg_dir=self.pkg_dir)
sync_conduit.set_progress = mock.Mock()
sync_conduit.set_progress = set_progress
status, summary, details = yi._sync_repo(repo, sync_conduit, config)
self.assertTrue(status)
self.assertEqual(updated_progress["comps"]["state"], "SKIPPED")
开发者ID:jlsherrill,项目名称:pulp,代码行数:25,代码来源:test_comps.py
示例5: test_progress_sync
def test_progress_sync(self):
global updated_progress
updated_progress = None
def set_progress(progress):
global updated_progress
updated_progress = progress
feed_url = "http://repos.fedorapeople.org/repos/pulp/pulp/demo_repos/pulp_unittest/"
importer = YumImporter()
repo = mock.Mock(spec=Repository)
repo.working_dir = self.working_dir
repo.id = "test_progress_sync"
sync_conduit = importer_mocks.get_sync_conduit(pkg_dir=self.pkg_dir)
sync_conduit.set_progress = mock.Mock()
sync_conduit.set_progress.side_effect = set_progress
config = importer_mocks.get_basic_config(feed_url=feed_url)
status, summary, details = importer._sync_repo(repo, sync_conduit, config)
self.assertTrue(status)
self.assertEquals(summary["packages"]["num_synced_new_rpms"], 3)
self.assertTrue(updated_progress is not None)
self.assertTrue("metadata" in updated_progress)
self.assertTrue(updated_progress["metadata"].has_key("state"))
self.assertTrue("errata" in updated_progress)
self.assertTrue(updated_progress["errata"].has_key("state"))
self.assertTrue("content" in updated_progress)
self.assertTrue(updated_progress["content"].has_key("state"))
self.assertEquals(updated_progress["content"]["state"], "FINISHED")
for key in importer_rpm.PROGRESS_REPORT_FIELDS:
self.assertTrue(key in updated_progress["content"])
开发者ID:jessegonzalez,项目名称:pulp_rpm,代码行数:30,代码来源:test_rpms.py
示例6: test_cancel_sync
def test_cancel_sync(self):
global updated_progress
updated_progress = None
def set_progress(progress):
global updated_progress
updated_progress = progress
class SyncThread(threading.Thread):
def __init__(self, importer, repo, sync_conduit, config):
threading.Thread.__init__(self)
self.importer = importer
self.repo = repo
self.sync_conduit = sync_conduit
self.config = config
self.status = None
self.summary = None
self.details = None
self.finished = False
def run(self):
status, summary, details = self.importer._sync_repo(self.repo, self.sync_conduit, self.config)
self.status = status
self.summary = summary
self.details = details
self.finished = True
feed_url = "http://repos.fedorapeople.org/repos/pulp/pulp/v1/testing/6Server/x86_64/"
repo = mock.Mock(spec=Repository)
repo.working_dir = self.working_dir
repo.id = "test_cancel_sync"
sync_conduit = importer_mocks.get_sync_conduit(existing_units=[], pkg_dir=self.pkg_dir)
sync_conduit.set_progress = mock.Mock()
sync_conduit.set_progress.side_effect = set_progress
config = importer_mocks.get_basic_config(feed_url=feed_url, num_threads=1, max_speed=25)
importer = YumImporter()
sync_thread = SyncThread(importer, repo, sync_conduit, config)
sync_thread.start()
# Wait to confirm that sync has started and we are downloading packages
# We are intentionally setting the 'config' to use 1 thread and max_speed to be low so we will
# have a chance to cancel the sync before it completes
for i in range(30):
if updated_progress and updated_progress.has_key("content") and updated_progress["content"].has_key("state") \
and updated_progress["content"]["state"] == "IN_PROGRESS":
break
time.sleep(1)
self.assertEquals(updated_progress["metadata"]["state"], "FINISHED")
self.assertEquals(updated_progress["content"]["state"], "IN_PROGRESS")
###
### Issue Cancel
###
importer.cancel_sync_repo(None, None)
# Wait for cancel of sync
for i in range(45):
if sync_thread.finished:
break
time.sleep(1)
self.assertEquals(updated_progress["content"]["state"], "CANCELED")
self.assertFalse(sync_thread.status)
开发者ID:jessegonzalez,项目名称:pulp_rpm,代码行数:59,代码来源:test_rpms.py
示例7: test_feedless_repo_sync
def test_feedless_repo_sync(self):
repo = mock.Mock(spec=Repository)
repo.working_dir = self.working_dir
repo.id = "test_feedless_repo_sync"
sync_conduit = importer_mocks.get_sync_conduit(type_id=TYPE_ID_RPM, pkg_dir=self.pkg_dir)
sync_conduit.set_progress = mock.Mock()
config = importer_mocks.get_basic_config()
importer = YumImporter()
status, summary, details = importer._sync_repo(repo, sync_conduit, config)
self.assertFalse(status)
self.assertEquals(summary['error'], "Cannot perform repository sync on a repository with no feed")
开发者ID:jessegonzalez,项目名称:pulp_rpm,代码行数:11,代码来源:test_rpms.py
示例8: test_distribution_unit_import
def test_distribution_unit_import(self):
existing_units = []
dunit_key = {}
dunit_key['id'] = "ks-TestFamily-TestVariant-16-x86_64"
dunit_key['version'] = "16"
dunit_key['arch'] = "x86_64"
dunit_key['family'] = "TestFamily"
dunit_key['variant'] = "TestVariant"
metadata = { "files" : [{"checksumtype" : "sha256", "relativepath" : "images/fileA.txt", "fileName" : "fileA.txt",
"downloadurl" : "http://repos.fedorapeople.org/repos/pulp/pulp/demo_repos/pulp_unittest//images/fileA.txt",
"item_type" : "tree_file",
"savepath" : "%s/testr1/images" % self.data_dir,
"checksum" : "22603a94360ee24b7034c74fa13d70dd122aa8c4be2010fc1361e1e6b0b410ab",
"filename" : "fileA.txt",
"pkgpath" : "%s/ks-TestFamily-TestVariant-16-x86_64/images" % self.pkg_dir,
"size" : 0 },
{ "checksumtype" : "sha256", "relativepath" : "images/fileB.txt", "fileName" : "fileB.txt",
"downloadurl" : "http://repos.fedorapeople.org/repos/pulp/pulp/demo_repos/pulp_unittest//images/fileB.txt",
"item_type" : "tree_file",
"savepath" : "%s/testr1/images" % self.data_dir,
"checksum" : "8dc89e9883c098443f6616e60a8e489254bf239eeade6e4b4943b7c8c0c345a4",
"filename" : "fileB.txt",
"pkgpath" : "%s/ks-TestFamily-TestVariant-16-x86_64/images" % self.pkg_dir, "size" : 0 },
{ "checksumtype" : "sha256", "relativepath" : "images/fileC.iso", "fileName" : "fileC.iso",
"downloadurl" : "http://repos.fedorapeople.org/repos/pulp/pulp/demo_repos/pulp_unittest//images/fileC.iso",
"item_type" : "tree_file",
"savepath" : "%s/testr1/images" % self.data_dir,
"checksum" : "099f2bafd533e97dcfee778bc24138c40f114323785ac1987a0db66e07086f74",
"filename" : "fileC.iso",
"pkgpath" : "%s/ks-TestFamily-TestVariant-16-x86_64/images" % self.pkg_dir, "size" : 0 } ],}
distro_unit = [Unit(TYPE_ID_DISTRO, dunit_key, metadata, '')]
distro_unit[0].storage_path = "%s/ks-TestFamily-TestVariant-16-x86_64" % self.pkg_dir
existing_units += distro_unit
# REPO A (source)
repoA = mock.Mock(spec=Repository)
repoA.working_dir = self.data_dir
repoA.id = "test_distro_unit_copy"
# REPO B (target)
repoB = mock.Mock(spec=Repository)
repoB.working_dir = self.working_dir
repoB.id = "repoB"
conduit = importer_mocks.get_import_conduit([distro_unit], existing_units=existing_units)
config = importer_mocks.get_basic_config()
importer = YumImporter()
# Test
result = importer.import_units(repoA, repoB, conduit, config, distro_unit)
# Verify
print conduit.associate_unit.call_args_list
associated_units = [mock_call[0][0] for mock_call in conduit.associate_unit.call_args_list]
self.assertEqual(len(associated_units), len(distro_unit))
for u in associated_units:
self.assertTrue(u in distro_unit)
开发者ID:ehelms,项目名称:pulp_rpm,代码行数:52,代码来源:test_import_units.py
示例9: test_upload_package_group
def test_upload_package_group(self):
repo = mock.Mock(spec=Repository)
repo.working_dir = self.working_dir
repo.id = "test_upload_package_group"
upload_conduit = importer_mocks.get_upload_conduit(pkg_dir=self.pkg_dir)
config = importer_mocks.get_basic_config()
importer = YumImporter()
file_path = None
type_id = TYPE_ID_PKG_CATEGORY
unit = self.get_pkg_group_or_category(repo, type_id)
status, summary, details = importer._upload_unit(repo, type_id, unit.unit_key, unit.metadata, file_path, upload_conduit, config)
self.assertTrue(status)
self.assertEqual(summary['state'], 'FINISHED')
开发者ID:jessegonzalez,项目名称:pulp_rpm,代码行数:13,代码来源:test_upload_unit.py
示例10: test_validate_config
def test_validate_config(self):
feed_url = "http://repos.fedorapeople.org/repos/pulp/pulp/demo_repos/pulp_unittest/"
importer = YumImporter()
config = importer_mocks.get_basic_config(feed_url=feed_url)
repo = mock.Mock(spec=Repository)
state, msg = importer.validate_config(repo, config, [])
self.assertTrue(state)
# Test that an unknown argument in the config throws an error
# and the unknown arg is identified in the message
config = importer_mocks.get_basic_config(feed_url=feed_url, bad_unknown_arg="blah")
state, msg = importer.validate_config(repo, config, [])
self.assertFalse(state)
self.assertTrue("bad_unknown_arg" in msg)
开发者ID:jessegonzalez,项目名称:pulp_rpm,代码行数:14,代码来源:test_rpms.py
示例11: test_upload_erratum
def test_upload_erratum(self):
repo = mock.Mock(spec=Repository)
repo.working_dir = self.working_dir
repo.id = "test_upload_errata"
upload_conduit = importer_mocks.get_upload_conduit(pkg_dir=self.pkg_dir)
config = importer_mocks.get_basic_config()
importer = YumImporter()
file_path = []
type_id = "erratum"
unit_key = dict()
unit_key['id'] = "RHBA-2012:0101"
metadata = {"pkglist" : [],}
status, summary, details = importer._upload_unit(repo, type_id, unit_key, metadata, file_path, upload_conduit, config)
self.assertTrue(status)
self.assertEqual(summary['state'], 'FINISHED')
开发者ID:jessegonzalez,项目名称:pulp_rpm,代码行数:15,代码来源:test_upload_unit.py
示例12: setUp
def setUp(self):
super(TestValidateConfig, self).setUp()
self.temp_dir = tempfile.mkdtemp()
self.repo = mock.Mock(spec=Repository)
self.repo.working_dir = os.path.join(self.temp_dir, "repo_working_dir")
os.makedirs(self.repo.working_dir)
self.importer = YumImporter()
self.init()
开发者ID:jessegonzalez,项目名称:pulp_rpm,代码行数:8,代码来源:test_importer_validate_config.py
示例13: setup_source_repo
def setup_source_repo(self):
# Sync a sample repository to populate and setup up Source Repo
source_repo = mock.Mock(spec=Repository)
source_repo.id = "repo_a"
source_repo.working_dir = os.path.join(self.working_dir, source_repo.id)
importer = YumImporter()
feed_url = "file://%s/pulp_unittest/" % (self.data_dir)
config = importer_mocks.get_basic_config(feed_url=feed_url)
sync_conduit = importer_mocks.get_sync_conduit(existing_units=[], pkg_dir=self.pkg_dir)
status, summary, details = importer._sync_repo(source_repo, sync_conduit, config)
self.assertTrue(status)
self.assertEquals(summary["packages"]["num_synced_new_rpms"], 3)
# Confirm regular RPM files exist under self.pkg_dir
pkgs = self.get_files_in_dir("*.rpm", self.pkg_dir)
self.assertEquals(len(pkgs), 3)
for p in pkgs:
self.assertTrue(os.path.isfile(p))
# Confirm symlinks to RPMs exist under repo.working_dir
sym_links = self.get_files_in_dir("*.rpm", source_repo.working_dir)
self.assertEquals(len(pkgs), 3)
for link in sym_links:
self.assertTrue(os.path.islink(link))
#
# Now we have some test data in the source repo
#
# Simulate what import_conduit.get_source_repos would return
#
metadata = {}
source_units = []
storage_path = '%s/pulp-dot-2.0-test/0.1.2/1.fc11/x86_64/435d92e6c09248b501b8d2ae786f92ccfad69fab8b1bc774e2b66ff6c0d83979/pulp-dot-2.0-test-0.1.2-1.fc11.x86_64.rpm' % (self.pkg_dir)
filename = os.path.basename(storage_path)
unit_key = {"filename":filename}
source_units.append(Unit(TYPE_ID_RPM, unit_key, metadata, storage_path))
storage_path = '%s/pulp-test-package/0.3.1/1.fc11/x86_64/6bce3f26e1fc0fc52ac996f39c0d0e14fc26fb8077081d5b4dbfb6431b08aa9f/pulp-test-package-0.3.1-1.fc11.x86_64.rpm' % (self.pkg_dir)
filename = os.path.basename(storage_path)
unit_key = {"filename":filename}
source_units.append(Unit(TYPE_ID_RPM, unit_key, metadata, storage_path))
storage_path = '%s/pulp-test-package/0.2.1/1.fc11/x86_64/4dbde07b4a8eab57e42ed0c9203083f1d61e0b13935d1a569193ed8efc9ecfd7/pulp-test-package-0.2.1-1.fc11.x86_64.rpm' % (self.pkg_dir)
filename = os.path.basename(storage_path)
unit_key = {"filename":filename}
source_units.append(Unit(TYPE_ID_RPM, unit_key, metadata, storage_path))
# Pass in the simulated source_units to the import_conduit
import_conduit = importer_mocks.get_import_conduit(source_units=source_units)
return importer, source_repo, source_units, import_conduit, config
开发者ID:stpierre,项目名称:pulp,代码行数:44,代码来源:test_import_units.py
示例14: test_upload_rpm
def test_upload_rpm(self):
repo = mock.Mock(spec=Repository)
repo.working_dir = self.working_dir
repo.id = "test_upload_rpm"
upload_conduit = importer_mocks.get_upload_conduit(pkg_dir=self.pkg_dir)
config = importer_mocks.get_basic_config()
importer = YumImporter()
file_path = "%s/%s" % (self.data_dir, "incisura-7.1.4-1.elfake.noarch.rpm")
metadata = {'filename' : "incisura-7.1.4-1.elfake.noarch.rpm", 'checksum' : 'e0e98e76e4e06dad65a82b0111651d7aca5b00fe'}
unit_key = {'name' : 'incisura', 'version' : '7.1.4', 'release' : '1', 'arch' : 'noarch', 'checksum' : 'e0e98e76e4e06dad65a82b0111651d7aca5b00fe', 'checksumtype' : 'sha1'}
type_id = "rpm"
status, summary, details = importer._upload_unit(repo, type_id, unit_key, metadata, file_path, upload_conduit, config)
self.assertTrue(status)
self.assertTrue(summary is not None)
self.assertTrue(details is not None)
self.assertEquals(summary["num_units_saved"], 1)
self.assertEquals(summary["num_units_processed"], 1)
self.assertEquals(summary["filename"], "incisura-7.1.4-1.elfake.noarch.rpm")
self.assertEquals(details["errors"], [])
开发者ID:ehelms,项目名称:pulp,代码行数:19,代码来源:test_upload_unit.py
示例15: test_local_sync_with_bad_url
def test_local_sync_with_bad_url(self):
feed_url = "file:///INTENTIONAL_BAD_URL/demo_repos/pulp_unittest/"
repo = mock.Mock(spec=Repository)
repo.working_dir = self.working_dir
repo.id = "test_local_sync_with_bad_url"
sync_conduit = importer_mocks.get_sync_conduit(existing_units=[], pkg_dir=self.pkg_dir)
config = importer_mocks.get_basic_config(feed_url=feed_url)
caught_exception = False
try:
importerRPM = importer_rpm.ImporterRPM()
status, summary, details = importerRPM.sync(repo, sync_conduit, config)
except:
caught_exception = True
self.assertTrue(caught_exception)
importer = YumImporter()
caught_exception = False
try:
report = importer.sync_repo(repo, sync_conduit, config)
except:
caught_exception = True
self.assertFalse(caught_exception)
开发者ID:jessegonzalez,项目名称:pulp_rpm,代码行数:22,代码来源:test_rpms.py
示例16: test_import
def test_import(self):
# Setup
existing_units = self.existing_units()
# REPO A (source)
repoA = mock.Mock(spec=Repository)
repoA.working_dir = self.data_dir
repoA.id = "test_resolve_deps"
# REPO B (target)
repoB = mock.Mock(spec=Repository)
repoB.working_dir = self.working_dir
repoB.id = "repoB"
units = [Unit(TYPE_ID_RPM, self.UNIT_KEY_B, {}, '')]
conduit = importer_mocks.get_import_conduit(units, existing_units=existing_units)
config = importer_mocks.get_basic_config()
importer = YumImporter()
# Test
result = importer.import_units(repoA, repoB, conduit, config, units)
# Verify
associated_units = [mock_call[0][0] for mock_call in conduit.associate_unit.call_args_list]
self.assertEqual(len(associated_units), len(units))
for u in associated_units:
self.assertTrue(u in units)
开发者ID:ehelms,项目名称:pulp,代码行数:22,代码来源:test_import_units.py
示例17: test_package_category_unit_import
def test_package_category_unit_import(self):
# REPO A (source)
repoA = mock.Mock(spec=Repository)
repoA.working_dir = self.data_dir
repoA.id = "test_pkg_cat_unit_copy"
# REPO B (target)
repoB = mock.Mock(spec=Repository)
repoB.working_dir = self.working_dir
repoB.id = "repoB"
# Create 2 pkg groups
grp_a = self.create_dummy_pkg_group_unit(repoA.id, "group_a")
# Create 2 pkg categories
cat_a = self.create_dummy_pkg_category_unit(repoA.id, "cat_a", ["group_a"])
# Add the grps/cats to the publish_conduit
existing_units=[grp_a, cat_a]
conduit = importer_mocks.get_import_conduit([cat_a], existing_units=existing_units)
config = importer_mocks.get_basic_config()
importer = YumImporter()
# Test
result = importer.import_units(repoA, repoB, conduit, config, [cat_a])
# Verify
associated_units = [mock_call[0][0] for mock_call in conduit.associate_unit.call_args_list]
for u in associated_units:
self.assertTrue(u in [cat_a, grp_a])
开发者ID:ehelms,项目名称:pulp_rpm,代码行数:24,代码来源:test_import_units.py
示例18: test_import_with_dependencies
def test_import_with_dependencies(self):
# Setup
existing_units = self.existing_units()
# REPO A (source)
repoA = mock.Mock(spec=Repository)
repoA.working_dir = "/tmp/test_resolve_deps"
repoA.id = "test_resolve_deps"
# REPO B (target)
repoB = mock.Mock(spec=Repository)
repoB.working_dir = "/tmp/test_resolve_deps"
repoB.id = "repo_b"
units = [Unit(TYPE_ID_RPM, self.UNIT_KEY_B, {}, '')]
conduit = importer_mocks.get_import_conduit(units, existing_units=existing_units)
config = importer_mocks.get_basic_config()
config.override_config['recursive'] = True
config.override_config['resolve_dependencies'] = True
importer = YumImporter()
# Test
result = importer.import_units(repoA, repoB, conduit, config, units)
# Verify
associated_units = [mock_call[0][0] for mock_call in conduit.associate_unit.call_args_list]
self.assertEqual(len(associated_units), len(existing_units))
for u in associated_units:
self.assertTrue(u in existing_units + units)
开发者ID:jwmatthews,项目名称:pulp_rpm,代码行数:24,代码来源:test_import_units.py
示例19: TestValidateConfig
class TestValidateConfig(rpm_support_base.PulpRPMTests):
def setUp(self):
super(TestValidateConfig, self).setUp()
self.temp_dir = tempfile.mkdtemp()
self.repo = mock.Mock(spec=Repository)
self.repo.working_dir = os.path.join(self.temp_dir, "repo_working_dir")
os.makedirs(self.repo.working_dir)
self.importer = YumImporter()
self.init()
def tearDown(self):
super(TestValidateConfig, self).tearDown()
shutil.rmtree(self.temp_dir)
def init(self):
self.data_dir = os.path.abspath(os.path.join(os.path.abspath(os.path.dirname(__file__)), "../data"))
def test_config_feed_url(self):
# test bad feed_url
feed_url = "fake://example.redhat.com/"
config = importer_mocks.get_basic_config(feed_url=feed_url)
state, msg = self.importer.validate_config(self.repo, config, [])
self.assertFalse(state)
feed_url = "http://example.redhat.com/"
config = importer_mocks.get_basic_config(feed_url=feed_url)
state, msg = self.importer.validate_config(self.repo, config, [])
self.assertTrue(state)
def test_config_ssl_verify(self):
feed_url = "http://example.redhat.com/"
ssl_verify = "fake"
config = importer_mocks.get_basic_config(feed_url=feed_url, ssl_verify=ssl_verify)
state, msg = self.importer.validate_config(self.repo, config, [])
self.assertFalse(state)
ssl_verify = True
config = importer_mocks.get_basic_config(feed_url=feed_url, ssl_verify=ssl_verify)
state, msg = self.importer.validate_config(self.repo, config, [])
self.assertTrue(state)
def test_config_ssl_ca_cert(self):
if not M2CRYPTO_HAS_CRL_SUPPORT:
return
feed_url = "http://example.redhat.com/"
ssl_ca_cert = "fake_path_to_ca"
config = importer_mocks.get_basic_config(feed_url=feed_url, ssl_ca_cert=ssl_ca_cert)
state, msg = self.importer.validate_config(self.repo, config, [])
self.assertFalse(state)
ssl_ca_cert = open(os.path.join(self.data_dir, "valid_ca.crt")).read()
config = importer_mocks.get_basic_config(feed_url=feed_url, ssl_ca_cert=ssl_ca_cert)
state, msg = self.importer.validate_config(self.repo, config, [])
self.assertTrue(state)
ssl_ca_cert_filename = os.path.join(self.repo.working_dir, "ssl_ca_cert")
self.assertTrue(os.path.exists(ssl_ca_cert_filename))
ca_cert_data = open(ssl_ca_cert_filename).read()
self.assertEqual(ca_cert_data, ssl_ca_cert)
def test_config_ssl_client_cert(self):
if not M2CRYPTO_HAS_CRL_SUPPORT:
return
feed_url = "http://example.redhat.com/"
ssl_client_cert = "fake_path_to_client_cert"
config = importer_mocks.get_basic_config(feed_url=feed_url, ssl_client_cert=ssl_client_cert)
state, msg = self.importer.validate_config(self.repo, config, [])
self.assertFalse(state)
ssl_client_cert = open(os.path.join(self.data_dir, "cert.crt")).read()
config = importer_mocks.get_basic_config(feed_url=feed_url, ssl_client_cert=ssl_client_cert)
state, msg = self.importer.validate_config(self.repo, config, [])
self.assertTrue(state)
ssl_client_cert_filename = os.path.join(self.repo.working_dir, "ssl_client_cert")
self.assertTrue(os.path.exists(ssl_client_cert_filename))
client_cert_data = open(ssl_client_cert_filename).read()
self.assertEqual(client_cert_data, ssl_client_cert)
def test_config_proxy_url(self):
feed_url = "http://example.redhat.com/"
proxy_url = "fake://proxy"
config = importer_mocks.get_basic_config(feed_url=feed_url, proxy_url=proxy_url)
state, msg = self.importer.validate_config(self.repo, config, [])
self.assertFalse(state)
proxy_url = "http://proxy"
config = importer_mocks.get_basic_config(feed_url=feed_url, proxy_url=proxy_url)
state, msg = self.importer.validate_config(self.repo, config, [])
self.assertTrue(state)
def test_config_proxy_port(self):
config = importer_mocks.get_basic_config(proxy_port=100)
state, msg = self.importer.validate_config(self.repo, config, [])
self.assertTrue(state)
config = importer_mocks.get_basic_config(proxy_port='port')
#.........这里部分代码省略.........
开发者ID:jessegonzalez,项目名称:pulp_rpm,代码行数:101,代码来源:test_importer_validate_config.py
示例20: test_package_group_unit_import
#.........这里部分代码省略.........
'name':'pulp-dot-1.0-test',
'version':'0.1.1',
'release':'1.fc11',
'epoch':'0',
'arch':'x86_64',
'checksum':'435d92e6c09248b501b8d2ae786f92ccfad69fab8b1bc774e2b66ff6c0d83979',
'checksumtype':'sha256',
}
metadata = {
'filename':filename
}
u = Unit(TYPE_ID_RPM, unit_key, metadata, storage_path)
source_units.append(u)
verify_units.append(u)
storage_path = '%s/pulp-dot-2.0-test/0.1.2/1.fc11/x86_64/435d92e6c09248b501b8d2ae786f92ccfad69fab8b1bc774e2b66ff6c0d83979/pulp-dot-2.0-test-0.1.2-1.fc11.x86_64.rpm' % (self.pkg_dir)
filename = os.path.basename(storage_path)
unit_key = {
'name':'pulp-dot-2.0-test',
'version':'0.1.2',
'release':'1.fc11',
'epoch':'0',
'arch':'x86_64',
'checksum':'435d92e6c09248b501b8d2ae786f92ccfad69fab8b1bc774e2b66ff6c0d83979',
'checksumtype':'sha256',
}
metadata = {
'filename':filename
}
u = Unit(TYPE_ID_RPM, unit_key, metadata, storage_path)
source_units.append(u)
verify_units.append(u)
storage_path = '%s/pulp-test-package/0.3.1/1.fc11/x86_64/6bce3f26e1fc0fc52ac996f39c0d0e14fc26fb8077081d5b4dbfb6431b08aa9f/pulp-test-package-0.3.1-1.fc11.x86_64.rpm' % (self.pkg_dir)
filename = os.path.basename(storage_path)
unit_key = {
'name':'pulp-test-package',
'version':'0.3.1',
'release':'1.fc11',
'epoch':'0',
'arch':'x86_64',
'checksum':'6bce3f26e1fc0fc52ac996f39c0d0e14fc26fb8077081d5b4dbfb6431b08aa9f',
'checksumtype':'sha256',
}
metadata = {
'filename':filename
}
u = Unit(TYPE_ID_RPM, unit_key, metadata, storage_path)
source_units.append(u)
verify_old_version_skipped = [u]
storage_path = '%s/pulp-test-package/0.3.2/1.fc11/x86_64/6bce3f26e1fc0fc52ac996f39c0d0e14fc26fb8077081d5b4dbfb6431b08aa9f/pulp-test-package-0.3.1-1.fc11.x86_64.rpm' % (self.pkg_dir)
filename = os.path.basename(storage_path)
unit_key = {
'name':'pulp-test-package',
'version':'0.3.2',
'release':'1.fc11',
'epoch':'0',
'arch':'x86_64',
'checksum':'6bce3f26e1fc0fc52ac996f39c0d0e14fc26fb8077081d5b4dbfb6431b08aa9f',
'checksumtype':'sha256',
}
metadata = {
'filename':filename
}
u = Unit(TYPE_ID_RPM, unit_key, metadata, storage_path)
source_units.append(u)
verify_units.append(u)
storage_path = '%s/pulp-test-optional-package/0.3.2/1.fc11/x86_64/6bce3f26e1fc0fc52ac996f39c0d0e14fc26fb8077081d5b4dbfb6431b08aa9f/pulp-test-package-0.3.1-1.fc11.x86_64.rpm' % (self.pkg_dir)
filename = os.path.basename(storage_path)
unit_key = {
'name':'pulp-optional-package',
'version':'0.1.1',
'release':'1.fc11',
'epoch':'0',
'arch':'x86_64',
'checksum':'6bce3f26e1fc0fc52ac996f39c0d0e14fc26fb8077081d5b4dbfb6431b08aa9f',
'checksumtype':'sha256',
}
metadata = {
'filename':filename
}
u = Unit(TYPE_ID_RPM, unit_key, metadata, storage_path)
source_units.append(u)
verify_units.append(u)
grp_a.metadata['mandatory_package_names'] = ["pulp-test-package",]
grp_a.metadata['default_package_names'] = ["pulp-dot-2.0-test"]
grp_a.metadata['optional_package_names'] = ["pulp-optional-package"]
grp_a.metadata['conditional_package_names'] = [('pulp-dot-1.0-test', [])]
existing_units = [grp_a]
conduit = importer_mocks.get_import_conduit(source_units + [grp_a], existing_units=source_units + existing_units)
config = importer_mocks.get_basic_config()
importer = YumImporter()
# Test
result = importer.import_units(repoA, repoB, conduit, config, [grp_a])
# Verify
associated_units = [mock_call[0][0] for mock_call in conduit.associate_unit.call_args_list]
# verify expected units are in associate units
for u in verify_units:
self.assertTrue(u in associated_units)
# verify that the version compare worked and skipped old versions
for u in verify_old_version_skipped:
self.assertFalse(u in associated_units)
开发者ID:ehelms,项目名称:pulp_rpm,代码行数:101,代码来源:test_import_units.py
注:本文中的yum_importer.importer.YumImporter类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论