init testu

dev_requirements.txt

@@ -0,0 +1,4 @@
+pytest
+mypy
+flake8
+pytest-asyncio

pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+asyncio_mode = auto

tests/conftest.py

@@ -0,0 +1,44 @@
+import os
+import sys
+from pathlib import Path
+import pytest
+from sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession
+
+ROOT_DIR = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT_DIR / "src"))
+
+from logger import log
+
+
+def pytest_sessionstart():
+    os.environ['MYSQL_HOST'] = 'localhost'
+    os.environ['MYSQL_USER'] = 'root'
+    os.environ['MYSQL_PASSWORD'] = 'root'
+
+
+@pytest.fixture(scope="function")
+async def db():
+    # from database.transaction import get_session
+    #
+    # async with get_session() as db:
+    #     yield db
+    #     await db.rollback()
+
+
+    import database
+    from database.config import create_db_engine
+
+    test_engine = create_db_engine()
+    test_async_session = async_sessionmaker(test_engine, expire_on_commit=True, class_=AsyncSession)
+
+    database.config.engine = test_engine
+    database.async_session = test_async_session
+
+    async with test_async_session() as session:
+        async with session.begin():
+            try:
+                yield session
+                await session.rollback()
+            except Exception as e:
+                log.exception("Error", exc_info=e)
+                await session.rollback()

tests/factories.py

@@ -0,0 +1,16 @@
+from typing import Optional
+
+from database import models
+from utils.str_utils import random_str
+
+
+async def create_pilot(db, data: Optional[dict] = None) -> models.User:
+    if not data:
+        data = {}
+
+    return await models.User.create(db, {
+        "email": data.pop("email", f"{random_str(5)}@{random_str(10)}.cz"),
+        "name": data.pop("name", "Pepa Tester"),
+        "description": data.pop("description", ""),
+        "password_hashed": data.pop("password_hashed", "***")
+    })

tests/resolvers/test_aircraft.py

@@ -0,0 +1,49 @@
+from collections import namedtuple
+from typing import Optional
+
+import pytest
+
+from database import models
+from graphql_schema.entities.resolvers.aircraft import AircraftQueryResolver
+from tests.factories import create_pilot
+from utils.str_utils import random_str
+
+AircraftData = namedtuple('AircraftData', ["aircraft", "user"])
+
+
+async def create_aircraft(db, data: Optional[dict] = None) -> AircraftData:
+    if not data:
+        data = {}
+
+    if not data.get("created_by_id"):
+        pilot = await create_pilot(db)
+        data["created_by_id"] = pilot.id
+    else:
+        pilot = await models.User.get_one(db, id=data['created_by_id'])
+
+    return AircraftData(
+        aircraft=await models.Aircraft.create(
+            db,
+            {
+                "call_sign": data.pop("call_sign", f"OK-{random_str(4)}"),
+                "is_public": data.pop("is_public", False),
+                "created_by_id": data.pop('created_by_id'),
+                "seats": data.pop("seats", 2),
+                **data
+            }
+        ),
+        user=pilot
+    )
+
+
+async def test_private_aircraft(db):
+    aircraft_data = await create_aircraft(db)
+
+    resolver = AircraftQueryResolver()
+
+    params =[
+        {"id": aircraft_data.aircraft.id, "created_by_id": 234}
+    ]
+    for param in params:
+        aircraft = await resolver.get_one(**param)
+        assert aircraft.id == aircraft_data.aircraft.id

tests/resolvers/test_basic.py

@@ -0,0 +1,3 @@
+async def test_simple_query(db):
+    # query_builder = QueryBuilder()
+    pass