"""Матрица причин и ряд по месяцам (§14.3; design quality-metrics §6, §7) — чистая функция, без БД.""" import datetime import uuid import pytest from src.domain.quality_metrics import QualityOutcomeFactRow, compute_matrix, compute_monthly_series from src.models.enums import QualityCause, QualityOutcome TWO = 2 HALF = 0.5 def make_outcome_row( *, primary_cause: QualityCause | None = QualityCause.AS_IS, outcome: QualityOutcome = QualityOutcome.CONFIRMED, outcome_at: datetime.datetime | None = None, received_at: datetime.datetime | None = None, document_id: uuid.UUID | None = None, ) -> QualityOutcomeFactRow: resolved_outcome_at = outcome_at or datetime.datetime(2026, 10, 15, tzinfo=datetime.UTC) return QualityOutcomeFactRow( document_id=document_id or uuid.uuid4(), outcome=outcome, outcome_at=resolved_outcome_at, received_at=received_at or resolved_outcome_at, primary_cause=primary_cause, ) def test_compute_matrix_groups_by_stored_primary_cause(): # атрибуция уже в факте — матрица только группирует matrix = compute_matrix( [ make_outcome_row(primary_cause=QualityCause.SERVICE_ERROR), make_outcome_row(primary_cause=QualityCause.OUR_ERROR), ] ) assert matrix.counts[QualityCause.SERVICE_ERROR] == 1 assert matrix.counts[QualityCause.OUR_ERROR] == 1 assert matrix.counts[QualityCause.AS_IS] == 0 def test_compute_matrix_denominator_excludes_removed_documents(): outcomes = [ make_outcome_row(primary_cause=QualityCause.WITHOUT_PARTICIPATION, outcome=QualityOutcome.AUTO_CONFIRMED), make_outcome_row(primary_cause=QualityCause.AS_IS), make_outcome_row(primary_cause=None, outcome=QualityOutcome.DELETED), make_outcome_row(primary_cause=None, outcome=QualityOutcome.SUPERSEDED), make_outcome_row(primary_cause=None, outcome=QualityOutcome.SUPERSEDED_AFTER_CONFIRM), ] matrix = compute_matrix(outcomes) assert matrix.denominator == TWO assert matrix.removed_before_confirm == TWO assert matrix.removed_after_confirm == 1 assert matrix.shares[QualityCause.AS_IS] == HALF assert sum(matrix.shares.values()) == pytest.approx(1.0) def test_compute_matrix_empty_has_no_division_by_zero(): matrix = compute_matrix([]) assert matrix.denominator == 0 assert set(matrix.counts) == set(QualityCause) assert all(share == 0.0 for share in matrix.shares.values()) def test_compute_matrix_takes_latest_fact_per_document(): doc = uuid.uuid4() first = make_outcome_row(document_id=doc, primary_cause=QualityCause.OUR_ERROR) later = make_outcome_row( document_id=doc, primary_cause=None, outcome=QualityOutcome.DELETED_AFTER_CONFIRM, outcome_at=datetime.datetime(2026, 10, 20, tzinfo=datetime.UTC), ) for ordered in ([first, later], [later, first]): matrix = compute_matrix(ordered) assert matrix.denominator == 0 assert matrix.removed_after_confirm == 1 def test_compute_matrix_confirmed_without_cause_is_invariant_violation(): with pytest.raises(ValueError): compute_matrix([make_outcome_row(primary_cause=None)]) def test_monthly_series_fills_gaps_and_ignores_removed(): def at(month: int) -> datetime.datetime: return datetime.datetime(2026, month, 5, tzinfo=datetime.UTC) outcomes = [ make_outcome_row(primary_cause=QualityCause.OUR_ERROR, received_at=at(8), outcome_at=at(11)), make_outcome_row(primary_cause=QualityCause.AS_IS, received_at=at(8), outcome_at=at(9)), make_outcome_row(primary_cause=QualityCause.OUR_ERROR, received_at=at(10), outcome_at=at(10)), make_outcome_row(primary_cause=None, outcome=QualityOutcome.DELETED, received_at=at(12), outcome_at=at(12)), ] series = compute_monthly_series(outcomes, QualityCause.OUR_ERROR) assert [(p.month.month, p.count, p.denominator, p.share) for p in series] == [ (8, 1, 2, 0.5), (9, 0, 0, 0.0), (10, 1, 1, 1.0), ] def test_monthly_series_empty_and_year_rollover(): assert compute_monthly_series([], QualityCause.AS_IS) == [] outcomes = [ make_outcome_row(received_at=datetime.datetime(2026, 12, 31, tzinfo=datetime.UTC)), make_outcome_row(received_at=datetime.datetime(2027, 1, 1, tzinfo=datetime.UTC)), ] months = [p.month for p in compute_monthly_series(outcomes, QualityCause.AS_IS)] assert months == [datetime.date(2026, 12, 1), datetime.date(2027, 1, 1)] def test_compute_matrix_equal_outcome_at_after_confirm_wins_regardless_of_order(): doc = uuid.uuid4() same_time = datetime.datetime(2026, 10, 20, tzinfo=datetime.UTC) confirmed = make_outcome_row(document_id=doc, outcome_at=same_time) deleted = make_outcome_row( document_id=doc, primary_cause=None, outcome=QualityOutcome.DELETED_AFTER_CONFIRM, outcome_at=same_time ) for ordered in ([confirmed, deleted], [deleted, confirmed]): matrix = compute_matrix(ordered) assert matrix.denominator == 0 assert matrix.removed_after_confirm == 1 def test_compute_matrix_unknown_outcome_raises(): row = make_outcome_row() object.__setattr__(row, "outcome", "UNKNOWN") with pytest.raises(ValueError, match="неизвестный исход"): compute_matrix([row]) def test_monthly_series_uses_received_month_not_outcome_month(): row = make_outcome_row( received_at=datetime.datetime(2026, 8, 20, tzinfo=datetime.UTC), outcome_at=datetime.datetime(2026, 10, 2, tzinfo=datetime.UTC), ) series = compute_monthly_series([row], QualityCause.AS_IS) assert [(p.month, p.count, p.denominator) for p in series] == [(datetime.date(2026, 8, 1), 1, 1)]