Python model.ContentSource类代码示例

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

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

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

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

示例1: test_enabled

 def test_enabled(self):
     source = ContentSource('s-1', {constants.ENABLED: 'true'})
     self.assertTrue(source.enabled)
     source.descriptor[constants.ENABLED] = 'false'
     self.assertFalse(source.enabled)
     source.descriptor = {}
     self.assertFalse(source.enabled)

开发者ID:nareshbatthula，项目名称:pulp，代码行数:7，代码来源:test_model.py

示例2: test_refresh_canceled

    def test_refresh_canceled(self, fake_urls):
        url = 'http://xyz.com'
        urls = ['url-1', 'url-2']

        fake_urls.__get__ = Mock(return_value=urls)

        canceled = Mock()
        canceled.isSet = Mock(return_value=True)
        conduit = Mock()
        cataloger = Mock()

        source = ContentSource('s-1', {constants.BASE_URL: url})
        source.get_conduit = Mock(return_value=conduit)
        source.get_cataloger = Mock(return_value=cataloger)

        # test

        report = source.refresh(canceled)

        # validation

        self.assertEqual(canceled.isSet.call_count, 1)
        self.assertEqual(conduit.reset.call_count, 0)
        self.assertEqual(cataloger.refresh.call_count, 0)
        self.assertEqual(report, [])

开发者ID:kevinaloys，项目名称:pulp，代码行数:25，代码来源:test_model.py

示例3: test_refresh_raised

    def test_refresh_raised(self, fake_urls):
        url = 'http://xyz.com'
        urls = ['url-1', 'url-2']
        fake_urls.__get__ = Mock(return_value=urls)

        canceled = Mock()
        canceled.isSet = Mock(return_value=False)
        conduit = Mock()
        cataloger = Mock()
        cataloger.refresh.side_effect = ValueError('just failed')

        source = ContentSource('s-1', {constants.BASE_URL: url})
        source.get_conduit = Mock(return_value=conduit)
        source.get_cataloger = Mock(return_value=cataloger)

        # test

        report = source.refresh(canceled)

        # validation

        self.assertEqual(canceled.isSet.call_count, len(urls))
        self.assertEqual(conduit.reset.call_count, len(urls))
        self.assertEqual(cataloger.refresh.call_count, len(urls))

        n = 0
        for _url in source.urls:
            cataloger.refresh.assert_any(conduit, source.descriptor, _url)
            self.assertEqual(report[n].source_id, source.id)
            self.assertEqual(report[n].url, _url)
            self.assertFalse(report[n].succeeded)
            self.assertEqual(report[n].errors, ['just failed'])
            self.assertEqual(report[n].added_count, 0)
            self.assertEqual(report[n].deleted_count, 0)
            n += 1

开发者ID:kevinaloys，项目名称:pulp，代码行数:35，代码来源:test_model.py

示例4: test_conduit

    def test_conduit(self):
        source = ContentSource('s-1', {constants.EXPIRES: '1h'})

        conduit = source.get_conduit()

        self.assertEqual(conduit.source_id, source.id)
        self.assertEqual(conduit.expires, 3600)
        self.assertTrue(isinstance(conduit, CatalogerConduit))

开发者ID:kevinaloys，项目名称:pulp，代码行数:8，代码来源:test_model.py

示例5: test_download

    def test_download(self, fake_load):
        sources = []
        for n in range(3):
            s = ContentSource('s-%d' % n, {})
            s.get_downloader = Mock()
            sources.append(s)

        fake_load.return_value = sources

        request_list = []
        for n in range(6):
            r = Request('T', {}, 'url-%d' % n, 'path-%d' % n)
            r.find_sources = Mock(return_value=sources[n % 3:])
            request_list.append(r)

        collated = [
            {
                sources[0]: ['nectar-1'],
                sources[1]: ['nectar-2', 'nectar-3', 'nectar-4'],
                sources[2]: ['nectar-5', 'nectar-6']
            },
            {}
        ]
        fake_collated = Mock(side_effect=collated)

        fake_listener = Mock()
        canceled = FakeEvent()
        fake_primary = PrimarySource(Mock())

        # test
        container = ContentContainer('')
        container.refresh = Mock()
        container.collated = fake_collated
        report = container.download(canceled, fake_primary, request_list, fake_listener)

        # validation
        container.refresh.assert_called_with(canceled)

        for r in request_list:
            r.find_sources.assert_called_with(fake_primary, container.sources)

        self.assertEqual(report.total_passes, 1)
        self.assertEqual(report.total_sources, len(sources))
        self.assertEqual(len(report.downloads), 3)
        for source in sources:
            self.assertEqual(report.downloads[source.id].total_succeeded, 0)
            self.assertEqual(report.downloads[source.id].total_failed, 0)

        for source in sources:
            source.get_downloader.assert_called_with()
            downloader = source.get_downloader()
            listener = downloader.event_listener
            self.assertEqual(listener.cancel_event, canceled)
            self.assertEqual(listener.downloader, downloader)
            self.assertEqual(listener.listener, fake_listener)
            downloader.download.assert_called_with(collated[0][source])

开发者ID:aweiteka，项目名称:pulp，代码行数:56，代码来源:test_container.py

示例6: test_dict

    def test_dict(self):
        descriptor = {'A': 1, 'B': 2}

        # test
        source = ContentSource('s-1', descriptor)

        # validation
        expected = {}
        expected.update(descriptor)
        expected[constants.SOURCE_ID] = source.id
        self.assertEqual(source.dict(), expected)

开发者ID:kevinaloys，项目名称:pulp，代码行数:11，代码来源:test_model.py

示例7: test_download_canceled_after_collated

    def test_download_canceled_after_collated(self, fake_load):
        sources = []
        for n in range(3):
            s = ContentSource('s-%d' % n, {})
            s.get_downloader = Mock()
            sources.append(s)

        fake_load.return_value = sources

        request_list = []
        for n in range(6):
            r = Request('T', {}, 'url-%d' % n, 'path-%d' % n)
            r.find_sources = Mock(return_value=sources[n % 3:])
            request_list.append(r)

        collated = [
            {
                sources[0]: ['nectar-1'],
                sources[1]: ['nectar-2', 'nectar-3', 'nectar-4'],
                sources[2]: ['nectar-5', 'nectar-6']
            },
            {}
        ]
        fake_collated = Mock(side_effect=collated)

        fake_listener = Mock()
        canceled = Mock()
        canceled.isSet.side_effect = [False, True, True]
        fake_primary = PrimarySource(Mock())

        # test
        container = ContentContainer('')
        container.refresh = Mock()
        container.collated = fake_collated
        report = container.download(canceled, fake_primary, request_list, fake_listener)

        # validation
        container.refresh.assert_called_with(canceled)

        for r in request_list:
            r.find_sources.assert_called_with(fake_primary, container.sources)

        called = 0
        for s in sources:
            if s.get_downloader.called:
                called += 1

        self.assertEqual(called, 1)
        self.assertEqual(report.total_passes, 1)
        self.assertEqual(report.total_sources, len(sources))
        self.assertEqual(len(report.downloads), 1)
        self.assertEqual(report.downloads[sources[2].id].total_succeeded, 0)
        self.assertEqual(report.downloads[sources[2].id].total_failed, 0)

开发者ID:aweiteka，项目名称:pulp，代码行数:53，代码来源:test_container.py

示例8: test_is_valid_bad_descriptor

    def test_is_valid_bad_descriptor(self, mock_descriptor_is_valid):
        source = ContentSource('s-1', {'A': 1})
        source.get_downloader = Mock()
        source.get_cataloger = Mock()
        mock_descriptor_is_valid.side_effect = ValueError()

        # Test

        valid = source.is_valid()

        # validation

        self.assertFalse(valid)

开发者ID:kevinaloys，项目名称:pulp，代码行数:13，代码来源:test_model.py

示例9: test_cataloger

    def test_cataloger(self, fake_plugins):
        plugin = Mock()
        fake_plugins.get_cataloger_by_id.return_value = plugin, {}

        # test

        source = ContentSource('s-1', {constants.TYPE: 1234})
        cataloger = source.get_cataloger()

        # validation

        fake_plugins.get_cataloger_by_id.assert_called_with(1234)
        self.assertEqual(plugin, cataloger)

开发者ID:kevinaloys，项目名称:pulp，代码行数:13，代码来源:test_model.py

示例10: test_is_valid_no_downloader

    def test_is_valid_no_downloader(self, mock_descriptor_is_valid):
        source = ContentSource('s-1', {'A': 1})
        source.get_downloader = Mock(side_effect=NotImplementedError())
        source.get_cataloger = Mock()

        # Test

        valid = source.is_valid()

        # validation

        source.get_cataloger.assert_called_with()
        source.get_downloader.assert_called_with()
        self.assertFalse(mock_descriptor_is_valid.called)
        self.assertFalse(valid)

开发者ID:aweiteka，项目名称:pulp，代码行数:15，代码来源:test_model.py

示例11: test_is_valid

    def test_is_valid(self, mock_descriptor_is_valid):
        source = ContentSource('s-1', {'A': 1})
        source.get_downloader = Mock()
        source.get_cataloger = Mock()

        # Test

        valid = source.is_valid()

        # validation

        source.get_cataloger.assert_called_with()
        source.get_downloader.assert_called_with()
        mock_descriptor_is_valid.assert_called_with(source.id, source.descriptor)
        self.assertTrue(valid)

开发者ID:kevinaloys，项目名称:pulp，代码行数:15，代码来源:test_model.py

示例12: test_is_valid_no_plugin

    def test_is_valid_no_plugin(self, mock_descriptor_is_valid):
        mock_descriptor_is_valid.return_value = True
        source = ContentSource("s-1", {"A": 1})
        source.get_downloader = Mock()
        source.get_cataloger = Mock(side_effect=NotImplementedError())

        # Test

        valid = source.is_valid()

        # validation

        source.get_cataloger.assert_called_with()
        self.assertFalse(source.get_downloader.called)
        self.assertTrue(mock_descriptor_is_valid.called)
        self.assertFalse(valid)

开发者ID:maxamillion，项目名称:pulp，代码行数:16，代码来源:test_model.py

示例13: test_load_all

    def test_load_all(self, fake_valid, fake_enabled, fake_parser, fake_listdir, fake_isfile):
        conf_d = '/fake/conf_d'
        files = ['one.conf', 'other']
        fake_listdir.return_value = files

        fake_valid.side_effect = [
            True,  # s-0
                   # s-1 not enabled
            True,  # s-2
            False  # s-3
        ]

        fake_isfile.side_effect = [True, False]

        fake_enabled.__get__ = Mock(side_effect=[d[1]['enabled'] for d in DESCRIPTOR])

        parser = Mock()
        parser.items.side_effect = [d[1].items() for d in DESCRIPTOR]
        parser.sections.return_value = [d[0] for d in DESCRIPTOR]
        fake_parser.return_value = parser

        # test

        sources = ContentSource.load_all(conf_d)

        # validation

        fake_listdir.assert_called_with(conf_d)
        fake_parser.assert_called_with()
        fake_parser().read.assert_called_with(os.path.join(conf_d, files[0]))

        self.assertEqual(len(sources), 2)
        self.assertTrue(DESCRIPTOR[0][0] in sources)
        self.assertTrue(DESCRIPTOR[2][0] in sources)

开发者ID:kevinaloys，项目名称:pulp，代码行数:34，代码来源:test_model.py

示例14: init

 def __init__(self, path=None):
     """
     :param path: The absolute path to a directory containing
         content source descriptor files.
     :type path: str
     """
     self.sources = ContentSource.load_all(path)

开发者ID:aweiteka，项目名称:pulp，代码行数:7，代码来源:container.py

示例15: test_forced_refresh

    def test_forced_refresh(self, fake_manager, fake_load):
        sources = {}
        for n in range(3):
            s = ContentSource('s-%d' % n, {})
            s.refresh = Mock()
            sources[s.id] = s

        fake_manager().has_entries.return_value = True
        fake_load.return_value = sources

        # test
        container = ContentContainer('')
        container.refresh(force=True)

        # validation
        for s in sources.values():
            s.refresh.assert_called_with()

开发者ID:alanoe，项目名称:pulp，代码行数:17，代码来源:test_container.py

示例16: test_refresh_canceled

    def test_refresh_canceled(self, fake_load):
        sources = {}
        for n in range(3):
            s = ContentSource('s-%d' % n, {})
            s.refresh = Mock()
            sources[s.id] = s

        fake_load.return_value = sources

        # test
        canceled = FakeEvent()
        canceled.set()
        container = ContentContainer('')
        container.refresh(canceled, force=True)

        # validation
        for s in sources.values():
            self.assertFalse(s.refresh.called)

开发者ID:aweiteka，项目名称:pulp，代码行数:18，代码来源:test_container.py

示例17: test_urls

 def test_urls(self):
     sources = ContentSource.load_all(self.tmp_dir)
     underground = sources[UNDERGROUND]
     urls = underground.urls
     self.assertEqual(len(urls), 4)
     self.assertEqual(urls[0], 'file:///underground/fedora/18/x86_64/')
     self.assertEqual(urls[1], 'file:///underground/fedora/18/i386/')
     self.assertEqual(urls[2], 'file:///underground/fedora/19/x86_64/')
     self.assertEqual(urls[3], 'file:///underground/fedora/19/i386/')

开发者ID:alanoe，项目名称:pulp，代码行数:9，代码来源:test_content_sources.py

示例18: test_downloader

    def test_downloader(self):
        url = 'http://xyz.com'
        fake_conduit = Mock()
        fake_cataloger = Mock()
        fake_downloader = Mock()
        fake_cataloger.get_downloader = Mock(return_value=fake_downloader)

        source = ContentSource('s-1', {constants.BASE_URL: url})
        source.get_conduit = Mock(return_value=fake_conduit)
        source.get_cataloger = Mock(return_value=fake_cataloger)

        # test
        downloader = source.get_downloader()

        # validation
        source.get_cataloger.assert_called_with()
        fake_cataloger.get_downloader.assert_called_with(fake_conduit, source.descriptor, url)
        self.assertEqual(downloader, fake_downloader)

开发者ID:AndreaGiardini，项目名称:pulp，代码行数:18，代码来源:test_model.py

示例19: init

 def __init__(self, path=None, threaded=True):
     """
     :param path:     The absolute path to a directory containing
                      content source descriptor files.
     :type  path:     str
     :param threaded: Whether or not to use the threaded download method.
     :type  threaded: bool
     """
     self.sources = ContentSource.load_all(path)
     self.threaded = threaded

开发者ID:maxamillion，项目名称:pulp，代码行数:10，代码来源:container.py

示例20: test_refresh

    def test_refresh(self, fake_manager, fake_load):
        sources = {}
        for n in range(3):
            s = ContentSource('s-%d' % n, {})
            s.refresh = Mock(return_value=[n])
            s.get_downloader = Mock()
            sources[s.id] = s

        fake_manager().has_entries.return_value = False
        fake_load.return_value = sources

        # test
        container = ContentContainer('')
        report = container.refresh()

        # validation
        for s in sources.values():
            s.refresh.assert_called_with()

        self.assertEqual(sorted(report), [0, 1, 2])

开发者ID:alanoe，项目名称:pulp，代码行数:20，代码来源:test_container.py

注：本文中的pulp.server.content.sources.model.ContentSource类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。