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
2 changes: 0 additions & 2 deletions src/app/portfolio_truth.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@
load_release_count_by_name,
load_repo_status_from_audit_by_name,
load_security_alerts_by_name,
warn_if_warehouse_report_stale,
)
from src.producer_preflight import load_producer_evidence

Expand Down Expand Up @@ -99,7 +98,6 @@ def run_portfolio_truth_mode(args: Any) -> None:
f"(registry {'updated' if result.registry_changed else 'unchanged'}, "
f"report {'updated' if result.report_changed else 'unchanged'})"
)
warn_if_warehouse_report_stale(output_dir, args.username)


def run_portfolio_context_recovery_mode(args: Any) -> None:
Expand Down
30 changes: 0 additions & 30 deletions src/portfolio_truth_status.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,16 +4,11 @@

import json
import logging
import re
from datetime import date
from pathlib import Path

from src.cache import ResponseCache
from src.cli_output import print_warning
from src.github_client import GitHubClient

WAREHOUSE_REPORT_STALE_DAYS = 7


def load_release_count_by_name(*, output_dir: Path, username: str) -> dict[str, int] | None:
audit_files = sorted(
Expand Down Expand Up @@ -150,28 +145,3 @@ def load_security_alerts_by_name(
)
return None
return {name: entry for name, entry in data.items() if isinstance(entry, dict)}


def warn_if_warehouse_report_stale(output_dir: Path, username: str) -> None:
report_path = latest_audit_report_path(output_dir=output_dir, username=username)
if report_path is None:
print_warning(
f"No audit-report-{username}-*.json in {output_dir}: Notion's Repo Auditor "
f"signal reads that warehouse report and this --portfolio-truth run did not "
f"create one. Run `audit report {username}` to generate it (F2)."
)
return
match = re.search(r"(\d{4}-\d{2}-\d{2})", report_path.name)
if not match:
return
try:
report_date = date.fromisoformat(match.group(1))
except ValueError:
return
age = (date.today() - report_date).days
if age > WAREHOUSE_REPORT_STALE_DAYS:
print_warning(
f"Newest warehouse report {report_path.name} is {age}d old: Notion's Repo "
f"Auditor signal reads it and is now stale. Run `audit report {username}` to "
f"refresh the warehouse report (F2 — both artifacts kept live by decision)."
)
42 changes: 0 additions & 42 deletions tests/test_portfolio_truth.py
Original file line number Diff line number Diff line change
Expand Up @@ -2017,9 +2017,6 @@ def fake_publish(**kwargs):
monkeypatch.setattr(
"src.app.portfolio_truth.load_live_repo_status_by_name", lambda **_kwargs: {}
)
monkeypatch.setattr(
"src.app.portfolio_truth.warn_if_warehouse_report_stale", lambda *_args: None
)
monkeypatch.setenv("GHRA_REQUIRE_PRODUCER_EVIDENCE", "1")
monkeypatch.setenv("GHRA_PRODUCER_EVIDENCE", str(receipt))
monkeypatch.setenv("GHRA_PRODUCER_REPO_ROOT", str(tmp_path / "producer-repo"))
Expand Down Expand Up @@ -2781,42 +2778,3 @@ def test_git_default_branch_empty_when_origin_head_unset(tmp_path: Path) -> None

# A freshly init'd repo has no origin/HEAD → "" so callers fall back.
assert _git_default_branch(repo) == ""


# ── F2: warehouse-report staleness reminder ────────────────────────────────
from src.portfolio_truth_status import warn_if_warehouse_report_stale # noqa: E402


def _write_warehouse_report(d: Path, username: str, date_str: str) -> None:
(d / f"audit-report-{username}-{date_str}.json").write_text("{}", encoding="utf-8")


class TestWarehouseStalenessReminder:
"""F2 (keep-dual): --portfolio-truth mode warns when the warehouse report Notion
reads is missing or stale, so the operator refreshes it."""

def test_missing_report_warns(self, tmp_path: Path, capsys) -> None:
import re

warn_if_warehouse_report_stale(tmp_path, "saagpatel")
captured = capsys.readouterr()
# print_warning word-wraps, so normalize whitespace before substring checks
combined = re.sub(r"\s+", " ", captured.out + captured.err)
assert "No audit-report-saagpatel" in combined
assert "audit report saagpatel" in combined

def test_stale_report_warns(self, tmp_path: Path, capsys) -> None:
_write_warehouse_report(tmp_path, "saagpatel", "2020-01-01")
warn_if_warehouse_report_stale(tmp_path, "saagpatel")
captured = capsys.readouterr()
assert "stale" in (captured.out + captured.err).lower()

def test_fresh_report_no_warning(self, tmp_path: Path, capsys) -> None:
from datetime import date

_write_warehouse_report(tmp_path, "saagpatel", date.today().isoformat())
warn_if_warehouse_report_stale(tmp_path, "saagpatel")
captured = capsys.readouterr()
combined = captured.out + captured.err
assert "stale" not in combined.lower()
assert "No audit-report" not in combined