How to use test_include_key method in assertpy

Best Python code snippet using assertpy_python

test_dict_compare.py

Source:test_dict_compare.py

...172        assert_that({(1,):'a'}).is_equal_to({(1,):'b'}, ignore=((2,),))173        fail('should have raised error')174    except AssertionError as ex:175        assert_that(str(ex)).is_equal_to("Expected <{(1,): 'a'}> to be equal to <{(1,): 'b'}> ignoring keys <2>, but was not.")176def test_include_key():177    assert_that({'a':1,'b':2}).is_equal_to({'a':1}, include='a')178    assert_that({'a':1,'b':{'x':2,'y':3}}).is_equal_to({'a':1}, include='a')179    assert_that({'a':1,'b':{'x':2,'y':3}}).is_equal_to({'b':{'x':2,'y':3}}, include='b')180def test_include_list_of_keys():181    assert_that({'a':1,'b':2,'c':3}).is_equal_to({'a':1,'b':2,'c':3}, include=['a','b','c'])182    assert_that({'a':1,'b':2,'c':3}).is_equal_to({'a':1,'b':2}, include=['a','b'])183    assert_that({'a':1,'b':2,'c':3}).is_equal_to({'a':1}, include=['a'])184    assert_that({'a':1,'b':2,'c':3}).is_equal_to({'b':2}, include=['b'])185    assert_that({'a':1,'b':2,'c':3}).is_equal_to({'c':3}, include=['c'])186def test_include_deep_key():187    assert_that({'a':1,'b':{'x':2,'y':3}}).is_equal_to({'b':{'x':2,'y':3}}, include=('b'))188    assert_that({'a':1,'b':{'x':2,'y':3}}).is_equal_to({'b':{'x':2}}, include=('b','x'))189    assert_that({'a':1,'b':{'c':2,'d':{'e':3,'f':{'x':4,'y':5}}}}).is_equal_to({'b':{'c':2,'d':{'e':3,'f':{'x':4,'y':5}}}}, include=('b'))190    assert_that({'a':1,'b':{'c':2,'d':{'e':3,'f':{'x':4,'y':5}}}}).is_equal_to({'b':{'c':2}}, include=('b','c'))...

test_cromwell_api.py

Source:test_cromwell_api.py

1#!/usr/bin/env python2import io3import json4import os5import requests6import requests_mock7import six8import tempfile9import unittest10six.add_move(six.MovedModule('mock', 'mock', 'unittest.mock'))11from six.moves import mock  # noqa12from cromwell_tools.cromwell_api import CromwellAPI  # noqa13from cromwell_tools.cromwell_auth import CromwellAuth  # noqa14from cromwell_tools import utilities as utils  # noqa15class TestAPI(unittest.TestCase):16    @classmethod17    def setUpClass(cls):18        # Change to test directory, as tests may have been invoked from another dir19        dir_ = os.path.abspath(os.path.dirname(__file__))20        os.chdir(dir_)21    def setUp(self):22        self.wdl_file = io.BytesIO(b"wdl_file_content")23        self.zip_file = io.BytesIO(b"zip_file_content")24        self.inputs_file = io.BytesIO(b"inputs_file_content")25        self.options_file = io.BytesIO(b"options_file_content")26        self.label = io.BytesIO(b'{"test-label-key": "test-label-value"}')27        self.auth_options = self.set_up_auth()28    @mock.patch(29        'cromwell_tools.cromwell_auth.CromwellAuth.from_service_account_key_file'30    )31    def set_up_auth(self, mock_header):32        # set up authentication options for the tests33        temp_dir = tempfile.mkdtemp()34        secrets_file = temp_dir + 'fake_secrets.json'35        service_account_key = os.path.join(temp_dir, 'fake_key.json')36        username = "fake_user"37        password = "fake_password"38        url = "https://fake_url"39        auth = {"url": url, "username": username, "password": password}40        with open(secrets_file, 'w') as f:41            json.dump(auth, f)42        mock_header.return_value = CromwellAuth(43            url=url, header={"Authorization": "bearer fake_token"}, auth=None44        )45        auth_options = (46            CromwellAuth.harmonize_credentials(**auth),  # HTTPBasicAuth47            CromwellAuth.harmonize_credentials(48                **{"secrets_file": secrets_file}49            ),  # Secret file50            CromwellAuth.harmonize_credentials(51                **{"service_account_key": service_account_key, "url": url}52            ),  # OAuth53            CromwellAuth.harmonize_credentials(url=url),  # No Auth54        )55        return auth_options56    def _submit_workflows(self, cromwell_auth, mock_request, _request_callback):57        mock_request.post(58            cromwell_auth.url + '/api/workflows/v1', json=_request_callback59        )60        return CromwellAPI.submit(61            auth=cromwell_auth,62            wdl_file=self.wdl_file,63            inputs_files=self.inputs_file,64            options_file=self.options_file,65            dependencies=self.zip_file,66            label_file=self.label,67        )68    @requests_mock.mock()69    def test_submit_workflow(self, mock_request):70        def _request_callback(request, context):71            context.status_code = 20072            context.headers['test'] = 'header'73            return {'request': {'body': "content"}}74        for cromwell_auth in self.auth_options:75            result = self._submit_workflows(76                cromwell_auth, mock_request, _request_callback77            )78            self.assertEqual(result.status_code, 200)79            self.assertEqual(result.headers.get('test'), 'header')80    @requests_mock.mock()81    def test_submit_workflow_handlers_error_response(self, mock_request):82        def _request_callback(request, context):83            context.status_code = 50084            context.headers['test'] = 'header'85            return {'status': 'error', 'message': 'Internal Server Error'}86        # Check request actions87        for cromwell_auth in self.auth_options:88            with self.assertRaises(requests.HTTPError):89                self._submit_workflows(90                    cromwell_auth, mock_request, _request_callback91                ).raise_for_status()92    @requests_mock.mock()93    def test_query_workflows_returns_200(self, mock_request):94        query_dict = {95            'status': ['Running', 'Failed'],96            'label': {'label_key1': 'label_value1', 'label_key2': 'label_value2'},97        }98        def _request_callback(request, context):99            context.status_code = 200100            context.headers['test'] = 'header'101            return {102                'results': [103                    {104                        'name': 'workflow1',105                        'submission': 'submission1',106                        'id': 'id1',107                        'status': 'Failed',108                        'start': 'start1',109                        'end': 'end1',110                    },111                    {112                        'name': 'workflow2',113                        'submission': 'submission2',114                        'id': 'id2',115                        'status': 'Running',116                        'start': 'start2',117                        'end': 'end2',118                    },119                ],120                'totalResultsCount': 2,121            }122        for cromwell_auth in self.auth_options:123            mock_request.post(124                '{}/api/workflows/v1/query'.format(cromwell_auth.url),125                json=_request_callback,126            )127            result = CromwellAPI.query(query_dict, cromwell_auth)128            self.assertEqual(result.status_code, 200)129            self.assertEqual(result.json()['totalResultsCount'], 2)130    def test_compose_query_params_can_compose_simple_query_dicts(self):131        query_dict = {132            'status': 'Running',133            'start': '2018-01-01T00:00:00.000Z',134            'end': '2018-01-01T12:00:00.000Z',135            'label': {'Comment': 'test'},136            'page': 1,137            'pageSize': 10,138        }139        expect_params = [140            {'status': 'Running'},141            {'start': '2018-01-01T00:00:00.000Z'},142            {'end': '2018-01-01T12:00:00.000Z'},143            {'label': 'Comment:test'},144            {'page': '1'},145            {'pageSize': '10'},146        ]147        six.assertCountEqual(148            self, CromwellAPI._compose_query_params(query_dict), expect_params149        )150    def test_compose_query_params_can_compose_nested_query_dicts(self):151        query_dict = {152            'status': ['Running', 'Failed', 'Submitted'],153            'start': '2018-01-01T00:00:00.000Z',154            'end': '2018-01-01T12:00:00.000Z',155            'label': {'Comment1': 'test1', 'Comment2': 'test2', 'Comment3': 'test3'},156        }157        expect_params = [158            {'status': 'Running'},159            {'status': 'Failed'},160            {'status': 'Submitted'},161            {'start': '2018-01-01T00:00:00.000Z'},162            {'end': '2018-01-01T12:00:00.000Z'},163            {'label': 'Comment1:test1'},164            {'label': 'Comment2:test2'},165            {'label': 'Comment3:test3'},166        ]167        six.assertCountEqual(168            self, CromwellAPI._compose_query_params(query_dict), expect_params169        )170    def test_compose_query_params_can_convert_bools_within_query_dicts(self):171        query_dict = {172            'status': ['Running', 'Failed', 'Submitted'],173            'start': '2018-01-01T00:00:00.000Z',174            'end': '2018-01-01T12:00:00.000Z',175            'label': {'Comment1': 'test1', 'Comment2': 'test2', 'Comment3': 'test3'},176            'includeSubworkflows': True,177        }178        expect_params = [179            {'status': 'Running'},180            {'status': 'Failed'},181            {'status': 'Submitted'},182            {'start': '2018-01-01T00:00:00.000Z'},183            {'end': '2018-01-01T12:00:00.000Z'},184            {'label': 'Comment1:test1'},185            {'label': 'Comment2:test2'},186            {'label': 'Comment3:test3'},187            {'includeSubworkflows': 'true'},188        ]189        six.assertCountEqual(190            self, CromwellAPI._compose_query_params(query_dict), expect_params191        )192    def test_compose_query_params_raises_error_for_invalid_query_dict_that_has_multiple_values_for_exclusive_keys(193        self,194    ):195        query_dict = {196            'status': ['Running', 'Failed', 'Submitted'],197            'start': ['2018-01-01T00:00:00.000Z', '2018-01-02T00:00:00.000Z'],198            'end': '2018-01-01T12:00:00.000Z',199            'label': {'Comment1': 'test1', 'Comment2': 'test2', 'Comment3': 'test3'},200        }201        with self.assertRaises(ValueError):202            CromwellAPI._compose_query_params(query_dict)203    @requests_mock.mock()204    def test_path_labels_returns_200(self, mock_request):205        workflow_id = 'labeltest'206        new_label = {'foo': 'bar'}207        def _request_callback(request, context):208            context.status_code = 200209            context.headers['test'] = 'header'210            return {'id': request.url.split('/')[-2], 'labels': new_label}211        for cromwell_auth in self.auth_options:212            mock_request.patch(213                '{0}/api/workflows/v1/{1}/labels'.format(214                    cromwell_auth.url, workflow_id215                ),216                json=_request_callback,217            )218            result = CromwellAPI.patch_labels(workflow_id, new_label, cromwell_auth)219            self.assertEqual(result.status_code, 200)220            self.assertEqual(result.json()['id'], workflow_id)221            self.assertEqual(result.json()['labels'], new_label)222    @requests_mock.mock()223    def test_release_onhold_returns_200(self, mock_request):224        workflow_id = '12345abcde'225        def _request_callback(request, context):226            context.status_code = 200227            context.headers['test'] = 'header'228            return {'id': request.url.split('/')[-2], 'status': 'Submitted'}229        for cromwell_auth in self.auth_options:230            mock_request.post(231                '{0}/api/workflows/v1/{1}/releaseHold'.format(232                    cromwell_auth.url, workflow_id233                ),234                json=_request_callback,235            )236            result = CromwellAPI.release_hold(workflow_id, cromwell_auth)237            self.assertEqual(result.status_code, 200)238            self.assertEqual(result.json()['id'], workflow_id)239            self.assertEqual(result.json()['status'], 'Submitted')240    @requests_mock.mock()241    def test_release_workflow_that_is_not_on_hold_returns_error(self, mock_request):242        workflow_id = 'test'243        def _request_callback(request, context):244            context.status_code = 403245            context.headers['test'] = 'header'246            return {247                'status': 'error',248                'message': 'Couldn\'t change status of workflow {} to \'Submitted\' because the workflow is not in '249                '\'On Hold\' state'.format(request.url.split('/')[-2]),250            }251        for cromwell_auth in self.auth_options:252            mock_request.post(253                '{0}/api/workflows/v1/{1}/releaseHold'.format(254                    cromwell_auth.url, workflow_id255                ),256                json=_request_callback,257            )258            with self.assertRaises(requests.exceptions.HTTPError):259                CromwellAPI.release_hold(workflow_id, cromwell_auth).raise_for_status()260    @requests_mock.mock()261    def test_metadata_returns_200(self, mock_request):262        workflow_id = '12345abcde'263        test_include_key = 'workflow'264        def _request_callback(request, context):265            context.status_code = 200266            context.headers['test'] = 'header'267            return {'id': '12345abcde', 'actualWorkflowLanguageVersion': 'draft-2'}268        for cromwell_auth in self.auth_options:269            mock_request.get(270                '{0}/api/workflows/v1/{1}/metadata?expandSubWorkflows=false&includeKey={2}'.format(271                    cromwell_auth.url, workflow_id, test_include_key272                ),273                json=_request_callback,274            )275            result = CromwellAPI.metadata(276                workflow_id, cromwell_auth, includeKey=test_include_key277            )278            self.assertEqual(result.status_code, 200)279            self.assertEqual(result.json()['id'], workflow_id)280    @requests_mock.mock()281    def test_health_returns_200(self, mock_request):282        expected = {283            "DockerHub": {"ok": "true"},284            "Engine Database": {"ok": "true"},285            "PAPI": {"ok": "true"},286            "GCS": {"ok": "true"},287        }288        def _request_callback(request, context):289            context.status_code = 200290            context.headers['test'] = 'header'291            return expected292        for cromwell_auth in self.auth_options:293            mock_request.get(294                '{0}/engine/v1/status'.format(cromwell_auth.url), json=_request_callback295            )296            result = CromwellAPI.health(cromwell_auth)297            self.assertEqual(result.status_code, 200)298            self.assertEqual(result.json(), expected)299    @requests_mock.mock()300    def test_abort(self, mock_request):301        workflow_id = "01234"302        expected = {"id": workflow_id, "status": "Aborting"}303        def _request_callback(request, context):304            context.status_code = 200305            context.headers['test'] = 'header'306            return expected307        for cromwell_auth in self.auth_options:308            mock_request.post(309                cromwell_auth.url + '/api/workflows/v1/{}/abort'.format(workflow_id),310                json=_request_callback,311            )312            result = CromwellAPI.abort(workflow_id, cromwell_auth)313            self.assertEqual(result.json(), expected)314    @requests_mock.mock()315    def test_status(self, mock_request):316        def _request_callback_status(request, context):317            context.status_code = 200318            context.headers['test'] = 'header'319            return {'status': 'Succeeded'}320        workflow_id = "01234"321        for cromwell_auth in self.auth_options:322            mock_request.get(323                cromwell_auth.url + '/api/workflows/v1/{}/status'.format(workflow_id),324                json=_request_callback_status,325            )326            result = CromwellAPI.status(workflow_id, cromwell_auth)327            self.assertEqual(result.json()['status'], 'Succeeded')328if __name__ == '__main__':...

expect_test.py

Source:expect_test.py

...19        expect(SIMPLE_DICT).to.include({'c': 3})20    def test_not_include(self):21        expect(SIMPLE_DICT).to_not.include({'c' : 3})22class TestDictIncludesKey(object):23    def test_include_key(self):24        expect(SIMPLE_DICT).to.include.key('a')25    @assert_raises("assert 'c' in ['a', 'b']")26    def test_include_key_is_missing(self):27        expect(SIMPLE_DICT).to.include.key('c')28    def test_not_include_key(self):29        expect(SIMPLE_DICT).to_not.include.key('c')30class TestListIncludesValue(object):31    def test_include_in_list(self):32        expect(['a', 'b']).to.include('a')33    @assert_raises("assert 'c' in ['a', 'b']")34    def test_include_in_list_fail(self):35        expect(['a', 'b']).to.include('c')36class expect:37    def __init__(self, actual):...

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.