我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用requests_mock.mock()。
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)
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)
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)
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)
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('{}')
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))
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)
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, )
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, )
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, )
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)
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)
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'])
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'])
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)
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)
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)
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)
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)
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)
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 )
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} }
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)
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)
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)
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)
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])
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')
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)
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)
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']
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
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
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
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
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
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
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]
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
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
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]
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']
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']
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']
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' )
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()
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)
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)
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')
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)