Python config.set函数代码示例

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Python›Python编程经验

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了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；未经允许，请勿转载。