#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#

"""Offline checks for the ``scripts/provider-watch`` tooling.

``inventory.py`` must keep finding every provider service and its default model
as the services tree evolves; ``digest.py`` must render whatever frontmatter the
researcher agents write; ``probe.py`` must refuse to run without credentials
instead of failing later with a provider error. None of these touch the network.
"""

import json
import os
import re
import subprocess
import sys
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).parent.parent
SCRIPTS = REPO_ROOT / "scripts" / "provider-watch"
sys.path.insert(0, str(SCRIPTS))

import digest  # noqa: E402
import inventory  # noqa: E402


@pytest.fixture(scope="module")
def units():
    found = inventory.scan_services()
    inventory.enrich(found)
    return found


class TestInventory:
    def test_covers_the_services_tree(self, units):
        providers = {u.provider for u in units}
        assert len(providers) >= 55
        assert len(units) >= 90
        assert {u.type for u in units} >= {"llm", "stt", "tts", "realtime", "image"}

    def test_unit_ids_are_unique_and_sorted(self, units):
        ids = [u.id for u in units]
        assert ids == sorted(ids)
        assert len(ids) == len(set(ids))

    @pytest.mark.parametrize(
        "unit_id, class_name, expect_model",
        [
            ("openai/llm", "OpenAILLMService", True),
            ("openai/responses-llm", "OpenAIResponsesLLMService", True),
            ("openai/realtime", "OpenAIRealtimeLLMService", True),
            ("cartesia/tts", "CartesiaTTSService", True),
            ("deepgram/stt", "DeepgramSTTService", True),
            ("google/realtime", "GeminiLiveLLMService", True),
            ("groq/llm", "GroqLLMService", True),
            ("azure/tts", "AzureTTSService", False),
        ],
    )
    def test_known_units(self, units, unit_id, class_name, expect_model):
        unit = next(u for u in units if u.id == unit_id)
        assert class_name in [c.name for c in unit.classes]
        assert (unit.default_model is not None) == expect_model

    def test_every_settings_model_literal_is_captured(self, units):
        """A class whose ``__init__`` writes ``model="..."`` into its Settings has a default."""
        literal = re.compile(r"self\.Settings\([^)]*?\bmodel=\"", re.DOTALL)
        for unit in units:
            for cls in unit.classes:
                source = (REPO_ROOT / "src" / Path(*cls.module.split("."))).with_suffix(".py")
                text = source.read_text()
                if f"class {cls.name}(" not in text:
                    continue
                section = text.split(f"class {cls.name}(")[1].split("\nclass ")[0]
                if literal.search(section):
                    assert cls.default_model, cls.name

    def test_thin_wrappers_are_flagged(self, units):
        groq = next(u for u in units if u.id == "groq/llm")
        openai = next(u for u in units if u.id == "openai/llm")
        assert groq.is_thin_wrapper
        assert groq.classes[0].base_url == "https://api.groq.com/openai/v1"
        assert not openai.is_thin_wrapper

    def test_base_classes_and_shims_are_excluded(self, units):
        names = {c.name for u in units for c in u.classes}
        assert "BaseOpenAILLMService" not in names
        assert "AzureBaseTTSService" not in names
        assert "BaseWhisperSTTService" not in names
        assert "BasetenLLMService" in names
        assert not any(u.provider == "grok" for u in units)

    def test_joins(self, units):
        deepgram = next(u for u in units if u.id == "deepgram/stt")
        assert any(e["value"] == "deepgram_stt" for e in deepgram.registry)
        assert "DEEPGRAM_API_KEY" in deepgram.env_vars
        assert any("deepgram" in bot for bot in deepgram.example_bots)
        assert (
            deepgram.docs_url
            == "https://docs.pipecat.ai/api-reference/server/services/stt/deepgram"
        )

    def test_select_and_cli(self, units):
        picked = inventory.select(units, ["openai", "deepgram/stt"], None)
        assert {u.provider for u in picked} == {"openai", "deepgram"}
        assert [u.id for u in picked if u.provider == "deepgram"] == ["deepgram/stt"]
        assert len(inventory.select(units, None, 3)) == 3

        result = subprocess.run(
            [sys.executable, str(SCRIPTS / "inventory.py"), "--json", "--only", "cartesia"],
            capture_output=True,
            text=True,
            check=True,
        )
        data = json.loads(result.stdout)
        assert {u["id"] for u in data} == {"cartesia/stt", "cartesia/tts", "cartesia/turns-stt"}


class TestDigest:
    @pytest.fixture
    def reports_dir(self, tmp_path):
        def write(unit, body):
            path = tmp_path / "reports" / unit / "2026-08-20.md"
            path.parent.mkdir(parents=True)
            path.write_text(body)

        write(
            "openai/llm",
            "---\nservice: openai/llm\ndefault_model: gpt-4.1\nsummary: gpt-5 is GA and faster\n"
            "gaps:\n  - item: gpt-5 supersedes gpt-4.1\n    first_seen: 2026-08-20\n    action: pr\n"
            "prs:\n  - url: https://github.com/pipecat-ai/pipecat/pull/1\n    state: open\n"
            "    summary: bump default to gpt-5\nerror: null\n---\n# OpenAI LLM\n",
        )
        write(
            "cartesia/tts",
            "---\nservice: cartesia/tts\ndefault_model: sonic-3.5\n"
            "gaps:\n  - item: sonic-4 preview needs a voice migration\n    first_seen: 2026-07-30\n"
            "    action: consider\n    priority: low\n    note: re-check when GA\n"
            "  - item: emotion controls unreachable\n    first_seen: 2026-08-20\n"
            "    action: consider\n    priority: high\n"
            "    needs: which voices support emotion\nprs: []\nerror: null\n---\n",
        )
        write(
            "groq/llm",
            "---\nservice: groq/llm\ndefault_model: openai/gpt-oss-120b\ngaps: []\nprs: []\nerror: null\n---\n",
        )
        write(
            "fireworks/llm",
            "---\nservice: fireworks/llm\ndefault_model: accounts/fireworks/models/firefunction-v2\n"
            "gaps:\n  - item: firefunction-v2 is retired\n    first_seen: 2026-08-20\n    action: pr\n"
            "prs:\n  - branch: provider-watch/fireworks-llm-default\n    state: branch\n"
            "    summary: Default FireworksLLMService to gpt-oss-120b\nerror: null\n---\n"
            "\n## PRs\n- `provider-watch/fireworks-llm-default` — review: "
            "`git show provider-watch/fireworks-llm-default` — Default FireworksLLMService to gpt-oss-120b\n",
        )
        write("mistral/stt", "---\nservice: mistral/stt\nerror: missing MISTRAL_API_KEY\n---\n")
        write("broken/tts", "no frontmatter at all\n")
        return tmp_path

    def test_render_sections(self, reports_dir):
        reports = digest.load_reports(reports_dir, "2026-08-20")
        text = digest.render(
            reports, date="2026-08-20", highlights="- Big week for LLMs", repo_url="https://x/y"
        )

        assert text.startswith("# Provider watch — 2026-08-20\n\n- Big week for LLMs")
        assert (
            "**6 units researched** — 1 PRs, 1 branches, 2 changes to consider, 2 errors, 1 with nothing new."
            in text
        )
        sections = [line for line in text.splitlines() if line.startswith("## ")]
        assert sections == [
            "## PRs to review",
            "## Branches not opened as PRs (dry run)",
            "## Changes to consider",
            "## Did not complete",
            "## Nothing new",
        ]
        assert "https://github.com/pipecat-ai/pipecat/pull/1 — bump default to gpt-5" in text
        assert "`git show provider-watch/fireworks-llm-default`" in text
        assert (
            "sonic-4 preview needs a voice migration (since 2026-07-30, 3 weeks) — re-check when GA"
            in text
        )
        consider = text.split("## Changes to consider")[1].split("## Did not complete")[0]
        assert (
            "emotion controls unreachable — *needs a call: which voices support emotion*"
            in consider
        )
        assert consider.index("**High**") < consider.index("emotion controls unreachable")
        assert consider.index("emotion controls unreachable") < consider.index(
            "<details><summary><b>Low</b> (1)"
        )
        assert consider.index("<details>") < consider.index("sonic-4 preview")
        assert "missing MISTRAL_API_KEY" in text and "report has no frontmatter" in text
        assert (
            "[groq/llm](https://x/y/blob/main/reports/groq/llm/2026-08-20.md)"
            in text.split("## Nothing new")[1]
        )
        assert text.rstrip().endswith("#the-weekly-loop).")

    def test_cli_writes_file(self, reports_dir):
        out = reports_dir / "digests" / "2026-08-20.md"
        subprocess.run(
            [
                sys.executable,
                str(SCRIPTS / "digest.py"),
                "--reports",
                str(reports_dir),
                "--date",
                "2026-08-20",
                "--out",
                str(out),
                "--repo-url",
                "",
            ],
            check=True,
        )
        assert out.read_text().startswith("# Provider watch — 2026-08-20")
        assert "`openai/llm`" in out.read_text()


class TestProbe:
    def _run(self, *argv, env_overrides=None):
        env = {k: v for k, v in os.environ.items() if not k.endswith("_API_KEY")}
        env.update(env_overrides or {})
        return subprocess.run(
            [sys.executable, str(SCRIPTS / "probe.py"), *argv],
            capture_output=True,
            text=True,
            env=env,
            cwd=REPO_ROOT,
        )

    def test_missing_credentials_exit_2_and_name_only(self):
        result = self._run("--no-dotenv", "run", "--service", "CartesiaTTSService", "--model", "x")
        assert result.returncode == 2, result.stderr
        assert "CARTESIA_API_KEY" in result.stderr
        assert "***" not in result.stderr

    def test_research_only_types_exit_3(self):
        result = self._run("--no-dotenv", "run", "--service", "FalImageGenService", "--model", "x")
        assert result.returncode == 3
        assert "research-only" in result.stderr

    def test_unknown_provider_catalogue_exit_3(self):
        result = self._run("--no-dotenv", "list-models", "--provider", "nosuchprovider")
        assert result.returncode == 3

    def test_merge_probe_results_medians(self):
        import probe

        def result(ok=True, ttfat=None, ttfb=None, error=None):
            return probe.ProbeResult(
                service="AnthropicLLMService",
                model="m",
                ok=ok,
                ttfb_ms=ttfb,
                ttfat_ms=ttfat,
                thinking_ms=0,
                frames={},
                error=error,
            )

        single = result(ttfat=100, ttfb=100)
        assert probe.merge_probe_results([single]) is single

        merged = probe.merge_probe_results(
            [
                result(ttfat=998, ttfb=998),
                result(ttfat=2401, ttfb=2401),
                result(ttfat=933, ttfb=933),
            ]
        )
        assert merged.ok and merged.ttfat_ms == 998 and merged.ttfb_ms == 998
        assert merged.note == "median of 3/3 runs, ttfat 933–2401 ms"

        partial = probe.merge_probe_results(
            [result(ttfat=100, ttfb=100), result(ok=False, error="boom")]
        )
        assert not partial.ok and partial.ttfat_ms == 100
        assert partial.note.startswith("median of 1/2 runs")

    def test_setting_values_parse_json_and_scalars(self):
        import probe

        assert probe._kv_pairs(['extra={"reasoning_effort": "low"}', "speed=1.5", "x=none"]) == {
            "extra": {"reasoning_effort": "low"},
            "speed": 1.5,
            "x": None,
        }
        with pytest.raises(SystemExit):
            probe._kv_pairs(["extra={not json"])

    def test_too_many_models_rejected(self):
        result = self._run(
            "run", "--service", "OpenAILLMService", *sum([["--model", m] for m in "abcd"], [])
        )
        assert result.returncode != 0
        assert "at most 3" in result.stderr


class TestPublish:
    """publish.py against a fake git/gh so nothing leaves the machine."""

    class FakeShell:
        def __init__(self, open_prs=None, branches=None, commits=None, fragments=None):
            self.calls = []
            self.open_prs = open_prs or {}
            self.branches = set(branches or [])
            self.commits = commits or {}  # branch -> [(subject, body), ...]
            self.fragments = fragments or {}  # branch -> [changelog paths added]
            self.pr_counter = 100

        def run(self, *args, cwd=None, check=True):
            self.calls.append(args)
            if args[:3] == ("gh", "pr", "list"):
                head = args[args.index("--head") + 1]
                url = self.open_prs.get(head)
                return json.dumps([{"url": url}] if url else [])
            if args[:3] == ("gh", "pr", "create"):
                self.pr_counter += 1
                return f"https://github.com/pipecat-ai/pipecat/pull/{self.pr_counter}\n"
            if args[:2] == ("git", "diff") and "--name-only" in args:
                branch = args[4].split("..")[-1]
                return "\n".join(self.fragments.get(branch, []))
            if args[:2] == ("git", "rev-list"):
                branch = args[-1].split("..")[-1]
                return "\n".join(f"{branch}@{i}" for i in range(len(self.commits.get(branch, ()))))
            if args[:2] == ("git", "log"):
                ref, _, index = args[-1].rpartition("@")
                if ref in self.commits:
                    subject, body = self.commits[ref][int(index)]
                    return (subject if "--format=%s" in args else body) + "\n"
                return (
                    "Default FireworksLLMService to gpt-oss-120b\n"
                    if "--format=%s" in args
                    else "Body.\n"
                )
            if args[:3] == ("gh", "issue", "list"):
                return "[]"
            if args[:3] == ("gh", "issue", "create"):
                return "https://github.com/pipecat-ai/provider-watch-reports/issues/1\n"
            return ""

        def ok(self, *args, cwd=None):
            self.calls.append(args)
            if args[:2] == ("git", "rev-parse"):
                return args[-1] in self.branches
            if args[:2] == ("git", "diff"):
                return False  # something is staged
            return True

    @pytest.fixture
    def reports_dir(self, tmp_path):
        import publish

        def write(unit, branch):
            path = tmp_path / "reports" / unit / "2026-08-20.md"
            path.parent.mkdir(parents=True)
            path.write_text(
                f"---\nservice: {unit}\nprs:\n  - branch: {branch}\n"
                f"    state: branch\n    summary: s\n---\n\n# R\n\n## PRs\n"
                f"- `{branch}` — review: `git show {branch}` — s\n"
            )

        write("fireworks/llm", "provider-watch/fireworks-llm-default")
        write("groq/llm", "provider-watch/groq-llm-example")
        write("ollama/llm", "provider-watch/ollama-llm-default")
        return tmp_path, publish

    def test_opens_adopts_and_caps(self, reports_dir):
        tmp_path, publish = reports_dir
        sh = self.FakeShell(
            open_prs={
                "provider-watch/groq-llm-example": "https://github.com/pipecat-ai/pipecat/pull/7"
            },
            branches=[
                "provider-watch/fireworks-llm-default",
                "provider-watch/groq-llm-example",
                "provider-watch/ollama-llm-default",
            ],
        )
        reports = publish.load_reports(tmp_path, "2026-08-20")
        outcome = publish.publish_prs(
            reports,
            sh=sh,
            repo_root=tmp_path,
            pipecat_repo="pipecat-ai/pipecat",
            reports_repo="pipecat-ai/provider-watch-reports",
            date="2026-08-20",
            cap=1,
        )
        assert outcome.opened == ["https://github.com/pipecat-ai/pipecat/pull/101"]
        assert outcome.adopted == ["https://github.com/pipecat-ai/pipecat/pull/7"]
        assert outcome.capped == ["provider-watch/ollama-llm-default"]

        fireworks = (tmp_path / "reports/fireworks/llm/2026-08-20.md").read_text()
        assert (
            "state: open" in fireworks
            and "url: https://github.com/pipecat-ai/pipecat/pull/101" in fireworks
        )
        assert "- https://github.com/pipecat-ai/pipecat/pull/101 — s" in fireworks
        assert "git diff" not in fireworks
        ollama = (tmp_path / "reports/ollama/llm/2026-08-20.md").read_text()
        assert "capped: true" in ollama and "git show provider-watch/ollama-llm-default" in ollama

        pushes = [c for c in sh.calls if c[:2] == ("git", "push")]
        assert pushes == [("git", "push", "-u", "origin", "provider-watch/fireworks-llm-default")]
        create = next(c for c in sh.calls if c[:3] == ("gh", "pr", "create"))
        assert "--draft" in create and "--label" in create
        assert create[create.index("--title") + 1] == "Default FireworksLLMService to gpt-oss-120b"
        assert "reports/fireworks/llm/2026-08-20.md" in create[create.index("--body") + 1]

    def test_multi_commit_branch_stitches_pr(self, tmp_path):
        import publish

        branch = "provider-watch/cartesia-tts-updates"
        path = tmp_path / "reports/cartesia/tts/2026-08-20.md"
        path.parent.mkdir(parents=True)
        path.write_text(
            f"---\nservice: cartesia/tts\nprs:\n  - branch: {branch}\n"
            f"    state: branch\n    summary: Track Cartesia's sonic-4 rollout\n---\n\n"
            f"# R\n\n## PRs\n- `{branch}` — review: `git show {branch}` — s\n"
        )
        sh = self.FakeShell(
            branches=[branch],
            commits={
                branch: [
                    ("Default CartesiaTTSService to sonic-4", "Sonic body."),
                    ("Fix the retired sonic-2 in the eval fallback", "Eval body."),
                ]
            },
        )
        reports = publish.load_reports(tmp_path, "2026-08-20")
        outcome = publish.publish_prs(
            reports,
            sh=sh,
            repo_root=tmp_path,
            pipecat_repo="pipecat-ai/pipecat",
            reports_repo="pipecat-ai/provider-watch-reports",
            date="2026-08-20",
            cap=8,
        )
        assert len(outcome.opened) == 1
        create = next(c for c in sh.calls if c[:3] == ("gh", "pr", "create"))
        assert create[create.index("--title") + 1] == "Track Cartesia's sonic-4 rollout"
        body = create[create.index("--body") + 1]
        assert "## Default CartesiaTTSService to sonic-4\n\nSonic body." in body
        assert "## Fix the retired sonic-2 in the eval fallback\n\nEval body." in body
        assert body.index("sonic-4") < body.index("retired sonic-2")

    def test_renames_changelog_fragments_to_pr_number(self, tmp_path):
        import publish

        branch = "provider-watch/cartesia-tts-updates"
        path = tmp_path / "reports/cartesia/tts/2026-08-20.md"
        path.parent.mkdir(parents=True)
        path.write_text(
            f"---\nservice: cartesia/tts\nprs:\n  - branch: {branch}\n"
            f"    state: branch\n    summary: s\n---\n\n# R\n\n## PRs\n"
            f"- `{branch}` — review: `git show {branch}` — s\n"
        )
        sh = self.FakeShell(
            branches=[branch],
            fragments={
                branch: [
                    "changelog/+cartesia-normalization.added.md",
                    "changelog/+cartesia-locale.added.md",
                    "changelog/+cartesia-sonic.fixed.md",
                    "changelog/999.changed.md",
                ]
            },
        )
        reports = publish.load_reports(tmp_path, "2026-08-20")
        outcome = publish.publish_prs(
            reports,
            sh=sh,
            repo_root=tmp_path,
            pipecat_repo="pipecat-ai/pipecat",
            reports_repo="pipecat-ai/provider-watch-reports",
            date="2026-08-20",
            cap=8,
        )
        assert len(outcome.opened) == 1 and not outcome.skipped
        moves = [c for c in sh.calls if c[:2] == ("git", "mv")]
        assert moves == [
            ("git", "mv", "changelog/+cartesia-locale.added.md", "changelog/101.added.md"),
            ("git", "mv", "changelog/+cartesia-normalization.added.md", "changelog/101.added.2.md"),
            ("git", "mv", "changelog/+cartesia-sonic.fixed.md", "changelog/101.fixed.md"),
            ("git", "mv", "changelog/999.changed.md", "changelog/101.changed.md"),
        ]
        assert any(c[:2] == ("git", "commit") and "#101" in c[-1] for c in sh.calls)
        assert ("git", "push", "origin", branch) in [
            c[:4] for c in sh.calls if c[:2] == ("git", "push")
        ]

    def test_second_pass_is_a_no_op(self, reports_dir):
        tmp_path, publish = reports_dir
        sh = self.FakeShell(
            branches=[
                "provider-watch/fireworks-llm-default",
                "provider-watch/groq-llm-example",
                "provider-watch/ollama-llm-default",
            ]
        )
        kwargs = dict(
            sh=sh,
            repo_root=tmp_path,
            pipecat_repo="p/p",
            reports_repo="p/r",
            date="2026-08-20",
            cap=8,
        )
        publish.publish_prs(publish.load_reports(tmp_path, "2026-08-20"), **kwargs)
        before = len([c for c in sh.calls if c[:3] == ("gh", "pr", "create")])
        publish.publish_prs(publish.load_reports(tmp_path, "2026-08-20"), **kwargs)
        after = len([c for c in sh.calls if c[:3] == ("gh", "pr", "create")])
        assert before == 3 and after == 3

    def test_worth_an_issue(self, reports_dir, tmp_path):
        _, publish = reports_dir
        assert publish.worth_an_issue(publish.load_reports(tmp_path, "2026-08-20"))
        quiet = tmp_path / "quiet" / "reports" / "groq" / "llm"
        quiet.mkdir(parents=True)
        (quiet / "2026-08-20.md").write_text(
            "---\nservice: groq/llm\ngaps: []\nprs: []\nerror: null\n---\n"
        )
        assert not publish.worth_an_issue(publish.load_reports(tmp_path / "quiet", "2026-08-20"))
        (quiet / "2026-08-20.md").write_text("---\nservice: groq/llm\nerror: boom\n---\n")
        assert publish.worth_an_issue(publish.load_reports(tmp_path / "quiet", "2026-08-20"))

    def test_missing_branch_is_skipped(self, reports_dir):
        tmp_path, publish = reports_dir
        sh = self.FakeShell(branches=[])
        outcome = publish.publish_prs(
            publish.load_reports(tmp_path, "2026-08-20"),
            sh=sh,
            repo_root=tmp_path,
            pipecat_repo="p/p",
            reports_repo="p/r",
            date="2026-08-20",
            cap=8,
        )
        assert len(outcome.skipped) == 3 and not outcome.opened


class TestSdkVersions:
    """probe.py sdk-versions: SDK derivation from pyproject, no network."""

    @pytest.fixture
    def probe(self):
        import probe

        return probe

    def test_sdk_requirements_come_from_pyproject_extras(self, probe, units):
        deepgram = [u for u in units if u.provider == "deepgram"]
        reqs = probe.sdk_requirements("deepgram", deepgram)
        assert any(r.startswith("deepgram-sdk") for r in reqs)
        assert not any(r.startswith("pipecat-ai") for r in reqs)

        google = [u for u in units if u.provider == "google"]
        names = {
            r.split(">")[0].split("<")[0].split("=")[0]
            for r in probe.sdk_requirements("google", google)
        }
        assert {"google-genai", "google-cloud-speech", "google-cloud-texttospeech"} <= names

    def test_thin_wrappers_fall_back_to_openai(self, probe, units):
        groq = [u for u in units if u.provider == "groq"]
        reqs = probe.sdk_requirements("groq", groq)
        assert any(r.startswith("groq") for r in reqs)  # groq has its own extra
        cerebras = [u for u in units if u.provider == "cerebras"]
        reqs = probe.sdk_requirements("cerebras", cerebras)
        assert reqs and all(r.startswith("openai") for r in reqs)