Spaces:

Gaurav711
/

Rail_Mind

Sleeping

Rail_Mind / tests /unit /test_api_endpoints.py
deploy: clean release with secured container paths and dynamic model version
70fb2b5 5 months ago
31.2 kB
import pytest
import pytest_asyncio
from pathlib import Path
from datetime import datetime, timezone
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

from app.main import app
from app.db.database import (
    get_db,
    Base,
    DBDisruption,
    DBRecommendation,
    DBStation,
    DBSection,
    DBUser,
    DBAuditEntry,
)
from app.config import settings

TEST_UNIT_DB_URL = "sqlite+aiosqlite:///./test_api_unit.db"
engine = create_async_engine(TEST_UNIT_DB_URL, echo=False)
SessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)


@pytest_asyncio.fixture(scope="module", autouse=True)
async def setup_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await engine.dispose()

    db_path = Path("./test_api_unit.db")
    if db_path.exists():
        try:
            db_path.unlink()
        except Exception:
            pass


@pytest_asyncio.fixture
async def unit_session():
    async with SessionLocal() as session:
        yield session
        await session.rollback()


@pytest_asyncio.fixture
async def unit_client(unit_session):
    async def override_get_db():
        yield unit_session

    app.dependency_overrides[get_db] = override_get_db

    original_rbac = settings.ENFORCE_RBAC
    settings.ENFORCE_RBAC = False

    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://testserver",
    ) as ac:
        yield ac

    app.dependency_overrides.clear()
    settings.ENFORCE_RBAC = original_rbac


# ─────────────────────────────────────────────────────────────
#  Auth Tests
# ─────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_auth_endpoints_unit(unit_client, unit_session):
    # Test Register
    reg_payload = {"username": "admin_unit", "password": "securepassword"}
    resp = await unit_client.post("/api/v1/auth/register", json=reg_payload)
    assert resp.status_code == 200
    user_data = resp.json()
    assert user_data["username"] == "admin_unit"

    # Test Register Duplicate
    resp_dup = await unit_client.post("/api/v1/auth/register", json=reg_payload)
    assert resp_dup.status_code == 400

    # Test Login Success
    login_payload = {"username": "admin_unit", "password": "securepassword"}
    resp_login = await unit_client.post("/api/v1/auth/login", json=login_payload)
    assert resp_login.status_code == 200
    tokens = resp_login.json()
    assert "access_token" in tokens
    assert "refresh_token" in tokens

    # Test Login Fail
    bad_login = {"username": "admin_unit", "password": "wrongpassword"}
    resp_bad = await unit_client.post("/api/v1/auth/login", json=bad_login)
    assert resp_bad.status_code == 401

    # Test GET /me with auth header
    # First set ENFORCE_RBAC to True to test auth logic
    settings.ENFORCE_RBAC = True
    headers = {"Authorization": f"Bearer {tokens['access_token']}"}
    resp_me = await unit_client.get("/api/v1/auth/me", headers=headers)
    assert resp_me.status_code == 200
    assert resp_me.json()["username"] == "admin_unit"

    # Test GET /me unauthenticated
    resp_me_unauth = await unit_client.get("/api/v1/auth/me")
    assert resp_me_unauth.status_code == 401
    settings.ENFORCE_RBAC = False

    # Test Operator Performance
    resp_perf = await unit_client.get("/api/v1/auth/operator-performance")
    assert resp_perf.status_code == 200
    assert "variance_score" in resp_perf.json()

    # Test Refresh Token
    refresh_payload = {"refresh_token": tokens["refresh_token"]}
    resp_refresh = await unit_client.post("/api/v1/auth/refresh", json=refresh_payload)
    assert resp_refresh.status_code == 200
    assert "access_token" in resp_refresh.json()

    # Test Bad Refresh Token
    bad_refresh = {"refresh_token": "invalid_token"}
    resp_bad_refresh = await unit_client.post("/api/v1/auth/refresh", json=bad_refresh)
    assert resp_bad_refresh.status_code == 401

    # Test Logout (requires active user session)
    settings.ENFORCE_RBAC = True
    resp_logout = await unit_client.post("/api/v1/auth/logout", headers=headers)
    assert resp_logout.status_code == 200
    settings.ENFORCE_RBAC = False


# ─────────────────────────────────────────────────────────────
#  Cascade Tests
# ─────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_cascade_scenario_endpoints(unit_client, unit_session):
    # Scenario State
    resp = await unit_client.get("/api/v1/cascade/scenario")
    assert resp.status_code == 200

    # Corridor Metrics
    resp_metrics = await unit_client.get("/api/v1/cascade/corridor-metrics")
    assert resp_metrics.status_code == 200

    # Weather endpoints
    resp_weather = await unit_client.get("/api/v1/cascade/weather")
    assert resp_weather.status_code == 200
    assert resp_weather.json()["visibility_meters"] == 2200

    resp_weather_post = await unit_client.post(
        "/api/v1/cascade/weather?visibility=400&fog_density=35&speed_limit=60"
    )
    assert resp_weather_post.status_code == 200
    assert resp_weather_post.json()["active_warning"] == "SEVERE_FOG_WARNING"

    # Kavach toggle
    resp_kavach = await unit_client.post(
        "/api/v1/cascade/kavach-toggle?section_code=DLI-GZB&active=false"
    )
    assert resp_kavach.status_code == 200
    assert resp_kavach.json()["active"] is False

    resp_kavach_bad = await unit_client.post(
        "/api/v1/cascade/kavach-toggle?section_code=INVALID&active=false"
    )
    assert resp_kavach_bad.status_code == 404

    # Disruption details
    resp_disp_det = await unit_client.get(
        "/api/v1/cascade/disruption-details?disruption_id=nonexistent"
    )
    assert resp_disp_det.status_code == 200
    assert (
        "Clearance" in resp_disp_det.json().get("details", "")
        or "nominal" in resp_disp_det.json().get("details", "").lower()
        or "active disruptions" in resp_disp_det.json().get("details", "").lower()
    )

    # Demo run
    resp_demo = await unit_client.post("/api/v1/cascade/scenario/demo-run")
    assert resp_demo.status_code == 200
    assert resp_demo.json()["status"] == "demo_complete"

    # Impact Summary
    resp_impact = await unit_client.get("/api/v1/cascade/impact-summary")
    assert resp_impact.status_code == 200

    # Simulate GET (Scenario context)
    # Put a mock disruption in scenario first
    from app.core.scenario_engine import scenario_engine

    scenario_engine.current_step = 1
    state = scenario_engine.get_state()
    disp_id = state["disruptions"][0]["id"]

    resp_sim_get = await unit_client.get(f"/api/v1/cascade/simulate?disruption_id={disp_id}")
    assert (
        resp_sim_get.shape_code
        if hasattr(resp_sim_get, "shape_code")
        else resp_sim_get.status_code == 200
    )
    assert resp_sim_get.json()["root_disruption_id"] == disp_id

    # Simulate GET Not Found
    resp_sim_get_bad = await unit_client.get("/api/v1/cascade/simulate?disruption_id=invalid")
    assert resp_sim_get_bad.status_code == 404


@pytest.mark.asyncio
async def test_recommendations_approve_override_scenario_and_db(unit_client, unit_session):
    # 1. SCENARIO MODE = True
    # Seed scenario engine state
    from app.core.scenario_engine import scenario_engine

    scenario_engine.current_step = 4
    state = scenario_engine.get_state()
    rec_id = state["recommendations"][0]["id"]

    # Test Approve Recommendation
    resp = await unit_client.post(f"/api/v1/cascade/recommendations/{rec_id}/approve")
    assert resp.status_code == 200
    assert resp.json()["is_approved"] is True

    # Test Override Recommendation
    resp_over = await unit_client.post(
        f"/api/v1/cascade/recommendations/{rec_id}/override?override_reason=SafetyOverride"
    )
    assert resp_over.status_code == 200
    assert resp_over.json()["is_approved"] is False
    assert resp_over.json()["override_reason"] == "SafetyOverride"

    # Approve Nonexistent
    resp_bad = await unit_client.post("/api/v1/cascade/recommendations/nonexistent/approve")
    assert resp_bad.status_code == 404

    # Override Nonexistent
    resp_over_bad = await unit_client.post(
        "/api/v1/cascade/recommendations/nonexistent/override?override_reason=Bypass"
    )
    assert resp_over_bad.status_code == 404

    # 2. SCENARIO MODE = False
    original_mode = settings.SCENARIO_MODE
    settings.SCENARIO_MODE = False
    try:
        # Seed DB Recommendation
        db_rec = DBRecommendation(
            id="rec-db-test",
            disruption_id="disp-001",
            type="HOLD",
            target_train="12002",
            target_section="GZB-ALJN",
            reasoning="Reasoning text",
            confidence=0.85,
            tier=1,
            is_approved=False,
            generated_at=datetime.utcnow(),
        )
        unit_session.add(db_rec)
        await unit_session.commit()

        # Approve DB
        resp_db = await unit_client.post("/api/v1/cascade/recommendations/rec-db-test/approve")
        assert resp_db.status_code == 200
        assert resp_db.json()["is_approved"] is True

        # Override DB
        resp_db_over = await unit_client.post(
            "/api/v1/cascade/recommendations/rec-db-test/override?override_reason=OverrideDBText"
        )
        assert resp_db_over.status_code == 200
        assert resp_db_over.json()["is_approved"] is False
        assert resp_db_over.json()["override_reason"] == "OverrideDBText"

        # Nonexistent DB
        resp_db_bad = await unit_client.post(
            "/api/v1/cascade/recommendations/rec-db-nonexistent/approve"
        )
        assert resp_db_bad.status_code == 404

        resp_db_over_bad = await unit_client.post(
            "/api/v1/cascade/recommendations/rec-db-nonexistent/override?override_reason=Bypass"
        )
        assert resp_db_over_bad.status_code == 404

        # Simulate GET in DB mode (returns 501)
        resp_sim_db = await unit_client.get("/api/v1/cascade/simulate?disruption_id=any")
        assert resp_sim_db.status_code == 501
    finally:
        settings.SCENARIO_MODE = original_mode


# ─────────────────────────────────────────────────────────────
#  Disruptions Tests (DB Mode)
# ─────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_disruptions_db_mode_unit(unit_client, unit_session):
    original_mode = settings.SCENARIO_MODE
    settings.SCENARIO_MODE = False
    try:
        # Seed Disruption
        db_disp = DBDisruption(
            id="disp-unit-db-001",
            train_no="12002",
            section_from="NDLS",
            section_to="GZB",
            disruption_type="SIGNAL_FAILURE",
            severity="HIGH",
            cascade_depth=1,
            trains_affected_json="[]",
            passengers_affected=150,
            status="ACTIVE",
            detected_at=datetime.utcnow(),
        )
        unit_session.add(db_disp)
        await unit_session.commit()

        # List Disruptions
        resp = await unit_client.get("/api/v1/disruptions")
        assert resp.status_code == 200
        assert len(resp.json()) >= 1
        assert any(d["id"] == "disp-unit-db-001" for d in resp.json())

        # Get Disruption
        resp_get = await unit_client.get("/api/v1/disruptions/disp-unit-db-001")
        assert resp_get.status_code == 200
        assert resp_get.json()["id"] == "disp-unit-db-001"

        # Create Disruption
        payload = {
            "id": "disp-unit-db-created",
            "train_no": "22415",
            "section_from": "GZB",
            "section_to": "ALJN",
            "disruption_type": "WEATHER",
            "severity": "MEDIUM",
            "cascade_depth": 0,
            "trains_affected": [],
            "passengers_affected": 200,
            "status": "ACTIVE",
            "detected_at": datetime.now(timezone.utc).isoformat(),
        }
        resp_create = await unit_client.post("/api/v1/disruptions", json=payload)
        assert resp_create.status_code == 200
        assert resp_create.json()["id"] == "disp-unit-db-created"
    finally:
        settings.SCENARIO_MODE = original_mode


# ─────────────────────────────────────────────────────────────
#  RAC / Recommendations / Rerouting / Stream / Health Tests
# ─────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_rac_extended_routes(unit_client):
    # Historical trends
    resp = await unit_client.get("/api/v1/rac/historical-trends?train_no=12002")
    assert resp.status_code == 200
    assert isinstance(resp.json(), list)

    # Model health
    resp_health = await unit_client.get("/api/v1/rac/model-health")
    assert resp_health.status_code == 200
    assert "model_loaded" in resp_health.json()

    # Drift report
    resp_drift = await unit_client.get("/api/v1/rac/drift-report")
    assert resp_drift.status_code == 200
    assert "dataset_drift" in resp_drift.json()
    assert "drift_by_columns" in resp_drift.json()



@pytest.mark.asyncio
async def test_recommendations_escalated_routes(unit_client):
    # List all
    resp = await unit_client.get("/api/v1/recommendations")
    assert resp.status_code == 200
    assert len(resp.json()) >= 1

    # List active
    resp_act = await unit_client.get("/api/v1/recommendations/active")
    assert resp_act.status_code == 200

    # Approve
    resp_app = await unit_client.post("/api/v1/recommendations/rec-hold-001/approve")
    assert resp_app.status_code == 200
    assert resp_app.json()["is_approved"] is True

    # Override
    resp_over = await unit_client.post(
        "/api/v1/recommendations/rec-hold-001/override", json={"reason": "SafetyFirst"}
    )
    assert resp_over.status_code == 200
    assert resp_over.json()["override_reason"] == "SafetyFirst"

    # Approve Bad
    resp_app_bad = await unit_client.post("/api/v1/recommendations/rec-hold-nonexistent/approve")
    assert resp_app_bad.status_code == 404

    # Override Bad
    resp_over_bad = await unit_client.post(
        "/api/v1/recommendations/rec-hold-nonexistent/override", json={"reason": "Bypass"}
    )
    assert resp_over_bad.status_code == 404


@pytest.mark.asyncio
async def test_rerouting_extended_routes_scenario_and_db(unit_client, unit_session):
    # 1. SCENARIO MODE = True
    resp = await unit_client.get("/api/v1/rerouting")
    assert resp.status_code == 200
    assert len(resp.json()) == 1

    resp_single = await unit_client.get("/api/v1/rerouting/disp-001")
    assert resp_single.status_code == 200

    resp_single_bad = await unit_client.get("/api/v1/rerouting/nonexistent")
    assert resp_single_bad.status_code == 200  # scenario fallback matches suggestions

    # 2. SCENARIO MODE = False
    original_mode = settings.SCENARIO_MODE
    settings.SCENARIO_MODE = False
    try:
        # Seed Stations and Sections
        st1 = DBStation(code="NDLS", name="New Delhi", zone="NR")
        st2 = DBStation(code="GZB", name="Ghaziabad", zone="NR")
        sec1 = DBSection(
            id=1, from_station="NDLS", to_station="GZB", distance_km=25.0, max_speed_kmh=110
        )
        unit_session.add_all([st1, st2, sec1])
        await unit_session.commit()

        # Network State
        resp_net = await unit_client.get("/api/v1/rerouting/network-state")
        assert resp_net.status_code == 200
        assert len(resp_net.json()["nodes"]) >= 2

        # Rerouting suggestions in DB mode
        resp_list = await unit_client.get("/api/v1/rerouting")
        assert resp_list.status_code == 200
        assert len(resp_list.json()) == 0

        # suggest reroute
        payload = {"from_station": "NDLS", "to_station": "GZB", "train_no": "12002"}
        resp_sug = await unit_client.post("/api/v1/rerouting/suggest", json=payload)
        assert resp_sug.status_code == 200
        assert "NDLS" in resp_sug.json()["advisory_text"]
    finally:
        settings.SCENARIO_MODE = original_mode


@pytest.mark.asyncio
async def test_stream_sse_endpoints(unit_client):
    from unittest.mock import AsyncMock, patch

    # Mock the infinite event generators to yield once and exit cleanly, preventing event loop hanging/timeouts
    async def mock_agent_event_generator():
        yield "data: {}\n\n"

    async def mock_position_event_generator():
        yield "data: {}\n\n"

    with (
        patch("app.api.v1.routes.stream._agent_event_generator", mock_agent_event_generator),
        patch("app.api.v1.routes.stream._position_event_generator", mock_position_event_generator),
    ):
        # Test SSE route `/positions`
        async with unit_client.stream("GET", "/api/v1/stream/positions") as response:
            assert response.status_code == 200
            async for line in response.aiter_lines():
                if line.strip():
                    assert "data:" in line
                    break

        # Test SSE route `/agents`
        async with unit_client.stream("GET", "/api/v1/stream/agents") as response:
            assert response.status_code == 200
            async for line in response.aiter_lines():
                if line.strip():
                    assert "data:" in line
                    break


@pytest.mark.asyncio
async def test_trains_endpoints_extended(unit_client, unit_session):
    # 1. SCENARIO MODE = True
    resp_list = await unit_client.get("/api/v1/trains")
    assert resp_list.status_code == 200
    assert len(resp_list.json()) > 0

    resp_single = await unit_client.get("/api/v1/trains/12002")
    assert resp_single.status_code == 200
    assert resp_single.json()["train_no"] == "12002"

    resp_bad = await unit_client.get("/api/v1/trains/invalid_train")
    assert resp_bad.status_code == 404

    # Update speed lock
    resp_speed = await unit_client.post(
        "/api/v1/trains/speed-lock?section_code=DLI-GZB&speed_limit=60"
    )
    assert resp_speed.status_code == 200
    assert resp_speed.json()["speed_limit"] == 60

    resp_speed_bad = await unit_client.post(
        "/api/v1/trains/speed-lock?section_code=INVALID&speed_limit=60"
    )
    assert resp_speed_bad.status_code == 404

    # Search & Schedule
    from unittest.mock import AsyncMock, patch

    with patch(
        "app.services.rapidapi_irctc.rapidapi_irctc.get", new_callable=AsyncMock
    ) as mock_get:
        mock_get.return_value = {"status": "success", "data": {}}

        resp_search = await unit_client.get("/api/v1/trains/rapidapi/search-station?query=NDLS")
        assert resp_search.status_code == 200

        resp_sched = await unit_client.get("/api/v1/trains/rapidapi/train-schedule?train_no=12002")
        assert resp_sched.status_code == 200

    # 2. SCENARIO MODE = False
    original_mode = settings.SCENARIO_MODE
    settings.SCENARIO_MODE = False
    try:
        # DB trains list (empty)
        resp_db_list = await unit_client.get("/api/v1/trains")
        assert resp_db_list.status_code == 200

        # DB train single (not found)
        resp_db_single = await unit_client.get("/api/v1/trains/99999")
        assert resp_db_single.status_code == 404
    finally:
        settings.SCENARIO_MODE = original_mode


@pytest.mark.asyncio
async def test_health_endpoints_unit_route(unit_client):
    resp = await unit_client.get("/api/v1/health")
    assert resp.status_code == 200
    assert resp.json()["status"] == "healthy"

    resp_sys = await unit_client.get("/api/v1/health/system")
    assert resp_sys.status_code == 200

    resp_fresh = await unit_client.get("/api/v1/health/data-freshness")
    assert resp_fresh.status_code == 200

    resp_agents = await unit_client.get("/api/v1/health/agents")
    assert resp_agents.status_code == 200


# ─────────────────────────────────────────────────────────────
#  Additional Coverage Tests
# ─────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_rac_all_endpoints(unit_client):
    # Predict with dummy query
    payload = {
        "train_no": "12002",
        "from_station": "NDLS",
        "to_station": "GZB",
        "date": "2026-06-12",
        "current_waitlist_position": 10,
        "current_rac_count": 5,
        "days_to_journey": 15,
        "quota": "GN",
    }
    resp = await unit_client.post("/api/v1/rac/predict", json=payload)
    assert resp.status_code == 200
    data = resp.json()
    assert "confirmation_probability" in data
    assert "key_factors" in data

    # Alternative suggestions
    resp_alt = await unit_client.get(
        "/api/v1/rac/alternative-suggestions?train_no=12002&from_station=NDLS&to_station=GZB"
    )
    assert resp_alt.status_code == 200
    assert isinstance(resp_alt.json(), list)

    # Quota heatmap
    resp_heat = await unit_client.get("/api/v1/rac/quota-heatmap?train_no=12002&waitlist_pos=10")
    assert resp_heat.status_code == 200
    assert isinstance(resp_heat.json(), list)

    # Model health
    resp_health = await unit_client.get("/api/v1/rac/model-health")
    assert resp_health.status_code == 200
    assert "model_version" in resp_health.json()

    # Train stats
    resp_stats = await unit_client.get("/api/v1/rac/train-stats/12002")
    assert resp_stats.status_code == 200
    assert resp_stats.json()["train_no"] == "12002"

    # Engine error mock
    from unittest.mock import patch

    with patch(
        "app.ml.rac_predictor.rac_predictor.predict", side_effect=Exception("ML engine crash")
    ):
        resp_err = await unit_client.post("/api/v1/rac/predict", json=payload)
        assert resp_err.status_code == 500


@pytest.mark.asyncio
async def test_audit_scenario_and_corrupted_chain(unit_client, unit_session):
    # 1. Test empty DB under scenario mode
    original_mode = settings.SCENARIO_MODE
    settings.SCENARIO_MODE = True
    try:
        # Clear DBAuditEntry first (in case integration test seeded anything)
        from sqlalchemy import delete

        await unit_session.execute(delete(DBAuditEntry))
        await unit_session.commit()

        # Advance scenario step to have some audit entries
        from app.core.scenario_engine import scenario_engine

        scenario_engine.current_step = 1

        # List audit logs (scenario mode fallback)
        resp_list = await unit_client.get("/api/v1/audit")
        assert resp_list.status_code == 200
        assert len(resp_list.json()) > 0

        # Statistics (scenario mode fallback)
        resp_stats = await unit_client.get("/api/v1/audit/statistics")
        assert resp_stats.status_code == 200
        assert resp_stats.json()["total_blocks_sealed"] > 0

        # Verify chain on empty DB
        resp_verify = await unit_client.get("/api/v1/audit/verify")
        assert resp_verify.status_code == 200
        assert resp_verify.json()["chain_valid"] is True
    finally:
        settings.SCENARIO_MODE = original_mode

    # 2. Test calculate_content_hash directly
    from app.api.v1.routes.audit import calculate_content_hash

    test_entry = DBAuditEntry(
        id=1001,
        agent_name="TestAgent",
        action_type="TEST",
        target="None",
        reasoning="Test reason",
        confidence=0.95,
        prev_hash="a" * 64,
        current_hash="b" * 64,
        timestamp=datetime.utcnow(),
    )
    h = calculate_content_hash(test_entry)
    assert isinstance(h, str)

    # 3. Test corrupted chains in DB mode
    settings.SCENARIO_MODE = False
    try:
        # Add corrupted genesis (prev_hash is not zero)
        bad_genesis = DBAuditEntry(
            id=101,
            agent_name="Agent",
            action_type="ACTION",
            target="T",
            reasoning="Reason",
            confidence=0.9,
            prev_hash="1" * 64,
            current_hash="2" * 64,
            timestamp=datetime(2026, 6, 1, 12, 0, 0),
        )
        # Add out-of-order timestamp or link breakage
        bad_link = DBAuditEntry(
            id=102,
            agent_name="Agent",
            action_type="ACTION",
            target="T",
            reasoning="Reason",
            confidence=0.9,
            prev_hash="3" * 64,  # broken link
            current_hash="4" * 64,
            timestamp=datetime(2026, 5, 1, 12, 0, 0),  # out of order timestamp
        )
        unit_session.add_all([bad_genesis, bad_link])
        await unit_session.commit()

        resp_verify = await unit_client.get("/api/v1/audit/verify")
        assert resp_verify.status_code == 200
        verify_data = resp_verify.json()
        assert verify_data["chain_valid"] is False
        assert verify_data["genesis_valid"] is False
        assert verify_data["links_valid"] is False
        assert verify_data["timestamps_valid"] is False
        assert "101" in verify_data["corrupted_records"]
    finally:
        settings.SCENARIO_MODE = original_mode


@pytest.mark.asyncio
async def test_rerouting_edge_cases_and_500(unit_client, unit_session):
    # Test get_routing_for_disruption not found (DB mode empty suggestions)
    original_mode = settings.SCENARIO_MODE
    settings.SCENARIO_MODE = False
    try:
        resp = await unit_client.get("/api/v1/rerouting/disp-nonexistent")
        assert resp.status_code == 404
    finally:
        settings.SCENARIO_MODE = original_mode

    # Test 500 error on rerouting suggest when db error happens
    from unittest.mock import patch

    with patch.object(unit_session, "execute", side_effect=Exception("Database error")):
        payload = {"from_station": "NDLS", "to_station": "GZB", "train_no": "12002"}
        resp_suggest = await unit_client.post("/api/v1/rerouting/suggest", json=payload)
        assert resp_suggest.status_code == 500

        resp_net = await unit_client.get("/api/v1/rerouting/network-state")
        assert resp_net.status_code == 500


@pytest.mark.asyncio
async def test_auth_edge_cases_rbac(unit_client, unit_session):
    from fastapi import HTTPException
    from app.api.v1.routes.auth import (
        verify_password,
        get_password_hash,
        create_access_token,
        create_refresh_token,
    )

    # 1. verify_password exception
    assert verify_password("pass", None) is False

    # 2. get_current_user credentials exception with invalid token
    original_rbac = settings.ENFORCE_RBAC
    settings.ENFORCE_RBAC = True
    try:
        resp = await unit_client.get(
            "/api/v1/auth/me", headers={"Authorization": "Bearer invalid_token"}
        )
        assert resp.status_code == 401

        # 3. inactive user
        inactive_user = DBUser(
            username="inactive_usr",
            email="inactive@test.com",
            password_hash=get_password_hash("password"),
            role="PASSENGER",
            is_active=False,
        )
        unit_session.add(inactive_user)
        await unit_session.commit()

        token = create_access_token(data={"sub": "inactive_usr", "role": "PASSENGER"})
        resp_inactive = await unit_client.get(
            "/api/v1/auth/me", headers={"Authorization": f"Bearer {token}"}
        )
        assert resp_inactive.status_code == 400
        assert resp_inactive.json()["detail"] == "Inactive user"

        # 4. test role checker logic directly to avoid import-time dependency override issues
        from app.api.v1.routes.auth import require_roles

        role_checker = require_roles("CONTROLLER", "ADMIN")

        class MockRequest:
            def __init__(self, headers):
                self.headers = headers

        # Bypassed when ENFORCE_RBAC is False
        settings.ENFORCE_RBAC = False
        mock_req = MockRequest(headers={})
        assert await role_checker(mock_req, db=unit_session) is None
        settings.ENFORCE_RBAC = True

        # Passenger is forbidden (raises 403)
        passenger_token = create_access_token(data={"sub": "passenger_usr", "role": "PASSENGER"})
        passenger_user = DBUser(
            username="passenger_usr",
            email="passenger@test.com",
            password_hash=get_password_hash("password"),
            role="PASSENGER",
            is_active=True,
        )
        unit_session.add(passenger_user)
        await unit_session.commit()

        mock_req_p = MockRequest(headers={"Authorization": f"Bearer {passenger_token}"})
        with pytest.raises(HTTPException) as exc_info:
            await role_checker(mock_req_p, db=unit_session)
        assert exc_info.value.status_code == 403

        # Controller is allowed
        controller_token = create_access_token(data={"sub": "controller_usr", "role": "CONTROLLER"})
        controller_user = DBUser(
            username="controller_usr",
            email="controller@test.com",
            password_hash=get_password_hash("password"),
            role="CONTROLLER",
            is_active=True,
        )
        unit_session.add(controller_user)
        await unit_session.commit()

        mock_req_c = MockRequest(headers={"Authorization": f"Bearer {controller_token}"})
        res_user = await role_checker(mock_req_c, db=unit_session)
        assert res_user.username == "controller_usr"

        # 5. bad refresh token missing "type": "refresh"
        bad_ref = create_access_token(data={"sub": "passenger_usr"})
        resp_ref_bad = await unit_client.post(
            "/api/v1/auth/refresh", json={"refresh_token": bad_ref}
        )
        assert resp_ref_bad.status_code == 401

        # 6. Inactive user refresh
        inactive_ref = create_refresh_token(data={"sub": "inactive_usr"})
        resp_ref_inactive = await unit_client.post(
            "/api/v1/auth/refresh", json={"refresh_token": inactive_ref}
        )
        assert resp_ref_inactive.status_code == 401
    finally:
        settings.ENFORCE_RBAC = original_rbac