298 lines
13 KiB
Python
298 lines
13 KiB
Python
# #region Test.TranslationEventLog [C:3] [TYPE Module] [SEMANTICS test, event, log, audit]
|
|
# @BRIEF Verify TranslationEventLog contracts — log_event, query_events, prune_expired, get_run_event_summary.
|
|
# @RELATION BINDS_TO -> [TranslationEventLog]
|
|
# @TEST_EDGE: invalid_event_type -> Raises ValueError
|
|
# @TEST_EDGE: terminal_event_twice -> Raises ValueError
|
|
# @TEST_EDGE: missing_run_started -> Raises ValueError for run events
|
|
# @TEST_EDGE: prune_expired_noop -> Returns zero when no expired events
|
|
# @TEST_EDGE: prune_expired_creates_snapshot -> MetricSnapshot created before pruning
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from src.models.translate import (
|
|
TranslationEvent, MetricSnapshot, TranslationRun,
|
|
)
|
|
from src.plugins.translate.events import (
|
|
TranslationEventLog, VALID_EVENT_TYPES, TERMINAL_EVENT_TYPES,
|
|
)
|
|
|
|
from .conftest import JOB_ID, RUN_ID
|
|
|
|
|
|
class TestLogEvent:
|
|
"""Verify log_event method."""
|
|
|
|
def test_log_invalid_event_type(self, db_session):
|
|
"""Negative: unknown event type raises ValueError."""
|
|
el = TranslationEventLog(db_session)
|
|
with pytest.raises(ValueError, match="Invalid event_type"):
|
|
el.log_event(job_id=JOB_ID, event_type="UNKNOWN_TYPE")
|
|
|
|
def test_log_valid_event_no_run(self, db_session):
|
|
"""Happy: job-level event without run_id."""
|
|
el = TranslationEventLog(db_session)
|
|
event = el.log_event(job_id=JOB_ID, event_type="JOB_CREATED",
|
|
created_by="test_user")
|
|
assert event.job_id == JOB_ID
|
|
assert event.event_type == "JOB_CREATED"
|
|
assert event.created_by == "test_user"
|
|
|
|
def test_log_run_started(self, db_with_run):
|
|
"""Happy: RUN_STARTED event for a run."""
|
|
session, run_id = db_with_run
|
|
el = TranslationEventLog(session)
|
|
event = el.log_event(job_id=JOB_ID, run_id=run_id,
|
|
event_type="RUN_STARTED")
|
|
assert event.run_id == run_id
|
|
|
|
def test_log_run_event_without_started(self, db_session):
|
|
"""Negative: other run events require RUN_STARTED first."""
|
|
el = TranslationEventLog(db_session)
|
|
run = TranslationRun(id="orphan-run", job_id=JOB_ID, status="RUNNING",
|
|
started_at=datetime.now(UTC))
|
|
db_session.add(run)
|
|
db_session.commit()
|
|
with pytest.raises(ValueError, match="RUN_STARTED event must precede"):
|
|
el.log_event(job_id=JOB_ID, run_id="orphan-run",
|
|
event_type="BATCH_STARTED")
|
|
|
|
def test_log_terminal_event_twice(self, db_with_run):
|
|
"""Negative: cannot log two terminal events for same run."""
|
|
session, run_id = db_with_run
|
|
el = TranslationEventLog(session)
|
|
el.log_event(job_id=JOB_ID, run_id=run_id, event_type="RUN_STARTED")
|
|
el.log_event(job_id=JOB_ID, run_id=run_id, event_type="RUN_COMPLETED")
|
|
with pytest.raises(ValueError, match="already has a terminal event"):
|
|
el.log_event(job_id=JOB_ID, run_id=run_id, event_type="RUN_FAILED")
|
|
|
|
def test_log_terminal_event_first_time_ok(self, db_with_run):
|
|
"""Happy: first terminal event succeeds."""
|
|
session, run_id = db_with_run
|
|
el = TranslationEventLog(session)
|
|
el.log_event(job_id=JOB_ID, run_id=run_id, event_type="RUN_STARTED")
|
|
event = el.log_event(job_id=JOB_ID, run_id=run_id,
|
|
event_type="RUN_COMPLETED")
|
|
assert event.event_type == "RUN_COMPLETED"
|
|
|
|
def test_log_with_payload(self, db_with_run):
|
|
"""Happy: event with structured payload."""
|
|
session, run_id = db_with_run
|
|
el = TranslationEventLog(session)
|
|
payload = {"rows": 10, "language": "es"}
|
|
event = el.log_event(job_id=JOB_ID, run_id=run_id,
|
|
event_type="RUN_STARTED", payload=payload)
|
|
assert event.event_data == payload
|
|
|
|
def test_log_event_flushes(self, db_session):
|
|
"""Verify flush is called after adding event."""
|
|
el = TranslationEventLog(db_session)
|
|
event = el.log_event(job_id=JOB_ID, event_type="JOB_CREATED")
|
|
fetched = db_session.query(TranslationEvent).filter(
|
|
TranslationEvent.id == event.id
|
|
).first()
|
|
assert fetched is not None
|
|
|
|
def test_run_started_before_terminal(self, db_with_run):
|
|
"""Edge: RUN_STARTED precedes terminal event."""
|
|
session, run_id = db_with_run
|
|
el = TranslationEventLog(session)
|
|
el.log_event(job_id=JOB_ID, run_id=run_id, event_type="RUN_STARTED")
|
|
ev = el.log_event(job_id=JOB_ID, run_id=run_id, event_type="RUN_COMPLETED")
|
|
assert ev is not None
|
|
|
|
|
|
class TestQueryEvents:
|
|
"""Verify query_events method."""
|
|
|
|
def test_query_all_events(self, db_session):
|
|
"""Happy: query returns all events."""
|
|
el = TranslationEventLog(db_session)
|
|
el.log_event(job_id=JOB_ID, event_type="JOB_CREATED")
|
|
el.log_event(job_id=JOB_ID, event_type="JOB_UPDATED")
|
|
results = el.query_events()
|
|
assert len(results) >= 1
|
|
|
|
def test_query_filter_by_job(self, db_session):
|
|
"""Happy: filter by job_id."""
|
|
el = TranslationEventLog(db_session)
|
|
el.log_event(job_id=JOB_ID, event_type="JOB_CREATED")
|
|
results = el.query_events(job_id=JOB_ID)
|
|
assert len(results) >= 1
|
|
assert results[0]["job_id"] == JOB_ID
|
|
|
|
def test_query_filter_by_run(self, db_with_run):
|
|
"""Happy: filter by run_id."""
|
|
session, run_id = db_with_run
|
|
el = TranslationEventLog(session)
|
|
el.log_event(job_id=JOB_ID, run_id=run_id, event_type="RUN_STARTED")
|
|
results = el.query_events(run_id=run_id)
|
|
# db_with_run fixture already logs RUN_STARTED during setup, plus this test logs another = 2
|
|
assert len(results) == 2
|
|
|
|
def test_query_filter_by_type(self, db_with_run):
|
|
"""Happy: filter by event_type."""
|
|
session, run_id = db_with_run
|
|
el = TranslationEventLog(session)
|
|
el.log_event(job_id=JOB_ID, event_type="JOB_CREATED")
|
|
el.log_event(job_id=JOB_ID, run_id=run_id, event_type="RUN_STARTED")
|
|
results = el.query_events(event_type="JOB_CREATED")
|
|
assert len(results) == 1
|
|
|
|
def test_query_limit_offset(self, db_session):
|
|
"""Edge: pagination with limit and offset."""
|
|
el = TranslationEventLog(db_session)
|
|
for i in range(5):
|
|
el.log_event(job_id=JOB_ID, event_type="JOB_UPDATED",
|
|
payload={"seq": i})
|
|
results = el.query_events(limit=2, offset=0)
|
|
assert len(results) == 2
|
|
results_page2 = el.query_events(limit=2, offset=2)
|
|
assert len(results_page2) == 2
|
|
|
|
|
|
class TestPruneExpired:
|
|
"""Verify prune_expired method."""
|
|
|
|
def test_no_expired_events(self, db_session):
|
|
"""Edge: no events older than cutoff."""
|
|
el = TranslationEventLog(db_session)
|
|
el.log_event(job_id=JOB_ID, event_type="JOB_CREATED")
|
|
result = el.prune_expired(retention_days=365)
|
|
assert result["pruned"] == 0
|
|
assert result["snapshot_id"] is None
|
|
|
|
def test_prune_expired_events(self, db_session):
|
|
"""Happy: prune old events and create snapshot."""
|
|
el = TranslationEventLog(db_session)
|
|
old_event = TranslationEvent(
|
|
id="old-event", job_id=JOB_ID, event_type="JOB_CREATED",
|
|
event_data={}, created_at=datetime.now(UTC) - timedelta(days=400),
|
|
)
|
|
db_session.add(old_event)
|
|
db_session.commit()
|
|
|
|
result = el.prune_expired(retention_days=90)
|
|
assert result["pruned"] >= 1
|
|
assert result["snapshot_id"] is not None
|
|
remaining = db_session.query(TranslationEvent).all()
|
|
assert len(remaining) == 0
|
|
|
|
def test_prune_with_metric_snapshot_created(self, db_session):
|
|
"""Invariant: MetricSnapshot created before pruning."""
|
|
el = TranslationEventLog(db_session)
|
|
old_event = TranslationEvent(
|
|
id="old-with-metrics", job_id=JOB_ID, event_type="BATCH_COMPLETED",
|
|
event_data={"token_count": 500, "cost": 0.01,
|
|
"language_code": "es", "run_id": "run-1"},
|
|
created_at=datetime.now(UTC) - timedelta(days=400),
|
|
)
|
|
db_session.add(old_event)
|
|
db_session.commit()
|
|
|
|
result = el.prune_expired(retention_days=90)
|
|
assert result["snapshot_id"] is not None
|
|
snapshot = db_session.query(MetricSnapshot).first()
|
|
assert snapshot is not None
|
|
assert snapshot.total_records == 1
|
|
|
|
def test_prune_language_metrics_aggregation(self, db_session):
|
|
"""Edge: per-language metrics aggregated correctly."""
|
|
el = TranslationEventLog(db_session)
|
|
for i, lang in enumerate(["es", "es", "fr"]):
|
|
e = TranslationEvent(
|
|
id=f"old-{lang}-{i}", job_id=JOB_ID, event_type="BATCH_COMPLETED",
|
|
event_data={"token_count": 100, "cost": 0.01,
|
|
"language_code": lang},
|
|
created_at=datetime.now(UTC) - timedelta(days=400),
|
|
)
|
|
db_session.add(e)
|
|
db_session.commit()
|
|
|
|
result = el.prune_expired(retention_days=90)
|
|
assert result["pruned"] == 3
|
|
snapshot = db_session.query(MetricSnapshot).first()
|
|
assert snapshot is not None
|
|
metrics = snapshot.per_language_metrics or {}
|
|
assert "es" in metrics
|
|
assert metrics["es"]["cumulative_tokens"] == 200
|
|
assert "fr" in metrics
|
|
|
|
def test_prune_event_without_language(self, db_session):
|
|
"""Edge: event without language_code grouped as _unknown_."""
|
|
el = TranslationEventLog(db_session)
|
|
e = TranslationEvent(
|
|
id="no-lang", job_id=JOB_ID, event_type="BATCH_COMPLETED",
|
|
event_data={"token_count": 50},
|
|
created_at=datetime.now(UTC) - timedelta(days=400),
|
|
)
|
|
db_session.add(e)
|
|
db_session.commit()
|
|
result = el.prune_expired(retention_days=90)
|
|
assert result["pruned"] == 1
|
|
|
|
|
|
class TestGetRunEventSummary:
|
|
"""Verify get_run_event_summary method."""
|
|
|
|
def test_valid_run_summary(self, db_with_run):
|
|
"""Happy: run with started and completed events."""
|
|
session, run_id = db_with_run
|
|
el = TranslationEventLog(session)
|
|
el.log_event(job_id=JOB_ID, run_id=run_id, event_type="RUN_STARTED")
|
|
el.log_event(job_id=JOB_ID, run_id=run_id, event_type="RUN_COMPLETED")
|
|
summary = el.get_run_event_summary(run_id)
|
|
assert summary["run_id"] == run_id
|
|
assert summary["has_run_started"] is True
|
|
assert summary["terminal_event_count"] == 1
|
|
assert summary["invariant_valid"] is True
|
|
|
|
def test_empty_run_summary(self, db_session):
|
|
"""Edge: run with no events."""
|
|
el = TranslationEventLog(db_session)
|
|
summary = el.get_run_event_summary("run-empty")
|
|
assert summary["event_count"] == 0
|
|
assert summary["has_run_started"] is False
|
|
assert summary["invariant_valid"] is False
|
|
|
|
def test_run_missing_start(self, db_session):
|
|
"""Edge: run with terminal but no start."""
|
|
el = TranslationEventLog(db_session)
|
|
# Need a run record to satisfy FK constraint on translation_events.run_id
|
|
run = TranslationRun(id="run-no-start-2", job_id=JOB_ID, status="RUNNING",
|
|
started_at=datetime.now(UTC))
|
|
db_session.add(run)
|
|
db_session.flush()
|
|
e = TranslationEvent(
|
|
id="orphan-completed", job_id=JOB_ID, run_id="run-no-start-2",
|
|
event_type="RUN_COMPLETED", event_data={},
|
|
created_at=datetime.now(UTC),
|
|
)
|
|
db_session.add(e)
|
|
db_session.commit()
|
|
summary = el.get_run_event_summary("run-no-start-2")
|
|
assert summary["has_run_started"] is False
|
|
assert summary["invariant_valid"] is False
|
|
|
|
def test_run_two_terminals_violation(self, db_with_run):
|
|
"""Edge: run with two terminal events violates invariant."""
|
|
session, run_id = db_with_run
|
|
el = TranslationEventLog(session)
|
|
el.log_event(job_id=JOB_ID, run_id=run_id, event_type="RUN_STARTED")
|
|
el.log_event(job_id=JOB_ID, run_id=run_id, event_type="RUN_COMPLETED")
|
|
# Force-add a second terminal
|
|
e2 = TranslationEvent(
|
|
id="second-terminal", job_id=JOB_ID, run_id=run_id,
|
|
event_type="RUN_FAILED", event_data={},
|
|
created_at=datetime.now(UTC),
|
|
)
|
|
session.add(e2)
|
|
session.commit()
|
|
session.expire_all()
|
|
summary = el.get_run_event_summary(run_id)
|
|
assert summary["terminal_event_count"] > 1
|
|
assert summary["invariant_valid"] is False
|
|
# #endregion Test.TranslationEventLog
|