Python requests_mock 模块,mock() 实例源码

我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用requests_mock.mock()

项目:abodepy    作者:MisterWil    | 项目源码 | 文件源码
def tests_event_registration(self):
        """Tests that events register correctly."""
        # Get the event controller
        events = self.abode.events
        self.assertIsNotNone(events)

        # Create mock callback
        callback = Mock()

        # Test that a valid event registers
        self.assertTrue(
            events.add_event_callback(TIMELINE.ALARM_GROUP, callback))

        # Test that no event group returns false
        self.assertFalse(events.add_event_callback(None, callback))

        # Test that an invalid event throws exception
        with self.assertRaises(abodepy.AbodeException):
            events.add_event_callback("lol", callback)
项目:abodepy    作者:MisterWil    | 项目源码 | 文件源码
def tests_timeline_registration(self):
        """Tests that timeline events register correctly."""
        # Get the event controller
        events = self.abode.events
        self.assertIsNotNone(events)

        # Create mock callback
        callback = Mock()

        # Test that a valid timeline event registers
        self.assertTrue(
            events.add_timeline_callback(
                TIMELINE.CAPTURE_IMAGE, callback))

        # Test that no timeline event returns false
        self.assertFalse(events.add_timeline_callback(None, callback))

        # Test that an invalid timeline event string throws exception
        with self.assertRaises(abodepy.AbodeException):
            events.add_timeline_callback("lol", callback)

        # Test that an invalid timeline event dict throws exception
        with self.assertRaises(abodepy.AbodeException):
            events.add_timeline_callback({"lol": "lol"}, callback)
项目:abodepy    作者:MisterWil    | 项目源码 | 文件源码
def tests_multi_events_callback(self):
        """Tests that multiple event updates callback correctly."""
        # Get the event controller
        events = self.abode.events
        self.assertIsNotNone(events)

        # Create mock callback
        callback = Mock()

        # Register our events
        self.assertTrue(
            events.add_event_callback(
                [TIMELINE.ALARM_GROUP, TIMELINE.CAPTURE_GROUP],
                callback))

        # Call our events callback method and trigger a capture group event
        # pylint: disable=protected-access
        event_json = json.loads(IRCAMERA.timeline_event())
        events._on_timeline_update(event_json)

        # Ensure our callback was called
        callback.assert_called_with(event_json)
项目:abodepy    作者:MisterWil    | 项目源码 | 文件源码
def tests_multi_timeline_callback(self):
        """Tests that multiple timeline updates callback correctly."""
        # Get the event controller
        events = self.abode.events
        self.assertIsNotNone(events)

        # Create mock callback
        callback = Mock()

        # Register our events
        self.assertTrue(
            events.add_timeline_callback(
                [TIMELINE.CAPTURE_IMAGE, TIMELINE.OPENED], callback))

        # Call our events callback method and trigger a capture group event
        # pylint: disable=protected-access
        event_json = json.loads(IRCAMERA.timeline_event())
        events._on_timeline_update(event_json)

        # Ensure our callback was called
        callback.assert_called_with(event_json)
项目:abodepy    作者:MisterWil    | 项目源码 | 文件源码
def tests_automations_callback(self):
        """Tests that automation updates callback correctly."""
        # Get the event controller
        events = self.abode.events
        self.assertIsNotNone(events)

        # Create mock callbacks
        automation_callback = Mock()

        # Register our events
        self.assertTrue(
            events.add_event_callback(
                TIMELINE.AUTOMATION_EDIT_GROUP, automation_callback))

        # Call our events callback method and trigger a capture group event
        # pylint: disable=protected-access
        events._on_automation_update('{}')

        # Our capture callback should get one, but our alarm should not
        automation_callback.assert_called_with('{}')
项目:h1-python    作者:uber-common    | 项目源码 | 文件源码
def test_find_resources_yield_pages(self):
        page_one = json.dumps(load_resource_blob("responses/report-list-twopage-0"))
        page_two = json.dumps(load_resource_blob("responses/report-list-twopage-1"))
        with requests_mock.mock() as m:
            filter_str = "filter%5Bprogram%5D%5B%5D=foo"
            m.get(HackerOneClient.BASE_URL + "/reports?" + filter_str, text=page_one)
            m.get(HackerOneClient.BASE_URL + "/reports/nextpage", text=page_two)

            client = HackerOneClient("a", "b")
            listing = client.find_resources(Report, program=["foo"], yield_pages=True)

            # The listing should contain two pages each with one report
            self.assertEqual(len(listing), 2)
            # Flatten the pages into a single page
            all_items = list(itertools.chain(*listing))
            self.assertEqual(len(all_items), 2)
            self.assertTrue(all(isinstance(r, Report) for r in all_items))
项目:newrelic-cli    作者:NativeInstruments    | 项目源码 | 文件源码
def test_get_nonexistent_alert_condition(self, mock):
        policy_name = 'IAmNotHerePolicy'
        policies_response = {
            'policies': []
        }
        mock.get(
            '{}/v2/alerts_policies.json'.format(self.client.base_url),
            json=policies_response,
            status_code=200
        )
        # we don't even need a second endpoint here
        with self.assertRaisesRegexp(
            newrelic_cli.exceptions.ItemNotFoundError,
            '{}'.format(policy_name)
        ):
            self.client.get_alert_conditions(policy_name)
项目:newrelic-cli    作者:NativeInstruments    | 项目源码 | 文件源码
def test_create_synthetics_alert_condition_no_policy(self, mock):
        # Alerts policies list mock
        response = {
            'policies': []
        }
        mock.get(
            '{}/v2/alerts_policies.json'.format(self.client.base_url),
            json=response,
            status_code=200
        )
        # We don't need other endpoints
        with self.assertRaisesRegexp(
            newrelic_cli.exceptions.ItemNotFoundError,
            '{}'.format(self.policy_name)
        ):
            self.client.create_synthetics_alert_condition(
                policy_name=self.policy_name,
                condition_name=self.monitor_name,
                monitor_name=self.monitor_name,
            )
项目:newrelic-cli    作者:NativeInstruments    | 项目源码 | 文件源码
def test_delete_synthetics_alert_condition_no_policy(self, mock):
        response = {
            'policies': []
        }
        # Synthetics monitors list mock
        mock.get(
            'https://synthetics.newrelic.com/synthetics/api/v3/monitors',
            status_code=200,
            json=self.monitor_response
        )
        # Alerts policies list mock
        mock.get(
            '{}/v2/alerts_policies.json'.format(self.client.base_url),
            json=response,
            status_code=200
        )
        with self.assertRaisesRegexp(
            newrelic_cli.exceptions.ItemNotFoundError,
            '{}'.format(self.policy_name)
        ):
            self.client.delete_synthetics_alert_conditions(
                policy_name=self.policy_name,
                monitor_name=self.monitor_name,
            )
项目:newrelic-cli    作者:NativeInstruments    | 项目源码 | 文件源码
def test_delete_synthetics_alert_condition_no_monitor(self, mock):
        response = {
            'monitors': []
        }
        # Synthetics monitors list mock
        mock.get(
            'https://synthetics.newrelic.com/synthetics/api/v3/monitors',
            status_code=200,
            json=response
        )
        # Alerts policies list mock
        mock.get(
            '{}/v2/alerts_policies.json'.format(self.client.base_url),
            json=self.policy_response,
            status_code=200
        )
        with self.assertRaisesRegexp(
            newrelic_cli.exceptions.ItemNotFoundError,
            '{}'.format(self.monitor_name)
        ):
            self.client.delete_synthetics_alert_conditions(
                policy_name=self.policy_name,
                monitor_name=self.monitor_name,
            )
项目:newrelic-cli    作者:NativeInstruments    | 项目源码 | 文件源码
def test_create_alert_policy_with_incident_preference(self, mock):
        # Now we don't have any policies
        response = {
            'policies': []
        }
        mock.get(
            '{}/v2/alerts_policies.json'.format(self.client.base_url),
            json=response,
            status_code=200
        )
        # Policy created successfully
        policy_created_response = {
            'policy': self.first_policy
        }
        mock.post(
            '{}/v2/alerts_policies.json'.format(self.client.base_url),
            json=policy_created_response,
            status_code=201
        )

        created_policy = self.client.create_alert_policy(
            self.first_policy['name'],
            incident_preference=self.first_policy['incident_preference']
        )
        self.assertDictEqual(created_policy, self.first_policy)
项目:newrelic-cli    作者:NativeInstruments    | 项目源码 | 文件源码
def test_create_alert_policy_duplicate_success(self, mock):
        mock.get(
            '{}/v2/alerts_policies.json'.format(self.client.base_url),
            json=self.policy_response,
            status_code=200
        )
        # Policy already exist
        policy_created_response = {
            'policy': self.first_policy
        }
        mock.post(
            '{}/v2/alerts_policies.json'.format(self.client.base_url),
            json=policy_created_response,
            status_code=201
        )
        created_policy = self.client.create_alert_policy(
            self.first_policy['name'],
            check_unique=False
        )
        self.assertDictEqual(created_policy, self.first_policy)
项目:newrelic-cli    作者:NativeInstruments    | 项目源码 | 文件源码
def test_create_alert_policy_duplicate_failure(self, mock):
        mock.get(
            '{}/v2/alerts_policies.json'.format(self.client.base_url),
            json=self.policy_response,
            status_code=200
        )
        # Policy already exist
        policy_created_response = {
            'policy': self.first_policy
        }
        mock.post(
            '{}/v2/alerts_policies.json'.format(self.client.base_url),
            json=policy_created_response,
            status_code=201
        )
        with self.assertRaisesRegexp(
                newrelic_cli.exceptions.ItemAlreadyExistsError,
                '{}'.format(self.first_policy['name'])
        ):
            self.client.create_alert_policy(self.first_policy['name'])
项目:newrelic-cli    作者:NativeInstruments    | 项目源码 | 文件源码
def test_delete_nonexistent_alert_policy(self, mock):
        # Now we don't have any policies
        response = {
            'policies': []
        }
        mock.get(
            '{}/v2/alerts_policies.json?filter[name]={}'
            .format(self.client.base_url, self.first_policy['name']),
            json=response,
            status_code=200
        )
        mock.delete(
            '{}/v2/alerts_policies/{}.json'
            .format(self.client.base_url, self.first_policy['id'])
        )
        with self.assertRaisesRegexp(
                newrelic_cli.exceptions.ItemNotFoundError,
                '{}'.format(self.first_policy['name'])
        ):
            self.client.delete_alert_policy(self.first_policy['name'])
项目:newrelic-cli    作者:NativeInstruments    | 项目源码 | 文件源码
def test_get_other_error_with_message(self, mock):
        url = '{}/v2/i_am_bogus'.format(self.client.base_url)
        mock.get(
            url=url,
            status_code=400,
            json={
                'errors': [
                    {'error': 'I am bogus'},
                    {'error': 'I am verbose'}
                ]
            }
        )
        with self.assertRaisesRegexp(
            newrelic_cli.exceptions.NewRelicException,
            'The following errors were returned by server'
        ):
            self.client._get(url)
项目:newrelic-cli    作者:NativeInstruments    | 项目源码 | 文件源码
def test_create_monitor_with_sla_success(self, mock):
        create_monitor_endpoint = '{}/v3/monitors'.format(self.client.base_url)
        monitor_data = self.first_monitor
        monitor_data['slaThreshold'] = 42

        mock.post(
            url=create_monitor_endpoint,
            status_code=201,
            headers={'Location': self.monitor_location}
        )
        mock.get(
            url=self.monitor_location,
            status_code=200,
            json=monitor_data
        )
        r = self.client.create_monitor(self.monitor_name, slaThreshold=42)
        self.assertDictEqual(r, self.first_monitor)
项目:newrelic-cli    作者:NativeInstruments    | 项目源码 | 文件源码
def test_delete_monitor_success(self, mock):
        get_all_monitors_endpoint = '{}/v3/monitors'.format(
            self.client.base_url
        )
        monitor_endpooint = '{}/v3/monitors/{}'.format(
            self.client.base_url,
            self.monitor_id
        )
        mock.get(
            url=get_all_monitors_endpoint,
            status_code=200,
            json=self.all_monitors_response
        )
        mock.delete(
            url=monitor_endpooint,
            status_code=204
        )
        # We don't expect any response here.
        # Just make sure no exceptions raised
        self.client.delete_monitor(self.monitor_name)
项目:newrelic-cli    作者:NativeInstruments    | 项目源码 | 文件源码
def test_delete_nonexistent_monitor(self, mock):
        get_all_monitors_endpoint = '{}/v3/monitors'.format(
            self.client.base_url
        )
        monitor_endpooint = '{}/v3/monitors/{}'.format(
            self.client.base_url,
            self.monitor_id
        )
        # we don't have any monitors
        monitors_list = {'monitors': []}

        mock.get(
            url=get_all_monitors_endpoint,
            status_code=200,
            json=monitors_list
        )
        mock.delete(
            url=monitor_endpooint,
            status_code=204
        )
        with self.assertRaisesRegexp(
            newrelic_cli.exceptions.ItemNotFoundError,
            '{}'.format(self.monitor_name)
        ):
            self.client.delete_monitor(self.monitor_name)
项目:newrelic-cli    作者:NativeInstruments    | 项目源码 | 文件源码
def test_get_monitor_script_success(self, mock):
        get_all_monitors_endpoint = '{}/v3/monitors'.format(
            self.client.base_url
        )
        monitor_script_endpoint = '{}/v3/monitors/{}/script'.format(
            self.client.base_url,
            self.monitor_id
        )
        mock.get(
            url=get_all_monitors_endpoint,
            status_code=200,
            json=self.all_monitors_response
        )
        mock.get(
            url=monitor_script_endpoint,
            status_code=200,
            json={'scriptText': self.monitor_script_base64}
        )
        script = self.client.get_monitor_script(self.monitor_name)
        self.assertEquals(self.monitor_script_plain, script)
项目:newrelic-cli    作者:NativeInstruments    | 项目源码 | 文件源码
def test_get_monitor_without_script(self, mock):
        get_all_monitors_endpoint = '{}/v3/monitors'.format(
            self.client.base_url
        )
        monitor_script_endpoint = '{}/v3/monitors/{}/script'.format(
            self.client.base_url,
            self.monitor_id
        )
        mock.get(
            url=get_all_monitors_endpoint,
            status_code=200,
            json=self.all_monitors_response
        )
        mock.get(
            url=monitor_script_endpoint,
            status_code=404,
        )
        with self.assertRaisesRegexp(
            newrelic_cli.exceptions.ItemNotFoundError,
            '{}'.format(self.monitor_name)
        ):
            self.client.get_monitor_script(self.monitor_name)
项目:newrelic-cli    作者:NativeInstruments    | 项目源码 | 文件源码
def test_upload_monitor_script_success(self, mock):
        get_all_monitors_endpoint = '{}/v3/monitors'.format(
            self.client.base_url
        )
        monitor_script_endpoint = '{}/v3/monitors/{}/script'.format(
            self.client.base_url,
            self.monitor_id
        )
        mock.get(
            url=get_all_monitors_endpoint,
            status_code=200,
            json=self.all_monitors_response
        )
        mock.put(
            url=monitor_script_endpoint,
            status_code=204
        )
        # We don't expect any response here.
        # Just make sure no exceptions raised
        self.client.upload_monitor_script(
            self.monitor_name,
            self.monitor_script_plain
        )
项目:bothub-sdk-python    作者:bothub-studio    | 项目源码 | 文件源码
def test_handle_message_should_add_channel_cmd_buff():
    event = {'content': 'hi!',
             'channel': 'mychannel',
             'sender': {'id': 'abcd1234',
                        'name': 'myname'}}

    context = {'channel': {'endpoint': 'http://localhost'}}

    with requests_mock.mock() as m:
        m.post('http://localhost/messages')
        response = handle_message(event, context, Bot)
        assert response['response'] is None
        assert m.called
        assert m.request_history[0].json() == {
            'channel': None,
            'receiver': None,
            'message': 'hello, you said: hi!',
            'event': event,
            'extra': None,
            'context': {'api_key': '', 'project_id': None}
        }
项目:flask_atlassian_connect    作者:halkeye    | 项目源码 | 文件源码
def test_lifecycle_installed_multiple_no_auth(self):
        """Multiple requests should fail unless auth is provided the second time"""
        with requests_mock.mock() as m:
            m.get('https://gavindev.atlassian.net/plugins/servlet/oauth/consumer-info',
                  text=consumer_info_response)

            client = dict(
                baseUrl='https://gavindev.atlassian.net',
                clientKey='abc123',
                publicKey='public123')
            rv = self.client.post('/atlassian_connect/lifecycle/installed',
                                  data=json.dumps(client),
                                  content_type='application/json')
            self.assertEqual(204, rv.status_code)
            rv = self.client.post('/atlassian_connect/lifecycle/installed',
                                  data=json.dumps(client),
                                  content_type='application/json')
            self.assertEqual(401, rv.status_code)
项目:python-cartolafc    作者:vicenteneto    | 项目源码 | 文件源码
def test_liga_com_nome(self):
        # Arrange and Act
        with requests_mock.mock() as m:
            url = '{api_url}/auth/liga/{slug}'.format(api_url=self.api_url, slug='falydos-fc')
            m.get(url, text=self.LIGA)
            liga = self.api.liga(nome='Falydos FC')
            primeiro_time = liga.times[0]

            # Assert
            self.assertIsInstance(liga, Liga)
            self.assertEqual(liga.id, 6407)
            self.assertEqual(liga.nome, 'Virtus Premier League')
            self.assertEqual(liga.slug, 'virtus-premier-league')
            self.assertEqual(liga.descricao,
                             u'Prêmios para: \n\n- Melhor de cada Mês (R$50,00)\n- Melhor do 1º e 2º Turno (R$150,00)\n- 2º Lugar Geral (R$50)\n- 1º Lugar Geral (R$250,00)\n\nBoa sorte!')
            self.assertIsInstance(liga.times, list)
            self.assertIsInstance(primeiro_time, TimeInfo)
            self.assertEqual(primeiro_time.id, 453420)
            self.assertEqual(primeiro_time.nome, 'Mosqueteiros JPB')
            self.assertEqual(primeiro_time.nome_cartola, 'Erick Costa')
            self.assertEqual(primeiro_time.slug, 'mosqueteiros-jpb')
            self.assertTrue(primeiro_time.assinante)
项目:python-cartolafc    作者:vicenteneto    | 项目源码 | 文件源码
def test_pontuacao_atleta(self):
        # Arrange and Act
        with requests_mock.mock() as m:
            url = '{api_url}/auth/mercado/atleta/{id}/pontuacao'.format(api_url=self.api_url, id=81682)
            m.get(url, text=self.PONTUACAO_ATLETA)
            pontuacoes = self.api.pontuacao_atleta(81682)
            primeira_rodada = pontuacoes[0]

            # Assert
            self.assertIsInstance(pontuacoes, list)
            self.assertIsInstance(primeira_rodada, PontuacaoInfo)
            self.assertEqual(primeira_rodada.atleta_id, 81682)
            self.assertEqual(primeira_rodada.rodada_id, 1)
            self.assertEqual(primeira_rodada.pontos, 1.1)
            self.assertEqual(primeira_rodada.preco, 6.44)
            self.assertEqual(primeira_rodada.variacao, -1.56)
            self.assertEqual(primeira_rodada.media, 1.1)
项目:python-cartolafc    作者:vicenteneto    | 项目源码 | 文件源码
def test_ligas(self):
        # Arrange and Act
        with requests_mock.mock() as m:
            url = '{api_url}/ligas'.format(api_url=self.api_url)
            m.get(url, text=self.LIGAS)
            ligas = self.api.ligas(query='premiere')
            primeira_liga = ligas[0]

            # Assert
            self.assertIsInstance(ligas, list)
            self.assertIsInstance(primeira_liga, Liga)
            self.assertEqual(primeira_liga.id, 36741)
            self.assertEqual(primeira_liga.nome, 'PREMIERE_LIGA_ENTEL')
            self.assertEqual(primeira_liga.slug, 'premiere-liga-entel')
            self.assertEqual(primeira_liga.descricao, u'“Vale tudo, só não vale...”')
            self.assertIsNone(primeira_liga.times)
项目:python-cartolafc    作者:vicenteneto    | 项目源码 | 文件源码
def test_mercado_atletas(self):
        # Arrange and Act
        with requests_mock.mock() as m:
            url = '{api_url}/atletas/mercado'.format(api_url=self.api_url)
            m.get(url, text=self.MERCADO_ATLETAS)
            mercado = self.api.mercado_atletas()
            primeiro_atleta = mercado[0]

            # Assert
            self.assertIsInstance(mercado, list)
            self.assertIsInstance(primeiro_atleta, Atleta)
            self.assertEqual(primeiro_atleta.id, 86935)
            self.assertEqual(primeiro_atleta.apelido, 'Rodrigo')
            self.assertEqual(primeiro_atleta.pontos, 0)
            self.assertEqual(primeiro_atleta.scout, {'CA': 1, 'FC': 3, 'FS': 1, 'PE': 2, 'RB': 2})
            self.assertEqual(primeiro_atleta.posicao, _posicoes[4])
            self.assertIsInstance(primeiro_atleta.clube, Clube)
            self.assertEqual(primeiro_atleta.clube.id, 292)
            self.assertEqual(primeiro_atleta.clube.nome, 'Sport')
            self.assertEqual(primeiro_atleta.clube.abreviacao, 'SPO')
            self.assertEqual(primeiro_atleta.status, _atleta_status[6])
项目:python-cartolafc    作者:vicenteneto    | 项目源码 | 文件源码
def test_parciais_mercado_fechado(self):
        # Arrange
        with requests_mock.mock() as m:
            url = '{api_url}/mercado/status'.format(api_url=self.api_url)
            m.get(url, text=self.MERCADO_STATUS_FECHADO)

            url = '{api_url}/atletas/pontuados'.format(api_url=self.api_url)
            m.get(url, text=self.PARCIAIS)

            # Act
            parciais = self.api.parciais()
            parcial_juan = parciais[36540]

            # Assert
            self.assertIsInstance(parciais, dict)
            self.assertIsInstance(parcial_juan, Atleta)
            self.assertEqual(parcial_juan.id, 36540)
            self.assertEqual(parcial_juan.apelido, 'Juan')
            self.assertEqual(parcial_juan.pontos, 2.9)
            self.assertEqual(parcial_juan.scout, {'CA': 1, 'FC': 1, 'FS': 2, 'PE': 2, 'SG': 1})
            self.assertEqual(parcial_juan.posicao, _posicoes[3])
            self.assertIsInstance(parcial_juan.clube, Clube)
            self.assertEqual(parcial_juan.clube.id, 262)
            self.assertEqual(parcial_juan.clube.nome, 'Flamengo')
            self.assertEqual(parcial_juan.clube.abreviacao, 'FLA')
项目:python-cartolafc    作者:vicenteneto    | 项目源码 | 文件源码
def test_pos_rodada_destaques_com_mercado_aberto(self):
        # Arrange and Act
        with requests_mock.mock() as m:
            url = '{api_url}/mercado/status'.format(api_url=self.api_url)
            m.get(url, text=self.MERCADO_STATUS_ABERTO)

            url = '{api_url}/pos-rodada/destaques'.format(api_url=self.api_url)
            m.get(url, text=self.POS_RODADA_DESTAQUES)
            destaque_rodada = self.api.pos_rodada_destaques()

            # Assert
            self.assertIsInstance(destaque_rodada, DestaqueRodada)
            self.assertEqual(destaque_rodada.media_cartoletas, 115.8235753058391)
            self.assertEqual(destaque_rodada.media_pontos, 46.6480728839843)
            self.assertIsInstance(destaque_rodada.mito_rodada, TimeInfo)
            self.assertEqual(destaque_rodada.mito_rodada.id, 896224)
            self.assertEqual(destaque_rodada.mito_rodada.nome, 'gama campos fc')
            self.assertEqual(destaque_rodada.mito_rodada.nome_cartola, 'malmal')
            self.assertEqual(destaque_rodada.mito_rodada.slug, 'gama-campos-fc')
            self.assertFalse(destaque_rodada.mito_rodada.assinante)
项目:python-cartolafc    作者:vicenteneto    | 项目源码 | 文件源码
def test_times(self):
        # Arrange and Act
        with requests_mock.mock() as m:
            url = '{api_url}/times'.format(api_url=self.api_url)
            m.get(url, text=self.TIMES)
            times = self.api.times(query='Faly')
            primeiro_time = times[0]

            # Assert
            self.assertIsInstance(times, list)
            self.assertIsInstance(primeiro_time, TimeInfo)
            self.assertEqual(primeiro_time.id, 4626963)
            self.assertEqual(primeiro_time.nome, 'Falysson29')
            self.assertEqual(primeiro_time.nome_cartola, 'Alysson')
            self.assertEqual(primeiro_time.slug, 'falysson29')
            self.assertFalse(primeiro_time.assinante)
项目:python-zhmcclient    作者:zhmcclient    | 项目源码 | 文件源码
def test_create(self):
        """
        This tests the 'Create Virtual Function' operation.
        """
        vf_mgr = self.partition.virtual_functions
        with requests_mock.mock() as m:
            result = {
                'element-uri':
                    '/api/partitions/fake-part-id-1/virtual-functions/'
                    'fake-vf-id-1'
            }
            m.post('/api/partitions/fake-part-id-1/virtual-functions',
                   json=result)

            vf = vf_mgr.create(properties={})

            assert isinstance(vf, VirtualFunction)
            assert vf.properties == result
            assert vf.uri == result['element-uri']
项目:python-zhmcclient    作者:zhmcclient    | 项目源码 | 文件源码
def test_check_complete_success_noresult(self):
        """Test check_for_completion() with successful complete job without
        result."""
        with requests_mock.mock() as m:
            self.mock_server_1(m)
            session = Session('fake-host', 'fake-user', 'fake-pw')
            op_method = 'POST'
            op_uri = '/api/foo'
            job = Job(session, self.job_uri, op_method, op_uri)
            query_job_status_result = {
                'status': 'complete',
                'job-status-code': 200,
                # 'job-reason-code' omitted because HTTP status good
                # 'job-results' is optional and is omitted
            }
            m.get(self.job_uri, json=query_job_status_result)
            m.delete(self.job_uri, status_code=204)

            job_status, op_result = job.check_for_completion()

            assert job_status == 'complete'
            assert op_result is None
项目:python-zhmcclient    作者:zhmcclient    | 项目源码 | 文件源码
def test_check_complete_success_result(self):
        """Test check_for_completion() with successful complete job with a
        result."""
        with requests_mock.mock() as m:
            self.mock_server_1(m)
            session = Session('fake-host', 'fake-user', 'fake-pw')
            op_method = 'POST'
            op_uri = '/api/foo'
            job = Job(session, self.job_uri, op_method, op_uri)
            exp_op_result = {
                'foo': 'bar',
            }
            query_job_status_result = {
                'status': 'complete',
                'job-status-code': 200,
                # 'job-reason-code' omitted because HTTP status good
                'job-results': exp_op_result,
            }
            m.get(self.job_uri, json=query_job_status_result)
            m.delete(self.job_uri, status_code=204)

            job_status, op_result = job.check_for_completion()

            assert job_status == 'complete'
            assert op_result == exp_op_result
项目:python-zhmcclient    作者:zhmcclient    | 项目源码 | 文件源码
def test_check_complete_error1(self):
        """Test check_for_completion() with complete job in error (1)."""
        with requests_mock.mock() as m:
            self.mock_server_1(m)
            session = Session('fake-host', 'fake-user', 'fake-pw')
            op_method = 'POST'
            op_uri = '/api/foo'
            job = Job(session, self.job_uri, op_method, op_uri)
            query_job_status_result = {
                'status': 'complete',
                'job-status-code': 500,
                'job-reason-code': 42,
                # no 'job-results' field (it is not guaranteed to be there)
            }

            m.get(self.job_uri, json=query_job_status_result)
            m.delete(self.job_uri, status_code=204)

            with pytest.raises(HTTPError) as exc_info:
                job_status, op_result = job.check_for_completion()
            exc = exc_info.value

            assert exc.http_status == 500
            assert exc.reason == 42
            assert exc.message is None
项目:python-zhmcclient    作者:zhmcclient    | 项目源码 | 文件源码
def test_check_complete_error2(self):
        """Test check_for_completion() with complete job in error (2)."""
        with requests_mock.mock() as m:
            self.mock_server_1(m)
            session = Session('fake-host', 'fake-user', 'fake-pw')
            op_method = 'POST'
            op_uri = '/api/foo'
            job = Job(session, self.job_uri, op_method, op_uri)
            query_job_status_result = {
                'status': 'complete',
                'job-status-code': 500,
                'job-reason-code': 42,
                'job-results': {},  # it is not guaranteed to have any content
            }

            m.get(self.job_uri, json=query_job_status_result)
            m.delete(self.job_uri, status_code=204)

            with pytest.raises(HTTPError) as exc_info:
                job_status, op_result = job.check_for_completion()
            exc = exc_info.value

            assert exc.http_status == 500
            assert exc.reason == 42
            assert exc.message is None
项目:python-zhmcclient    作者:zhmcclient    | 项目源码 | 文件源码
def test_wait_complete1_success_result(self):
        """Test wait_for_completion() with successful complete job with a
        result."""
        with requests_mock.mock() as m:
            self.mock_server_1(m)
            session = Session('fake-host', 'fake-user', 'fake-pw')
            op_method = 'POST'
            op_uri = '/api/foo'
            job = Job(session, self.job_uri, op_method, op_uri)
            exp_op_result = {
                'foo': 'bar',
            }
            query_job_status_result = {
                'status': 'complete',
                'job-status-code': 200,
                # 'job-reason-code' omitted because HTTP status good
                'job-results': exp_op_result,
            }
            m.get(self.job_uri, json=query_job_status_result)
            m.delete(self.job_uri, status_code=204)

            op_result = job.wait_for_completion()

            assert op_result == exp_op_result
项目:python-zhmcclient    作者:zhmcclient    | 项目源码 | 文件源码
def test_wait_complete3_success_result(self):
        """Test wait_for_completion() with successful complete job with a
        result."""
        with requests_mock.mock() as m:
            self.mock_server_1(m)
            session = Session('fake-host', 'fake-user', 'fake-pw')
            op_method = 'POST'
            op_uri = '/api/foo'
            job = Job(session, self.job_uri, op_method, op_uri)
            exp_op_result = {
                'foo': 'bar',
            }
            m.get(self.job_uri,
                  [
                      {'text': result_running_callback},
                      {'text': result_complete_callback},
                  ])
            m.delete(self.job_uri, status_code=204)

            op_result = job.wait_for_completion()

            assert op_result == exp_op_result
项目:python-zhmcclient    作者:zhmcclient    | 项目源码 | 文件源码
def setup_method(self):
        self.session = Session('vswitch-dpm-host', 'vswitch-user',
                               'vswitch-pwd')
        self.client = Client(self.session)
        with requests_mock.mock() as m:
            # Because logon is deferred until needed, we perform it
            # explicitly in order to keep mocking in the actual test simple.
            m.post('/api/sessions', json={'api-session': 'vswitch-session-id'})
            self.session.logon()

        self.cpc_mgr = self.client.cpcs
        with requests_mock.mock() as m:
            result = {
                'cpcs': [
                    {
                        'object-uri': '/api/cpcs/vswitch-cpc-id-1',
                        'name': 'CPC',
                        'status': 'service-required',
                    }
                ]
            }
            m.get('/api/cpcs', json=result)

            cpcs = self.cpc_mgr.list()
            self.cpc = cpcs[0]
项目:kripodb    作者:3D-e-Chem    | 项目源码 | 文件源码
def test_fragments_by_id(base_url, client):
    with requests_mock.mock() as m:
        expected = [
            {'smiles': '[*]C1OC(COP(=O)([O-])OP(=O)([O-])OCC2OC(N3C=CCC(C(N)=O)=C3)C(O)C2O)C(O)C1[*]',
             'pdb_code': '3j7u',
             'pdb_title': 'Catalase structure determined by electron crystallography of thin 3D crystals',
             'atom_codes': 'PA,O1A,O2A,O5B,C5B,C4B,O4B,C3B,O3B,C2B,C1B,O3,PN,O1N,O2N,O5D,C5D,C4D,O4D,C3D,O3D,C2D,O2D,C1D,N1N,C2N,C3N,C7N,O7N,N7N,C4N,C5N,C6N',
             'uniprot_acc': 'P00432',
             'mol': '3j7u_NDP_frag24\n     RDKit          3D\n\n 35 37  0  0  0  0  0  0  0  0999 V2000\n  -15.1410  -11.1250  -79.4200 P   0  0  0  0  0  0  0  0  0  0  0  0\n  -14.6900  -10.9960  -80.8600 O   0  0  0  0  0  0  0  0  0  0  0  0\n  -16.5040  -11.6890  -79.0770 O   0  0  0  0  0  0  0  0  0  0  0  0\n  -14.9990   -9.6870  -78.7060 O   0  0  0  0  0  0  0  0  0  0  0  0\n  -15.1870   -8.4550  -79.4050 C   0  0  0  0  0  0  0  0  0  0  0  0\n  -14.6700   -7.3160  -78.5260 C   0  0  0  0  0  0  0  0  0  0  0  0\n  -13.2400   -7.2390  -78.5880 O   0  0  0  0  0  0  0  0  0  0  0  0\n  -15.2130   -5.9510  -78.9460 C   0  0  0  0  0  0  0  0  0  0  0  0\n  -16.1600   -5.4570  -77.9880 O   0  0  0  0  0  0  0  0  0  0  0  0\n  -14.0000   -5.0420  -79.0650 C   0  0  0  0  0  0  0  0  0  0  0  0\n  -14.1790   -3.8250  -78.3260 R   0  0  0  0  0  1  0  0  0  0  0  0\n  -12.8370   -5.8690  -78.5180 C   0  0  0  0  0  0  0  0  0  0  0  0\n  -11.5470   -5.6210  -79.2410 R   0  0  0  0  0  1  0  0  0  0  0  0\n  -14.0270  -11.9960  -78.6490 O   0  0  0  0  0  0  0  0  0  0  0  0\n  -14.1810  -13.5930  -78.4870 P   0  0  0  0  0  0  0  0  0  0  0  0\n  -14.5480  -14.2030  -79.8230 O   0  0  0  0  0  0  0  0  0  0  0  0\n  -15.0330  -13.8500  -77.2690 O   0  0  0  0  0  0  0  0  0  0  0  0\n  -12.6800  -14.0730  -78.1770 O   0  0  0  0  0  0  0  0  0  0  0  0\n  -12.1840  -14.2350  -76.8490 C   0  0  0  0  0  0  0  0  0  0  0  0\n  -11.1340  -13.1670  -76.6050 C   0  0  0  0  0  0  0  0  0  0  0  0\n  -11.6880  -11.8550  -76.6770 O   0  0  0  0  0  0  0  0  0  0  0  0\n  -10.5070  -13.2750  -75.2350 C   0  0  0  0  0  0  0  0  0  0  0  0\n   -9.4070  -14.1780  -75.3000 O   0  0  0  0  0  0  0  0  0  0  0  0\n  -10.0970  -11.8400  -74.9280 C   0  0  0  0  0  0  0  0  0  0  0  0\n   -8.6920  -11.6460  -75.1050 O   0  0  0  0  0  0  0  0  0  0  0  0\n  -10.8280  -10.9760  -75.9460 C   0  0  0  0  0  0  0  0  0  0  0  0\n  -11.5890   -9.8540  -75.3660 N   0  0  0  0  0  0  0  0  0  0  0  0\n  -12.7860  -10.0630  -74.7850 C   0  0  0  0  0  0  0  0  0  0  0  0\n  -13.5340   -9.0090  -74.2510 C   0  0  0  0  0  0  0  0  0  0  0  0\n  -14.8620   -9.2740  -73.5990 C   0  0  0  0  0  0  0  0  0  0  0  0\n  -15.1890  -10.4300  -73.3940 O   0  0  0  0  0  0  0  0  0  0  0  0\n  -15.6600   -8.2650  -73.2400 N   0  0  0  0  0  0  0  0  0  0  0  0\n  -13.0230   -7.5870  -74.3390 C   0  0  0  0  0  0  0  0  0  0  0  0\n  -11.7130   -7.4960  -74.9740 C   0  0  0  0  0  0  0  0  0  0  0  0\n  -11.0640   -8.6200  -75.4710 C   0  0  0  0  0  0  0  0  0  0  0  0\n  1  2  2  0\n  1  3  1  0\n  1  4  1  0\n  1 14  1  0\n  4  5  1  0\n  5  6  1  0\n  6  7  1  0\n  6  8  1  0\n  7 12  1  0\n  8  9  1  0\n  8 10  1  0\n 10 11  1  0\n 10 12  1  0\n 12 13  1  0\n 14 15  1  0\n 15 16  2  0\n 15 17  1  0\n 15 18  1  0\n 18 19  1  0\n 19 20  1  0\n 20 21  1  0\n 20 22  1  0\n 21 26  1  0\n 22 23  1  0\n 22 24  1  0\n 24 25  1  0\n 24 26  1  0\n 26 27  1  0\n 27 28  1  0\n 27 35  1  0\n 28 29  2  0\n 29 30  1  0\n 29 33  1  0\n 30 31  2  0\n 30 32  1  0\n 33 34  1  0\n 34 35  2  0\nM  CHG  2   3  -1  17  -1\nM  END\n',
             'prot_chain': 'A', 'het_seq_nr': 602, 'het_code': 'NDP', 'prot_name': 'Catalase',
             'ec_number': '1.11.1.6', 'frag_nr': 24, 'frag_id': '3j7u_NDP_frag24', 'rowid': 7059,
             'uniprot_name': 'Catalase', 'nr_r_groups': 2, 'het_chain': 'A', 'hash_code': '6ef5a609fb192dba'}
        ]
        url = base_url + '/fragments?fragment_ids=3j7u_NDP_frag24,3j7u_NDP_frag23'
        m.get(url, json=expected)

        response = client.fragments_by_id(fragment_ids=['3j7u_NDP_frag24', '3j7u_NDP_frag23'])

        assert isinstance(response[0]['mol'], Mol)
        del response[0]['mol']
        del expected[0]['mol']
        assert response == expected
项目:kripodb    作者:3D-e-Chem    | 项目源码 | 文件源码
def test_fragments_by_id_withmolisnone(base_url, client):
    with requests_mock.mock() as m:
        expected = [
            {'smiles': None,
             'pdb_code': '3j7u',
             'pdb_title': 'Catalase structure determined by electron crystallography of thin 3D crystals',
             'atom_codes': 'PA,O1A,O2A,O5B,C5B,C4B,O4B,C3B,O3B,C2B,C1B,O3,PN,O1N,O2N,O5D,C5D,C4D,O4D,C3D,O3D,C2D,O2D,C1D,N1N,C2N,C3N,C7N,O7N,N7N,C4N,C5N,C6N',
             'uniprot_acc': 'P00432',
             'mol': None,
             'prot_chain': 'A', 'het_seq_nr': 602, 'het_code': 'NDP', 'prot_name': 'Catalase',
             'ec_number': '1.11.1.6', 'frag_nr': 24, 'frag_id': '3j7u_NDP_frag24', 'rowid': 7059,
             'uniprot_name': 'Catalase', 'nr_r_groups': 2, 'het_chain': 'A', 'hash_code': '6ef5a609fb192dba'}
        ]
        url = base_url + '/fragments?fragment_ids=3j7u_NDP_frag24,3j7u_NDP_frag23'
        m.get(url, json=expected)

        response = client.fragments_by_id(fragment_ids=['3j7u_NDP_frag24', '3j7u_NDP_frag23'])

        assert response == expected
项目:kripodb    作者:3D-e-Chem    | 项目源码 | 文件源码
def test_pharmacophores_somenotfound_incomplete(base_url, client, example1_phar):
    with requests_mock.mock() as m:
        m.get(base_url + '/fragments/3j7u_NDP_frag24.phar', text=example1_phar)
        notfound = {
            'detail': "Fragment with identifier '3j7u_NDP_frag23' not found",
            'identifier': '3j7u_NDP_frag23',
            'status': 404,
            'title': 'Not Found',
            'type': 'about:blank'
        }
        m.get(base_url + '/fragments/3j7u_NDP_frag23.phar', status_code=404, json=notfound, headers={'Content-Type': 'application/problem+json'})

        with pytest.raises(IncompletePharmacophores) as excinfo:
            client.pharmacophores(['3j7u_NDP_frag24', '3j7u_NDP_frag23'])

        assert excinfo.value.absent_identifiers == ['3j7u_NDP_frag23']
        assert excinfo.value.pharmacophores == [example1_phar, None]
项目:kripodb    作者:3D-e-Chem    | 项目源码 | 文件源码
def test_similarities__webbased_partbadid(base_url):
    queries = pd.Series(['3j7u_NDP_frag24', 'foo-bar'])
    body = [
        {'query_frag_id': '3j7u_NDP_frag24', 'hit_frag_id': '3j7u_NDP_frag23', 'score': 0.8991},
    ]

    with requests_mock.mock() as m:
        m.get(base_url + '/fragments/' + 'foo-bar' + '/similar?cutoff=0.55', status_code=404)
        url = base_url + '/fragments/' + '3j7u_NDP_frag24' + '/similar?cutoff=0.55'
        m.get(url, json=body)

        with pytest.raises(IncompleteHits) as e:
            similarities(queries, base_url, 0.55)

    assert_frame_equal(e.value.hits, pd.DataFrame(body, columns=['query_frag_id', 'hit_frag_id', 'score']))
    assert e.value.absent_identifiers == ['foo-bar']
项目:kripodb    作者:3D-e-Chem    | 项目源码 | 文件源码
def test_fragments_by_pdb_codes__usingwebservice_withbadid(base_url):
    with requests_mock.mock() as m:
        url = base_url + '/fragments?pdb_codes=0000'
        body = {
            'detail': "Fragment with identifier '0000' not found",
            'absent_identifiers': ['0000'],
            'fragments': [],
            'status': 404,
            'title': 'Not Found',
            'type': 'about:blank'
        }
        m.get(url, json=body, status_code=404, headers={'Content-Type': 'application/problem+json'})

        with pytest.raises(IncompleteFragments) as e:
            pdb_codes = pd.Series(['0000'])
            fragments_by_pdb_codes(pdb_codes, base_url)

        assert e.value.fragments.empty
        assert e.value.absent_identifiers == ['0000']
项目:kripodb    作者:3D-e-Chem    | 项目源码 | 文件源码
def test_fragments_by_id__usingwebservice_withbadid(base_url):
    with requests_mock.mock() as m:
        url = base_url + '/fragments?fragment_ids=foo-bar'
        body = {
            'detail': "Fragment with identifier 'foo-bar' not found",
            'absent_identifiers': ['foo-bar'],
            'fragments': [],
            'status': 404,
            'title': 'Not Found',
            'type': 'about:blank'
        }
        m.get(url, json=body, status_code=404, headers={'Content-Type': 'application/problem+json'})

        with pytest.raises(IncompleteFragments) as e:
            frag_ids = pd.Series(['foo-bar'])
            fragments_by_id(frag_ids, base_url)

        assert e.value.fragments.empty
        assert e.value.absent_identifiers == ['foo-bar']
项目:directory-ui-buyer    作者:uktrade    | 项目源码 | 文件源码
def test_verify_oauth2_code():
    with requests_mock.mock() as mock:
        mock.post(
            'https://account.companieshouse.gov.uk/oauth2/token',
            status_code=http.client.OK,
        )
        response = helpers.CompaniesHouseClient.verify_oauth2_code(
            code='123',
            redirect_uri='http://redirect.com',
        )
        assert response.status_code == 200

    request = mock.request_history[0]
    assert request.url == (
        'https://account.companieshouse.gov.uk/oauth2/token'
        '?grant_type=authorization_code'
        '&code=123'
        '&client_id=debug-client-id'
        '&client_secret=debug-client-secret'
        '&redirect_uri=http%3A%2F%2Fredirect.com'
    )
项目:hatchery    作者:ajk8    | 项目源码 | 文件源码
def test_convert_readme_to_rst(tmpdir):

    def _mock_pypandoc_convert_OSError(filename, format):
        raise OSError('this would happen if pandoc were not installed!')

    with tmpdir.as_cwd():
        with pytest.raises(project.ProjectError):
            project.convert_readme_to_rst()
        open('README', 'w').close()
        with pytest.raises(project.ProjectError):
            project.convert_readme_to_rst()
        os.remove('README')
        open('README.rst', 'w').close()
        with pytest.raises(project.ProjectError):
            project.convert_readme_to_rst()
        os.remove('README.rst')
        with open('README.md', 'w') as readme_md:
            readme_md.write('# heading')
        project.convert_readme_to_rst()
        assert helpers.regex_in_file(r'=======', 'README.rst') is True
        os.remove('README.rst')
        with mock.patch('pypandoc.convert', _mock_pypandoc_convert_OSError):
            with pytest.raises(project.ProjectError):
                project.convert_readme_to_rst()
项目:pyoozie    作者:Shopify    | 项目源码 | 文件源码
def test_test_connection(self, oozie_config):
        with requests_mock.mock() as m:
            m.get('http://localhost:11000/oozie/versions', text='[0, 1, 2]')
            client.OozieClient(**oozie_config)

        with pytest.raises(exceptions.OozieException) as err:
            with requests_mock.mock() as m:
                m.get('http://localhost:11000/oozie/versions', text='[0, 1]')
                client.OozieClient(**oozie_config)
        assert 'does not support API version 2' in str(err)

        with pytest.raises(exceptions.OozieException) as err:
            with requests_mock.mock() as m:
                m.get('http://localhost:11000/oozie/versions', status_code=404)
                client.OozieClient(**oozie_config)
        assert 'Unable to contact Oozie server' in str(err)

        with pytest.raises(exceptions.OozieException) as err:
            with requests_mock.mock() as m:
                m.get('http://localhost:11000/oozie/versions', text='>>> fail <<<')
                client.OozieClient(**oozie_config)
        assert 'Invalid response from Oozie server' in str(err)
项目:pyoozie    作者:Shopify    | 项目源码 | 文件源码
def test_request(self, api):
        with requests_mock.mock() as m:
            m.get('http://localhost:11000/oozie/v2/endpoint', text='{"result": "pass"}')
            result = api._request('GET', 'endpoint', None, None)
            assert result['result'] == 'pass'

        with requests_mock.mock() as m:
            m.get('http://localhost:11000/oozie/v2/endpoint')
            result = api._request('GET', 'endpoint', None, None)
            assert result is None

        with requests_mock.mock() as m:
            m.get('http://localhost:11000/oozie/v2/endpoint', text='>>> fail <<<')
            with pytest.raises(exceptions.OozieException) as err:
                api._request('GET', 'endpoint', None, None)
            assert 'Invalid response from Oozie server' in str(err)
项目:pyoozie    作者:Shopify    | 项目源码 | 文件源码
def test_admin_list_sharelib(self, api):
        reply = {
            'sharelib': [
                {'name': 'oozie'},
                {'name': 'hive'},
                {'name': 'distcp'},
                {'name': 'hcatalog'},
                {'name': 'sqoop'},
                {'name': 'mapreduce-streaming'},
                {'name': 'spark'},
                {'name': 'hive2'},
                {'name': 'pig'}
            ]
        }
        expected = ['oozie', 'hive', 'distcp', 'hcatalog', 'sqoop', 'mapreduce-streaming', 'spark', 'hive2', 'pig']
        with mock.patch.object(api, '_get', return_value=reply) as mock_get:
            assert api.admin_list_sharelib() == expected
            mock_get.assert_called_with('admin/list_sharelib')
项目:pyoozie    作者:Shopify    | 项目源码 | 文件源码
def test_jobs_query_coordinator_pagination(self, _, api):
        mock_results = iter(
            [
                {
                    'total': 501,
                    'coordinatorjobs': [{'coordJobId': '1-C'}, {'coordJobId': '2-C'}]
                },
                {
                    'total': 501,
                    'coordinatorjobs': [{'coordJobId': '3-C'}]
                }
            ]
        )

        with mock.patch.object(api, '_get') as mock_get:
            mock_get.side_effect = lambda url: next(mock_results)
            result = api._jobs_query(model.ArtifactType.Coordinator)
            assert len(result) == 3
            mock_get.assert_any_call('jobs?jobtype=coordinator&offset=1&len=500')
            mock_get.assert_any_call('jobs?jobtype=coordinator&offset=501&len=500')
            with pytest.raises(StopIteration):
                next(mock_results)