apidemo, test_api: basically working with simple (old) config

This commit is contained in:
Helmut Merz 2026-09-18 16:43:06 +02:00
parent b3d6512987
commit 5478ce98a7
4 changed files with 78 additions and 9 deletions

2
.gitignore vendored
View file

@ -8,10 +8,12 @@
*.sublime-project *.sublime-project
*.sublime-workspace *.sublime-workspace
*.ropeproject *.ropeproject
uv.lock
.env .env
.private* .private*
.pytest.ini .pytest.ini
*#*# *#*#
*.#* *.#*
__pycache__ __pycache__
log
var var

48
apidemo/config.py Normal file
View file

@ -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),
)

View file

@ -1,10 +1,25 @@
# scopes.api.main # scopes.api.main
import sys
if '' not in sys.path:
sys.path = [''] + sys.path
import config import config
config.storageFactory = config.StorageFactory(config)
from scopes.storage import tracking from scopes.storage import tracking
from scopes.storage.common import Storage
import transaction 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() app = FastAPI()
@ -17,8 +32,8 @@ async def read_person(name: str, q: str | None = None):
return {'name': name, 'query': q} return {'name': name, 'query': q}
@app.get('/tracks') @app.get('/tracks')
async def get_tracks(): async def get_tracks(storage: Annotated[Storage, Depends(storage)]):
storage = config.storageFactory(config.dbschema) #storage = config.storageFactory(config.dbschema)
tracks = storage.create(tracking.Container) tracks = storage.create(tracking.Container)
result = [tr.asDict() for tr in tracks.query()] result = [tr.asDict() for tr in tracks.query()]
transaction.commit() transaction.commit()

View file

@ -21,10 +21,13 @@ import pytest
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from httpx2 import ASGITransport, AsyncClient from httpx2 import ASGITransport, AsyncClient
from scopes.api import main
from scopes.api.main import app from scopes.api.main import app
def setup(): @pytest.fixture(scope='module')
storage = config.storageFactory(config.dbschema) def storage():
#storage = config.storageFactory(config.dbschema)
storage = main.storage()
storage.dropTable('tracks') storage.dropTable('tracks')
tracks = storage.create(tracking.Container) tracks = storage.create(tracking.Container)
tr01 = tracking.Track('t01', 'john') tr01 = tracking.Track('t01', 'john')
@ -41,17 +44,18 @@ def async_client():
#async with async_client() as clt: #async with async_client() as clt:
#response = await clt.get('/') #response = await clt.get('/')
def test_root(): def test_root(storage):
storage = setup()
with TestClient(app) as clt: with TestClient(app) as clt:
response = clt.get('/') response = clt.get('/')
assert response.status_code == 200 assert response.status_code == 200
assert response.json() == {'Hello': 'World'} assert response.json() == {'Hello': 'World'}
def test_tracking(storage):
with TestClient(app) as clt:
tracks = clt.get('/tracks').json() tracks = clt.get('/tracks').json()
assert len(tracks) == 1 assert len(tracks) == 1
tr = tracks[0] tr = tracks[0]
tr['timeStamp'] = '...' tr['timeStamp'] = '...'
assert tr == {'uid': 'rec-1', 'head': {'taskId': 't01', 'userName': 'john'}, assert tr == {'uid': 'rec-1', 'head': {'taskId': 't01', 'userName': 'john'},
'data': {'activity': 'testing'}, 'timeStamp': '...'} 'data': {'activity': 'testing'}, 'timeStamp': '...'}
assert tr['uid'] == 'rec-1'