Spaces:
Sleeping
Sleeping
File size: 4,798 Bytes
b84ea83 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 | """
Shared pytest fixtures for the Apl_GS test suite.
Provides:
- async_engine: in-memory SQLite engine for isolation
- async_session: transactional session rolled back after each test
- sample_visit / sample_checklist / sample_photo / sample_outlet / sample_assessment
"""
import asyncio
from datetime import datetime, timezone
from typing import AsyncGenerator
import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import (
AsyncSession,
async_sessionmaker,
create_async_engine,
)
# Import all models so SQLAlchemy registers them before create_all
from models.base import Base
import models.visit_model # noqa: F401
import models.checklist_model # noqa: F401
import models.photo_model # noqa: F401
import models.assessment_model # noqa: F401
import models.outlet_model # noqa: F401
from models.visit_model import Visit
from models.checklist_model import Checklist
from models.photo_model import Photo
from models.assessment_model import Assessment
from models.outlet_model import Outlet
# ── In-memory async SQLite engine ─────────────────────────────────────────
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
@pytest.fixture(scope="session")
def event_loop():
"""Use a single event loop for the whole session."""
loop = asyncio.new_event_loop()
yield loop
loop.close()
@pytest_asyncio.fixture(scope="session")
async def async_engine():
"""Create engine + tables once per session."""
engine = create_async_engine(
TEST_DATABASE_URL,
connect_args={"check_same_thread": False},
)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield engine
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await engine.dispose()
@pytest_asyncio.fixture
async def async_session(async_engine) -> AsyncGenerator[AsyncSession, None]:
"""Provide a transactional session that rolls back after each test."""
session_factory = async_sessionmaker(
async_engine, expire_on_commit=False
)
async with session_factory() as session:
async with session.begin():
yield session
await session.rollback()
# ── Sample data factories ──────────────────────────────────────────────────
@pytest_asyncio.fixture
async def sample_outlet(async_session: AsyncSession) -> Outlet:
"""Persist and return a sample Outlet."""
outlet = Outlet(
name="Outlet Gedung A",
location="Jl. Sudirman No. 1",
description="Outlet pusat",
)
async_session.add(outlet)
await async_session.flush()
return outlet
@pytest_asyncio.fixture
async def sample_visit(async_session: AsyncSession, sample_outlet: Outlet) -> Visit:
"""Persist and return a sample Visit linked to sample_outlet."""
visit = Visit(
title="Test Site Inspection",
description="Monthly check",
location="Gedung A",
visit_date=datetime(2026, 5, 1, 9, 0, tzinfo=timezone.utc),
status="pending",
outlet_id=sample_outlet.id,
ghost_shopper="Auditor Test",
visit_type="dine_in",
)
async_session.add(visit)
await async_session.flush()
return visit
@pytest_asyncio.fixture
async def sample_checklist(
async_session: AsyncSession, sample_visit: Visit
) -> Checklist:
"""Persist and return a sample Checklist item linked to sample_visit."""
item = Checklist(
visit_id=sample_visit.id,
item_name="Panel listrik",
is_checked=False,
)
async_session.add(item)
await async_session.flush()
return item
@pytest_asyncio.fixture
async def sample_photo(
async_session: AsyncSession, sample_visit: Visit
) -> Photo:
"""Persist and return a sample Photo linked to sample_visit."""
photo = Photo(
visit_id=sample_visit.id,
file_path="/uploads/test.jpg",
file_name="test.jpg",
file_size=102400,
mime_type="image/jpeg",
)
async_session.add(photo)
await async_session.flush()
return photo
@pytest_asyncio.fixture
async def sample_assessment(
async_session: AsyncSession, sample_visit: Visit
) -> Assessment:
"""Persist and return a sample Assessment item linked to sample_visit."""
assessment = Assessment(
visit_id=sample_visit.id,
category="A",
item_no=1,
criteria="Welcoming Kasir (Senyum, Sapa, Salam)",
score=4.0,
raw_value="4",
notes="Baik",
)
async_session.add(assessment)
await async_session.flush()
return assessment
|