last30days-skill/tests/test_competitor_fanout.py

# ruff: noqa: E402
"""Tests for scripts/lib/fanout.run_competitor_fanout."""

from __future__ import annotations

import io
import sys
import threading
import time
import unittest
from contextlib import redirect_stderr
from pathlib import Path
from unittest import mock

REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT / "skills" / "last30days" / "scripts"))

from lib import fanout


def _fake_report(topic: str):
    """Build a lightweight Report stand-in. Tests only check identity."""
    class _R:
        pass

    r = _R()
    r.topic = topic
    return r


class FanoutOrchestratorTests(unittest.TestCase):
    def test_main_plus_two_competitors_all_succeed(self):
        def main_runner():
            return _fake_report("OpenAI")

        def comp_runner(entity):
            return _fake_report(entity)

        err = io.StringIO()
        with redirect_stderr(err):
            results = fanout.run_competitor_fanout(
                main_topic="OpenAI",
                main_runner=main_runner,
                competitors=["Anthropic", "xAI"],
                competitor_runner=comp_runner,
            )
        labels = [label for label, _ in results]
        self.assertEqual(labels, ["OpenAI", "Anthropic", "xAI"])
        self.assertEqual(results[0][1].topic, "OpenAI")
        self.assertEqual(results[1][1].topic, "Anthropic")

    def test_one_competitor_failure_degrades_gracefully(self):
        def main_runner():
            return _fake_report("OpenAI")

        def comp_runner(entity):
            if entity == "BrokenCo":
                raise RuntimeError("upstream offline")
            return _fake_report(entity)

        err = io.StringIO()
        with redirect_stderr(err):
            results = fanout.run_competitor_fanout(
                main_topic="OpenAI",
                main_runner=main_runner,
                competitors=["Anthropic", "BrokenCo", "xAI"],
                competitor_runner=comp_runner,
            )
        labels = [label for label, _ in results]
        self.assertEqual(labels, ["OpenAI", "Anthropic", "xAI"])
        self.assertIn("BrokenCo", err.getvalue())
        self.assertIn("upstream offline", err.getvalue())

    def test_main_topic_failure_leaves_only_competitors(self):
        def main_runner():
            raise RuntimeError("main exploded")

        def comp_runner(entity):
            return _fake_report(entity)

        err = io.StringIO()
        with redirect_stderr(err):
            results = fanout.run_competitor_fanout(
                main_topic="OpenAI",
                main_runner=main_runner,
                competitors=["Anthropic", "xAI"],
                competitor_runner=comp_runner,
            )
        labels = [label for label, _ in results]
        self.assertEqual(labels, ["Anthropic", "xAI"])
        self.assertIn("main exploded", err.getvalue())

    def test_empty_competitor_list_runs_only_main(self):
        def main_runner():
            return _fake_report("OpenAI")

        def comp_runner(_entity):
            raise AssertionError("should not be called when competitors=[]")

        err = io.StringIO()
        with redirect_stderr(err):
            results = fanout.run_competitor_fanout(
                main_topic="OpenAI",
                main_runner=main_runner,
                competitors=[],
                competitor_runner=comp_runner,
            )
        self.assertEqual([label for label, _ in results], ["OpenAI"])

    def test_sub_runs_execute_in_parallel(self):
        """Wall clock should be closer to max(latency) than sum(latency)."""
        delay = 0.2
        call_count = 3  # main + 2 competitors

        def make_runner(_label):
            def runner():
                time.sleep(delay)
                return _fake_report(_label)
            return runner

        def comp_runner(entity):
            return make_runner(entity)()

        start = time.monotonic()
        with redirect_stderr(io.StringIO()):
            results = fanout.run_competitor_fanout(
                main_topic="OpenAI",
                main_runner=make_runner("OpenAI"),
                competitors=["Anthropic", "xAI"],
                competitor_runner=comp_runner,
            )
        elapsed = time.monotonic() - start
        self.assertEqual(len(results), 3)
        # Generous margin: parallel execution should finish well under
        # sum(call_count * delay) == 0.6s. We accept anything under 0.5s.
        self.assertLess(
            elapsed, delay * call_count,
            f"Expected parallel execution < {delay * call_count:.2f}s, "
            f"got {elapsed:.2f}s (sub-runs likely serialized)",
        )

    def test_all_competitors_fail_leaves_main_only(self):
        def main_runner():
            return _fake_report("OpenAI")

        def comp_runner(_entity):
            raise RuntimeError("all offline")

        with redirect_stderr(io.StringIO()):
            results = fanout.run_competitor_fanout(
                main_topic="OpenAI",
                main_runner=main_runner,
                competitors=["A", "B", "C"],
                competitor_runner=comp_runner,
            )
        self.assertEqual([label for label, _ in results], ["OpenAI"])


if __name__ == "__main__":
    unittest.main()