• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号
登陆 注册
描述文字

Python distributor.NodesHttpDistributor类代码示例

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

本文整理汇总了Python中pulp_node.distributors.http.distributor.NodesHttpDistributor类的典型用法代码示例。如果您正苦于以下问题：Python NodesHttpDistributor类的具体用法？Python NodesHttpDistributor怎么用？Python NodesHttpDistributor使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。



在下文中一共展示了NodesHttpDistributor类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: test_publish

    def test_publish(self, mock_repo_ctrl):
        # Setup
        self.populate()

        with mock_config.patch({'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:maxamillion，项目名称:pulp，代码行数:32，代码来源:test_plugins.py


示例2: test_payload

    def test_payload(self, mock_repo_model):
        mock_repo = mock_repo_model.objects.get_repo_or_missing_resource.return_value
        self.populate()

        with mock_config.patch({'server': {'storage_dir': self.parentfs}}):
            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.assertEqual(repository['id'], mock_repo.repo_id)
        self.assertEqual(repository['display_name'], mock_repo.display_name)
        self.assertEqual(repository['description'], mock_repo.description)
        self.assertEqual(repository['notes'], mock_repo.notes)
        self.assertEqual(repository['scratchpad'], mock_repo.scratchpad)
        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:maxamillion，项目名称:pulp，代码行数:28，代码来源:test_plugins.py


示例3: 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


示例4: test_import_cached_manifest_missing_units

 def test_import_cached_manifest_missing_units(self, *unused):
     # Setup
     self.populate()
     with mock_config.patch({'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)
         model.Distributor.objects.delete()
         RepoContentUnit.get_collection().remove()
         unit_db.clean()
         self.define_plugins()
         publisher = dist.publisher(repo, configuration)
         manifest_path = publisher.manifest_path()
         manifest = Manifest(manifest_path)
         manifest.read()
         shutil.copy(manifest_path, os.path.join(working_dir, MANIFEST_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, Mock())
     with mock_config.patch({'server': {'storage_dir': self.childfs}}):
         with patch('pulp_node.constants.CONTENT_PATH', self.parentfs):
             importer.sync_repo(repo, conduit, configuration)
         # Verify
         units = conduit.get_units()
         self.assertEquals(len(units), self.NUM_UNITS)
开发者ID:maxamillion，项目名称:pulp，代码行数:35，代码来源:test_plugins.py


示例5: test_get_publish_dir

    def test_get_publish_dir(self):
        dist = NodesHttpDistributor()
        repo = plugin_model.Repository(self.REPO_ID)
        config = PluginCallConfiguration(self.VALID_CONFIGURATION, {})

        ret = dist._get_publish_dir(repo.id, config)

        self.assertEqual(ret, '/var/www/pulp/nodes/https/repos/%s' % repo.id)
开发者ID:maxamillion，项目名称:pulp，代码行数:8，代码来源:test_plugins.py


示例6: test_distributor_removed

    def test_distributor_removed(self, mock_system):
        dist = NodesHttpDistributor()
        repo = plugin_model.Repository(self.REPO_ID)
        config = PluginCallConfiguration(self.VALID_CONFIGURATION, {})

        dist.distributor_removed(repo, config)

        mock_system.assert_called_once_with('rm -rf /var/www/pulp/nodes/https/repos/%s' % repo.id)
开发者ID:maxamillion，项目名称:pulp，代码行数:8，代码来源:test_plugins.py


示例7: 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


示例8: test_valid_config

 def test_valid_config(self):
     # Test
     dist = NodesHttpDistributor()
     repo = plugin_model.Repository(self.REPO_ID)
     report = dist.validate_config(repo, self.VALID_CONFIGURATION, [])
     # Verify
     self.assertTrue(isinstance(report, tuple))
     self.assertTrue(len(report), 2)
     self.assertTrue(isinstance(report[0], bool))
     self.assertTrue(report[0])
     self.assertEqual(report[1], None)
开发者ID:maxamillion，项目名称:pulp，代码行数:11，代码来源:test_plugins.py


示例9: test_config_missing_protocol

 def test_config_missing_protocol(self):
     # Test
     conf = deepcopy(self.VALID_CONFIGURATION)
     del conf[constants.PROTOCOL_KEYWORD]
     dist = NodesHttpDistributor()
     repo = plugin_model.Repository(self.REPO_ID)
     report = dist.validate_config(repo, {}, [])
     # Verify
     self.assertTrue(isinstance(report, tuple))
     self.assertTrue(len(report), 2)
     self.assertTrue(isinstance(report[0], bool))
     self.assertFalse(report[0])
     self.assertFalse(report[1] is None)
开发者ID:maxamillion，项目名称:pulp，代码行数:13，代码来源:test_plugins.py


示例10: test_config_missing_invalid_alias

 def test_config_missing_invalid_alias(self):
     # Test
     conf = deepcopy(self.VALID_CONFIGURATION)
     conf['https']['alias'] = None
     dist = NodesHttpDistributor()
     repo = plugin_model.Repository(self.REPO_ID)
     report = dist.validate_config(repo, {}, [])
     # Verify
     self.assertTrue(isinstance(report, tuple))
     self.assertTrue(len(report), 2)
     self.assertTrue(isinstance(report[0], bool))
     self.assertFalse(report[0])
     self.assertFalse(report[1] is None)
开发者ID:maxamillion，项目名称:pulp，代码行数:13，代码来源:test_plugins.py


示例11: 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


示例12: 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


示例13: test_import_modified_units

 def test_import_modified_units(self, mock_get_working, *mocks):
     # Setup
     self.populate()
     mock_get_working.return_value = self.temp_dir
     max_concurrency = 5
     max_bandwidth = 12345
     with mock_config.patch({'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))
         # N=0 (no file)
         unit = collection.find_one({'N': 0})
         unit['age'] = 84  # this will be updated back to 42.
         unit['_last_updated'] -= 1
         unit['_storage_path'] = None
         collection.update({'N': 0}, unit)
         # N=1
         unit = collection.find_one({'N': 1})
         unit['age'] = 85   # this will be updated back to 42.
         unit['_last_updated'] -= 1
         collection.update({'N': 1}, unit)
         # 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, Mock())
     with mock_config.patch({'server': {'storage_dir': self.childfs}}):
         with patch('pulp_node.constants.CONTENT_PATH', self.parentfs):
             importer.sync_repo(repo, conduit, configuration)
         # Verify
         unit = collection.find_one({'N': 0})
         self.assertEqual(unit['age'], 42)
         unit = collection.find_one({'N': 1})
         self.assertEqual(unit['age'], 42)
开发者ID:BrnoPCmaniak，项目名称:pulp，代码行数:47，代码来源:test_plugins.py


示例14: 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


示例15: 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


示例16: 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


示例17: 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


示例18: 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


示例19: test_import_unit_files_already_exist

 def test_import_unit_files_already_exist(self, mock_get_working, *mocks):
     # Setup
     self.populate()
     mock_get_working.return_value = self.temp_dir
     with mock_config.patch({'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)
         model.Distributor.objects.delete()
         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)
         # 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, Mock())
     with mock_config.patch({'server': {'storage_dir': self.childfs}}):
         with patch('pulp_node.constants.CONTENT_PATH', self.parentfs):
             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:BrnoPCmaniak，项目名称:pulp，代码行数:37，代码来源:test_plugins.py


示例20: 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_node.distributors.http.distributor.NodesHttpDistributor类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。

鲜花
握手
雷人
路过
鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
More+
上一篇：
Python extension.missing_resources函数代码示例发布时间：2022-05-25
下一篇：
Python importer.DockerImporter类代码示例发布时间：2022-05-25
热门推荐
More+
热门话题
More+
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服（服务时间 9:00～18:00）

在线QQ客服
地址：深圳市南山区西丽大学城创智工业园
电邮：jeky_zhao#qq.com
移动电话：139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap