From 5478ce98a74cd785314acc2991cf23fa670ddd29 Mon Sep 17 00:00:00 2001 From: Helmut Merz Date: Fri, 18 Sep 2026 16:43:06 +0200 Subject: [PATCH] apidemo, test_api: basically working with simple (old) config --- .gitignore | 2 ++ apidemo/config.py | 48 ++++++++++++++++++++++++++++++++++++++++ scopes/api/main.py | 21 +++++++++++++++--- scopes/tests/test_api.py | 16 +++++++++----- 4 files changed, 78 insertions(+), 9 deletions(-) create mode 100644 apidemo/config.py diff --git a/.gitignore b/.gitignore index 0fb90a1..20b387b 100644 --- a/.gitignore +++ b/.gitignore @@ -8,10 +8,12 @@ *.sublime-project *.sublime-workspace *.ropeproject +uv.lock .env .private* .pytest.ini *#*# *.#* __pycache__ +log var diff --git a/apidemo/config.py b/apidemo/config.py new file mode 100644 index 0000000..0f1c89f --- /dev/null +++ b/apidemo/config.py @@ -0,0 +1,48 @@ +# py-scopes/apidemo/config.py + +from dotenv import load_dotenv +import logging +from os import getenv +from scopes.web.app import zope_app_factory + +load_dotenv() + +log_file = 'log/scopes.log' +log_level = logging.DEBUG +log_format = '%(asctime)s %(levelname)s %(name)s %(message)s' +log_dateformat = '%Y-%m-%dT%H:%M:%S' +logging.basicConfig(filename=log_file, level=log_level, + format=log_format, datefmt=log_dateformat) + +server_port = getenv('SERVER_PORT', '8099') +base_url = getenv('BASE_URL', 'https://demo.cy7.de') + +app_factory = zope_app_factory + +# storage settings +from scopes.storage.db.postgres import StorageFactory +dbengine = 'postgresql+psycopg' +dbname = getenv('DBNAME', 'demo') +dbuser = getenv('DBUSER', 'demo') +dbpassword = getenv('DBPASSWORD', 'secret') +dbschema = getenv('DBSCHEMA', 'demo') + +# authentication settings +oidc_provider = getenv('OIDC_PROVIDER', 'https://a1.cy7.de') +oidc_client_id = getenv('OIDC_CLIENT_ID', '12345') +oidc_params = dict( + op_config_url=oidc_provider + '/.well-known/openid-configuration', + op_uris=None, + op_keys=None, + op_project_scope='urn:zitadel:iam:org:project:id:zitadel:aud', + callback_url=getenv('OIDC_CALLBACK_URL', base_url + '/auth/callback'), + client_id = oidc_client_id, + cookie_name=getenv('OIDC_COOKIE_NAME', 'oidc_' + oidc_client_id), + cookie_domain=getenv('OIDC_COOKIE_DOMAIN', None), + cookie_lifetime=getenv('OIDC_COOKIE_LIFETIME', '86400'), + cookie_crypt=getenv('OIDC_COOKIE_CRYPT', None), + private_key_file=getenv('OIDC_SERVICE_USER_PRIVATE_KEY_FILE', '.private-key.json'), + organization_id=getenv('OIDC_ORGANIZATION_ID', '12346'), + project_id=getenv('OIDC_PROJECT_ID', None), +) + diff --git a/scopes/api/main.py b/scopes/api/main.py index d981402..702f03e 100644 --- a/scopes/api/main.py +++ b/scopes/api/main.py @@ -1,10 +1,25 @@ # scopes.api.main +import sys +if '' not in sys.path: + sys.path = [''] + sys.path + import config +config.storageFactory = config.StorageFactory(config) + from scopes.storage import tracking +from scopes.storage.common import Storage import transaction -from fastapi import FastAPI +from fastapi import Depends, FastAPI +from typing import Annotated + +# dependencies + +def storage(): + return config.storageFactory(config.dbschema) + +# app and routes app = FastAPI() @@ -17,8 +32,8 @@ async def read_person(name: str, q: str | None = None): return {'name': name, 'query': q} @app.get('/tracks') -async def get_tracks(): - storage = config.storageFactory(config.dbschema) +async def get_tracks(storage: Annotated[Storage, Depends(storage)]): + #storage = config.storageFactory(config.dbschema) tracks = storage.create(tracking.Container) result = [tr.asDict() for tr in tracks.query()] transaction.commit() diff --git a/scopes/tests/test_api.py b/scopes/tests/test_api.py index 096b687..11eb66b 100644 --- a/scopes/tests/test_api.py +++ b/scopes/tests/test_api.py @@ -21,10 +21,13 @@ import pytest from fastapi.testclient import TestClient from httpx2 import ASGITransport, AsyncClient +from scopes.api import main from scopes.api.main import app -def setup(): - storage = config.storageFactory(config.dbschema) +@pytest.fixture(scope='module') +def storage(): + #storage = config.storageFactory(config.dbschema) + storage = main.storage() storage.dropTable('tracks') tracks = storage.create(tracking.Container) tr01 = tracking.Track('t01', 'john') @@ -41,17 +44,18 @@ def async_client(): #async with async_client() as clt: #response = await clt.get('/') -def test_root(): - storage = setup() +def test_root(storage): with TestClient(app) as clt: response = clt.get('/') assert response.status_code == 200 assert response.json() == {'Hello': 'World'} + +def test_tracking(storage): + with TestClient(app) as clt: tracks = clt.get('/tracks').json() assert len(tracks) == 1 tr = tracks[0] tr['timeStamp'] = '...' assert tr == {'uid': 'rec-1', 'head': {'taskId': 't01', 'userName': 'john'}, - 'data': {'activity': 'testing'}, 'timeStamp': '...'} - assert tr['uid'] == 'rec-1' + 'data': {'activity': 'testing'}, 'timeStamp': '...'}