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类mock()的实例源码
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)
