Testsuite works with local infrastructure

Author: MrMatAP

.idea/runConfigurations/test__resources_.xml

@@ -44,18 +44,18 @@ def _extract_identity() -> Tuple:
 @bp.route('/', methods=['GET'])
 @oidc.accept_token(require_token=True, scopes_required=['mpaf-read'])
 def get_all():
-    identity = _extract_identity()
-    logger.info(f'Called by {identity[1]} ({identity[0]}')
+    (client_id, name) = _extract_identity()
+    logger.info(f'Called by {client_id} for {name}')
     a = Resource.query.all()
     return {'resources': resources_schema.dump(a)}, 200
 
 
 @bp.route('/<i>', methods=['GET'])
 @oidc.accept_token(require_token=True, scopes_required=['mpaf-read'])
 def get_one(i: int):
-    identity = _extract_identity()
-    logger.info(f'Called by {identity[1]} ({identity[0]}')
-    resource = Resource.query.filter(Resource.id == i).first_or_404()
+    (client_id, name) = _extract_identity()
+    logger.info(f'Called by {client_id} for {name}')
+    resource = Resource.query.filter(Resource.id == i).one_or_none()
     if resource is None:
         return {'status': 404, 'message': f'Unable to find entry with identifier {i} in database'}, 404
     return resource_schema.dump(resource), 200
@@ -65,7 +65,7 @@ def get_one(i: int):
 @oidc.accept_token(require_token=True, scopes_required=['mpaf-write'])
 def create():
     (client_id, name) = _extract_identity()
-    logger.info(f'Called by {name} ({client_id}')
+    logger.info(f'Called by {client_id} for {name}')
     try:
         json_body = request.get_json()
         if not json_body:
@@ -102,7 +102,7 @@ def create():
 @oidc.accept_token(require_token=True, scopes_required=['mpaf-write'])
 def modify(i: int):
     (client_id, name) = _extract_identity()
-    logger.info(f'Called by {name} ({client_id}')
+    logger.info(f'Called by {client_id} for {name}')
     body = resource_schema.load(request.get_json())
 
     resource = Resource.query.filter(Resource.id == i).one_or_none()
@@ -121,11 +121,11 @@ def modify(i: int):
 @oidc.accept_token(require_token=True, scopes_required=['mpaf-write'])
 def remove(i: int):
     (client_id, name) = _extract_identity()
-    logger.info(f'Called by {name} ({client_id}')
+    logger.info(f'Called by {client_id} for {name}')
 
     resource = Resource.query.filter(Resource.id == i).one_or_none()
     if resource is None:
-        return {'status': 410, 'message': 'Unable to find requested resource'}, 410
+        return {'status': 410, 'message': 'The requested resource is permanently deleted'}, 410
     if resource.owner.client_id != client_id:
         return {'status': 401, 'message': 'You do not own this resource'}, 401
 

.idea/runConfigurations/tests.xml

@@ -19,7 +19,7 @@
 #  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 #  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 #  SOFTWARE.
-import abc
+
 import os
 import logging
 import json
@@ -36,7 +36,7 @@
 import psycopg2.extensions
 
 from flask_migrate import upgrade
-from keycloak import KeycloakOpenID, KeycloakAdmin
+from keycloak.keycloak_admin import KeycloakOpenID, KeycloakAdmin
 from keycloak.exceptions import KeycloakOperationError
 
 from mrmat_python_api_flask import create_app, db
@@ -140,7 +140,7 @@ def app_dsn(self,
             self._pg_admin.commit()
             cur.close()
 
-            dsn_info = psycopg2.extensions.ConnectionInfo = psycopg2.extensions.parse_dsn(self._ti_config['keycloak'].
+            dsn_info = psycopg2.extensions.ConnectionInfo = psycopg2.extensions.parse_dsn(self._ti_config['pg'].
                                                                                           get('admin_dsn'))
             app_dsn = f"postgresql://{role}:{password}@{dsn_info['host']}:{dsn_info['port']}/{dsn_info['dbname']}"
             yield app_dsn
@@ -171,12 +171,13 @@ def app_auth(self,
             if scopes is None:
                 scopes = []
             for scope in scopes:
-                self._keycloak_admin.create_client_scope({
-                    'id': scope,
-                    'name': scope,
-                    'description': f'Test {scope}',
-                    'protocol': 'openid-connect'
-                }, skip_exists=True)
+                if not self._keycloak_admin.get_client_scope(scope):
+                    self._keycloak_admin.create_client_scope({
+                        'id': scope,
+                        'name': scope,
+                        'description': f'Test {scope}',
+                        'protocol': 'openid-connect'
+                    })
             if not self._keycloak_admin.get_client_id(client_id):
                 self._keycloak_admin.create_client({
                     'id': client_id,
@@ -255,8 +256,9 @@ def app(self,
             yield self._app
 
     @contextlib.contextmanager
-    def app_client(self):
-        yield self._app.test_client()
+    def app_client(self, app_dir):
+        with self.app(app_dir) as app:
+            yield app.test_client()
 
     @contextlib.contextmanager
     def user_token(self,
@@ -282,9 +284,8 @@ def user_token(self,
                                       client_secret_key=self._auth_info['ti_secret'],
                                       realm_name='master',
                                       verify=True)
-            token = keycloak.token(self._auth_info['user_id'],
-                                   self._auth_info['user_password'],
-                                   scope=scopes)
+            token = keycloak.token(user_id, user_password, scope=scopes)
+            token['user_id'] = user_id
             yield token
         finally:
             if drop_finally:
@@ -311,5 +312,5 @@ def local_test_infrastructure():
 
     """
     if 'TI_CONFIG' not in os.environ:
-        raise TIException(skip=True, msg='There is no TI_CONFIG environment variable to configure the environment')
+        pytest.skip('There is TI_CONFIG environment variable configuring local infrastructure to test with')
     yield LocalTestInfrastructure(ti_config_path=pathlib.Path(os.path.expanduser(os.getenv('TI_CONFIG'))))

.idea/runConfigurations/tests__no_infrastructure_.xml

@@ -35,7 +35,7 @@ def __init__(self, client: FlaskClient, token: Optional[Dict]):
         self.client = client
         self.token = token
         if token is not None:
-            self._headers = {'Authorization': f'Bearer {token["access_token"]}'}
+            self._headers = {'Authorization': f'Bearer {token}'}
 
     def get_all(self) -> Tuple:
         resp: Response = self.client.get('/api/resource/v1/', headers=self._headers)

mrmat_python_api_flask/apis/resource/v1/api.py

@@ -23,11 +23,14 @@
 import pytest
 
 from flask import Response
-from flask.testing import FlaskClient
 
-@pytest.mark.skip
-def test_greeting_v1(client: FlaskClient):
-    rv: Response = client.get('/api/greeting/v1/')
-    json_body = rv.get_json()
-    assert 'message' in json_body
-    assert json_body['message'] == 'Hello World'
+
+@pytest.mark.usefixtures('no_test_infrastructure')
+class TestWithoutInfrastructure:
+
+    def test_greeting_v1(self, no_test_infrastructure):
+        with no_test_infrastructure.app_client() as client:
+            rv: Response = client.get('/api/greeting/v1/')
+            json_body = rv.get_json()
+            assert 'message' in json_body
+            assert json_body['message'] == 'Hello World'

tests/conftest.py

@@ -23,12 +23,14 @@
 import pytest
 
 from flask import Response
-from flask.testing import FlaskClient
 
 
-@pytest.mark.skip
-def test_greeting_v2(client: FlaskClient):
-    rv: Response = client.get('/api/greeting/v2/?name=MrMat')
-    json_body = rv.get_json()
-    assert 'message' in json_body
-    assert json_body['message'] == 'Hello MrMat'
+@pytest.mark.usefixtures('no_test_infrastructure')
+class TestWithoutInfrastructure:
+
+    def test_greeting_v2(self, no_test_infrastructure):
+        with no_test_infrastructure.app_client() as client:
+            rv: Response = client.get('/api/greeting/v2/?name=MrMat')
+            json_body = rv.get_json()
+            assert 'message' in json_body
+            assert json_body['message'] == 'Hello MrMat'

tests/resource_api_client.py

@@ -21,19 +21,18 @@
 #  SOFTWARE.
 
 import pytest
-
-from typing import Optional, Dict
 from flask import Response
-from flask.testing import FlaskClient
 
 
-@pytest.mark.skip
-def test_greeting_v3(client: FlaskClient, test_config: Dict, oidc_token_read: Optional[Dict]):
-    if oidc_token_read is None:
-        pytest.skip('Skip test because there is no OIDC client configuration')
-    rv: Response = client.get('/api/greeting/v3/',
-                              headers={'Authorization': f'Bearer {oidc_token_read["access_token"]}'})
-    assert rv.status_code == 200
-    json_body = rv.get_json()
-    assert 'message' in json_body
-    assert json_body['message'] == f'Hello {test_config["client"]["preferred_name"]}'
+@pytest.mark.usefixtures('local_test_infrastructure')
+class TestWithLocalInfrastructure:
+
+    def test_greeting_v3(self, tmpdir, local_test_infrastructure):
+        with local_test_infrastructure.app_client(tmpdir) as client:
+            with local_test_infrastructure.user_token() as user_token:
+                rv: Response = client.get('/api/greeting/v3/',
+                                          headers={'Authorization': f'Bearer {user_token["access_token"]}'})
+                assert rv.status_code == 200
+                json_body = rv.get_json()
+                assert 'message' in json_body
+                assert json_body['message'] == f'Hello {user_token["user_id"]}'

tests/test_greeting_v1.py

@@ -23,23 +23,14 @@
 import pytest
 
 from flask import Response
-from flask.testing import FlaskClient
 
 
 @pytest.mark.usefixtures('no_test_infrastructure')
-class TestNoInfrastructure:
+class TestWithoutInfrastructure:
 
     def test_healthz(self, no_test_infrastructure):
         with no_test_infrastructure.app_client() as client:
             rv: Response = client.get('/healthz/')
             json_body = rv.get_json()
             assert 'status' in json_body
             assert json_body['status'] == 'OK'
-
-
-@pytest.mark.skip
-def test_healthz(client: FlaskClient):
-    rv: Response = client.get('/healthz/')
-    json_body = rv.get_json()
-    assert 'status' in json_body
-    assert json_body['status'] == 'OK'