reuse-surface/tests/test_stats.py

from __future__ import annotations

from pathlib import Path

import pytest

from reuse_surface.stats import (
    collect_roster_stats,
    collect_stats,
    format_roster_stats_markdown,
    format_stats_markdown,
)


@pytest.fixture
def stats_repo(tmp_path, monkeypatch):
    monkeypatch.delenv("REUSE_SURFACE_URL", raising=False)
    index = tmp_path / "registry/indexes/capabilities.yaml"
    index.parent.mkdir(parents=True)
    index.write_text("""version: 1
domain: infotech
capabilities:
  - id: capability.example.source
    vector: D3 / A2 / C2 / R2
    path: registry/capabilities/source.md
  - id: capability.example.service
    vector: D4 / A5 / C3 / R3
    path: registry/capabilities/service.md
""")
    return tmp_path


def test_collect_stats_on_repo_root(stats_repo):
    stats = collect_stats(stats_repo)
    assert stats["capability_count"] == 2
    assert stats["index_present"] is True
    assert stats["histograms"]["discovery"]["D3"] == 1
    assert stats["histograms"]["discovery"]["D4"] == 1
    assert stats["reliability"] == {"r0_r2": 1, "r3_plus": 1}


def test_format_stats_markdown_contains_count(stats_repo):
    text = format_stats_markdown(collect_stats(stats_repo))
    assert "**Capabilities:** 2" in text


def test_collect_roster_stats_federation_ready():
    root = Path(__file__).resolve().parent.parent
    roster = root / "registry/federation/local-repo-roster.yaml"
    stats = collect_roster_stats(roster, federation_ready=True)
    assert stats["counts"]["total"] == 62
    assert stats["counts"]["established"] == 62
    assert "federation_readiness" in stats
    text = format_roster_stats_markdown(stats)
    assert "publish pass ratio" in text

# --- T06: hub freshness (composed_at age + stale flag) ---

from datetime import datetime, timedelta, timezone

from reuse_surface.stats import _composed_at_age_days, _freshness_days, _hub_summary


def test_hub_summary_includes_freshness_fields(monkeypatch):
    monkeypatch.setenv("REUSE_SURFACE_URL", "http://example.invalid")
    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    monkeypatch.setattr(
        "reuse_surface.hub_client.hub_list",
        lambda base_url=None: (200, {"count": 3, "repos": [{"enabled": True}] * 3}),
    )
    monkeypatch.setattr(
        "reuse_surface.hub_client.hub_federated",
        lambda base_url=None: (200, {"composed_at": now, "stale": False}),
    )
    summary = _hub_summary(None)
    assert summary["composed_at"] == now
    assert summary["stale"] is False
    assert summary["age_days"] is not None
    assert summary["age_days"] < 1
    assert summary["stale_warning"] is False


def test_hub_summary_stale_warning_from_age(monkeypatch):
    monkeypatch.setenv("REUSE_SURFACE_URL", "http://example.invalid")
    old = (datetime.now(timezone.utc) - timedelta(days=10)).strftime("%Y-%m-%dT%H:%M:%SZ")
    monkeypatch.setattr(
        "reuse_surface.hub_client.hub_list",
        lambda base_url=None: (200, {"count": 1, "repos": [{"enabled": True}]}),
    )
    monkeypatch.setattr(
        "reuse_surface.hub_client.hub_federated",
        lambda base_url=None: (200, {"composed_at": old, "stale": False}),
    )
    summary = _hub_summary(None)
    assert summary["age_days"] > 7
    assert summary["stale_warning"] is True


def test_hub_summary_stale_warning_from_flag(monkeypatch):
    monkeypatch.setenv("REUSE_SURFACE_URL", "http://example.invalid")
    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    monkeypatch.setattr(
        "reuse_surface.hub_client.hub_list",
        lambda base_url=None: (200, {"count": 1, "repos": [{"enabled": True}]}),
    )
    monkeypatch.setattr(
        "reuse_surface.hub_client.hub_federated",
        lambda base_url=None: (200, {"composed_at": now, "stale": True}),
    )
    summary = _hub_summary(None)
    assert summary["stale_warning"] is True


def test_hub_summary_handles_null_composed_at(monkeypatch):
    monkeypatch.setenv("REUSE_SURFACE_URL", "http://example.invalid")
    monkeypatch.setattr(
        "reuse_surface.hub_client.hub_list",
        lambda base_url=None: (200, {"count": 0, "repos": []}),
    )
    monkeypatch.setattr(
        "reuse_surface.hub_client.hub_federated",
        lambda base_url=None: (200, {"composed_at": None, "stale": False}),
    )
    summary = _hub_summary(None)
    assert summary["age_days"] is None
    assert summary["stale_warning"] is False


def test_hub_summary_degrades_when_federated_unreachable(monkeypatch):
    monkeypatch.setenv("REUSE_SURFACE_URL", "http://example.invalid")
    monkeypatch.setattr(
        "reuse_surface.hub_client.hub_list",
        lambda base_url=None: (200, {"count": 1, "repos": [{"enabled": True}]}),
    )

    def _raise(base_url=None):
        import urllib.error

        raise urllib.error.URLError("no route")

    monkeypatch.setattr("reuse_surface.hub_client.hub_federated", _raise)
    summary = _hub_summary(None)
    assert summary["configured"] is True
    assert "composed_at" not in summary


def test_composed_at_age_days_none_when_missing():
    assert _composed_at_age_days(None) is None


def test_composed_at_age_days_none_when_malformed():
    assert _composed_at_age_days("not-a-timestamp") is None


def test_freshness_days_default():
    assert _freshness_days() == 7


def test_freshness_days_env_override(monkeypatch):
    monkeypatch.setenv("REUSE_SURFACE_FRESHNESS_DAYS", "3")
    assert _freshness_days() == 3


def test_freshness_days_env_invalid_falls_back(monkeypatch):
    monkeypatch.setenv("REUSE_SURFACE_FRESHNESS_DAYS", "not-a-number")
    assert _freshness_days() == 7


def test_format_stats_markdown_shows_stale_warning():
    from reuse_surface.stats import format_stats_markdown

    stats = {
        "repo_root": "/tmp/x",
        "capability_count": 0,
        "registry_present": True,
        "index_present": True,
        "sources_present": True,
        "reliability": {"r0_r2": 0, "r3_plus": 0},
        "histograms": {},
        "vector_drift": [],
        "hub": {
            "configured": True,
            "registration_count": 1,
            "enabled_count": 1,
            "composed_at": "2020-01-01T00:00:00Z",
            "stale": False,
            "age_days": 100.0,
            "freshness_threshold_days": 7,
            "stale_warning": True,
        },
    }
    text = format_stats_markdown(stats)
    assert "STALE" in text