LT_Janitorial_Quality_Control/tests/test_sla_candidate_query.py

"""
Equivalence test for the SQL prefilter in send_sla_alerts().

The cron used to read EVERY open issue and decide in Python which ones deserved
a notification. It now narrows to candidates in SQL first.

The contract is a CONSERVATIVE SUPERSET, not equality:

  * Nothing the old loop would have notified may be missed. This is the safety
    property -- a miss is a silently unsent SLA alert.
  * The SQL may return extra rows, because it deliberately does not replicate
    the "already notified at_risk and still only at_risk" skip (that would mean
    writing the per-severity deadline arithmetic a second time, in SQL). The
    Python loop still applies that skip, so no extra notification is ever sent
    -- only a handful of extra rows are read.

The reference predicate below is the old loop's skip logic, written out.
"""
from datetime import timedelta

import pytest

from app import db
from app.models.issue import Issue
from app.utils.sla import SLA_HOURS, AT_RISK_THRESHOLD, sla_status
from app.utils.time_utils import now_eastern


def _candidate_query():
    """The prefilter exactly as send_sla_alerts() builds it."""
    now = now_eastern()
    age_clauses = [
        db.and_(
            Issue.severity == severity,
            Issue.reported_at <= now - timedelta(hours=hours * AT_RISK_THRESHOLD),
        )
        for severity, hours in SLA_HOURS.items()
    ]
    return Issue.query.filter(
        Issue.status.in_(['open', 'in_progress', 'pending_verification']),
        Issue.reported_at.isnot(None),
        db.or_(Issue.sla_notified.is_(None), Issue.sla_notified != 'breached'),
        db.or_(*age_clauses),
    )


def _old_loop_would_act(issue):
    """The pre-change Python logic: which rows survived to send a notification."""
    if issue.status not in ('open', 'in_progress', 'pending_verification'):
        return False
    if issue.reported_at is None:
        return False          # would have raised TypeError -- see the NULL test
    status = sla_status(issue)
    if status not in ('at_risk', 'breached'):
        return False
    already = issue.sla_notified
    if already == 'breached':
        return False
    if already == 'at_risk' and status == 'at_risk':
        return False
    return True


def _add(severity, hours_ago, status='open', sla_notified=None):
    issue = Issue(
        severity=severity,
        description='x',
        status=status,
        sla_notified=sla_notified,
        reported_at=now_eastern() - timedelta(hours=hours_ago),
    )
    db.session.add(issue)
    return issue


@pytest.fixture
def population(app):
    """One issue per interesting combination of severity / age / state."""
    rows = []
    for severity, window in SLA_HOURS.items():
        at_risk_at = window * AT_RISK_THRESHOLD
        rows += [
            _add(severity, 0),                       # fresh -> ok
            _add(severity, at_risk_at * 0.5),        # halfway -> ok
            _add(severity, at_risk_at + 1),          # at risk
            _add(severity, window + 1),              # breached
            # already-notified variants
            _add(severity, at_risk_at + 1, sla_notified='at_risk'),
            _add(severity, window + 1, sla_notified='at_risk'),
            _add(severity, window + 1, sla_notified='breached'),
            # non-actionable / other statuses
            _add(severity, window + 1, status='resolved'),
            _add(severity, window + 1, status='in_progress'),
            _add(severity, window + 1, status='pending_verification'),
        ]
    db.session.commit()
    return rows


def test_prefilter_never_misses_an_actionable_issue(population):
    """Safety property: every row the old loop notified is still selected."""
    expected = {i.id for i in population if _old_loop_would_act(i)}
    actual = {i.id for i in _candidate_query().all()}

    assert expected, 'fixture built no actionable issues -- test is vacuous'
    assert expected <= actual, (
        'the prefilter drops issues the old loop would have alerted on: '
        f'{sorted(expected - actual)}'
    )


def test_prefilter_extras_are_all_skipped_by_the_loop(population):
    """The extra rows the SQL lets through produce no extra notifications.

    Each must be a row the Python loop independently skips, so the set of alerts
    actually sent is unchanged.
    """
    by_id = {i.id: i for i in population}
    expected = {i.id for i in population if _old_loop_would_act(i)}
    extras = {i.id for i in _candidate_query().all()} - expected

    for iid in extras:
        issue = by_id[iid]
        assert not _old_loop_would_act(issue)
        # and the only reason it is allowed through is the at_risk bookkeeping
        assert issue.sla_notified == 'at_risk' and sla_status(issue) == 'at_risk', (
            f'issue {iid} is an unexplained extra: status={issue.status} '
            f'severity={issue.severity} sla_notified={issue.sla_notified} '
            f'sla_status={sla_status(issue)}'
        )


def test_prefilter_excludes_the_bulk_of_open_issues(population):
    """The point of the change: most open issues are never read."""
    total_open = Issue.query.filter(
        Issue.status.in_(['open', 'in_progress', 'pending_verification'])
    ).count()
    candidates = _candidate_query().count()
    assert candidates < total_open


def test_null_reported_at_is_excluded_not_crashed(app):
    """reported_at is nullable and sla_status() raises TypeError on NULL.

    Before the prefilter one such row aborted the whole cron run; it must now be
    excluded in SQL instead.
    """
    issue = Issue(severity='high', description='x', status='open')
    db.session.add(issue)
    db.session.commit()

    # The column carries default=now_eastern, so an ORM insert can never leave
    # it NULL -- the row has to be forced, which is exactly how a legacy or
    # raw-SQL-inserted row would look.
    db.session.execute(
        db.text('UPDATE issues SET reported_at = NULL WHERE id = :i'),
        {'i': issue.id},
    )
    db.session.commit()
    db.session.expire(issue)

    assert issue.reported_at is None
    assert issue.id not in {i.id for i in _candidate_query().all()}
    with pytest.raises(TypeError):
        sla_status(issue)      # confirms the crash the filter is avoiding