63 lines
1.8 KiB
Python
63 lines
1.8 KiB
Python
# scopes.tests.test_api
|
|
|
|
"""Tests for the fastapi- (and asyncio-) based stuff."""
|
|
|
|
import os, sys
|
|
sys.path = [os.path.dirname(__file__)] + sys.path
|
|
|
|
import config
|
|
from scopes.storage.db import postgres
|
|
from scopes.storage import concept, folder, message, topic, tracking
|
|
import transaction
|
|
|
|
config.dbengine = 'postgresql+psycopg'
|
|
config.dbname = 'testdb'
|
|
config.dbuser = 'testuser'
|
|
config.dbpassword = 'secret'
|
|
config.dbschema = 'testing'
|
|
config.StorageFactory = postgres.StorageFactory
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from httpx2 import ASGITransport, AsyncClient
|
|
|
|
from scopes.api import main
|
|
from scopes.api.main import app
|
|
|
|
@pytest.fixture(scope='module')
|
|
def storage():
|
|
#storage = config.StorageFactory(config)(config.dbschema)
|
|
storage = main.storage()
|
|
storage.dropTable('tracks')
|
|
tracks = storage.create(tracking.Container)
|
|
tr01 = tracking.Track('t01', 'john')
|
|
tr01.update(dict(activity='testing'))
|
|
trid01 = tracks.save(tr01)
|
|
transaction.commit()
|
|
return storage
|
|
|
|
def async_client():
|
|
return AsyncClient(transport=ASGITransport(app=app), base_url='http://test')
|
|
|
|
#@pytest.mark.anyio
|
|
#async def test_root():
|
|
#async with async_client() as clt:
|
|
#response = await clt.get('/')
|
|
|
|
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, caplog):
|
|
with TestClient(app) as clt:
|
|
caplog.set_level(config.settings.log_level)
|
|
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': '...'}
|
|
print('\nLogs: ', caplog.text)
|
|
|