本文整理汇总了Python中pulp.server.config.config.set函数的典型用法代码示例。如果您正苦于以下问题：Python set函数的具体用法？Python set怎么用？Python set使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了set函数的16个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_import
def test_import(self):
# Setup
self.populate()
pulp_conf.set('server', 'storage_dir', self.parentfs)
dist = NodesHttpDistributor()
repo = Repository(self.REPO_ID)
cfg = {
'protocol':'file',
'http':{'alias':self.alias},
'https':{'alias':self.alias},
'file':{'alias':self.alias},
}
conduit = RepoPublishConduit(self.REPO_ID, constants.HTTP_DISTRIBUTOR)
dist.publish_repo(repo, conduit, cfg)
Repo.get_collection().remove()
RepoDistributor.get_collection().remove()
RepoContentUnit.get_collection().remove()
unit_db.clean()
# Test
importer = NodesHttpImporter()
publisher = dist.publisher(repo, cfg)
manifest_url = 'file://' + publisher.manifest_path()
cfg = dict(manifest_url=manifest_url, strategy=constants.MIRROR_STRATEGY)
conduit = RepoSyncConduit(
self.REPO_ID,
constants.HTTP_IMPORTER,
RepoContentUnit.OWNER_TYPE_IMPORTER,
constants.HTTP_IMPORTER)
importer.sync_repo(repo, conduit, cfg)
# Verify
units = conduit.get_units()
self.assertEquals(len(units), self.NUM_UNITS)
开发者ID:bartwo，项目名称:pulp，代码行数:32，代码来源:test_plugins.py
示例2: test_publish
def test_publish(self):
# Setup
self.populate()
pulp_conf.set('server', 'storage_dir', self.parentfs)
# Test
dist = NodesHttpDistributor()
repo = Repository(self.REPO_ID)
conduit = RepoPublishConduit(self.REPO_ID, constants.HTTP_DISTRIBUTOR)
dist.publish_repo(repo, conduit, self.dist_conf())
# Verify
conf = DownloaderConfig()
downloader = LocalFileDownloader(conf)
pub = dist.publisher(repo, self.dist_conf())
url = pathlib.url_join(pub.base_url, pub.manifest_path())
working_dir = self.childfs
manifest = RemoteManifest(url, downloader, working_dir)
manifest.fetch()
manifest.fetch_units()
units = [u for u, r in manifest.get_units()]
self.assertEqual(len(units), self.NUM_UNITS)
for n in range(0, self.NUM_UNITS):
unit = units[n]
created = self.units[n]
for p, v in unit['unit_key'].items():
self.assertEqual(created[p], v)
for p, v in unit['metadata'].items():
if p in ('_ns', '_content_type_id'):
continue
self.assertEqual(created[p], v)
self.assertEqual(created.get('_storage_path'), unit['storage_path'])
self.assertEqual(unit['type_id'], self.UNIT_TYPE_ID)
开发者ID:ipanova，项目名称:pulp，代码行数:31，代码来源:test_plugins.py
示例3: test_payload_with_ssl
def test_payload_with_ssl(self):
# Setup
self.populate()
pulp_conf.set('server', 'storage_dir', self.parentfs)
# Test
dist = NodesHttpDistributor()
repo = Repository(self.REPO_ID)
payload = dist.create_consumer_payload(repo, self.dist_conf_with_ssl(), {})
开发者ID:bartwo，项目名称:pulp，代码行数:8，代码来源:test_plugins.py
示例4: test_payload
def test_payload(self):
# Setup
self.populate()
pulp_conf.set('server', 'storage_dir', self.upfs)
# Test
dist = CitrusHttpDistributor()
repo = Repository(self.REPO_ID)
payload = dist.create_consumer_payload(repo, self.dist_conf())
开发者ID:pieska，项目名称:pulp，代码行数:8，代码来源:test_plugins.py
示例5: populate
def populate(self):
# make content/ dir.
os.makedirs(os.path.join(self.parentfs, 'content'))
pulp_conf.set('server', 'storage_dir', self.parentfs)
# create repo
manager = managers.repo_manager()
manager.create_repo(
self.REPO_ID, display_name=REPO_NAME, description=REPO_DESCRIPTION, notes=REPO_NOTES)
manager.set_repo_scratchpad(self.REPO_ID, REPO_SCRATCHPAD)
# add units
units = self.add_units(0, self.NUM_UNITS)
self.units = units
开发者ID:ipanova，项目名称:pulp，代码行数:12，代码来源:test_plugins.py
示例6: test_handler
def test_handler(*unused):
# publish
self.populate(constants.MIRROR_STRATEGY, ssl=True)
pulp_conf.set('server', 'storage_dir', self.parentfs)
dist = NodesHttpDistributor()
repo = Repository(self.REPO_ID)
conduit = RepoPublishConduit(self.REPO_ID, constants.HTTP_DISTRIBUTOR)
dist.publish_repo(repo, conduit, self.dist_conf())
units = []
handler = NodeHandler(self)
pulp_conf.set('server', 'storage_dir', self.childfs)
report = handler.update(Conduit(), units, {})
_report.append(report)
开发者ID:bartwo，项目名称:pulp，代码行数:13，代码来源:test_plugins.py
示例7: test_handler
def test_handler(*unused):
# publish
self.populate(ssl=True)
pulp_conf.set('server', 'storage_dir', self.upfs)
dist = CitrusHttpDistributor()
repo = Repository(self.REPO_ID)
conduit = RepoPublishConduit(self.REPO_ID, CITRUS_DISTRUBUTOR)
dist.publish_repo(repo, conduit, self.dist_conf())
units = []
options = {}
handler = NodeHandler(self)
pulp_conf.set('server', 'storage_dir', self.downfs)
report = handler.update(Conduit(), units, options)
_report.append(report)
开发者ID:pieska，项目名称:pulp，代码行数:14，代码来源:test_plugins.py
示例8: test_handler
def test_handler(*unused):
# publish
self.populate(constants.ADDITIVE_STRATEGY)
pulp_conf.set('server', 'storage_dir', self.parentfs)
dist = NodesHttpDistributor()
repo = Repository(self.REPO_ID)
conduit = RepoPublishConduit(self.REPO_ID, constants.HTTP_DISTRIBUTOR)
dist.publish_repo(repo, conduit, self.dist_conf())
units = []
options = dict(strategy=constants.MIRROR_STRATEGY)
handler = NodeHandler(self)
pulp_conf.set('server', 'storage_dir', self.childfs)
os.makedirs(os.path.join(self.childfs, 'content'))
report = handler.update(Conduit(), units, options)
_report.append(report)
开发者ID:juwu，项目名称:pulp，代码行数:15，代码来源:test_plugins.py
示例9: populate
def populate(self):
# make content/ dir.
os.makedirs(os.path.join(self.parentfs, 'content'))
pulp_conf.set('server', 'storage_dir', self.parentfs)
# create repo
manager = managers.repo_manager()
manager.create_repo(self.REPO_ID)
# add units
units = []
for n in range(0, self.NUM_UNITS):
unit_id = self.UNIT_ID % n
unit = dict(self.UNIT_METADATA)
unit['N'] = n
# add unit file
storage_dir = pulp_conf.get('server', 'storage_dir')
storage_path = \
os.path.join(storage_dir, 'content',
'.'.join((unit_id, self.UNIT_TYPE_ID)))
unit['_storage_path'] = storage_path
fp = open(storage_path, 'w+')
fp.write(unit_id)
fp.close()
# add unit
manager = managers.content_manager()
manager.add_content_unit(
self.UNIT_TYPE_ID,
unit_id,
unit)
manager = managers.repo_unit_association_manager()
# associate unit
manager.associate_unit_by_id(
self.REPO_ID,
self.UNIT_TYPE_ID,
unit_id,
RepoContentUnit.OWNER_TYPE_IMPORTER,
constants.HTTP_IMPORTER)
units.append(unit)
# CA
self.units = units
path = os.path.join(self.parentfs, 'ca.crt')
fp = open(path, 'w+')
fp.write(self.CA_CERT)
fp.close()
# client cert
path = os.path.join(self.parentfs, 'local.crt')
fp = open(path, 'w+')
fp.write(self.CLIENT_CERT)
fp.close()
开发者ID:bartwo，项目名称:pulp，代码行数:48，代码来源:test_plugins.py
示例10: test_import_unit_files_already_exist_size_mismatch
def test_import_unit_files_already_exist_size_mismatch(self, *mocks):
# Setup
self.populate()
pulp_conf.set('server', 'storage_dir', self.parentfs)
dist = NodesHttpDistributor()
working_dir = os.path.join(self.childfs, 'working_dir')
os.makedirs(working_dir)
repo = Repository(self.REPO_ID, working_dir)
cfg = self.dist_conf()
conduit = RepoPublishConduit(self.REPO_ID, constants.HTTP_DISTRIBUTOR)
dist.publish_repo(repo, conduit, cfg)
Repo.get_collection().remove()
RepoDistributor.get_collection().remove()
RepoContentUnit.get_collection().remove()
unit_db.clean()
self.define_plugins()
parent_content = os.path.join(self.parentfs, 'content')
child_content = os.path.join(self.childfs, 'content')
shutil.copytree(parent_content, child_content)
for fn in os.listdir(child_content):
path = os.path.join(child_content, fn)
if os.path.isdir(path):
continue
with open(path, 'w') as fp:
fp.truncate()
# Test
importer = NodesHttpImporter()
publisher = dist.publisher(repo, cfg)
manifest_url = pathlib.url_join(publisher.base_url, publisher.manifest_path())
configuration = {
constants.MANIFEST_URL_KEYWORD: manifest_url,
constants.STRATEGY_KEYWORD: constants.MIRROR_STRATEGY,
}
configuration = PluginCallConfiguration(configuration, {})
conduit = RepoSyncConduit(
self.REPO_ID,
constants.HTTP_IMPORTER,
RepoContentUnit.OWNER_TYPE_IMPORTER,
constants.HTTP_IMPORTER)
pulp_conf.set('server', 'storage_dir', self.childfs)
importer.sync_repo(repo, conduit, configuration)
# Verify
units = conduit.get_units()
self.assertEquals(len(units), self.NUM_UNITS)
mock_importer_config_to_nectar_config = mocks[0]
mock_importer_config_to_nectar_config.assert_called_with(configuration.flatten())
开发者ID:ipanova，项目名称:pulp，代码行数:46，代码来源:test_plugins.py
示例11: test_update_units
def test_update_units(self, mock_read_config):
# Setup
mock_read_config.return_value = self.node_configuration()
# Test
host = 'abc'
port = 443
units = [1, 2, 3]
options = {}
p = NodeProfiler()
pulp_conf.set('server', 'server_name', host)
_units = p.update_units(None, units, options, None, None)
# Verify
self.assertTrue(constants.PARENT_SETTINGS in options)
settings = options[constants.PARENT_SETTINGS]
self.assertEqual(settings[constants.HOST], host)
self.assertEqual(settings[constants.PORT], port)
self.assertEqual(settings[constants.NODE_CERTIFICATE], NODE_CERTIFICATE)
self.assertEqual(units, _units)
开发者ID:CUXIDUMDUM，项目名称:pulp，代码行数:18，代码来源:test_plugins.py
示例12: test_import_cached_manifest_matched
def test_import_cached_manifest_matched(self, mock_fetch, *unused):
# Setup
self.populate()
pulp_conf.set('server', 'storage_dir', self.parentfs)
dist = NodesHttpDistributor()
working_dir = os.path.join(self.childfs, 'working_dir')
os.makedirs(working_dir)
repo = Repository(self.REPO_ID, working_dir)
configuration = self.dist_conf()
conduit = RepoPublishConduit(self.REPO_ID, constants.HTTP_DISTRIBUTOR)
dist.publish_repo(repo, conduit, configuration)
Repo.get_collection().remove()
RepoDistributor.get_collection().remove()
RepoContentUnit.get_collection().remove()
unit_db.clean()
self.define_plugins()
publisher = dist.publisher(repo, configuration)
manifest_path = publisher.manifest_path()
units_path = os.path.join(os.path.dirname(manifest_path), UNITS_FILE_NAME)
manifest = Manifest(manifest_path)
manifest.read()
shutil.copy(manifest_path, os.path.join(working_dir, MANIFEST_FILE_NAME))
shutil.copy(units_path, os.path.join(working_dir, UNITS_FILE_NAME))
# Test
importer = NodesHttpImporter()
manifest_url = pathlib.url_join(publisher.base_url, manifest_path)
configuration = {
constants.MANIFEST_URL_KEYWORD: manifest_url,
constants.STRATEGY_KEYWORD: constants.MIRROR_STRATEGY,
}
configuration = PluginCallConfiguration(configuration, {})
conduit = RepoSyncConduit(
self.REPO_ID,
constants.HTTP_IMPORTER,
RepoContentUnit.OWNER_TYPE_IMPORTER,
constants.HTTP_IMPORTER)
pulp_conf.set('server', 'storage_dir', self.childfs)
importer.sync_repo(repo, conduit, configuration)
# Verify
units = conduit.get_units()
self.assertEquals(len(units), self.NUM_UNITS)
self.assertFalse(mock_fetch.called)
开发者ID:ipanova，项目名称:pulp，代码行数:42，代码来源:test_plugins.py
示例13: test_import
def test_import(self, *mocks):
# Setup
self.populate()
max_concurrency = 5
max_bandwidth = 12345
pulp_conf.set('server', 'storage_dir', self.parentfs)
dist = NodesHttpDistributor()
working_dir = os.path.join(self.childfs, 'working_dir')
os.makedirs(working_dir)
repo = Repository(self.REPO_ID, working_dir)
cfg = self.dist_conf()
conduit = RepoPublishConduit(self.REPO_ID, constants.HTTP_DISTRIBUTOR)
dist.publish_repo(repo, conduit, cfg)
Repo.get_collection().remove()
RepoDistributor.get_collection().remove()
RepoContentUnit.get_collection().remove()
unit_db.clean()
self.define_plugins()
# Test
importer = NodesHttpImporter()
publisher = dist.publisher(repo, cfg)
manifest_url = pathlib.url_join(publisher.base_url, publisher.manifest_path())
configuration = {
constants.MANIFEST_URL_KEYWORD: manifest_url,
constants.STRATEGY_KEYWORD: constants.MIRROR_STRATEGY,
importer_constants.KEY_MAX_DOWNLOADS: max_concurrency,
importer_constants.KEY_MAX_SPEED: max_bandwidth,
}
configuration = PluginCallConfiguration(configuration, {})
conduit = RepoSyncConduit(
self.REPO_ID,
constants.HTTP_IMPORTER,
RepoContentUnit.OWNER_TYPE_IMPORTER,
constants.HTTP_IMPORTER)
pulp_conf.set('server', 'storage_dir', self.childfs)
importer.sync_repo(repo, conduit, configuration)
# Verify
units = conduit.get_units()
self.assertEquals(len(units), self.NUM_UNITS)
mock_importer_config_to_nectar_config = mocks[0]
mock_importer_config_to_nectar_config.assert_called_with(configuration.flatten())
开发者ID:ipanova，项目名称:pulp，代码行数:41，代码来源:test_plugins.py
示例14: test_import_modified_units
def test_import_modified_units(self, *mocks):
# Setup
self.populate()
max_concurrency = 5
max_bandwidth = 12345
pulp_conf.set('server', 'storage_dir', self.parentfs)
dist = NodesHttpDistributor()
working_dir = os.path.join(self.childfs, 'working_dir')
os.makedirs(working_dir)
repo = Repository(self.REPO_ID, working_dir)
cfg = self.dist_conf()
conduit = RepoPublishConduit(self.REPO_ID, constants.HTTP_DISTRIBUTOR)
dist.publish_repo(repo, conduit, cfg)
# make the published unit have a newer _last_updated.
collection = connection.get_collection(unit_db.unit_collection_name(self.UNIT_TYPE_ID))
unit = collection.find_one({'N': 0})
unit['age'] = 84
unit['_last_updated'] -= 1
collection.update({'N': 0}, unit, safe=True)
# Test
importer = NodesHttpImporter()
publisher = dist.publisher(repo, cfg)
manifest_url = pathlib.url_join(publisher.base_url, publisher.manifest_path())
configuration = {
constants.MANIFEST_URL_KEYWORD: manifest_url,
constants.STRATEGY_KEYWORD: constants.MIRROR_STRATEGY,
importer_constants.KEY_MAX_DOWNLOADS: max_concurrency,
importer_constants.KEY_MAX_SPEED: max_bandwidth,
}
configuration = PluginCallConfiguration(configuration, {})
conduit = RepoSyncConduit(
self.REPO_ID,
constants.HTTP_IMPORTER,
RepoContentUnit.OWNER_TYPE_IMPORTER,
constants.HTTP_IMPORTER)
pulp_conf.set('server', 'storage_dir', self.childfs)
importer.sync_repo(repo, conduit, configuration)
# Verify
unit = collection.find_one({'N': 0})
self.assertEqual(unit['age'], 42)
开发者ID:ipanova，项目名称:pulp，代码行数:40，代码来源:test_plugins.py
示例15: test_payload
def test_payload(self):
# Setup
self.populate()
pulp_conf.set('server', 'storage_dir', self.parentfs)
# Test
dist = NodesHttpDistributor()
repo = Repository(self.REPO_ID)
payload = dist.create_consumer_payload(repo, self.dist_conf(), {})
f = open('/tmp/payload', 'w+')
f.write(repr(payload['importers']))
f.close()
# Verify
distributors = payload['distributors']
importers = payload['importers']
repository = payload['repository']
self.assertTrue(isinstance(distributors, list))
self.assertTrue(isinstance(importers, list))
self.assertTrue(isinstance(repository, dict))
self.assertTrue(len(importers), 1)
for key in ('id', 'importer_type_id', 'config'):
self.assertTrue(key in importers[0])
for key in (constants.MANIFEST_URL_KEYWORD, constants.STRATEGY_KEYWORD):
self.assertTrue(key in importers[0]['config'])
开发者ID:ipanova，项目名称:pulp，代码行数:23，代码来源:test_plugins.py
示例16: test_publish
def test_publish(self):
# Setup
self.populate()
pulp_conf.set('server', 'storage_dir', self.parentfs)
# Test
dist = NodesHttpDistributor()
repo = Repository(self.REPO_ID)
conduit = RepoPublishConduit(self.REPO_ID, constants.HTTP_DISTRIBUTOR)
dist.publish_repo(repo, conduit, self.dist_conf())
# Verify
conf = DownloaderConfig()
downloader = HTTPSCurlDownloader(conf)
manifest = Manifest()
pub = dist.publisher(repo, self.dist_conf())
url = '/'.join((pub.base_url, pub.manifest_path()))
units = list(manifest.read(url, downloader))
self.assertEqual(len(units), self.NUM_UNITS)
for n in range(0, self.NUM_UNITS):
unit = units[n]
created = self.units[n]
for p, v in unit['metadata'].items():
if p.startswith('_'):
continue
self.assertEqual(created[p], v)
开发者ID:bartwo，项目名称:pulp，代码行数:24，代码来源:test_plugins.py
注：本文中的pulp.server.config.config.set函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论