Source code for florist.tests.integration.api.utils
import contextlib
import os
from typing import Literal
import pytest
import time
import threading
import uvicorn
import requests
from motor.motor_asyncio import AsyncIOMotorClient
from starlette.requests import Request
from florist.api.db.client_entities import EntityDAO
from florist.api.db.config import DatabaseConfig
from florist.api.auth.token import Token, _simple_hash, DEFAULT_USERNAME, DEFAULT_PASSWORD
[docs]
class TestUvicornServer(uvicorn.Server):
[docs]
@contextlib.contextmanager
def run_in_thread(self):
thread = threading.Thread(target=self.run)
thread.start()
try:
while not self.started:
time.sleep(1e-3)
yield
finally:
self.should_exit = True
thread.join()
[docs]
class MockApp:
def __init__(self, database_name: str):
self.db_client = AsyncIOMotorClient(DatabaseConfig.get_mongodb_uri())
self.database = self.db_client[database_name]
self.clients_auth_tokens = {}
[docs]
class MockRequest(Request):
def __init__(self, app: MockApp):
super().__init__({"type": "http"})
self._app = app
@property
def app(self):
return self._app
@app.setter
def app(self, value):
self._app = value
TEST_DATABASE_NAME = "test-database"
TEST_SQLITE_DB_PATH = "florist/tests/integration/api/client.db"
[docs]
@pytest.fixture
async def mock_request() -> MockRequest:
print(f"Creating test detabase '{TEST_DATABASE_NAME}'")
app = MockApp(TEST_DATABASE_NAME)
request = MockRequest(app)
print(f"Creating test detabase '{TEST_SQLITE_DB_PATH}'")
real_db_path = EntityDAO.db_path
EntityDAO.db_path = TEST_SQLITE_DB_PATH
yield request
print(f"Deleting test detabase '{TEST_DATABASE_NAME}'")
await app.db_client.drop_database(TEST_DATABASE_NAME)
EntityDAO.db_path = real_db_path
if os.path.exists(TEST_SQLITE_DB_PATH):
print(f"Deleting test detabase '{TEST_SQLITE_DB_PATH}'")
os.remove(TEST_SQLITE_DB_PATH)
[docs]
@pytest.fixture
async def use_test_database() -> None:
os.environ["MONGODB_DB_NAME"] = TEST_DATABASE_NAME
os.environ["SQLITE_DB_PATH"] = TEST_SQLITE_DB_PATH
# replacing real database config with test database config
EntityDAO.db_path = DatabaseConfig.get_sqlite_db_path()
print(f"Using test database '{DatabaseConfig.get_sqlite_db_path()}'")
print(f"Using test database '{DatabaseConfig.get_mongodb_db_name()}'")
yield
os.environ.pop("MONGODB_DB_NAME")
os.environ.pop("SQLITE_DB_PATH")
# deleting test databases
print(f"Deleting test detabase '{TEST_DATABASE_NAME}'")
db_client = AsyncIOMotorClient(DatabaseConfig.get_mongodb_uri())
await db_client.drop_database(TEST_DATABASE_NAME)
if os.path.exists(TEST_SQLITE_DB_PATH):
print(f"Deleting test detabase '{TEST_SQLITE_DB_PATH}'")
os.remove(TEST_SQLITE_DB_PATH)
[docs]
def change_default_password(address: str, new_password: str, type: Literal["server", "client"]) -> None:
response = requests.post(
f"http://{address}/api/{type}/auth/change_password",
data={
"grant_type": "password",
"username": DEFAULT_USERNAME,
"current_password": _simple_hash(DEFAULT_PASSWORD),
"new_password": new_password,
},
)
assert response.status_code == 200, f"Failed to change {type} password: {response.json()}"
return Token(**response.json())