本文整理汇总了Python中yum_distributor.distributor.YumDistributor类的典型用法代码示例。如果您正苦于以下问题:Python YumDistributor类的具体用法?Python YumDistributor怎么用?Python YumDistributor使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了YumDistributor类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_yum_plugin_generate_yum_metadata_checksum_default
def test_yum_plugin_generate_yum_metadata_checksum_default(self, mock_YumMetadataGenerator,
mock_distributor_manager):
repo = mock.Mock(spec=Repository)
repo.working_dir = self.repo_working_dir
repo.id = "test_publish"
num_units = 10
relative_url = "rel_a/rel_b/rel_c/"
existing_units = self.get_units(count=num_units)
publish_conduit = distributor_mocks.get_publish_conduit(type_id="rpm",
existing_units=existing_units,
checksum_type=None,
pkg_dir=self.pkg_dir)
config = distributor_mocks.get_basic_config(https_publish_dir=self.https_publish_dir,
relative_url=relative_url,
http=False, https=True)
distributor = YumDistributor()
distributor.process_repo_auth_certificate_bundle = mock.Mock()
config_conduit = mock.Mock(spec=RepoConfigConduit)
config_conduit.get_repo_distributors_by_relative_url.return_value = MockCursor([])
metadata.generate_yum_metadata(repo.id, repo.working_dir, publish_conduit, config)
mock_YumMetadataGenerator.assert_called_with(ANY, checksum_type=metadata.DEFAULT_CHECKSUM,
skip_metadata_types=ANY, is_cancelled=ANY,
group_xml_path=ANY,
updateinfo_xml_path=ANY,
custom_metadata_dict=ANY)
self.assertFalse(mock_distributor_manager.called)
开发者ID:bechtoldt,项目名称:pulp_rpm,代码行数:26,代码来源:test_distributor.py
示例2: test_consumer_payload
def test_consumer_payload(self):
PAYLOAD_FIELDS = [
"server_name",
"relative_path",
"protocols",
"gpg_keys",
"client_cert",
"ca_cert",
"repo_name",
]
http = True
https = True
relative_url = "/pub/content/"
gpgkey = ["test_gpg_key"]
auth_cert = open(os.path.join(self.data_dir, "cert.crt")).read()
auth_ca = open(os.path.join(self.data_dir, "ca.key")).read()
config = distributor_mocks.get_basic_config(
relative_url=relative_url, http=http, https=https, auth_cert=auth_cert, auth_ca=auth_ca, gpgkey=gpgkey
)
distributor = YumDistributor()
repo = mock.Mock(spec=Repository)
repo.working_dir = self.repo_working_dir
repo.id = "test_payload"
repo.display_name = "Nice Repo"
payload = distributor.create_consumer_payload(repo, config)
for field in PAYLOAD_FIELDS:
print field
self.assertTrue(field in payload)
self.assertTrue("http" in payload["protocols"])
self.assertTrue("https" in payload["protocols"])
print payload
开发者ID:ehelms,项目名称:pulp_rpm,代码行数:32,代码来源:test_distributor.py
示例3: test_distributor_removed
def test_distributor_removed(self, delete_protected_repo, mock_factory):
"""
Make sure the distributor_removed() method cleans up the published files.
"""
# Create and publish repo to both http and https directories
repo = mock.Mock(spec=Repository)
repo.id = 'about_to_be_removed'
repo.working_dir = self.repo_working_dir
existing_units = self.get_units(count=5)
publish_conduit = distributor_mocks.get_publish_conduit(type_id="rpm",
existing_units=existing_units,
pkg_dir=self.pkg_dir)
config = distributor_mocks.get_basic_config(http_publish_dir=self.http_publish_dir,
https_publish_dir=self.https_publish_dir,
http=True,
https=True)
distributor = YumDistributor()
publish_conduit.repo_id = repo.id
publish_conduit.distributor_id = 'foo'
report = distributor.publish_repo(repo, publish_conduit, config)
publishing_paths = [os.path.join(directory, repo.id) \
for directory in [self.http_publish_dir, self.https_publish_dir]]
# The publishing paths should exist
self.assertTrue(all([os.path.exists(directory) for directory in publishing_paths]))
delete_protected_repo.reset_mock()
distributor.distributor_removed(repo, config)
# Neither publishing path should exist now
self.assertFalse(all([os.path.exists(directory) for directory in publishing_paths]))
# delete_protected_repo should have been called
delete_protected_repo.assert_called_once_with(repo.id)
开发者ID:bechtoldt,项目名称:pulp_rpm,代码行数:31,代码来源:test_distributor.py
示例4: test_publish
def test_publish(self):
repo = mock.Mock(spec=Repository)
repo.working_dir = self.repo_working_dir
repo.id = "test_publish"
num_units = 10
relative_url = "rel_a/rel_b/rel_c/"
existing_units = self.get_units(count=num_units)
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, relative_url=relative_url,
http=False, https=True)
distributor = YumDistributor()
distributor.process_repo_auth_certificate_bundle = mock.Mock()
status, msg = distributor.validate_config(repo, config, None)
self.assertTrue(status)
report = distributor.publish_repo(repo, publish_conduit, config)
self.assertTrue(report.success_flag)
summary = report.summary
self.assertEqual(summary["num_package_units_attempted"], num_units)
self.assertEqual(summary["num_package_units_published"], num_units)
self.assertEqual(summary["num_package_units_errors"], 0)
# Verify we did not attempt to publish to http
expected_repo_http_publish_dir = os.path.join(self.http_publish_dir, relative_url)
self.assertFalse(os.path.exists(expected_repo_http_publish_dir))
expected_repo_https_publish_dir = os.path.join(self.https_publish_dir, relative_url).rstrip('/')
self.assertEqual(summary["https_publish_dir"], expected_repo_https_publish_dir)
self.assertTrue(os.path.exists(expected_repo_https_publish_dir))
details = report.details
self.assertEqual(len(details["errors"]), 0)
#
# Add a verification of the publish directory
#
self.assertTrue(os.path.exists(summary["https_publish_dir"]))
self.assertTrue(os.path.islink(summary["https_publish_dir"].rstrip("/")))
source_of_link = os.readlink(expected_repo_https_publish_dir.rstrip("/"))
self.assertEquals(source_of_link, repo.working_dir)
#
# Verify the expected units
#
for u in existing_units:
expected_link = os.path.join(expected_repo_https_publish_dir, u.metadata["relativepath"])
self.assertTrue(os.path.exists(expected_link))
actual_target = os.readlink(expected_link)
expected_target = u.storage_path
self.assertEqual(actual_target, expected_target)
#
# Now test flipping so https is disabled and http is enabled
#
config = distributor_mocks.get_basic_config(https_publish_dir=self.https_publish_dir,
http_publish_dir=self.http_publish_dir, relative_url=relative_url, http=True, https=False)
report = distributor.publish_repo(repo, publish_conduit, config)
self.assertTrue(report.success_flag)
# Verify we did publish to http
self.assertTrue(os.path.exists(expected_repo_http_publish_dir))
# Verify we did not publish to https
self.assertFalse(os.path.exists(expected_repo_https_publish_dir))
# Verify we cleaned up the misc dirs under the https dir
self.assertEquals(len(os.listdir(self.https_publish_dir)), 0)
开发者ID:ehelms,项目名称:pulp,代码行数:60,代码来源:test_distributor.py
示例5: test_basic_repo_publish_rel_path_conflict
def test_basic_repo_publish_rel_path_conflict(self):
repo = mock.Mock(spec=Repository)
repo.working_dir = self.repo_working_dir
repo.id = "test_basic_repo_publish_rel_path_conflict"
num_units = 10
relative_url = "rel_a/rel_b/rel_a/"
config = distributor_mocks.get_basic_config(https_publish_dir=self.https_publish_dir,
relative_url=relative_url, http=False, https=True)
url_a = relative_url
config_a = PluginCallConfiguration({"relative_url":url_a}, {})
repo_a = RelatedRepository("repo_a_id", [config_a])
config_conduit = mock.Mock(spec=RepoConfigConduit)
conduit_return_cursor = MockCursor([{'repo_id': 'repo_a_id', 'config': {'relative_url': "rel_a/rel_b/rel_a/"}}])
config_conduit.get_repo_distributors_by_relative_url.return_value = conduit_return_cursor
# Simple check of direct conflict of a duplicate - varieties of duplicates are tested via the conduit tests
related_repos = [repo_a]
distributor = YumDistributor()
distributor.process_repo_auth_certificate_bundle = mock.Mock()
status, msg = distributor.validate_config(repo, config, config_conduit)
self.assertFalse(status)
expected_msg = "Relative url '%s' conflicts with existing relative_url of '%s' from repo '%s'" % \
(relative_url, url_a, repo_a.id)
self.assertEqual(expected_msg, msg)
# Ensure this test can handle a large number of repos
"""
开发者ID:bechtoldt,项目名称:pulp_rpm,代码行数:30,代码来源:test_distributor.py
示例6: test_custom_metadata_publish
def test_custom_metadata_publish(self):
distributor = YumDistributor()
repo = self._mock_repo('test-presto-delta-metadata')
repo_units = self._test_drpm_repo_units()
publish_conduit = mock_conduits.repo_publish_conduit(existing_units=repo_units)
config = mock_conduits.plugin_call_config(http_publish_dir=self.content_dir, relative_url='', http=True, https=False)
distributor.publish_repo(repo, publish_conduit, config)
# make sure the metadata unit was published
criteria = UnitAssociationCriteria(type_ids=[TYPE_ID_YUM_REPO_METADATA_FILE])
metadata_units = publish_conduit.get_units(criteria)
self.assertEqual(len(metadata_units), 1)
unit = metadata_units[0]
self.assertEqual(unit.type_id, TYPE_ID_YUM_REPO_METADATA_FILE)
self.assertEqual(unit.unit_key['data_type'], 'prestodelta')
# make sure the file was copied into place
repodata_path = os.path.join(self.content_dir, repo.id, 'repodata')
prestodelta_files = glob.glob(repodata_path + '/*prestodelta*')
self.assertEqual(len(prestodelta_files), 1)
prestodelta_path = os.path.join(repodata_path, prestodelta_files[0])
self.assertTrue(os.path.exists(prestodelta_path))
开发者ID:bechtoldt,项目名称:pulp_rpm,代码行数:28,代码来源:test_custom_metadata.py
示例7: test_yum_plugin_generate_yum_metadata_checksum_from_conduit_sha1_conversion
def test_yum_plugin_generate_yum_metadata_checksum_from_conduit_sha1_conversion(self,
mock_YumMetadataGenerator,
mock_distributor_manager):
repo = mock.Mock(spec=Repository)
repo.working_dir = self.repo_working_dir
repo.id = "test_publish"
num_units = 10
relative_url = "rel_a/rel_b/rel_c/"
existing_units = self.get_units(count=num_units)
publish_conduit = distributor_mocks.get_publish_conduit(type_id="rpm", existing_units=existing_units, pkg_dir=self.pkg_dir)
publish_conduit.repo_id = 'foo'
publish_conduit.distributor_id = TYPE_ID_DISTRIBUTOR_YUM
config = distributor_mocks.get_basic_config(https_publish_dir=self.https_publish_dir, relative_url=relative_url,
http=False, https=True)
distributor = YumDistributor()
distributor.process_repo_auth_certificate_bundle = mock.Mock()
config_conduit = mock.Mock(spec=RepoConfigConduit)
config_conduit.get_repo_distributors_by_relative_url.return_value = MockCursor([])
metadata.generate_yum_metadata(repo.id, repo.working_dir, publish_conduit, config,
repo_scratchpad={'checksum_type': 'sha'})
mock_YumMetadataGenerator.assert_called_with(ANY, checksum_type='sha1',
skip_metadata_types=ANY, is_cancelled=ANY,
group_xml_path=ANY,
updateinfo_xml_path=ANY,
custom_metadata_dict=ANY)
mock_distributor_manager.return_value.update_distributor_config.\
assert_called_with(ANY, ANY, {'checksum_type': 'sha1'})
开发者ID:bechtoldt,项目名称:pulp_rpm,代码行数:27,代码来源:test_distributor.py
示例8: setUp
def setUp(self):
super(TestValidateConfig, self).setUp()
self.repo = mock.Mock(spec=Repository)
self.repo.id = "testrepo"
self.distributor = YumDistributor()
self.distributor.process_repo_auth_certificate_bundle = mock.Mock()
self.init()
开发者ID:ehelms,项目名称:pulp_rpm,代码行数:7,代码来源:test_distributor_validate_config.py
示例9: test_publish_progress
def test_publish_progress(self):
global progress_status
progress_status = None
def set_progress(progress):
global progress_status
progress_status = progress
PROGRESS_FIELDS = ["num_success", "num_error", "items_left", "items_total", "error_details"]
publish_conduit = distributor_mocks.get_publish_conduit(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,
relative_url="rel_temp/",
generate_metadata=True,
http=True,
https=False,
)
distributor = YumDistributor()
repo = mock.Mock(spec=Repository)
repo.working_dir = self.repo_working_dir
repo.id = "test_progress_sync"
publish_conduit.set_progress = mock.Mock()
publish_conduit.set_progress.side_effect = set_progress
distributor.publish_repo(repo, publish_conduit, config)
self.assertTrue(progress_status is not None)
self.assertTrue("packages" in progress_status)
self.assertTrue(progress_status["packages"].has_key("state"))
self.assertEqual(progress_status["packages"]["state"], "FINISHED")
for field in PROGRESS_FIELDS:
self.assertTrue(field in progress_status["packages"])
self.assertTrue("distribution" in progress_status)
self.assertTrue(progress_status["distribution"].has_key("state"))
self.assertEqual(progress_status["distribution"]["state"], "FINISHED")
for field in PROGRESS_FIELDS:
self.assertTrue(field in progress_status["distribution"])
self.assertTrue("metadata" in progress_status)
self.assertTrue(progress_status["metadata"].has_key("state"))
self.assertEqual(progress_status["metadata"]["state"], "FINISHED")
self.assertTrue("publish_http" in progress_status)
self.assertEqual(progress_status["publish_http"]["state"], "FINISHED")
self.assertTrue("publish_https" in progress_status)
self.assertEqual(progress_status["publish_https"]["state"], "SKIPPED")
开发者ID:ehelms,项目名称:pulp_rpm,代码行数:47,代码来源:test_distributor.py
示例10: TestPublishThread
class TestPublishThread(threading.Thread):
def __init__(self, working_dir, pkg_dir, config):
threading.Thread.__init__(self)
self.repo = mock.Mock(spec=Repository)
self.repo.working_dir = working_dir
self.repo.id = "test_cancel_publish"
self.publish_conduit = distributor_mocks.get_publish_conduit(pkg_dir=pkg_dir)
self.publish_conduit.set_progress = mock.Mock()
self.publish_conduit.set_progress.side_effect = set_progress
self.config = config
self.distributor = YumDistributor()
def run(self):
self.distributor.publish_repo(self.repo, self.publish_conduit, self.config)
def cancel(self):
return self.distributor.cancel_publish_repo(None, None)
开发者ID:ehelms,项目名称:pulp,代码行数:17,代码来源:test_distributor.py
示例11: __init__
def __init__(self, working_dir, pkg_dir, config):
threading.Thread.__init__(self)
self.repo = mock.Mock(spec=Repository)
self.repo.working_dir = working_dir
self.repo.id = "test_cancel_publish"
self.publish_conduit = distributor_mocks.get_publish_conduit(pkg_dir=pkg_dir)
self.publish_conduit.set_progress = mock.Mock()
self.publish_conduit.set_progress.side_effect = set_progress
self.config = config
self.distributor = YumDistributor()
开发者ID:ehelms,项目名称:pulp,代码行数:10,代码来源:test_distributor.py
示例12: test_consumer_payload
def test_consumer_payload(self):
PAYLOAD_FIELDS = [ 'server_name', 'relative_path',
'protocols', 'gpg_keys', 'client_cert', 'ca_cert', 'repo_name']
http = True
https = True
relative_url = "/pub/content/"
gpgkey = ["test_gpg_key",]
auth_cert = open(os.path.join(self.data_dir, "cert.crt")).read()
auth_ca = open(os.path.join(self.data_dir, "ca.key")).read()
config = distributor_mocks.get_basic_config(relative_url=relative_url, http=http, https=https, auth_cert=auth_cert, auth_ca=auth_ca, gpgkey=gpgkey)
distributor = YumDistributor()
repo = mock.Mock(spec=Repository)
repo.working_dir = self.repo_working_dir
repo.id = "test_payload"
repo.display_name = 'Nice Repo'
payload = distributor.create_consumer_payload(repo, config, None)
for field in PAYLOAD_FIELDS:
self.assertTrue(field in payload)
self.assertTrue('http' in payload['protocols'])
self.assertTrue('https' in payload['protocols'])
开发者ID:bechtoldt,项目名称:pulp_rpm,代码行数:21,代码来源:test_distributor.py
示例13: test_empty_publish
def test_empty_publish(self):
repo = mock.Mock(spec=Repository)
repo.working_dir = self.repo_working_dir
repo.id = "test_empty_publish"
existing_units = []
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,
http=True, https=True)
distributor = YumDistributor()
report = distributor.publish_repo(repo, publish_conduit, config)
self.assertTrue(report.success_flag)
summary = report.summary
self.assertEqual(summary["num_package_units_attempted"], 0)
self.assertEqual(summary["num_package_units_published"], 0)
self.assertEqual(summary["num_package_units_errors"], 0)
expected_repo_https_publish_dir = os.path.join(self.https_publish_dir, repo.id).rstrip('/')
expected_repo_http_publish_dir = os.path.join(self.http_publish_dir, repo.id).rstrip('/')
self.assertEqual(summary["https_publish_dir"], expected_repo_https_publish_dir)
self.assertEqual(summary["http_publish_dir"], expected_repo_http_publish_dir)
details = report.details
self.assertEqual(len(details["errors"]), 0)
开发者ID:ehelms,项目名称:pulp,代码行数:21,代码来源:test_distributor.py
示例14: test_handle_symlinks
def test_handle_symlinks(self):
distributor = YumDistributor()
units = []
symlink_dir = os.path.join(self.temp_dir, "symlinks")
num_links = 5
for index in range(0,num_links):
relpath = "file_%s.rpm" % (index)
sp = os.path.join(self.pkg_dir, relpath)
open(sp, "a") # Create an empty file
if index % 2 == 0:
# Ensure we can support symlinks in subdirs
relpath = os.path.join("a", "b", "c", relpath)
u = Unit("rpm", "unit_key_%s" % (index), {"relativepath":relpath}, sp)
units.append(u)
status, errors = distributor.handle_symlinks(units, symlink_dir)
self.assertTrue(status)
self.assertEqual(len(errors), 0)
for u in units:
symlink_path = os.path.join(symlink_dir, u.metadata["relativepath"])
self.assertTrue(os.path.exists(symlink_path))
self.assertTrue(os.path.islink(symlink_path))
target = os.readlink(symlink_path)
self.assertEqual(target, u.storage_path)
# Test republish is successful
status, errors = distributor.handle_symlinks(units, symlink_dir)
self.assertTrue(status)
self.assertEqual(len(errors), 0)
for u in units:
symlink_path = os.path.join(symlink_dir, u.metadata["relativepath"])
self.assertTrue(os.path.exists(symlink_path))
self.assertTrue(os.path.islink(symlink_path))
target = os.readlink(symlink_path)
self.assertEqual(target, u.storage_path)
# Simulate a package is deleted
os.unlink(units[0].storage_path)
status, errors = distributor.handle_symlinks(units, symlink_dir)
self.assertFalse(status)
self.assertEqual(len(errors), 1)
开发者ID:ehelms,项目名称:pulp,代码行数:39,代码来源:test_distributor.py
示例15: test_split_path
def test_split_path(self):
distributor = YumDistributor()
test_path = "/a"
pieces = distributor.split_path(test_path)
self.assertEqual(len(pieces), 1)
self.assertTrue(pieces[0], test_path)
test_path = "/a/"
pieces = distributor.split_path(test_path)
self.assertEqual(len(pieces), 1)
self.assertTrue(pieces[0], test_path)
test_path = "/a"
pieces = distributor.split_path(test_path)
self.assertEqual(len(pieces), 1)
self.assertTrue(pieces[0], test_path)
test_path = "a/"
pieces = distributor.split_path(test_path)
self.assertEqual(len(pieces), 1)
self.assertTrue(pieces[0], test_path)
test_path = "/a/bcde/f/ghi/j"
pieces = distributor.split_path(test_path)
self.assertEqual(len(pieces), 5)
self.assertTrue(os.path.join(*pieces), test_path)
test_path = "a/bcde/f/ghi/j"
pieces = distributor.split_path(test_path)
self.assertEqual(len(pieces), 5)
self.assertTrue(os.path.join(*pieces), test_path)
test_path = "a/bcde/f/ghi/j/"
pieces = distributor.split_path(test_path)
self.assertEqual(len(pieces), 5)
self.assertTrue(os.path.join(*pieces), test_path)
test_path = "/a/bcde/f/ghi/j/"
pieces = distributor.split_path(test_path)
self.assertEqual(len(pieces), 5)
self.assertTrue(os.path.join(*pieces), test_path)
开发者ID:ehelms,项目名称:pulp,代码行数:41,代码来源:test_distributor.py
示例16: test_validate_config
def test_validate_config(self):
repo = mock.Mock(spec=Repository)
repo.id = "testrepo"
distributor = YumDistributor()
distributor.process_repo_auth_certificate_bundle = mock.Mock()
# Confirm that required keys are successful
req_kwargs = {}
req_kwargs['http'] = True
req_kwargs['https'] = False
req_kwargs['relative_url'] = "sample_value"
config = distributor_mocks.get_basic_config(**req_kwargs)
state, msg = distributor.validate_config(repo, config, [])
self.assertTrue(state)
# Confirm required and optional are successful
optional_kwargs = dict(req_kwargs)
optional_kwargs['auth_ca'] = open(os.path.join(self.data_dir, "valid_ca.crt")).read()
optional_kwargs['https_ca'] = open(os.path.join(self.data_dir, "valid_ca.crt")).read()
optional_kwargs['protected'] = True
optional_kwargs['generate_metadata'] = True
optional_kwargs['checksum_type'] = "sha"
optional_kwargs['skip'] = []
optional_kwargs['auth_cert'] = open(os.path.join(self.data_dir, "cert.crt")).read()
config = distributor_mocks.get_basic_config(**optional_kwargs)
state, msg = distributor.validate_config(repo, config, [])
self.assertTrue(state)
# Test that config fails when a bad value for non_existing_dir is used
optional_kwargs["http_publish_dir"] = "non_existing_dir"
config = distributor_mocks.get_basic_config(**optional_kwargs)
state, msg = distributor.validate_config(repo, config, [])
self.assertFalse(state)
# Test config succeeds with a good value of https_publish_dir
optional_kwargs["http_publish_dir"] = self.temp_dir
config = distributor_mocks.get_basic_config(**optional_kwargs)
state, msg = distributor.validate_config(repo, config, [])
self.assertTrue(state)
del optional_kwargs["http_publish_dir"]
# Test that config fails when a bad value for non_existing_dir is used
optional_kwargs["https_publish_dir"] = "non_existing_dir"
config = distributor_mocks.get_basic_config(**optional_kwargs)
state, msg = distributor.validate_config(repo, config, [])
self.assertFalse(state)
# Test config succeeds with a good value of https_publish_dir
optional_kwargs["https_publish_dir"] = self.temp_dir
config = distributor_mocks.get_basic_config(**optional_kwargs)
state, msg = distributor.validate_config(repo, config, [])
self.assertTrue(state)
del optional_kwargs["https_publish_dir"]
# Confirm an extra key fails
optional_kwargs["extra_arg_not_used"] = "sample_value"
config = distributor_mocks.get_basic_config(**optional_kwargs)
state, msg = distributor.validate_config(repo, config, [])
self.assertFalse(state)
self.assertTrue("extra_arg_not_used" in msg)
# Confirm missing a required fails
del optional_kwargs["extra_arg_not_used"]
config = distributor_mocks.get_basic_config(**optional_kwargs)
state, msg = distributor.validate_config(repo, config, [])
self.assertTrue(state)
del optional_kwargs["relative_url"]
config = distributor_mocks.get_basic_config(**optional_kwargs)
state, msg = distributor.validate_config(repo, config, [])
self.assertFalse(state)
self.assertTrue("relative_url" in msg)
开发者ID:ehelms,项目名称:pulp,代码行数:66,代码来源:test_distributor.py
示例17: test_metadata
def test_metadata(self):
metadata = YumDistributor.metadata()
self.assertEquals(metadata["id"], TYPE_ID_DISTRIBUTOR_YUM)
self.assertTrue(TYPE_ID_RPM in metadata["types"])
self.assertTrue(TYPE_ID_SRPM in metadata["types"])
开发者ID:ehelms,项目名称:pulp,代码行数:5,代码来源:test_distributor.py
示例18: test_basic_repo_publish_rel_path_conflict
def test_basic_repo_publish_rel_path_conflict(self):
repo = mock.Mock(spec=Repository)
repo.working_dir = self.repo_working_dir
repo.id = "test_basic_repo_publish_rel_path_conflict"
num_units = 10
relative_url = "rel_a/rel_b/rel_a/"
config = distributor_mocks.get_basic_config(https_publish_dir=self.https_publish_dir,
relative_url=relative_url, http=False, https=True)
url_a = relative_url
config_a = PluginCallConfiguration({"relative_url":url_a}, {})
repo_a = RelatedRepository("repo_a_id", [config_a])
# Simple check of direct conflict of a duplicate
related_repos = [repo_a]
distributor = YumDistributor()
distributor.process_repo_auth_certificate_bundle = mock.Mock()
status, msg = distributor.validate_config(repo, config, related_repos)
self.assertFalse(status)
expected_msg = "Relative url '%s' conflicts with existing relative_url of '%s' from repo '%s'" % (relative_url, url_a, repo_a.id)
self.assertEqual(expected_msg, msg)
# Check conflict with a subdir
url_b = "rel_a/rel_b/"
config_b = PluginCallConfiguration({"relative_url":url_b}, {})
repo_b = RelatedRepository("repo_b_id", [config_b])
related_repos = [repo_b]
distributor = YumDistributor()
distributor.process_repo_auth_certificate_bundle = mock.Mock()
status, msg = distributor.validate_config(repo, config, related_repos)
self.assertFalse(status)
expected_msg = "Relative url '%s' conflicts with existing relative_url of '%s' from repo '%s'" % (relative_url, url_b, repo_b.id)
self.assertEqual(expected_msg, msg)
# Check no conflict with a pieces of a common subdir
url_c = "rel_a/rel_b/rel_c"
config_c = PluginCallConfiguration({"relative_url":url_c}, {})
repo_c = RelatedRepository("repo_c_id", [config_c])
url_d = "rel_a/rel_b/rel_d"
config_d = PluginCallConfiguration({"relative_url":url_d}, {})
repo_d = RelatedRepository("repo_d_id", [config_d])
url_e = "rel_a/rel_b/rel_e/rel_e"
config_e = PluginCallConfiguration({"relative_url":url_e}, {})
repo_e = RelatedRepository("repo_e_id", [config_e])
# Add a repo with no relative_url
config_f = PluginCallConfiguration({"relative_url":None}, {})
repo_f = RelatedRepository("repo_f_id", [config_f])
related_repos = [repo_c, repo_d, repo_e, repo_f]
distributor = YumDistributor()
distributor.process_repo_auth_certificate_bundle = mock.Mock()
status, msg = distributor.validate_config(repo, config, related_repos)
self.assertTrue(status)
self.assertEqual(msg, None)
# Test with 2 repos and no relative_url
config_h = PluginCallConfiguration({}, {})
repo_h = RelatedRepository("repo_h_id", [config_h])
config_i = PluginCallConfiguration({}, {})
repo_i = RelatedRepository("repo_i_id", [config_i])
status, msg = distributor.validate_config(repo_i, config, [repo_h])
self.assertTrue(status)
self.assertEqual(msg, None)
# TODO: Test, repo_1 has no rel url, so repo_1_id is used
# Then 2nd repo is configured with rel_url of repo_1_id
# should result in a conflict
# Ensure this test can handle a large number of repos
test_repos = []
for index in range(0,10000):
test_url = "rel_a/rel_b/rel_e/repo_%s" % (index)
test_config = PluginCallConfiguration({"relative_url":test_url}, {})
r = RelatedRepository("repo_%s_id" % (index), [test_config])
test_repos.append(r)
related_repos = test_repos
distributor = YumDistributor()
distributor.process_repo_auth_certificate_bundle = mock.Mock()
status, msg = distributor.validate_config(repo, config, related_repos)
self.assertTrue(status)
self.assertEqual(msg, None)
开发者ID:ehelms,项目名称:pulp,代码行数:89,代码来源:test_distributor.py
示例19: test_form_rel_url_lookup_table
def test_form_rel_url_lookup_table(self):
distributor = YumDistributor()
existing_urls = distributor.form_rel_url_lookup_table(None)
self.assertEqual(existing_urls, {})
url_a = "/abc/de/fg/"
config_a = PluginCallConfiguration({"relative_url":url_a}, {})
repo_a = RelatedRepository("repo_a_id", [config_a])
conflict_url_a = "/abc/de/"
conflict_config_a = PluginCallConfiguration({"relative_url":conflict_url_a}, {})
conflict_repo_a = RelatedRepository("conflict_repo_id_a", [conflict_config_a])
url_b = "/abc/de/kj/"
config_b = PluginCallConfiguration({"relative_url":url_b}, {})
repo_b = RelatedRepository("repo_b_id", [config_b])
repo_b_dup = RelatedRepository("repo_b_dup_id", [config_b])
url_c = "/abc/jk/fg/gfgf/gfgf/gfre/"
config_c = PluginCallConfiguration({"relative_url":url_c}, {})
repo_c = RelatedRepository("repo_c_id", [config_c])
url_d = "simple"
config_d = PluginCallConfiguration({"relative_url":url_d}, {})
repo_d = RelatedRepository("repo_d_id", [config_d])
url_e = ""
config_e = PluginCallConfiguration({"relative_url":url_e}, {})
repo_e = RelatedRepository("repo_e_id", [config_e])
url_f = "/foo"
config_f = PluginCallConfiguration({"relative_url":url_f}, {})
repo_f = RelatedRepository("repo_f_id", [config_f])
conflict_url_f = "foo/"
conflict_config_f = PluginCallConfiguration({"relative_url":conflict_url_f}, {})
conflict_repo_f = RelatedRepository("conflict_repo_f_id", [conflict_config_f])
url_g = "bar/"
config_g = PluginCallConfiguration({"relative_url":url_g}, {})
repo_g = RelatedRepository("repo_g_id", [config_g])
# Try with url set to None
url_h = None
config_h = PluginCallConfiguration({"relative_url":url_h}, {})
repo_h = RelatedRepository("repo_h_id", [config_h])
# Try with relative_url not existing
config_i = PluginCallConfiguration({}, {})
repo_i = RelatedRepository("repo_i_id", [config_i])
existing_urls = distributor.form_rel_url_lookup_table([repo_a, repo_d, repo_e, repo_f, repo_g, repo_h])
self.assertEqual(existing_urls, {'simple': {'repo_id': repo_d.id, 'url': url_d},
'abc': {'de': {'fg': {'repo_id': repo_a.id, 'url': url_a}}},
repo_e.id : {'repo_id': repo_e.id, 'url': repo_e.id}, # url_e is empty so we default to use repo id
repo_h.id : {'repo_id': repo_h.id, 'url': repo_h.id}, # urk_h is None so we default to use repo id
'bar': {'repo_id': repo_g.id, 'url':url_g}, 'foo': {'repo_id': repo_f.id, 'url': url_f}})
existing_urls = distributor.form_rel_url_lookup_table([repo_a])
self.assertEqual(existing_urls, {'abc': {'de': {'fg': {'repo_id': repo_a.id, 'url': url_a}}}})
existing_urls = distributor.form_rel_url_lookup_table([repo_a, repo_b])
self.assertEqual(existing_urls, {'abc': {'de': {'kj': {'repo_id': repo_b.id, 'url': url_b}, 'fg': {'repo_id': repo_a.id, 'url': url_a}}}})
existing_urls = distributor.form_rel_url_lookup_table([repo_a, repo_b, repo_c])
self.assertEqual(existing_urls, {'abc': {'de': {'kj': {'repo_id': repo_b.id, 'url':url_b},
'fg': {'repo_id': repo_a.id, 'url':url_a}}, 'jk': {'fg': {'gfgf': {'gfgf': {'gfre': {'repo_id': repo_c.id, 'url':url_c}}}}}}})
# Add test for exception on duplicate with repos passed in
caught = False
try:
existing_urls = distributor.form_rel_url_lookup_table([repo_a, repo_b, repo_b_dup, repo_c])
except Exception, e:
caught = True
开发者ID:ehelms,项目名称:pulp,代码行数:74,代码来源:test_distributor.py
示例20: TestValidateConfig
class TestValidateConfig(rpm_support_base.PulpRPMTests):
def setUp(self):
super(TestValidateConfig, self).setUp()
self.repo = mock.Mock(spec=Repository)
self.repo.id = "testrepo"
self.distributor = YumDistributor()
self.distributor.process_repo_auth_certificate_bundle = mock.Mock()
self.init()
def tearDown(self):
super(TestValidateConfig, self).tearDown()
def init(self):
self.data_dir = os.path.abspath(os.path.join(os.path.abspath(os.path.dirname(__file__)), "./data"))
def test_config_relative_path(self):
http = True
https = False
relative_url = 123
config = distributor_mocks.get_basic_config(relative_url=relative_url, http=http, https=https)
state, msg = self.distributor
|
请发表评论