Coverage for apps/workflow/ports.py: 100%

1"""Tireless tracker-mirror seam (ADR-002 / details/07-trackers-bridge.md). 

2 

3Lumen owns every milestone status. For `trid` / `title` / `hoi` only, Tireless's pipeline 

4produces a real sub-status worth surfacing as a read-only supplementary signal. That signal is 

5fetched here, through a port whose real implementation (polling `GET /v1/loan-automation`) is 

6**deferred** — today it is a no-op. 

7 

8Contract (defensive by construction): a null, stale, or malformed snapshot is normal input. The 

9caller logs and ignores it; it is never raised, and it NEVER overwrites a Lumen-owned 

10`LoanTracker.status`. It may only populate the read-only `external_*` fields. 

11""" 

12 

13from __future__ import annotations 

14 

15import logging 

16from typing import Protocol 

17 

18from django.conf import settings 

19 

20logger = logging.getLogger(__name__) 

21 

22# The three Lumen tracker keys that have a real Tireless pipeline counterpart. 

23MIRRORED_TRACKER_KEYS = ("trid", "title", "hoi") 

24 

25 

26class TirelessTrackerPort(Protocol): 

27 """Reads a loan's Tireless tracker snapshot (or None when unavailable).""" 

28 

29 def fetch_overview(self, tireless_loan_id: str | None) -> dict | None: 

30 """Return the `LoanTrackersOverview` dict for the ticket, or None. 

31 

32 Implementations MUST NOT raise for a missing/unknown loan or a transport error — 

33 they return None so the caller falls back to Lumen's own status. 

34 """ 

35 ... 

36 

37 

38class NoopTirelessTrackerPort: 

39 """Default: no connection to Tireless yet — always returns None (ADR-001 deferral).""" 

40 

41 def fetch_overview(self, tireless_loan_id: str | None) -> dict | None: 

42 return None 

43 

44 

45_port: TirelessTrackerPort | None = None 

46 

47 

48def get_tracker_port() -> TirelessTrackerPort: 

49 """The single swap point. Defaults to the no-op until a real Tireless client is wired 

50 (via the add-integrations-config catalog, per ADR-001/ADR-002).""" 

51 global _port 

52 if _port is None: 

53 # A future real port is selected here by setting (e.g. TIRELESS_TRACKER_PORT); until 

54 # then the no-op keeps the mirror empty and the LOS independent of Tireless. 

55 mode = getattr(settings, "TIRELESS_TRACKER_PORT", "noop") 

56 if mode != "noop": # pragma: no cover 

57 logger.warning("Tireless tracker port %r unimplemented; using no-op", mode) 

58 _port = NoopTirelessTrackerPort() 

59 return _port