本文整理汇总了Python中pulp.plugins.model.Unit类的典型用法代码示例。如果您正苦于以下问题：Python Unit类的具体用法？Python Unit怎么用？Python Unit使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了Unit类的19个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_handle_erratum_with_link
def test_handle_erratum_with_link(self, mock_link):
# Setup
unit_key = {'id': 'test-erratum'}
metadata = {'a': 'a'}
config = PluginCallConfiguration({}, {})
mock_repo = mock.MagicMock()
mock_conduit = mock.MagicMock()
inited_unit = Unit(models.Errata.TYPE, unit_key, metadata, None)
saved_unit = Unit(models.Errata.TYPE, unit_key, metadata, None)
saved_unit.id = 'ihaveanidnow'
mock_conduit.init_unit.return_value = inited_unit
mock_conduit.save_unit.return_value = saved_unit
# Test
upload._handle_erratum(mock_repo, models.Errata.TYPE, unit_key, metadata, None,
mock_conduit, config)
# Verify
mock_conduit.init_unit.assert_called_once_with(models.Errata.TYPE, unit_key,
metadata, None)
mock_conduit.save_unit.assert_called_once_with(inited_unit)
mock_link.assert_called_once()
self.assertEqual(mock_link.call_args[0][0], mock_conduit)
self.assertTrue(isinstance(mock_link.call_args[0][1], models.Errata))
# it is very important that this is the saved_unit, and not the inited_unit,
# because the underlying link logic requires it to have an "id".
self.assertTrue(mock_link.call_args[0][2] is saved_unit)
开发者ID:dkliban，项目名称:pulp_rpm，代码行数:29，代码来源:test_upload.py
示例2: to_plugin_unit
def to_plugin_unit(pulp_unit, type_def):
"""
Parses the raw dictionary of a content unit into its plugin representation.
@param pulp_unit: raw dictionary of unit metadata
@type pulp_unit: dict
@param type_def: Pulp stored definition for the unit type
@type type_def: pulp.server.db.model.content.ContentType
@return: plugin unit representation of the given unit
@rtype: pulp.plugins.model.Unit
"""
# Copy so we don't mangle the original unit
pulp_unit = dict(pulp_unit)
key_list = type_def['unit_key']
unit_key = {}
for k in key_list:
unit_key[k] = pulp_unit.pop(k)
storage_path = pulp_unit.pop('_storage_path', None)
unit_id = pulp_unit.pop('_id', None)
u = Unit(type_def['id'], unit_key, pulp_unit, storage_path)
u.id = unit_id
return u
开发者ID:bartwo，项目名称:pulp，代码行数:31，代码来源:_common.py
示例3: to_plugin_unit
def to_plugin_unit(pulp_unit, unit_type_id, unit_key_fields):
"""
Parses the raw dictionary of a content unit into its plugin representation.
:param pulp_unit: raw dictionary of unit metadata
:type pulp_unit: dict
:param unit_type_id: unique identifier for the type of unit
:type unit_type_id: str
:param unit_key_fields: collection of keys required for the type's unit key
:type unit_key_fields: list or tuple
:return: plugin unit representation of the given unit
:rtype: pulp.plugins.model.Unit
"""
# Copy so we don't mangle the original unit
pulp_unit = dict(pulp_unit)
unit_key = {}
for k in unit_key_fields:
unit_key[k] = pulp_unit.pop(k)
storage_path = pulp_unit.pop('_storage_path', None)
unit_id = pulp_unit.pop('_id', None)
u = Unit(unit_type_id, unit_key, pulp_unit, storage_path)
u.id = unit_id
return u
开发者ID:BrnoPCmaniak，项目名称:pulp，代码行数:30，代码来源:_common.py
示例4: 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
示例5: test_distribution_exports
def test_distribution_exports(self):
feed_url = "file://%s/pulp_unittest/" % self.data_dir
repo = mock.Mock(spec=Repository)
repo.working_dir = self.repo_working_dir
repo.id = "pulp_unittest"
repo.checksumtype = 'sha'
sync_conduit = importer_mocks.get_sync_conduit(type_id=TYPE_ID_RPM, existing_units=[], pkg_dir=self.pkg_dir)
config = importer_mocks.get_basic_config(feed_url=feed_url)
importerRPM = importer_rpm.ImporterRPM()
status, summary, details = importerRPM.sync(repo, sync_conduit, config)
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.repo_working_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.repo_working_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.repo_working_dir,
"checksum" : "099f2bafd533e97dcfee778bc24138c40f114323785ac1987a0db66e07086f74",
"filename" : "fileC.iso",
"pkgpath" : "%s/ks-TestFamily-TestVariant-16-x86_64/images" % self.pkg_dir, "size" : 0 } ],}
distro_unit = Unit(distribution.TYPE_ID_DISTRO, dunit_key, metadata, '')
distro_unit.storage_path = "%s/ks-TestFamily-TestVariant-16-x86_64" % self.pkg_dir
symlink_dir = "%s/%s" % (self.repo_working_dir, "isos")
iso_distributor = ISODistributor()
publish_conduit = distributor_mocks.get_publish_conduit(existing_units=[distro_unit], pkg_dir=self.pkg_dir)
config = distributor_mocks.get_basic_config(https_publish_dir=self.https_publish_dir, http=False, https=True)
repo_exporter = RepoExporter(symlink_dir)
# status, errors = iso_distributor._export_distributions([distro_unit], symlink_dir)
status, errors = repo_exporter.export_distributions([distro_unit])
print status, errors
self.assertTrue(status)
for file in metadata['files']:
print os.path.isfile("%s/%s" % (symlink_dir, file['relativepath']))
self.assertTrue(os.path.isfile("%s/%s" % (symlink_dir, file['relativepath'])))
开发者ID:ehelms，项目名称:pulp_rpm，代码行数:52，代码来源:test_iso_distributor.py
示例6: existing_units
def existing_units(self):
units = []
# for unit in [self.UNIT_KEY_A, self.UNIT_KEY_B]:
# unit = Unit(TYPE_ID_RPM, unit, {}, '')
# units.append(unit)
unit = Unit(TYPE_ID_RPM, self.UNIT_KEY_A, {}, '')
unit.metadata = constants.PULP_SERVER_RPM_METADATA
units.append(unit)
unit = Unit(TYPE_ID_RPM, self.UNIT_KEY_B, {}, '')
unit.metadata = constants.PULP_RPM_SERVER_RPM_METADATA
units.append(unit)
return units
开发者ID:jwmatthews，项目名称:pulp_rpm，代码行数:14，代码来源:test_import_units.py
示例7: test_orphaned_distributions
def test_orphaned_distributions(self):
feed_url = "http://repos.fedorapeople.org/repos/pulp/pulp/demo_repos/pulp_unittest/"
repo = mock.Mock(spec=Repository)
repo.working_dir = self.working_dir
repo.id = "test_repo"
sync_conduit = importer_mocks.get_sync_conduit(pkg_dir=self.pkg_dir)
config = importer_mocks.get_basic_config(feed_url=feed_url)
importerRPM = importer_rpm.ImporterRPM()
status, summary, details = importerRPM.sync(repo, sync_conduit, config)
self.assertTrue(status)
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.working_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.working_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.working_dir,
"checksum" : "099f2bafd533e97dcfee778bc24138c40f114323785ac1987a0db66e07086f74",
"filename" : "fileC.iso",
"pkgpath" : "%s/ks-TestFamily-TestVariant-16-x86_64/images" % self.pkg_dir, "size" : 0 } ],}
distro_unit = Unit(distribution.TYPE_ID_DISTRO, dunit_key, metadata, '')
distro_unit.storage_path = "%s/ks-TestFamily-TestVariant-16-x86_64" % self.pkg_dir
new_feed_url = "http://repos.fedorapeople.org/repos/pulp/pulp/demo_repos/zoo/"
sync_conduit = importer_mocks.get_sync_conduit(pkg_dir=self.pkg_dir, existing_units=[distro_unit])
config = importer_mocks.get_basic_config(feed_url=new_feed_url)
status, summary, details = importerRPM.sync(repo, sync_conduit, config)
print status, summary, details
self.assertTrue(status)
self.assertTrue(summary is not None)
self.assertTrue(details is not None)
self.assertEquals(summary["num_orphaned_distributions"], 1)
开发者ID:jessegonzalez，项目名称:pulp_rpm，代码行数:49，代码来源:test_distributions.py
示例8: test_link_unit
def test_link_unit(self, mock_link):
# Setup
from_unit = Unit('t1', {'k' : 'v1'}, {'m' : 'm'}, 'p')
from_unit.id = 'from-unit'
to_unit = Unit('t2', {'k' : 'v2'}, {'m' : 'm'}, 'p')
to_unit.id = 'to-unit'
# Test
self.mixin.link_unit(from_unit, to_unit)
# Verify
self.assertEqual(1, mock_link.call_count)
self.assertEqual(mock_link.call_args[0][0], from_unit.type_id)
self.assertEqual(mock_link.call_args[0][1], from_unit.id)
self.assertEqual(mock_link.call_args[0][2], to_unit.type_id)
self.assertEqual(mock_link.call_args[0][3], [to_unit.id])
开发者ID:preethit，项目名称:pulp-1，代码行数:16，代码来源:test_conduit_mixins.py
示例9: test_link_unit
def test_link_unit(self, mock_link):
# Setup
from_unit = Unit("t1", {"k": "v1"}, {"m": "m"}, "p")
from_unit.id = "from-unit"
to_unit = Unit("t2", {"k": "v2"}, {"m": "m"}, "p")
to_unit.id = "to-unit"
# Test
self.mixin.link_unit(from_unit, to_unit)
# Verify
self.assertEqual(1, mock_link.call_count)
self.assertEqual(mock_link.call_args[0][0], from_unit.type_id)
self.assertEqual(mock_link.call_args[0][1], from_unit.id)
self.assertEqual(mock_link.call_args[0][2], to_unit.type_id)
self.assertEqual(mock_link.call_args[0][3], [to_unit.id])
开发者ID:harrisongu，项目名称:pulp，代码行数:16，代码来源:test_conduit_mixins.py
示例10: test_unit_not_applicable_not_in_repo
def test_unit_not_applicable_not_in_repo(self):
# Errata refers to RPMs which ARE part of our test consumer's profile,
# but are not in the repo.
errata_obj = self.get_test_errata_object()
errata_unit = Unit(TYPE_ID_ERRATA, {"id": errata_obj["id"]}, errata_obj, None)
errata_unit.id = 'an_errata'
test_repo = profiler_mocks.get_repo("test_repo_id")
prof = YumProfiler()
errata_rpms = prof._get_rpms_from_errata(errata_unit)
conduit = profiler_mocks.get_profiler_conduit(repo_units=[errata_unit],
repo_bindings=[test_repo],
errata_rpms=errata_rpms)
unit_profile = self.test_consumer.profiles[TYPE_ID_RPM]
bound_repo_id = "test_repo_id"
report_list = prof.calculate_applicable_units(unit_profile, bound_repo_id, None, conduit)
self.assertEqual(report_list, {TYPE_ID_RPM: [], TYPE_ID_ERRATA: []})
开发者ID:BrnoPCmaniak，项目名称:pulp_rpm，代码行数:17，代码来源:test_yum.py
示例11: test_repo_export_isos
def test_repo_export_isos(self):
feed_url = "file://%s/pulp_unittest/" % self.data_dir
repo = mock.Mock(spec=Repository)
repo.working_dir = self.repo_working_dir
repo.id = "pulp_unittest"
repo.checksumtype = 'sha'
sync_conduit = importer_mocks.get_sync_conduit(type_id=TYPE_ID_RPM, existing_units=[], pkg_dir=self.pkg_dir)
config = importer_mocks.get_basic_config(feed_url=feed_url)
importerRPM = importer_rpm.ImporterRPM()
status, summary, details = importerRPM.sync(repo, sync_conduit, config)
unit_key_a = {'id' : '','name' :'pulp-dot-2.0-test', 'version' :'0.1.2', 'release' : '1.fc11', 'epoch':'0', 'arch' : 'x86_64', 'checksumtype' : 'sha256',
'checksum': '435d92e6c09248b501b8d2ae786f92ccfad69fab8b1bc774e2b66ff6c0d83979', 'type_id' : 'rpm'}
unit_a = Unit(TYPE_ID_RPM, unit_key_a, {}, '')
unit_a.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
unit_key_b = {'id' : '', 'name' :'pulp-test-package', 'version' :'0.2.1', 'release' :'1.fc11', 'epoch':'0','arch' : 'x86_64', 'checksumtype' :'sha256',
'checksum': '4dbde07b4a8eab57e42ed0c9203083f1d61e0b13935d1a569193ed8efc9ecfd7', 'type_id' : 'rpm', }
unit_b = Unit(TYPE_ID_RPM, unit_key_b, {}, '')
unit_b.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
unit_key_c = {'id' : '', 'name' :'pulp-test-package', 'version' :'0.3.1', 'release' :'1.fc11', 'epoch':'0','arch' : 'x86_64', 'checksumtype' :'sha256',
'checksum': '6bce3f26e1fc0fc52ac996f39c0d0e14fc26fb8077081d5b4dbfb6431b08aa9f', 'type_id' : 'rpm', }
unit_c = Unit(TYPE_ID_RPM, unit_key_c, {}, '')
unit_c.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
existing_units = []
for unit in [unit_a, unit_b, unit_c]:
existing_units.append(unit)
sync_conduit = importer_mocks.get_sync_conduit(type_id=TYPE_ID_RPM, existing_units=existing_units, pkg_dir=self.pkg_dir)
importerErrata = errata.ImporterErrata()
importerErrata.sync(repo, sync_conduit, config)
repo.working_dir = "%s/%s" % (self.repo_working_dir, "export")
iso_distributor = ISODistributor()
publish_conduit = distributor_mocks.get_publish_conduit(existing_units=existing_units, pkg_dir=self.pkg_dir)
# test https publish
config = distributor_mocks.get_basic_config(http_publish_dir=self.http_publish_dir, https_publish_dir=self.https_publish_dir, http=False, https=True, generate_metadata=True)
report = iso_distributor.publish_repo(repo, publish_conduit, config)
print report
self.assertTrue(os.path.exists("%s/%s" % (self.https_publish_dir, repo.id)))
self.assertEquals(len(os.listdir(self.http_publish_dir)), 0)
# test http publish
config = distributor_mocks.get_basic_config(http_publish_dir=self.http_publish_dir, https_publish_dir=self.https_publish_dir, http=True, https=False)
report = iso_distributor.publish_repo(repo, publish_conduit, config)
self.assertTrue(os.path.exists("%s/%s" % (self.http_publish_dir, repo.id)))
self.assertEquals(len(os.listdir(self.https_publish_dir)), 0)
isos_list = os.listdir("%s/%s" % (self.http_publish_dir, repo.id))
self.assertEqual(len(isos_list), 1)
# make sure the iso name defaults to repoid
self.assertTrue( isos_list[-1].startswith(repo.id))
# test isoprefix:
iso_prefix = "mock-iso-prefix"
config = distributor_mocks.get_basic_config(http_publish_dir=self.http_publish_dir, https_publish_dir=self.https_publish_dir, http=True, https=False, iso_prefix=iso_prefix)
report = iso_distributor.publish_repo(repo, publish_conduit, config)
self.assertTrue(os.path.exists("%s/%s" % (self.http_publish_dir, repo.id)))
self.assertEquals(len(os.listdir(self.https_publish_dir)), 0)
isos_list = os.listdir("%s/%s" % (self.http_publish_dir, repo.id))
self.assertEqual(len(isos_list), 2)
print isos_list
# make sure the iso name uses the prefix
self.assertTrue( isos_list[-1].startswith(iso_prefix))
开发者ID:stpierre，项目名称:pulp，代码行数:59，代码来源:test_iso_distributor.py
示例12: test_link_unit_bidirectional
def test_link_unit_bidirectional(self, mock_link):
# Setup
from_unit = Unit('t1', {'k' : 'v1'}, {'m' : 'm'}, 'p')
from_unit.id = 'from-unit'
to_unit = Unit('t2', {'k' : 'v2'}, {'m' : 'm'}, 'p')
to_unit.id = 'to-unit'
# Test
self.mixin.link_unit(from_unit, to_unit, bidirectional=True)
# Verify
self.assertEqual(2, mock_link.call_count)
call_1_args = mock_link.call_args_list[0][0]
self.assertEqual(call_1_args[0], from_unit.type_id)
self.assertEqual(call_1_args[1], from_unit.id)
self.assertEqual(call_1_args[2], to_unit.type_id)
self.assertEqual(call_1_args[3], [to_unit.id])
call_2_args = mock_link.call_args_list[1][0]
self.assertEqual(call_2_args[0], to_unit.type_id)
self.assertEqual(call_2_args[1], to_unit.id)
self.assertEqual(call_2_args[2], from_unit.type_id)
self.assertEqual(call_2_args[3], [from_unit.id])
开发者ID:preethit，项目名称:pulp-1，代码行数:24，代码来源:test_conduit_mixins.py
示例13: test_link_unit_bidirectional
def test_link_unit_bidirectional(self, mock_link):
# Setup
from_unit = Unit("t1", {"k": "v1"}, {"m": "m"}, "p")
from_unit.id = "from-unit"
to_unit = Unit("t2", {"k": "v2"}, {"m": "m"}, "p")
to_unit.id = "to-unit"
# Test
self.mixin.link_unit(from_unit, to_unit, bidirectional=True)
# Verify
self.assertEqual(2, mock_link.call_count)
call_1_args = mock_link.call_args_list[0][0]
self.assertEqual(call_1_args[0], from_unit.type_id)
self.assertEqual(call_1_args[1], from_unit.id)
self.assertEqual(call_1_args[2], to_unit.type_id)
self.assertEqual(call_1_args[3], [to_unit.id])
call_2_args = mock_link.call_args_list[1][0]
self.assertEqual(call_2_args[0], to_unit.type_id)
self.assertEqual(call_2_args[1], to_unit.id)
self.assertEqual(call_2_args[2], from_unit.type_id)
self.assertEqual(call_2_args[3], [from_unit.id])
开发者ID:harrisongu，项目名称:pulp，代码行数:24，代码来源:test_conduit_mixins.py
示例14: test_generate_isos
def test_generate_isos(self):
repo = mock.Mock(spec=Repository)
repo.id = "test_repo_for_export"
repo.working_dir = self.repo_iso_working_dir + "/" + repo.id
unit_key_a = {'id' : '','name' :'pulp-dot-2.0-test', 'version' :'0.1.2', 'release' : '1.fc11', 'epoch':'0', 'arch' : 'x86_64', 'checksumtype' : 'sha256',
'checksum': '435d92e6c09248b501b8d2ae786f92ccfad69fab8b1bc774e2b66ff6c0d83979', 'type_id' : 'rpm'}
unit_a = Unit(TYPE_ID_RPM, unit_key_a, {'updated' : ''}, '')
unit_a.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
unit_key_b = {'id' : '', 'name' :'pulp-test-package', 'version' :'0.2.1', 'release' :'1.fc11', 'epoch':'0','arch' : 'x86_64', 'checksumtype' :'sha256',
'checksum': '4dbde07b4a8eab57e42ed0c9203083f1d61e0b13935d1a569193ed8efc9ecfd7', 'type_id' : 'rpm', }
unit_b = Unit(TYPE_ID_RPM, unit_key_b, {'updated' : ''}, '')
unit_b.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
unit_key_c = {'id' : '', 'name' :'pulp-test-package', 'version' :'0.3.1', 'release' :'1.fc11', 'epoch':'0','arch' : 'x86_64', 'checksumtype' :'sha256',
'checksum': '6bce3f26e1fc0fc52ac996f39c0d0e14fc26fb8077081d5b4dbfb6431b08aa9f', 'type_id' : 'rpm', }
unit_c = Unit(TYPE_ID_RPM, unit_key_c, {'updated' : ''}, '')
unit_c.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
existing_units = []
for unit in [unit_a, unit_b, unit_c]:
existing_units.append(unit)
global progress_status
progress_status = None
def set_progress(progress):
global progress_status
progress_status = progress
publish_conduit = distributor_mocks.get_publish_conduit(pkg_dir=self.pkg_dir, existing_units=existing_units)
config = distributor_mocks.get_basic_config(https_publish_dir=self.https_publish_dir, http_publish_dir=self.http_publish_dir,
generate_metadata=True, http=True, https=False, iso_prefix="test-isos")
distributor = ISODistributor()
def cleanup(repo_working_dir):
return
iso_util.cleanup_working_dir = mock.Mock()
iso_util.cleanup_working_dir.side_effect = cleanup
publish_conduit.set_progress = mock.Mock()
publish_conduit.set_progress.side_effect = set_progress
progress_status = distributor.init_progress()
distributor.publish_repo(repo, publish_conduit, config)
self.assertTrue("isos" in progress_status)
self.assertTrue(progress_status["isos"].has_key("state"))
self.assertEqual(progress_status["isos"]["state"], "FINISHED")
self.assertEqual(progress_status["isos"]["num_success"], 1)
self.assertTrue(progress_status["isos"]["size_total"] is not None)
self.assertEqual(progress_status["isos"]["size_left"], 0)
self.assertEqual(progress_status["isos"]["items_total"], 1)
self.assertEqual(progress_status["isos"]["items_left"], 0)
self.assertTrue(os.path.exists("%s/%s" % (self.http_publish_dir, repo.id)))
self.assertEquals(len(os.listdir(self.https_publish_dir)), 0)
isos_list = os.listdir("%s/%s" % (self.http_publish_dir, repo.id))
self.assertEqual(len(isos_list), 1)
# make sure the iso name defaults to repoid
self.assertTrue( isos_list[0].startswith("test-isos"))
开发者ID:jessegonzalez，项目名称:pulp_rpm，代码行数:51，代码来源:test_iso_distributor.py
示例15: test_get_relpath_from_unit
def test_get_relpath_from_unit(self):
distributor = YumDistributor()
test_unit = Unit("rpm", "unit_key", {}, "")
test_unit.unit_key = {"fileName" : "test_1"}
rel_path = util.get_relpath_from_unit(test_unit)
self.assertEqual(rel_path, "test_1")
test_unit.unit_key = {}
test_unit.storage_path = "test_0"
rel_path = util.get_relpath_from_unit(test_unit)
self.assertEqual(rel_path, "test_0")
test_unit.metadata["filename"] = "test_2"
rel_path = util.get_relpath_from_unit(test_unit)
self.assertEqual(rel_path, "test_2")
test_unit.metadata["relativepath"] = "test_3"
rel_path = util.get_relpath_from_unit(test_unit)
self.assertEqual(rel_path, "test_3")
开发者ID:ehelms，项目名称:pulp，代码行数:20，代码来源:test_distributor.py
示例16: test_export_rpm
def test_export_rpm(self):
feed_url = "file://%s/test_repo_for_export/" % (self.data_dir)
repo = mock.Mock(spec=Repository)
repo.working_dir = self.repo_working_dir
repo.id = "test_repo_for_export"
sync_conduit = importer_mocks.get_sync_conduit(existing_units=[], pkg_dir=self.pkg_dir)
config = importer_mocks.get_basic_config(feed_url=feed_url)
importerRPM = importer_rpm.ImporterRPM()
status, summary, details = importerRPM.sync(repo, sync_conduit, config)
self.assertTrue(summary is not None)
self.assertTrue(details is not None)
self.assertTrue(status)
unit_key_a = {'id' : '','name' :'pulp-dot-2.0-test', 'version' :'0.1.2', 'release' : '1.fc11', 'epoch':'0', 'arch' : 'x86_64', 'checksumtype' : 'sha256',
'checksum': '435d92e6c09248b501b8d2ae786f92ccfad69fab8b1bc774e2b66ff6c0d83979', 'type_id' : 'rpm'}
unit_a = Unit(TYPE_ID_RPM, unit_key_a, {}, '')
unit_a.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
unit_key_b = {'id' : '', 'name' :'pulp-test-package', 'version' :'0.2.1', 'release' :'1.fc11', 'epoch':'0','arch' : 'x86_64', 'checksumtype' :'sha256',
'checksum': '4dbde07b4a8eab57e42ed0c9203083f1d61e0b13935d1a569193ed8efc9ecfd7', 'type_id' : 'rpm', }
unit_b = Unit(TYPE_ID_RPM, unit_key_b, {}, '')
unit_b.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
unit_key_c = {'id' : '', 'name' :'pulp-test-package', 'version' :'0.3.1', 'release' :'1.fc11', 'epoch':'0','arch' : 'x86_64', 'checksumtype' :'sha256',
'checksum': '6bce3f26e1fc0fc52ac996f39c0d0e14fc26fb8077081d5b4dbfb6431b08aa9f', 'type_id' : 'rpm', }
unit_c = Unit(TYPE_ID_RPM, unit_key_c, {}, '')
unit_c.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
existing_units = []
for unit in [unit_a, unit_b, unit_c]:
existing_units.append(unit)
symlink_dir = "%s/%s" % (self.repo_working_dir, "isos")
iso_distributor = ISODistributor()
publish_conduit = distributor_mocks.get_publish_conduit(existing_units=existing_units, pkg_dir=self.pkg_dir)
config = distributor_mocks.get_basic_config(https_publish_dir=self.https_publish_dir, http=False, https=True)
print symlink_dir
# status, errors = iso_distributor._export_rpms(existing_units, symlink_dir)
repo_exporter = RepoExporter(symlink_dir)
status, errors = repo_exporter.export_rpms(existing_units)
print status, errors
self.assertTrue(status)
self.assertEquals(len(os.listdir(symlink_dir)), 3)
开发者ID:ehelms，项目名称:pulp_rpm，代码行数:38，代码来源:test_iso_distributor.py
示例17: test_group_publish_isos
def test_group_publish_isos(self):
feed_url = "file://%s/pulp_unittest/" % self.data_dir
repo_1 = mock.Mock(spec=Repository)
repo_1.id = "test_repo_for_export_1"
repo_1.working_dir = self.repo_working_dir
repo_1.checksumtype = 'sha'
repo_2 = mock.Mock(spec=Repository)
repo_2.id = "test_repo_for_export_2"
repo_2.working_dir = self.repo_working_dir
repo_2.checksumtype = 'sha'
sync_conduit = importer_mocks.get_sync_conduit(type_id=TYPE_ID_RPM, existing_units=[], pkg_dir=self.pkg_dir)
config = importer_mocks.get_basic_config(feed_url=feed_url)
importerRPM = importer_rpm.ImporterRPM()
status, summary, details = importerRPM.sync(repo_1, sync_conduit, config)
status, summary, details = importerRPM.sync(repo_2, sync_conduit, config)
unit_key_a = {'id' : '','name' :'pulp-dot-2.0-test', 'version' :'0.1.2', 'release' : '1.fc11', 'epoch':'0', 'arch' : 'x86_64', 'checksumtype' : 'sha256',
'checksum': '435d92e6c09248b501b8d2ae786f92ccfad69fab8b1bc774e2b66ff6c0d83979', 'type_id' : 'rpm'}
unit_a = Unit(TYPE_ID_RPM, unit_key_a, {}, '')
unit_a.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
unit_key_b = {'id' : '', 'name' :'pulp-test-package', 'version' :'0.2.1', 'release' :'1.fc11', 'epoch':'0','arch' : 'x86_64', 'checksumtype' :'sha256',
'checksum': '4dbde07b4a8eab57e42ed0c9203083f1d61e0b13935d1a569193ed8efc9ecfd7', 'type_id' : 'rpm', }
unit_b = Unit(TYPE_ID_RPM, unit_key_b, {}, '')
unit_b.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
unit_key_c = {'id' : '', 'name' :'pulp-test-package', 'version' :'0.3.1', 'release' :'1.fc11', 'epoch':'0','arch' : 'x86_64', 'checksumtype' :'sha256',
'checksum': '6bce3f26e1fc0fc52ac996f39c0d0e14fc26fb8077081d5b4dbfb6431b08aa9f', 'type_id' : 'rpm', }
unit_c = Unit(TYPE_ID_RPM, unit_key_c, {}, '')
unit_c.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
existing_units = []
for unit in [unit_a, unit_b, unit_c]:
existing_units.append(unit)
sync_conduit = importer_mocks.get_sync_conduit(type_id=TYPE_ID_RPM, existing_units=existing_units, pkg_dir=self.pkg_dir)
importerErrata = errata.ImporterErrata()
importerErrata.sync(repo_1, sync_conduit, config)
importerErrata.sync(repo_2, sync_conduit, config)
repo_group = mock.Mock(spec=RepositoryGroup)
repo_group.id = "test_group"
repo_group.repo_ids = [repo_1.id, repo_2.id]
repo_group.working_dir = self.group_working_dir
global progress_status
progress_status = None
def set_progress(progress):
global progress_status
progress_status = progress
publish_conduit = distributor_mocks.get_publish_conduit(existing_units=existing_units, pkg_dir=self.pkg_dir)
config = distributor_mocks.get_basic_config(https_publish_dir=self.https_publish_dir, http_publish_dir=self.http_publish_dir,
generate_metadata=True, http=True, https=False, iso_prefix="test-isos")
distributor = GroupISODistributor()
def cleanup(repo_working_dir):
return
iso_util.cleanup_working_dir.cleanup = mock.Mock()
iso_util.cleanup_working_dir.side_effect = cleanup
publish_conduit.set_progress = mock.Mock()
publish_conduit.set_progress.side_effect = set_progress
distributor.publish_group(repo_group, publish_conduit, config)
self.assertTrue("isos" in progress_status)
self.assertTrue(progress_status["isos"].has_key("state"))
self.assertEqual(progress_status["isos"]["state"], "FINISHED")
self.assertTrue(os.path.exists("%s/%s" % (self.http_publish_dir, repo_group.id)))
self.assertEquals(len(os.listdir(self.https_publish_dir)), 0)
isos_list = os.listdir("%s/%s" % (self.http_publish_dir, repo_group.id))
print isos_list
self.assertEqual(len(isos_list), 1)
# make sure the iso name defaults to repoid
self.assertTrue( isos_list[0].startswith("test-isos"))
开发者ID:ehelms，项目名称:pulp，代码行数:66，代码来源:test_group_distributor.py
示例18: test_iso_export_by_date_range
def test_iso_export_by_date_range(self):
feed_url = "file://%s/test_errata_local_sync/" % self.data_dir
repo = mock.Mock(spec=Repository)
repo.working_dir = self.repo_working_dir
repo.id = "test_errata_local_sync"
repo.checksumtype = "sha"
sync_conduit = importer_mocks.get_sync_conduit(type_id=TYPE_ID_RPM, existing_units=[], pkg_dir=self.pkg_dir)
config = importer_mocks.get_basic_config(feed_url=feed_url)
importerRPM = importer_rpm.ImporterRPM()
status, summary, details = importerRPM.sync(repo, sync_conduit, config)
metadata = {}
existing_units = []
unit_key_a = {
"id": "",
"name": "patb",
"version": "0.1",
"release": "2",
"epoch": "0",
"arch": "x86_64",
"checksumtype": "sha",
"checksum": "017c12050a97cf6095892498750c2a39d2bf535e",
}
rpm_unit_a = Unit(TYPE_ID_RPM, unit_key_a, metadata, "")
rpm_unit_a.storage_path = (
"%s/patb/0.1/2/noarch/017c12050a97cf6095892498750c2a39d2bf535e/patb-0.1-2.noarch.rpm" % self.pkg_dir
)
existing_units.append(rpm_unit_a)
unit_key_b = {
"id": "",
"name": "emoticons",
"version": "0.1",
"release": "2",
"epoch": "0",
"arch": "x86_64",
"checksumtype": "sha",
"checksum": "663c89b0d29bfd5479d8736b716d50eed9495dbb",
}
rpm_unit_b = Unit(TYPE_ID_RPM, unit_key_b, metadata, "")
rpm_unit_b.storage_path = (
"%s/emoticons/0.1/2/noarch/663c89b0d29bfd5479d8736b716d50eed9495dbb/emoticons-0.1-2.noarch.rpm"
% self.pkg_dir
)
existing_units.append(rpm_unit_b)
sync_conduit = importer_mocks.get_sync_conduit(
type_id=TYPE_ID_RPM, existing_units=existing_units, pkg_dir=self.pkg_dir
)
importerErrata = errata.ImporterErrata()
status, summary, details = importerErrata.sync(repo, sync_conduit, config)
unit_key = dict()
unit_key["id"] = "RHEA-2010:9999"
mdata = {
"description": "test",
"from_str": "[email protected]",
"issued": "2010-03-30 08:07:30",
"pkglist": [
{
"name": "RHEL Virtualization (v. 5 for 32-bit x86)",
"packages": [
{
"arch": "x86_64",
"epoch": "0",
"filename": "patb-0.1-2.x86_64.rpm",
"name": "patb",
|
客服电话
APP下载
官方微信
请发表评论