tests/test_beta_telemetry.py

from __future__ import annotations

from sqlalchemy import select
from app.models.telemetry import TelemetryEvent
from app.services.telemetry_service import log_telemetry_event


def test_log_telemetry_event_directly(client):
    """Test that log_telemetry_event creates a database row."""
    from app.core.database import SessionLocal
    with SessionLocal() as db:
        event = log_telemetry_event(
            db,
            "custom_test_trigger",
            user_id="usr_demo_student",
            metadata={"source": "pytest", "score": 100},
        )
        assert event is not None
        assert event.id.startswith("tel")
        assert event.event_type == "custom_test_trigger"
        assert "pytest" in event.event_metadata

        # Verify it persisted in database
        stmt = select(TelemetryEvent).where(TelemetryEvent.id == event.id)
        db_event = db.scalar(stmt)
        assert db_event is not None
        assert db_event.user_id == "usr_demo_student"


def test_log_telemetry_event_uses_isolated_session(client):
    """Telemetry must not depend on the request session being writable."""
    from app.core.database import SessionLocal

    class PoisonedCallerSession:
        def add(self, *_args, **_kwargs):
            raise AssertionError("telemetry should not write through the caller session")

        def commit(self):
            raise AssertionError("telemetry should not commit the caller session")

        def rollback(self):
            raise AssertionError("rollback should not be needed for a successful isolated write")

    event = log_telemetry_event(
        PoisonedCallerSession(),  # type: ignore[arg-type]
        "isolated_session_test",
        user_id="usr_demo_student",
        metadata={"source": "pytest"},
    )

    assert event is not None
    with SessionLocal() as db:
        db_event = db.scalar(select(TelemetryEvent).where(TelemetryEvent.id == event.id))
        assert db_event is not None
        assert db_event.event_type == "isolated_session_test"


def test_upload_completed_telemetry(auth_client, monkeypatch):
    """Test that document upload writes an upload_completed telemetry event."""
    from app.core.database import SessionLocal

    # Mock the file extraction to keep the test fast
    monkeypatch.setattr(
        "app.routes.documents.extract_text_from_file",
        lambda *args, **kwargs: "Mocked text context for telemetry test.",
    )

    # Sign up and upload a document
    signup_response = auth_client.post(
        "/auth/signup",
        json={"name": "Telemetry Student", "email": "tel-student@example.test", "password": "Pass123!beta"},
    )
    assert signup_response.status_code == 201
    token = signup_response.json()["access_token"]
    headers = {"Authorization": f"Bearer {token}"}

    file_payload = {"file": ("test.txt", b"Photosynthesis formula is active.", "text/plain")}
    form_payload = {"title": "Test Chapter", "subject": "Biology"}

    upload_response = auth_client.post(
        "/documents/upload",
        headers=headers,
        files=file_payload,
        data=form_payload,
    )
    assert upload_response.status_code == 201

    # Verify upload_completed was written to the telemetry table
    with SessionLocal() as db:
        events = db.scalars(
            select(TelemetryEvent)
            .where(TelemetryEvent.event_type == "upload_completed")
            .order_by(TelemetryEvent.created_at.desc())
        ).all()
        assert len(events) >= 1
        assert "Biology" in events[0].event_metadata
        assert "doc_" in events[0].event_metadata
        assert "status" in events[0].event_metadata


def test_grounding_guard_telemetry(auth_client):
    """Test that triggering grounding guard logs grounding_guard_frequency."""
    from app.core.database import SessionLocal

    signup_response = auth_client.post(
        "/auth/signup",
        json={"name": "Guard Student", "email": "guard-student@example.test", "password": "Pass123!beta"},
    )
    assert signup_response.status_code == 201
    token = signup_response.json()["access_token"]
    headers = {"Authorization": f"Bearer {token}"}

    # Ask a question with no source -> triggers grounding guard
    ask_response = auth_client.post(
        "/ask",
        headers=headers,
        json={"question": "What is Lenz's law?", "mode": "exam_answer"},
    )
    assert ask_response.status_code == 200

    # Verify grounding_guard_frequency telemetry log
    with SessionLocal() as db:
        events = db.scalars(
            select(TelemetryEvent)
            .where(TelemetryEvent.event_type == "grounding_guard_frequency")
            .order_by(TelemetryEvent.created_at.desc())
        ).all()
        assert len(events) >= 1
        assert "Lenz" in events[0].event_metadata
        assert "no_source" in events[0].event_metadata


def test_quiz_retry_behavior_telemetry(auth_client):
    """Test that generating a quiz logs quiz_retry_behavior telemetry."""
    from app.core.database import SessionLocal

    signup_response = auth_client.post(
        "/auth/signup",
        json={"name": "Quiz Student", "email": "quiz-student@example.test", "password": "Pass123!beta"},
    )
    assert signup_response.status_code == 201
    token = signup_response.json()["access_token"]
    headers = {"Authorization": f"Bearer {token}"}

    # Generate a quiz
    quiz_response = auth_client.post(
        "/generate/quiz",
        headers=headers,
        json={
            "topic": "Photosynthesis",
            "subject": "Biology",
            "level": "Plus Two",
            "goal": "A+",
            "time_left": "tonight",
            "options": {"num_questions": 5},
        },
    )
    assert quiz_response.status_code == 200

    # Verify quiz_retry_behavior was logged
    with SessionLocal() as db:
        events = db.scalars(
            select(TelemetryEvent)
            .where(TelemetryEvent.event_type == "quiz_retry_behavior")
            .order_by(TelemetryEvent.created_at.desc())
        ).all()
        assert len(events) >= 1
        assert "Photosynthesis" in events[0].event_metadata
        assert "Biology" in events[0].event_metadata


def test_last_night_plan_usage_telemetry(auth_client):
    """Test that generating a last-night plan logs last_night_plan_usage telemetry."""
    from app.core.database import SessionLocal

    signup_response = auth_client.post(
        "/auth/signup",
        json={"name": "Revision Student", "email": "rev-student@example.test", "password": "Pass123!beta"},
    )
    assert signup_response.status_code == 201
    token = signup_response.json()["access_token"]
    headers = {"Authorization": f"Bearer {token}"}

    # Generate a last night plan
    plan_response = auth_client.post(
        "/generate/last-night-plan",
        headers=headers,
        json={
            "topic": "Organic Chemistry",
            "subject": "Chemistry",
            "level": "Plus Two",
            "goal": "A+",
            "time_left": "tonight",
            "options": {},
        },
    )
    assert plan_response.status_code == 200

    # Verify last_night_plan_usage was logged
    with SessionLocal() as db:
        events = db.scalars(
            select(TelemetryEvent)
            .where(TelemetryEvent.event_type == "last_night_plan_usage")
            .order_by(TelemetryEvent.created_at.desc())
        ).all()
        assert len(events) >= 1
        assert "Organic Chemistry" in events[0].event_metadata
        assert "Chemistry" in events[0].event_metadata