Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use _create_fixtures method in tempest
Best Python code snippet using tempest_python
test_marathon.py
Source:test_marathon.py 
...13def test_add_pod_returns_parsed_response_body():14 _assert_add_pod_returns_parsed_response_body({"id": "i-am-a-pod"})15 _assert_add_pod_returns_parsed_response_body(["another", "pod", "json"])16def test_remove_pod_has_default_force_value():17 marathon_client, rpc_client = _create_fixtures()18 marathon_client.remove_pod('foo')19 rpc_client.http_req.assert_called_with(20 http.delete, 'v2/pods/foo', params=None)21def test_remove_pod_builds_rpc_correctly():22 _assert_remove_pod_builds_rpc_correctly(23 pod_id='foo', force=False, path='v2/pods/foo', params=None)24 _assert_remove_pod_builds_rpc_correctly(25 pod_id='foo', force=True, path='v2/pods/foo', params={'force': 'true'})26 _assert_remove_pod_builds_rpc_correctly(27 pod_id='/foo bar/', force=False, path='v2/pods/foo%20bar', params=None)28def test_remove_pod_propagates_dcos_exception():29 _assert_method_propagates_rpc_dcos_exception(30 lambda marathon_client: marathon_client.remove_pod('bad'))31def test_show_pod_builds_rpc_correctly():32 _assert_show_pod_builds_rpc_correctly(33 pod_id='foo', path='v2/pods/foo::status')34 _assert_show_pod_builds_rpc_correctly(35 pod_id='/bar', path='v2/pods/bar::status')36 _assert_show_pod_builds_rpc_correctly(37 pod_id='baz/', path='v2/pods/baz::status')38 _assert_show_pod_builds_rpc_correctly(39 pod_id='foo bar', path='v2/pods/foo%20bar::status')40def test_show_pod_returns_response_json():41 _assert_show_pod_returns_response_json({'some': 'json'})42 _assert_show_pod_returns_response_json(['another', 'json', 'value'])43def test_show_pod_propagates_dcos_exception():44 _assert_method_propagates_rpc_dcos_exception(45 lambda marathon_client: marathon_client.show_pod('bad-req'))46def test_show_pod_raises_dcos_exception_for_json_parse_errors():47 _assert_method_raises_dcos_exception_for_json_parse_errors(48 lambda marathon_client: marathon_client.show_pod('bad-json'))49def test_list_pod_builds_rpc_correctly():50 marathon_client, rpc_client = _create_fixtures()51 marathon_client.list_pod()52 rpc_client.http_req.assert_called_with(http.get, 'v2/pods/::status')53def test_list_pod_returns_success_response_json():54 _assert_list_pod_returns_success_response_json(body_json={'some': 'json'})55 _assert_list_pod_returns_success_response_json(body_json=['a', 'b', 'c'])56def test_list_pod_propagates_rpc_dcos_exception():57 _assert_method_propagates_rpc_dcos_exception(58 lambda marathon_client: marathon_client.list_pod())59def test_list_pod_raises_dcos_exception_for_json_parse_errors():60 _assert_method_raises_dcos_exception_for_json_parse_errors(61 lambda marathon_client: marathon_client.list_pod())62def test_update_pod_executes_successfully():63 _assert_update_pod_executes_successfully(64 pod_id='foo',65 pod_json={'some', 'json'},66 force=False,67 path='v2/pods/foo',68 params=None,69 deployment_id='pod-deployment-id')70 _assert_update_pod_executes_successfully(71 pod_id='/foo bar/',72 pod_json={'some', 'json'},73 force=False,74 path='v2/pods/foo%20bar',75 params=None,76 deployment_id='pod-deployment-id')77 _assert_update_pod_executes_successfully(78 pod_id='foo',79 pod_json={'some', 'json'},80 force=True,81 path='v2/pods/foo',82 params={'force': 'true'},83 deployment_id='pod-deployment-id')84 _assert_update_pod_executes_successfully(85 pod_id='foo',86 pod_json={'something', 'different'},87 force=False,88 path='v2/pods/foo',89 params=None,90 deployment_id='pod-deployment-id')91 _assert_update_pod_executes_successfully(92 pod_id='foo',93 pod_json={'some', 'json'},94 force=False,95 path='v2/pods/foo',96 params=None,97 deployment_id='an-arbitrary-value')98def test_update_pod_has_default_force_value():99 marathon_client, rpc_client = _create_fixtures()100 marathon_client.update_pod('foo', {'some': 'json'})101 rpc_client.http_req.assert_called_with(102 http.put, 'v2/pods/foo', params=None, json={'some': 'json'})103def test_update_pod_propagates_rpc_dcos_exception():104 _assert_method_propagates_rpc_dcos_exception(105 lambda marathon_client:106 marathon_client.update_pod('foo', {'some': 'json'}))107def test_add_pod_raises_dcos_exception_if_deployment_id_missing():108 _assert_add_pod_raises_dcos_exception_if_deployment_id_missing(109 headers={'foo': 'bar'})110 _assert_add_pod_raises_dcos_exception_if_deployment_id_missing(111 headers={'marathon-deployment_ID': 'misspelled-field', 'zzz': 'aaa'})112def test_update_pod_raises_dcos_exception_if_deployment_id_missing():113 _assert_update_pod_raises_dcos_exception_if_deployment_id_missing(114 headers={'foo': 'bar'})115 _assert_update_pod_raises_dcos_exception_if_deployment_id_missing(116 headers={'marathon-deployment_ID': 'misspelled-field', 'zzz': 'aaa'})117def test_kill_pod_instances_executes_successfully():118 pod_id = 'foo'119 instance_ids = ['instance1', 'instance2']120 path = 'v2/pods/foo::instances'121 response_json = {'some': ['instance', 'status']}122 mock_response = mock.create_autospec(requests.Response)123 mock_response.json.return_value = response_json124 marathon_client, rpc_client = _create_fixtures()125 rpc_client.http_req.return_value = mock_response126 path_format_method_name = 'dcos.marathon.Client._marathon_id_path_format'127 path_format = mock.MagicMock()128 path_format.return_value = path129 with mock.patch(path_format_method_name, new=path_format):130 actual_json = marathon_client.kill_pod_instances(pod_id, instance_ids)131 path_format.assert_called_with('v2/pods/{}::instances', pod_id)132 rpc_client.http_req.assert_called_with(133 http.delete, path, json=instance_ids)134 assert actual_json == response_json135def test_kill_pod_instances_propagates_rpc_dcos_exception():136 _assert_method_propagates_rpc_dcos_exception(137 lambda marathon_client:138 marathon_client.kill_pod_instances('foo', ['inst1', 'inst2']))139def test_kill_pod_instances_raises_dcos_exception_for_json_parse_errors():140 _assert_method_raises_dcos_exception_for_json_parse_errors(141 lambda marathon_client: marathon_client.kill_pod_instances(142 'foo', ['bar', 'baz']))143@mock.patch('dcos.http.head')144def test_pod_feature_supported_gets_success_response(head_fn):145 def invoke_test_case(status_code):146 mock_response = mock.create_autospec(requests.Response)147 mock_response.status_code = status_code148 head_fn.return_value = mock_response149 rpc_client = rpcclient.RpcClient('http://base/url', timeout=42)150 marathon_client = marathon.Client(rpc_client)151 is_supported = marathon_client.pod_feature_supported()152 head_fn.assert_called_with('http://base/url/v2/pods', timeout=42)153 return is_supported154 assert invoke_test_case(status_code=200)155 assert invoke_test_case(status_code=204)156 assert not invoke_test_case(status_code=100)157 assert not invoke_test_case(status_code=302)158@mock.patch('dcos.http.head')159def test_pod_feature_supported_gets_404_response(head_fn):160 mock_response = mock.create_autospec(requests.Response)161 mock_response.status_code = 404162 head_fn.side_effect = DCOSHTTPException(mock_response)163 rpc_client = rpcclient.RpcClient('http://base/url', timeout=24)164 marathon_client = marathon.Client(rpc_client)165 assert not marathon_client.pod_feature_supported()166 head_fn.assert_called_with('http://base/url/v2/pods', timeout=24)167def test_pod_feature_supported_converts_http_exceptions_to_dcos_exceptions():168 @mock.patch('dcos.http.head')169 def test_case(head_fn, status_code):170 request = requests.Request(method='ANY', url='http://arbitrary/url')171 mock_response = mock.create_autospec(requests.Response)172 mock_response.status_code = status_code173 mock_response.reason = 'Arbitrary Reason'174 mock_response.request = request175 mock_response.json.side_effect = ValueError('empty body')176 head_fn.side_effect = DCOSHTTPException(mock_response)177 rpc_client = rpcclient.RpcClient('http://does/not/matter')178 marathon_client = marathon.Client(rpc_client)179 with pytest.raises(DCOSException) as exception_info:180 marathon_client.pod_feature_supported()181 message = rpcclient.RpcClient.response_error_message(182 status_code,183 reason='Arbitrary Reason',184 request_method='ANY',185 request_url='http://arbitrary/url',186 json_body=None)187 assert str(exception_info.value).endswith(message)188 test_case(status_code=400)189 test_case(status_code=401)190 test_case(status_code=403)191 test_case(status_code=409)192 test_case(status_code=422)193 test_case(status_code=500)194def test_pod_feature_supported_propagates_other_exceptions():195 _assert_pod_feature_supported_raises_exception(196 exception=DCOSException("BOOM!"))197 _assert_pod_feature_supported_raises_exception(198 exception=Exception("Uh oh"))199def test_rpc_client_http_req_calls_method_fn():200 def test_case(base_url, path, full_url):201 method_fn = mock.Mock()202 rpc_client = rpcclient.RpcClient(base_url)203 rpc_client.http_req(method_fn, path)204 method_fn.assert_called_with(full_url,205 timeout=http.DEFAULT_TIMEOUT)206 test_case(207 base_url='http://base/url',208 path='some/path',209 full_url='http://base/url/some/path')210 test_case(211 base_url='http://base/url',212 path='different/thing',213 full_url='http://base/url/different/thing')214 test_case(215 base_url='gopher://different/thing',216 path='some/path',217 full_url='gopher://different/thing/some/path')218 test_case(219 base_url='http://base/without/slash',220 path='/path/with/slash',221 full_url='http://base/without/slash/path/with/slash')222 test_case(223 base_url='http://base/with/slash/',224 path='path/without/slash',225 full_url='http://base/with/slash/path/without/slash')226 test_case(227 base_url='http://base/with/slash/',228 path='/path/with/slash',229 full_url='http://base/with/slash/path/with/slash')230def test_rpc_client_http_req_passes_args_to_method_fn():231 method_fn = mock.Mock()232 rpc_client = rpcclient.RpcClient('http://base/url')233 rpc_client.http_req(method_fn, 'some/path', 'foo', 42)234 method_fn.assert_called_with('http://base/url/some/path',235 'foo',236 42,237 timeout=http.DEFAULT_TIMEOUT)238def test_rpc_client_http_req_passes_kwargs_to_method_fn():239 method_fn = mock.Mock()240 rpc_client = rpcclient.RpcClient('http://base/url')241 rpc_client.http_req(method_fn, 'some/path', foo='bar', baz=42)242 method_fn.assert_called_with('http://base/url/some/path',243 foo='bar',244 baz=42,245 timeout=http.DEFAULT_TIMEOUT)246def test_rpc_client_http_req_kwarg_timeout_overrides_default():247 method_fn = mock.Mock()248 rpc_client = rpcclient.RpcClient('http://base/url')249 rpc_client.http_req(method_fn, 'some/path', timeout=42)250 method_fn.assert_called_with('http://base/url/some/path', timeout=42)251def test_rpc_client_http_req_set_timeout_in_constructor():252 method_fn = mock.Mock()253 rpc_client = rpcclient.RpcClient('http://base/url', 24)254 rpc_client.http_req(method_fn, 'some/path')255 method_fn.assert_called_with('http://base/url/some/path', timeout=24)256def test_rpc_client_http_req_returns_method_fn_result():257 def test_case(expected):258 def method_fn(*args, **kwargs):259 return expected260 rpc_client = rpcclient.RpcClient('http://base/url')261 actual = rpc_client.http_req(method_fn, 'some/path')262 assert actual == expected263 test_case(['the', 'result'])264 test_case({'another': 'result'})265def test_rpc_client_http_req_propagates_method_fn_exception_1():266 request = requests.Request(method='ANY', url='http://arbitrary/url')267 # Need the mock so that the `.json()` method can be overridden268 response = mock.create_autospec(requests.Response)269 response.status_code = 403270 response.reason = 'Forbidden'271 response.request = request272 response.json.side_effect = Exception('not JSON')273 def method_fn(*args, **kwargs):274 raise DCOSHTTPException(response)275 rpc_client = rpcclient.RpcClient('http://base/url')276 with pytest.raises(DCOSException) as e:277 rpc_client.http_req(method_fn, 'some/path')278 expected_message = rpcclient.RpcClient.response_error_message(279 status_code=403,280 reason='Forbidden',281 request_method='ANY',282 request_url='http://arbitrary/url',283 json_body=None)284 assert str(e).endswith(expected_message)285def test_rpc_client_http_req_propagates_method_fn_exception_2():286 request = requests.Request(method='NONE', url='http://host/path')287 # Need the mock so that the `.json()` method can be overridden288 response = mock.create_autospec(requests.Response)289 response.status_code = 422290 response.reason = 'Something Bad'291 response.request = request292 response.json.return_value = {'message': 'BOOM!'}293 def method_fn(*args, **kwargs):294 raise DCOSHTTPException(response)295 rpc_client = rpcclient.RpcClient('http://base/url')296 with pytest.raises(DCOSException) as e:297 rpc_client.http_req(method_fn, 'some/path')298 expected_message = rpcclient.RpcClient.response_error_message(299 status_code=422,300 reason='Something Bad',301 request_method='None',302 request_url='http://host/path',303 json_body={'message': 'BOOM!'})304 assert str(e).endswith(expected_message)305def test_error_json_schema_is_valid():306 error_json_schema = rpcclient.load_error_json_schema()307 jsonschema.Draft4Validator.check_schema(error_json_schema)308def test_response_error_message_with_400_status_no_json():309 _assert_response_error_message_with_400_status_no_json(310 reason=_REASON_1,311 request_method=_METHOD_1,312 request_url=_URL_1)313 _assert_response_error_message_with_400_status_no_json(314 reason=_REASON_2,315 request_method=_METHOD_2,316 request_url=_URL_2)317def test_response_error_message_with_400_status_json():318 _assert_response_error_message_with_400_status_json(319 _RESPONSE_JSON_1, _PRINTED_JSON_1)320 _assert_response_error_message_with_400_status_json(321 _RESPONSE_JSON_2, _PRINTED_JSON_2)322def test_res_err_msg_with_409_status():323 def test_case(request_url, expected_resource):324 actual = rpcclient.RpcClient.response_error_message(325 status_code=409,326 reason=_REASON_X,327 request_method=_METHOD_X,328 request_url=request_url,329 json_body=None)330 pattern = r'Changes blocked: deployment already in progress for (.*)\.'331 _assert_matches_with_groups(pattern, actual, (expected_resource,))332 test_case('http://marathon/v2/apps', 'app')333 test_case('http://marathon/v2/groups/group-id', 'group')334 test_case('http://marathon/v2/pods/', 'pod')335 test_case('http://marathon/v2/thingies/foo', 'resource')336 test_case('http://dcos/service/marathon/v2/apps/bar', 'app')337 test_case('http://pods-app.com/service/marathon/v2/groups/baz', 'group')338def test_response_error_message_with_other_status_no_json():339 _assert_response_error_message_with_other_status_no_json(340 status_code=401,341 reason=_REASON_1,342 request_url=_URL_1)343 _assert_response_error_message_with_other_status_no_json(344 status_code=401,345 reason=_REASON_2,346 request_url=_URL_2)347 _assert_response_error_message_with_other_status_no_json(348 status_code=403,349 reason=_REASON_X,350 request_url=_URL_X)351 _assert_response_error_message_with_other_status_no_json(352 status_code=404,353 reason=_REASON_X,354 request_url=_URL_X)355 _assert_response_error_message_with_other_status_no_json(356 status_code=422,357 reason=_REASON_X,358 request_url=_URL_X)359def test_response_error_message_with_other_status_json_has_message():360 _assert_response_error_message_with_other_status_json_has_message(361 status_code=401,362 json_message=_MESSAGE_1)363 _assert_response_error_message_with_other_status_json_has_message(364 status_code=401,365 json_message=_MESSAGE_2)366 _assert_response_error_message_with_other_status_json_has_message(367 status_code=403,368 json_message=_MESSAGE_X)369 _assert_response_error_message_with_other_status_json_has_message(370 status_code=404,371 json_message=_MESSAGE_X)372 _assert_response_error_message_with_other_status_json_has_message(373 status_code=422,374 json_message=_MESSAGE_X)375def test_res_err_msg_with_other_status_json_no_message_has_valid_errors():376 _assert_res_err_msg_with_other_status_json_no_message_has_valid_errors(377 status_code=401,378 errors_json=[{'error': 'err1'}, {'error': 'err2'}],379 errors_str='err1\nerr2')380 _assert_res_err_msg_with_other_status_json_no_message_has_valid_errors(381 status_code=401,382 errors_json=[{'error': 'foo'}, {'error': 'bar'}, {'error': 'baz'}],383 errors_str='foo\nbar\nbaz')384 _assert_res_err_msg_with_other_status_json_no_message_has_valid_errors(385 status_code=403,386 errors_json=[{'error': 'foo'}, {'error': 'bar'}, {'error': 'baz'}],387 errors_str='foo\nbar\nbaz')388 _assert_res_err_msg_with_other_status_json_no_message_has_valid_errors(389 status_code=404,390 errors_json=[{'error': 'foo'}, {'error': 'bar'}, {'error': 'baz'}],391 errors_str='foo\nbar\nbaz')392 _assert_res_err_msg_with_other_status_json_no_message_has_valid_errors(393 status_code=422,394 errors_json=[{'error': 'foo'}, {'error': 'bar'}, {'error': 'baz'}],395 errors_str='foo\nbar\nbaz')396def test_res_err_msg_with_other_status_invalid_error_json():397 # Is not an object398 _assert_res_err_msg_with_other_status_invalid_json(401, 'Error!')399 _assert_res_err_msg_with_other_status_invalid_json(401, ['Error!'])400 # Missing both message and errors fields401 _assert_res_err_msg_with_other_status_invalid_json(401, {})402 _assert_res_err_msg_with_other_status_invalid_json(403 401, {'foo': 5, 'bar': 'x'})404 # Has non-str message405 _assert_res_err_msg_with_other_status_invalid_json(401, {'message': 42})406 _assert_res_err_msg_with_other_status_invalid_json(407 401, {'message': ['Oops!']})408 # Has non-array errors409 _assert_res_err_msg_with_other_status_invalid_json(401, {'errors': 42})410 _assert_res_err_msg_with_other_status_invalid_json(411 401, {'errors': {'error': 5}})412 # Errors array has non-object elements413 _assert_res_err_msg_with_other_status_invalid_json(414 401, {'errors': [42, True]})415 _assert_res_err_msg_with_other_status_invalid_json(416 401, {'errors': [{'error': 'BOOM!'}, 'not_an_error']})417 # Errors array has objects without `error` field418 _assert_res_err_msg_with_other_status_invalid_json(419 401, {'errors': [{'error': 'BOOM!'}, {'foo': 'bar'}]})420 # Errors array has objects with non-str `error` field421 _assert_res_err_msg_with_other_status_invalid_json(422 401, {'errors': [{'error': 'BOOM!'}, {'error': 42}]})423 # Other status codes424 _assert_res_err_msg_with_other_status_invalid_json(425 403, {'errors': [{'error': 'BOOM!'}, {'error': 42}]})426 _assert_res_err_msg_with_other_status_invalid_json(427 404, {'errors': [{'error': 'BOOM!'}, {'error': 42}]})428 _assert_res_err_msg_with_other_status_invalid_json(429 422, {'errors': [{'error': 'BOOM!'}, {'error': 42}]})430def _assert_add_pod_puts_json_in_request_body(pod_json):431 rpc_client = mock.create_autospec(rpcclient.RpcClient)432 client = marathon.Client(rpc_client)433 client.add_pod(pod_json)434 rpc_client.http_req.assert_called_with(http.post, 'v2/pods', json=pod_json)435def _assert_add_pod_returns_parsed_response_body(response_json):436 headers = {'Marathon-Deployment-Id': "XYZ"}437 mock_response = _pod_response_fixture(headers)438 mock_response.json.return_value = response_json439 marathon_client, rpc_client = _create_fixtures()440 rpc_client.http_req.return_value = mock_response441 assert marathon_client.add_pod({'some': 'json'}) == "XYZ"442def _assert_remove_pod_builds_rpc_correctly(pod_id, force, path, params):443 marathon_client, rpc_client = _create_fixtures()444 marathon_client.remove_pod(pod_id, force)445 rpc_client.http_req.assert_called_with(http.delete, path, params=params)446def _assert_show_pod_builds_rpc_correctly(pod_id, path):447 marathon_client, rpc_client = _create_fixtures()448 marathon_client.show_pod(pod_id)449 rpc_client.http_req.assert_called_with(http.get, path)450def _assert_show_pod_returns_response_json(expected):451 marathon_client, rpc_client = _create_fixtures()452 mock_response = mock.create_autospec(requests.Response)453 mock_response.json.return_value = expected454 rpc_client.http_req.return_value = mock_response455 response_json = marathon_client.show_pod('arbitrary-id')456 assert response_json == expected457def _assert_list_pod_returns_success_response_json(body_json):458 marathon_client, rpc_client = _create_fixtures()459 mock_response = mock.create_autospec(requests.Response)460 mock_response.json.return_value = body_json461 rpc_client.http_req.return_value = mock_response462 assert marathon_client.list_pod() == body_json463def _assert_update_pod_executes_successfully(464 pod_id, pod_json, force, path, params, deployment_id):465 headers = {'Marathon-Deployment-Id': deployment_id}466 mock_response = _pod_response_fixture(headers)467 marathon_client, rpc_client = _create_fixtures()468 rpc_client.http_req.return_value = mock_response469 actual_return = marathon_client.update_pod(pod_id, pod_json, force)470 rpc_client.http_req.assert_called_with(471 http.put, path, params=params, json=pod_json)472 assert actual_return == deployment_id473def _assert_method_raises_dcos_exception_for_json_parse_errors(invoke_method):474 def assert_test_case(non_json):475 mock_response = mock.create_autospec(requests.Response)476 mock_response.json.side_effect = Exception()477 mock_response.text = non_json478 marathon_client, rpc_client = _create_fixtures()479 rpc_client.http_req.return_value = mock_response480 with pytest.raises(DCOSException) as exception_info:481 invoke_method(marathon_client)482 pattern = ('Error: Response from Marathon was not in expected JSON '483 'format:\n(.*)')484 actual_error = str(exception_info.value)485 _assert_matches_with_groups(pattern, actual_error, (non_json,))486 assert_test_case('not-json')487 assert_test_case('{"oops"}')488def _assert_add_pod_raises_dcos_exception_if_deployment_id_missing(headers):489 marathon_client, rpc_client = _create_fixtures()490 rpc_client.http_req.return_value = _pod_response_fixture(headers)491 result = marathon_client.add_pod({'some': 'json'})492 assert result is None493def _assert_update_pod_raises_dcos_exception_if_deployment_id_missing(headers):494 marathon_client, rpc_client = _create_fixtures()495 rpc_client.http_req.return_value = _pod_response_fixture(headers)496 with pytest.raises(DCOSException) as exception_info:497 marathon_client.update_pod('foo', {'some': 'json'})498 expected_error = ('Error: missing "Marathon-Deployment-Id" header from '499 'Marathon response')500 assert str(exception_info.value) == expected_error501def _assert_method_propagates_rpc_dcos_exception(invoke_method):502 marathon_client, rpc_client = _create_fixtures()503 rpc_client.http_req.side_effect = DCOSException('BOOM!')504 with pytest.raises(DCOSException) as exception_info:505 invoke_method(marathon_client)506 assert str(exception_info.value) == 'BOOM!'507@mock.patch('dcos.http.head')508def _assert_pod_feature_supported_raises_exception(head_fn, exception):509 rpc_client = rpcclient.RpcClient('http://base/url', timeout=22)510 marathon_client = marathon.Client(rpc_client)511 head_fn.side_effect = exception512 with pytest.raises(exception.__class__) as exception_info:513 marathon_client.pod_feature_supported()514 assert exception_info.value == exception515def _assert_response_error_message_with_400_status_no_json(516 reason, request_method, request_url):517 message = rpcclient.RpcClient.response_error_message(518 status_code=400,519 reason=reason,520 request_method=request_method,521 request_url=request_url,522 json_body=None)523 pattern = r'Error on request \[(.*) (.*)\]: HTTP 400: (.*)'524 groups = (request_method, request_url, reason)525 _assert_matches_with_groups(pattern, message, groups)526def _assert_response_error_message_with_400_status_json(527 response_json, printed_json):528 message = rpcclient.RpcClient.response_error_message(529 status_code=400,530 reason=_REASON_X,531 request_method=_METHOD_X,532 request_url=_URL_X,533 json_body=response_json)534 error_line, json_lines = message.split('\n', 1)535 pattern = r'Error on request \[(.*) (.*)\]: HTTP 400: (.*):'536 groups = (_METHOD_X, _URL_X, _REASON_X)537 # python 2.7 has trailing whitespace in pretty json538 json_lines = re.sub(r'\s+\n', '\n', json_lines)539 _assert_matches_with_groups(pattern, error_line, groups)540 assert json_lines == printed_json541def _assert_response_error_message_with_other_status_no_json(542 status_code, reason, request_url):543 message = rpcclient.RpcClient.response_error_message(544 status_code=status_code,545 reason=reason,546 request_method=_METHOD_X,547 request_url=request_url,548 json_body=None)549 pattern = r'Error decoding response from \[(.*)\]: HTTP (.*): (.*)'550 groups = (request_url, str(status_code), reason)551 _assert_matches_with_groups(pattern, message, groups)552def _assert_response_error_message_with_other_status_json_has_message(553 status_code, json_message):554 error_message = rpcclient.RpcClient.response_error_message(555 status_code=status_code,556 reason=_REASON_X,557 request_method=_METHOD_X,558 request_url=_URL_X,559 json_body={'message': json_message})560 pattern = r'Error: (.*)'561 _assert_matches_with_groups(pattern, error_message, (json_message,))562def _assert_res_err_msg_with_other_status_json_no_message_has_valid_errors(563 status_code, errors_json, errors_str):564 message = rpcclient.RpcClient.response_error_message(565 status_code=status_code,566 reason=_REASON_X,567 request_method=_METHOD_X,568 request_url=_URL_X,569 json_body={'errors': errors_json})570 pattern = (r'Service likely misconfigured\. Please check your proxy or '571 r'Service URL settings\. See dcos config --help\. (.*)')572 _assert_matches_with_groups(pattern, message, (errors_str,))573def _assert_res_err_msg_with_other_status_invalid_json(status_code, json_body):574 actual = rpcclient.RpcClient.response_error_message(575 status_code=status_code,576 reason=_REASON_X,577 request_method=_METHOD_X,578 request_url=_URL_X,579 json_body=json_body)580 expected = ('Service likely misconfigured. Please check your proxy or '581 'Service URL settings. See dcos config --help. ')582 assert actual == expected583def _assert_matches_with_groups(pattern, text, groups):584 match = re.match(pattern, text, flags=re.DOTALL)585 assert match586 assert match.groups() == groups587def _create_fixtures():588 rpc_client = mock.create_autospec(rpcclient.RpcClient)589 marathon_client = marathon.Client(rpc_client)590 return marathon_client, rpc_client591def _pod_response_fixture(headers=None):592 mock_response = mock.create_autospec(requests.Response)593 headers = CaseInsensitiveDict({} if headers is None else headers)594 mock_response.headers = headers595 return mock_response596_MESSAGE_1 = 'Oops!'597_MESSAGE_2 = 'Uh oh!'598_MESSAGE_X = "D'oh!"599_METHOD_1 = 'FOO'600_METHOD_2 = 'BAR'601_METHOD_X = 'ANY'...
polyglot_parser_test.py
Source:polyglot_parser_test.py
...21# Helper functions used in tests22def _parse_json(path):23 with open(path, 'r') as file:24 return json.loads('\n'.join(file.readlines()))25def _create_fixtures(test_folder, add_main=False):26 test_file = f'{test_folder}_main.py' if add_main else f'{test_folder}.py'27 test_folder_path = path.join(TEST_DATA_PATH, test_folder)28 repo_json = path.join(test_folder_path, 'polyglot_snippet_data.json')29 source_path = path.join(test_folder_path, test_file)30 source_methods_json = _parse_json(repo_json)['snippets']31 source_methods = [pdd.PolyglotDriftData(**json_dict)32 for json_dict in source_methods_json]33 source_region_tags = \34 polyglot_parser.get_region_tag_regions(source_path)[0]35 for idx, method in enumerate(source_methods):36 source_methods[idx] = \37 polyglot_parser.add_region_tags_to_method(38 method, source_region_tags39 )40 return source_methods41class PolyglotParserTests(unittest.TestCase):42 def test_ignores_ignored_method_names(self):43 source_methods = _create_fixtures('edge_cases')44 method = source_methods[0]45 # this should NOT contain 'main_method'46 self.assertEqual(method.region_tags, ['not_main'])47 def test_region_tags_nested(self):48 source_methods = _create_fixtures('nested_tags')49 method_1 = source_methods[0]50 method_2 = source_methods[1]51 method_1.region_tags.sort()52 self.assertEqual(53 method_1.region_tags, ['nested_tag', 'root_tag'])54 self.assertEqual(method_2.region_tags, ['root_tag'])55 def test_handle_multi_block_region_tags(self):56 source_methods = _create_fixtures('nested_tags')57 assert len(source_methods) == 358 method_1 = source_methods[1]59 method_2 = source_methods[2]60 self.assertEqual(method_1.region_tags, ['root_tag'])61 self.assertEqual(method_2.region_tags, ['root_tag'])62 def test_flask_router_parser(self):63 source_methods = _create_fixtures('flask', True)64 source_methods = [method for method in source_methods if65 method.parser == 'flask_router']66 method = source_methods[0]67 self.assertEqual(method.region_tags, ['sample_route'])68 def test_direct_invocation_parser(self):69 source_methods = _create_fixtures('http', True)70 method = source_methods[0]71 assert 'functions_helloworld_get' in method.region_tags72 def test_webapp2_parser(self):73 source_methods = _create_fixtures('webapp2', True)74 method = source_methods[-1]75 self.assertEqual(method.region_tags, ['sign_handler'])76 def test_ignores_exclude_tags(self):77 source_methods = _create_fixtures('exclude_tags', True)78 # make sure the file was parsed properly...
mixin.py
Source:mixin.py
...17 session = None18 fixtures = []19 fixtures_setup_class = False20 fixtures_paths = None21 def _create_fixtures(self):22 if self.base is None:23 return24 if self.session is None:25 return26 if self.fixtures_paths is None:27 return28 create_table(self.base, self.session)29 fixtures = load_fixture_files(self.fixtures_paths, self.fixtures)30 load_fixtures(self.session, fixtures)31 def _drop_fixtures(self):32 drop_table(self.base, self.session)33 @classmethod34 def setUpClass(cls):35 if cls.fixtures_setup_class is True:36 cls._create_fixtures(cls)37 @classmethod38 def tearDownClass(cls):39 if cls.fixtures_setup_class is True:40 cls._drop_fixtures(cls)41 def setUp(self):42 if self.fixtures_setup_class is True:43 return44 self._create_fixtures()45 def tearDown(self):46 if self.fixtures_setup_class is True:47 return...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run tempest automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
