File size: 7,549 Bytes
7c6ffa6 | 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 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 | 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
|