KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use get_credentials method in tempest

Best Python code snippet using tempest_python

api_base_test.py

Source:api_base_test.py

...67        '''68        Return the detail URL to use for testing69        '''70        return '{0}{1}/'.format(self.url, self.pk)71    def get_credentials(self, username=None):72        '''73        Authenticates a user74        '''75        if not username:76            username = self.user_access77        user_obj = User.objects.get(username=username)78        self.client.force_authenticate(user=user_obj)79class ApiGetTestCase(object):80    '''81    Base test case for testing GET access to the API82    '''83    def test_ordering(self):84        '''85        Test that ordering the resource works86        '''87        pass88        # TODO: implement this89    def test_get_detail(self):90        '''91        Tests accessing the detail view of a resource92        '''93        if self.private_resource:94            response = self.client.get(self.url_detail)95            self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)96            # Logged in owner user97            self.get_credentials()98            response = self.client.get(self.url_detail)99            self.assertEqual(response.status_code, status.HTTP_200_OK)100            # Different logged in user101            self.get_credentials(self.user_fail)102            response = self.client.get(self.url_detail)103            self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)104        else:105            response = self.client.get(self.url_detail)106            self.assertEqual(response.status_code, status.HTTP_200_OK)107    def test_get_overview(self):108        '''109        Test accessing the overview view of a resource110        '''111        if self.private_resource:112            response = self.client.get(self.url)113            self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)114            # Logged in owner user115            self.get_credentials()116            response = self.client.get(self.url)117            self.assertEqual(response.status_code, status.HTTP_200_OK)118        else:119            response = self.client.get(self.url)120            self.assertEqual(response.status_code, status.HTTP_200_OK)121    def test_special_endpoints(self):122        '''123        Test accessing any special endpoint the resource could have124        '''125        for endpoint in self.special_endpoints:126            url = self.url_detail + endpoint + '/'127            response = self.client.get(url)128            self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)129            # Logged in owner user130            self.get_credentials()131            response = self.client.get(url)132            self.assertEqual(response.status_code, status.HTTP_200_OK)133            # Different logged in user134            self.get_credentials(self.user_fail)135            response = self.client.get(self.url_detail)136            self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)137class ApiPostTestCase(object):138    '''139    Base test case for testing POST access to the API140    '''141    def test_post_detail(self):142        '''143        POSTing to a detail view is not allowed144        '''145        if self.private_resource:146            # Anonymous user147            response = self.client.post(self.url_detail, data=self.data)148            self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)149            # Authorized user (owner)150            self.get_credentials()151            response = self.client.post(self.url_detail, data=self.data)152            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)153            # Different logged in user154            self.get_credentials(self.user_fail)155            response = self.client.post(self.url_detail, data=self.data)156            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)157        else:158            # Anonymous user159            response = self.client.post(self.url_detail, data=self.data)160            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)161            # Authorized user (owner)162            self.get_credentials()163            response = self.client.post(self.url_detail, data=self.data)164            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)165            # Different logged in user166            self.get_credentials(self.user_fail)167            response = self.client.post(self.url_detail, data=self.data)168            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)169    def test_post(self):170        '''171        Tests POSTing (adding) a new object172        '''173        if self.private_resource:174            # Anonymous user175            response = self.client.post(self.url, data=self.data)176            self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)177            # Authorized user (owner)178            self.get_credentials()179            count_before = self.resource.objects.all().count()180            response = self.client.post(self.url, data=self.data)181            self.assertEqual(response.status_code, status.HTTP_201_CREATED)182            count_after = self.resource.objects.all().count()183            self.assertEqual(count_before + 1, count_after)184            # Different logged in user185            self.get_credentials(self.user_fail)186            response = self.client.post(self.url, data=self.data)187            # self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)188            # self.assertEqual(response.status_code, status.HTTP_201_CREATED)189        else:190            # Anonymous user191            response = self.client.post(self.url, data=self.data)192            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)193            # Logged in user194            self.get_credentials()195            response = self.client.post(self.url, data=self.data)196            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)197            # Different logged in user198            self.get_credentials(self.user_fail)199            response = self.client.post(self.url, data=self.data)200            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)201    def test_post_special_endpoints(self):202        '''203        Tests that it's not possible to POST to the special endpoints204        '''205        for endpoint in self.special_endpoints:206            url = self.url_detail + endpoint + '/'207            response = self.client.post(url, self.data)208            if self.private_resource:209                self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)210            else:211                self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)212            # Logged in owner user213            self.get_credentials()214            response = self.client.post(url, self.data)215            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)216            # Different logged in user217            self.get_credentials(self.user_fail)218            response = self.client.post(url, self.data)219            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)220class ApiPatchTestCase(object):221    '''222    Base test case for testing PATCH access to the API223    '''224    def test_patch_detail(self):225        '''226        Test PATCHING a detail view227        '''228        if self.private_resource:229            # Anonymous user230            response = self.client.patch(self.url_detail, data=self.data)231            self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)232            # Authorized user (owner)233            self.get_credentials()234            response = self.client.patch(self.url_detail, data=self.data)235            self.assertIn(response.status_code, (status.HTTP_201_CREATED, status.HTTP_200_OK))236            # Try updating each of the object's values237            for key in self.data:238                response = self.client.patch(self.url_detail, data={key: self.data[key]})239                self.assertEqual(response.status_code, status.HTTP_200_OK)240            # Different logged in user241            self.get_credentials(self.user_fail)242            response = self.client.patch(self.url_detail, data=self.data)243            self.assertIn(response.status_code,244                          (status.HTTP_403_FORBIDDEN, status.HTTP_404_NOT_FOUND))245        else:246            # Anonymous user247            response = self.client.patch(self.url_detail, data=self.data)248            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)249            # Authorized user (owner)250            self.get_credentials()251            response = self.client.patch(self.url_detail, data=self.data)252            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)253            # Different logged in user254            self.get_credentials(self.user_fail)255            response = self.client.patch(self.url_detail, data=self.data)256            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)257    def test_patch(self):258        '''259        PATCHING to the overview is not allowed260        '''261        if self.private_resource:262            # Anonymous user263            response = self.client.patch(self.url, data=self.data)264            self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)265        else:266            # Anonymous user267            response = self.client.patch(self.url, data=self.data)268            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)269        # Logged in user270        self.get_credentials()271        response = self.client.patch(self.url, data=self.data)272        self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)273        # Different logged in user274        self.get_credentials(self.user_fail)275        response = self.client.patch(self.url, data=self.data)276        self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)277    def test_patch_special_endpoints(self):278        '''279        Tests that it's not possible to patch to the special endpoints280        '''281        for endpoint in self.special_endpoints:282            url = self.url_detail + endpoint + '/'283            response = self.client.patch(url, self.data)284            if self.private_resource:285                self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)286            else:287                self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)288            # Logged in owner user289            self.get_credentials()290            response = self.client.patch(url, self.data)291            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)292            # Different logged in user293            self.get_credentials(self.user_fail)294            response = self.client.patch(url, self.data)295            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)296class ApiPutTestCase(object):297    '''298    Base test case for testing PUT access to the API299    '''300    def test_put_detail(self):301        '''302        PUTing to a detail view is allowed303        '''304        if self.private_resource:305            # Anonymous user306            response = self.client.put(self.url_detail, data=self.data)307            self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)308            # Authorized user (owner)309            self.get_credentials()310            response = self.client.put(self.url_detail, data=self.data)311            self.assertIn(response.status_code, (status.HTTP_200_OK, status.HTTP_201_CREATED))312            # Different logged in user313            count_before = self.resource.objects.all().count()314            self.get_credentials(self.user_fail)315            response = self.client.put(self.url_detail, data=self.data)316            count_after = self.resource.objects.all().count()317            # Even if we PUT to a detail resource that does not belong to us,318            # the created object will have the correct user assigned.319            #320            # Currently resources that have a 'user' field 'succeed'321            if response.status_code == status.HTTP_201_CREATED:322                # print('201: {0}'.format(self.url_detail))323                obj = self.resource.objects.get(pk=response.data['id'])324                obj2 = self.resource.objects.get(pk=self.pk)325                self.assertNotEqual(obj.get_owner_object().user.username,326                                    obj2.get_owner_object().user.username)327                self.assertEqual(obj.get_owner_object().user.username, self.user_fail)328                self.assertEqual(count_before + 1, count_after)329            elif response.status_code == status.HTTP_403_FORBIDDEN:330                # print('403: {0}'.format(self.url_detail))331                self.assertEqual(count_before, count_after)332        else:333            # Anonymous user334            response = self.client.put(self.url_detail, data=self.data)335            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)336            # Authorized user (owner)337            self.get_credentials()338            response = self.client.put(self.url_detail, data=self.data)339            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)340            # Different logged in user341            self.get_credentials(self.user_fail)342            response = self.client.put(self.url_detail, data=self.data)343            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)344    def test_put(self):345        '''346        Tests PUTTING (adding) a new object347        '''348        if self.private_resource:349            # Anonymous user350            response = self.client.put(self.url, data=self.data)351            self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)352        else:353            # Anonymous user354            response = self.client.put(self.url, data=self.data)355            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)356        # Authorized user (owner)357        self.get_credentials()358        response = self.client.put(self.url, data=self.data)359        self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)360        # Different logged in user361        self.get_credentials(self.user_fail)362        response = self.client.put(self.url, data=self.data)363        self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)364    def test_put_special_endpoints(self):365        '''366        Tests that it's not possible to PUT to the special endpoints367        '''368        for endpoint in self.special_endpoints:369            url = self.url_detail + endpoint + '/'370            response = self.client.put(url, self.data)371            if self.private_resource:372                self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)373            else:374                self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)375            # Logged in owner user376            self.get_credentials()377            response = self.client.put(url, self.data)378            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)379            # Different logged in user380            self.get_credentials(self.user_fail)381            response = self.client.put(url, self.data)382            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)383class ApiDeleteTestCase(object):384    '''385    Base test case for testing DELETE access to the API386    '''387    def test_delete_detail(self):388        '''389        Tests DELETEing an object390        '''391        if self.private_resource:392            # Anonymous user393            count_before = self.resource.objects.all().count()394            response = self.client.delete(self.url_detail)395            self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)396            count_after = self.resource.objects.all().count()397            self.assertEqual(count_before, count_after)398            # Authorized user (owner)399            self.get_credentials()400            count_before = self.resource.objects.all().count()401            response = self.client.delete(self.url_detail)402            self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)403            count_after = self.resource.objects.all().count()404            self.assertEqual(count_before - 1, count_after)405            # Different logged in user406            self.get_credentials(self.user_fail)407            response = self.client.delete(self.url_detail)408            self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)409        else:410            # Anonymous user411            response = self.client.delete(self.url_detail)412            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)413            # Logged in user414            self.get_credentials()415            response = self.client.delete(self.url_detail)416            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)417            # Different logged in user418            self.get_credentials(self.user_fail)419            response = self.client.delete(self.url_detail)420            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)421    def test_delete(self):422        '''423        DELETEing to the overview is not allowed424        '''425        if self.private_resource:426            # Anonymous user427            response = self.client.delete(self.url)428            self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)429            # Authorized user (owner)430            self.get_credentials()431            response = self.client.delete(self.url)432            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)433            # Different logged in user434            self.get_credentials(self.user_fail)435            response = self.client.delete(self.url)436            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)437        else:438            # Anonymous user439            response = self.client.delete(self.url)440            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)441            # Authorized user (owner)442            self.get_credentials()443            response = self.client.delete(self.url)444            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)445            # Different logged in user446            self.get_credentials(self.user_fail)447            response = self.client.delete(self.url)448            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)449    def test_delete_special_endpoints(self):450        '''451        Tests that it's not possible to delete to the special endpoints452        '''453        for endpoint in self.special_endpoints:454            url = self.url_detail + endpoint + '/'455            response = self.client.delete(url)456            if self.private_resource:457                self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)458            else:459                self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)460            # Logged in owner user461            self.get_credentials()462            response = self.client.delete(url)463            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)464            # Different logged in user465            self.get_credentials(self.user_fail)466            response = self.client.delete(url)467            self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)468class ApiBaseResourceTestCase(BaseTestCase,469                              ApiBaseTestCase,470                              ApiGetTestCase,471                              ApiPostTestCase,472                              ApiDeleteTestCase,473                              ApiPutTestCase,474                              ApiPatchTestCase):475    '''476    Base test case for the REST API477    All logic happens in the Api*TestCase classes478    '''479    pass

test_connection.py

Source:test_connection.py

...3from query_fixture import unique_table_name4from turbodbc import DatabaseError, InterfaceError, connect, make_options5@for_one_database6def test_cursor_on_closed_connection_raises(dsn, configuration):7    connection = connect(dsn, **get_credentials(configuration))8    connection.close()9    with pytest.raises(InterfaceError):10        connection.cursor()11@for_one_database12def test_closing_twice_is_ok(dsn, configuration):13    connection = connect(dsn, **get_credentials(configuration))14    connection.close()15    connection.close()16@for_one_database17def test_closing_connection_closes_all_cursors(dsn, configuration):18    connection = connect(dsn, **get_credentials(configuration))19    cursor_1 = connection.cursor()20    cursor_2 = connection.cursor()21    connection.close()22    with pytest.raises(InterfaceError):23        cursor_1.execute("SELECT 42")24    with pytest.raises(InterfaceError):25        cursor_2.execute("SELECT 42")26@for_one_database27def test_no_autocommit(dsn, configuration):28    connection = connect(dsn, **get_credentials(configuration))29    connection.cursor().execute("CREATE TABLE test_no_autocommit (a INTEGER)")30    connection.close()31    connection = connect(dsn, **get_credentials(configuration))32    with pytest.raises(DatabaseError):33        connection.cursor().execute("SELECT * FROM test_no_autocommit")34@for_one_database35def test_commit_on_closed_connection_raises(dsn, configuration):36    connection = connect(dsn, **get_credentials(configuration))37    connection.close()38    with pytest.raises(InterfaceError):39        connection.commit()40@for_one_database41def test_commit(dsn, configuration):42    table_name = unique_table_name()43    connection = connect(dsn, **get_credentials(configuration))44    connection.cursor().execute(f"CREATE TABLE {table_name} (a INTEGER)")45    connection.commit()46    connection.close()47    connection = connect(dsn, **get_credentials(configuration))48    cursor = connection.cursor()49    cursor.execute(f"SELECT * FROM {table_name}")50    results = cursor.fetchall()51    assert results == []52    cursor.execute(f"DROP TABLE {table_name}")53    connection.commit()54@for_one_database55def test_rollback_on_closed_connection_raises(dsn, configuration):56    connection = connect(dsn, **get_credentials(configuration))57    connection.close()58    with pytest.raises(InterfaceError):59        connection.rollback()60@for_one_database61def test_rollback(dsn, configuration):62    table_name = unique_table_name()63    connection = connect(dsn, **get_credentials(configuration))64    connection.cursor().execute(f"CREATE TABLE {table_name} (a INTEGER)")65    connection.rollback()66    with pytest.raises(DatabaseError):67        connection.cursor().execute(f"SELECT * FROM {table_name}")68@for_one_database69def test_autocommit_enabled_at_start(dsn, configuration):70    table_name = unique_table_name()71    options = make_options(autocommit=True)72    connection = connect(73        dsn, turbodbc_options=options, **get_credentials(configuration)74    )75    connection.cursor().execute(f"CREATE TABLE {table_name} (a INTEGER)")76    connection.close()77    connection = connect(dsn, **get_credentials(configuration))78    cursor = connection.cursor()79    cursor.execute(f"SELECT * FROM {table_name}")80    results = cursor.fetchall()81    assert results == []82    cursor.execute(f"DROP TABLE {table_name}")83    connection.commit()84@for_one_database85def test_autocommit_switching(dsn, configuration):86    table_name = unique_table_name()87    connection = connect(dsn, **get_credentials(configuration))88    connection.autocommit = True  # <---89    connection.cursor().execute(f"CREATE TABLE {table_name} (a INTEGER)")90    connection.close()91    options = make_options(autocommit=True)92    connection = connect(93        dsn, turbodbc_options=options, **get_credentials(configuration)94    )95    connection.autocommit = False  # <---96    connection.cursor().execute(f"INSERT INTO {table_name} VALUES (?)", [42])97    connection.close()98    # table is there, but data was not persisted99    connection = connect(dsn, **get_credentials(configuration))100    cursor = connection.cursor()101    cursor.execute(f"SELECT * FROM {table_name}")102    results = cursor.fetchall()103    assert results == []104    cursor.execute(f"DROP TABLE {table_name}")105    connection.commit()106@for_one_database107def test_autocommit_querying(dsn, configuration):108    connection = connect(dsn, **get_credentials(configuration))109    assert connection.autocommit is False110    connection.autocommit = True111    assert connection.autocommit is True112    connection.close()113@for_one_database114def test_pep343_with_statement(dsn, configuration):115    with connect(dsn, **get_credentials(configuration)) as connection:116        cursor = connection.cursor()117    # connection should be closed, test it with the cursor118    with pytest.raises(InterfaceError):...

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.