152 lines
4.2 KiB
Python
152 lines
4.2 KiB
Python
import os
|
|
from datetime import datetime
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import BigInteger, Integer, create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
os.environ.setdefault("DB_USER", "test_user")
|
|
os.environ.setdefault("DB_PASSWORD", "test_password")
|
|
os.environ.setdefault("DB_NAME", "test_db")
|
|
os.environ.setdefault("SECRET_KEY", "test-secret-key")
|
|
os.environ.setdefault("MAIL_HOST", "localhost")
|
|
os.environ.setdefault("MAIL_USERNAME", "test")
|
|
os.environ.setdefault("MAIL_PASSWORD", "test")
|
|
os.environ.setdefault("MAIL_FROM", "noreply@example.test")
|
|
os.environ.setdefault("MAIL_FROM_NAME", "PPR Tests")
|
|
os.environ.setdefault("DRONE_REQUEST_TOWER_EMAIL", "tower@swansea-airport.wales")
|
|
os.environ.setdefault("BASE_URL", "http://testserver")
|
|
os.environ.setdefault("ENVIRONMENT", "test")
|
|
|
|
from app.api import deps
|
|
from app.db import session as db_session
|
|
from app.db.session import Base
|
|
from app.main import app
|
|
from app.models.ppr import PPRRecord, PPRStatus, UserRole
|
|
|
|
|
|
engine = create_engine(
|
|
"sqlite://",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
|
|
def use_sqlite_integer_primary_keys():
|
|
for table in Base.metadata.tables.values():
|
|
for column in table.columns:
|
|
if column.primary_key and isinstance(column.type, BigInteger):
|
|
column.type = Integer()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def db_session_override():
|
|
Base.metadata.drop_all(bind=engine)
|
|
use_sqlite_integer_primary_keys()
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
def override_get_db():
|
|
db = TestingSessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
original_session_local = db_session.SessionLocal
|
|
db_session.SessionLocal = TestingSessionLocal
|
|
app.dependency_overrides[deps.get_db] = override_get_db
|
|
|
|
yield
|
|
|
|
app.dependency_overrides.clear()
|
|
db_session.SessionLocal = original_session_local
|
|
Base.metadata.drop_all(bind=engine)
|
|
|
|
|
|
@pytest.fixture
|
|
def db():
|
|
session = TestingSessionLocal()
|
|
try:
|
|
yield session
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
return TestClient(app)
|
|
|
|
|
|
@pytest.fixture
|
|
def auth_client():
|
|
test_user = SimpleNamespace(
|
|
id=1,
|
|
username="test-operator",
|
|
role=UserRole.ADMINISTRATOR,
|
|
is_active=1,
|
|
)
|
|
|
|
app.dependency_overrides[deps.get_current_user] = lambda: test_user
|
|
app.dependency_overrides[deps.get_current_active_user] = lambda: test_user
|
|
app.dependency_overrides[deps.get_current_read_user] = lambda: test_user
|
|
app.dependency_overrides[deps.get_current_operator_user] = lambda: test_user
|
|
app.dependency_overrides[deps.get_current_admin_user] = lambda: test_user
|
|
|
|
return TestClient(app)
|
|
|
|
|
|
@pytest.fixture
|
|
def ppr_payload():
|
|
return {
|
|
"ac_reg": "g-test",
|
|
"ac_type": "C172",
|
|
"ac_call": "GTEST",
|
|
"captain": "Test Pilot",
|
|
"fuel": "AVGAS",
|
|
"in_from": "EGLL",
|
|
"eta": "2026-06-20T10:00:00",
|
|
"pob_in": 2,
|
|
"out_to": "EGKK",
|
|
"etd": "2026-06-20T12:00:00",
|
|
"pob_out": 2,
|
|
"email": "pilot@example.com",
|
|
"phone": "0123456789",
|
|
"notes": "API test flight",
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def ppr_factory(db):
|
|
def create_ppr(**overrides):
|
|
values = {
|
|
"status": PPRStatus.NEW,
|
|
"ac_reg": "G-FACT",
|
|
"ac_type": "PA28",
|
|
"ac_call": "GFACT",
|
|
"captain": "Factory Pilot",
|
|
"fuel": "AVGAS",
|
|
"in_from": "EGLL",
|
|
"eta": datetime(2026, 6, 20, 10, 0),
|
|
"pob_in": 2,
|
|
"out_to": "EGKK",
|
|
"etd": datetime(2026, 6, 20, 12, 0),
|
|
"pob_out": 2,
|
|
"email": None,
|
|
"phone": None,
|
|
"notes": "Factory test flight",
|
|
"created_by": "factory",
|
|
"public_token": "token-factory",
|
|
}
|
|
values.update(overrides)
|
|
ppr = PPRRecord(**values)
|
|
db.add(ppr)
|
|
db.commit()
|
|
db.refresh(ppr)
|
|
return ppr
|
|
|
|
return create_ppr
|