""" Regression tests for schedule-driven slot reconciliation. Covers the production incident where changing the operating grid (hours / interval / capacity) or adding a closed holiday left stale appointment windows visible in Live Operations booking: - config update must drop future unbooked windows (booked kept), - closed holidays must drop unbooked windows in range, - GET /bills must honor the order_id filter (counter banner showed another order's invoice without it). """ from datetime import UTC, datetime, timedelta from uuid import uuid4 import pytest import pytest_asyncio from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from app.core.security import create_access_token, hash_password from app.models.appointment import AppointmentSlot from app.models.lab import Lab from app.models.user import User, UserRole from app.repositories.permission_repository import PermissionRepository from app.schemas.appointment import AppointmentSlotCreate from app.schemas.schedule import LabHolidayCreate, LabScheduleConfigUpdate from app.services.appointment_service import AppointmentService from app.services.schedule_service import ScheduleService @pytest_asyncio.fixture(autouse=True) async def seed_perms(db_session: AsyncSession) -> None: """Seed system permissions for RBAC enforcement.""" repo = PermissionRepository() await repo.seed_permissions(db_session) @pytest_asyncio.fixture async def sched_lab(db_session: AsyncSession) -> Lab: """Create primary test branch lab.""" lab = Lab(name="Schedule Reconcile Lab", address="Kathmandu") db_session.add(lab) await db_session.commit() await db_session.refresh(lab) return lab @pytest_asyncio.fixture async def sched_admin_token(db_session: AsyncSession, sched_lab: Lab) -> str: """Create branch admin user token.""" admin = User( lab_id=sched_lab.id, name="Schedule Admin", phone="9819000001", role=UserRole.SUPERADMIN, hashed_password=hash_password("AdminPass123!"), is_active=True, ) db_session.add(admin) await db_session.commit() await db_session.refresh(admin) return create_access_token( subject=str(admin.id), extra_claims={ "phone": admin.phone, "role": admin.role.value, "lab_id": str(sched_lab.id), }, ) async def _make_slot( db_session: AsyncSession, lab_id, start: datetime, booked: int = 0, ) -> AppointmentSlot: from app.repositories.appointment_repository import AppointmentRepository created = await AppointmentService().create_slot( db_session, AppointmentSlotCreate( lab_id=lab_id, start_at=start, end_at=start + timedelta(minutes=30), capacity=8, ), ) slot = await AppointmentRepository().get_slot_by_id(db_session, created.id) assert slot is not None slot.booked_count = booked await db_session.flush() return slot @pytest.mark.asyncio class TestSlotReconciliation: """Stale windows must not survive config/holiday changes.""" async def test_schedule_api_scopes_branch_and_audits_changes( self, client: AsyncClient, db_session: AsyncSession, sched_admin_token: str, sched_lab: Lab ) -> None: other_lab = Lab(name="Other Schedule Branch", address="Pokhara") db_session.add(other_lab) await db_session.commit() headers = {"Authorization": f"Bearer {sched_admin_token}"} update = await client.put( f"/api/v1/schedule/config?lab_id={other_lab.id}", json={"open_time": "08:00", "close_time": "17:00"}, headers=headers, ) assert update.status_code == 200 assert update.json()["data"]["lab_id"] == str(other_lab.id) assert update.json()["data"]["open_time"] == "08:00" selected = await client.get(f"/api/v1/schedule/config?lab_id={other_lab.id}", headers=headers) default = await client.get("/api/v1/schedule/config", headers=headers) assert selected.json()["data"]["open_time"] == "08:00" assert default.json()["data"]["lab_id"] == str(sched_lab.id) audit = await client.get( f"/api/v1/audit-logs?lab_id={other_lab.id}&search=operating hours", headers=headers ) assert audit.status_code == 200 assert any(entry["action"] == "LAB_SCHEDULE_UPDATED" for entry in audit.json()["data"]) async def test_config_update_clears_unbooked_but_keeps_booked( self, db_session: AsyncSession, sched_lab: Lab ) -> None: future = datetime.now(UTC) + timedelta(days=3) await _make_slot(db_session, sched_lab.id, future, booked=0) await _make_slot(db_session, sched_lab.id, future + timedelta(hours=1), booked=1) await db_session.commit() await ScheduleService().update_config( db_session, LabScheduleConfigUpdate(slot_capacity=2), lab_id=sched_lab.id ) await db_session.commit() remaining = ( ( await db_session.execute( __import__("sqlalchemy") .select(AppointmentSlot) .where( AppointmentSlot.lab_id == sched_lab.id, AppointmentSlot.deleted_at.is_(None), ) ) ) .scalars() .all() ) assert [s.booked_count for s in remaining] == [1] async def test_closed_holiday_clears_unbooked_in_range( self, db_session: AsyncSession, sched_lab: Lab ) -> None: from datetime import date target = date.today() + timedelta(days=5) start = datetime.combine(target, datetime.min.time(), tzinfo=UTC) + timedelta(hours=2) await _make_slot(db_session, sched_lab.id, start, booked=0) await db_session.commit() await ScheduleService().add_holiday( db_session, LabHolidayCreate(start_date=target, end_date=target, title="Test Closure", is_closed=True), lab_id=sched_lab.id, ) await db_session.commit() day_slots = await AppointmentService().ensure_day_slots(db_session, target, lab_id=sched_lab.id) assert day_slots == [] async def test_note_only_holiday_keeps_booking_slots_active( self, db_session: AsyncSession, sched_lab: Lab ) -> None: from datetime import date from app.repositories.appointment_repository import AppointmentRepository target = date.today() + timedelta(days=6) start = datetime.combine(target, datetime.min.time(), tzinfo=UTC) + timedelta(hours=2) slot = await _make_slot(db_session, sched_lab.id, start, booked=0) await db_session.commit() await ScheduleService().add_holiday( db_session, LabHolidayCreate(start_date=target, end_date=target, title="Note Only", is_closed=False), lab_id=sched_lab.id, ) await db_session.commit() refreshed = await AppointmentRepository().get_slot_by_id(db_session, slot.id) assert refreshed is not None assert refreshed.is_active is True async def test_bills_list_order_id_filter(self, client: AsyncClient, sched_admin_token: str) -> None: resp = await client.get( "/api/v1/bills", params={"order_id": str(uuid4()), "limit": 5}, headers={"Authorization": f"Bearer {sched_admin_token}"}, ) assert resp.status_code == 200, resp.text assert resp.json()["data"] == []