Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions services/graph/app/fsma_recall/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -122,8 +122,11 @@ def sla_status(self) -> SLAStatus:
if self.status == RecallStatus.COMPLETED:
duration = self.duration_seconds
# 0 seconds is a valid (instant) completion, so compare against
# None rather than relying on truthiness.
if duration is not None and duration <= sla_limit:
# None rather than relying on truthiness. A negative duration
# (completed_at before started_at) is corrupt timing data; it must
# never certify compliance, so fail closed as BREACHED rather than
# introducing a new status value API consumers don't expect.
if duration is not None and 0 <= duration <= sla_limit:
return SLAStatus.MET
return SLAStatus.BREACHED

Expand Down
48 changes: 48 additions & 0 deletions services/graph/tests/test_recall_sla_negative_duration.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
"""Data-free tests: a completed drill with completed_at before started_at.

A negative duration is corrupt timing data and must never certify 24-hour
compliance (21 CFR 1.1455(c)(1)). It fails closed as BREACHED, the existing
non-compliant status, rather than a new enum value.
Timestamps only; no records, CSV rows or EPCIS documents.
"""

from __future__ import annotations

from datetime import datetime, timedelta, timezone

from services.graph.app.fsma_recall.models import (
RecallDrill,
RecallSeverity,
RecallStatus,
RecallType,
SLAStatus,
)

START = datetime(2026, 1, 1, tzinfo=timezone.utc)


def _completed(elapsed: timedelta) -> RecallDrill:
return RecallDrill(
drill_id="",
tenant_id="",
created_at=START,
drill_type=list(RecallType)[0],
severity=list(RecallSeverity)[0],
status=RecallStatus.COMPLETED,
started_at=START,
completed_at=START + elapsed,
)


def test_negative_duration_is_not_met():
drill = _completed(timedelta(seconds=-1))
assert drill.duration_seconds == -1
assert drill.sla_status == SLAStatus.BREACHED


def test_large_negative_duration_is_breached():
assert _completed(timedelta(hours=-48)).sla_status == SLAStatus.BREACHED


def test_zero_still_met():
assert _completed(timedelta(0)).sla_status == SLAStatus.MET
Loading