tracer-cloud--opensre/tests/analytics/test_cli.py

from __future__ import annotations

import pytest

from platform.analytics import cli
from platform.analytics.events import Event
from platform.analytics.source import EntrypointSource, TriggerMode


def _assert_investigation_events_have_source(
    events: list[tuple[Event, dict[str, object] | None]],
) -> None:
    investigation_events = {
        Event.INVESTIGATION_STARTED,
        Event.INVESTIGATION_COMPLETED,
        Event.INVESTIGATION_FAILED,
        Event.INVESTIGATION_OUTCOME,
        Event.INVESTIGATION_CANCELLED,
    }
    for event, properties in events:
        if event not in investigation_events:
            continue
        assert properties is not None, f"{event.value} must include properties"
        source = properties.get("source")
        assert isinstance(source, str), f"{event.value} must include a string source"
        assert source.strip(), f"{event.value} source must be non-empty"
        assert properties.get("investigation_loop_count") is not None, (
            f"{event.value} must include investigation_loop_count"
        )
        assert properties.get("investigation_iteration_cap") is not None, (
            f"{event.value} must include investigation_iteration_cap"
        )


class _StubAnalytics:
    def __init__(self) -> None:
        self.events: list[tuple[Event, dict[str, object] | None]] = []
        self.identified: list[dict[str, object]] = []
        self.persistent_properties: dict[str, object] = {}

    def capture(self, event: Event, properties: dict[str, object] | None = None) -> None:
        self.events.append((event, properties))

    def identify(self, set_properties: dict[str, object]) -> None:
        self.identified.append(set_properties)

    def set_persistent_property(self, key: str, value: object) -> None:
        self.persistent_properties[key] = value


def test_capture_cli_invoked_uses_safe_capture(monkeypatch: pytest.MonkeyPatch) -> None:
    stub = _StubAnalytics()
    monkeypatch.setattr(cli, "get_analytics", lambda: stub)

    cli.capture_cli_invoked({"command_path": "opensre version"})

    assert stub.events == [
        (Event.CLI_INVOKED, {"command_path": "opensre version"}),
    ]


def test_capture_cli_invoked_reports_analytics_failures_to_sentry(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    captured_errors: list[BaseException] = []
    expected_error = RuntimeError("analytics unavailable")

    def raise_error() -> _StubAnalytics:
        raise expected_error

    monkeypatch.setattr(cli, "get_analytics", raise_error)
    monkeypatch.setattr(cli, "capture_exception", captured_errors.append)

    cli.capture_cli_invoked()

    assert captured_errors == [expected_error]


def test_identify_github_username_sets_person_property(monkeypatch: pytest.MonkeyPatch) -> None:
    stub = _StubAnalytics()
    monkeypatch.setattr(cli, "get_analytics", lambda: stub)

    cli.identify_github_username("octocat")

    assert stub.identified == [{"github_username": "octocat"}]
    assert stub.persistent_properties == {"github_username": "octocat"}


def test_identify_github_username_noop_on_empty(monkeypatch: pytest.MonkeyPatch) -> None:
    stub = _StubAnalytics()
    monkeypatch.setattr(cli, "get_analytics", lambda: stub)

    cli.identify_github_username("")

    assert stub.identified == []


def test_identify_saved_github_username_reads_store(monkeypatch: pytest.MonkeyPatch) -> None:
    stub = _StubAnalytics()
    monkeypatch.setattr(cli, "get_analytics", lambda: stub)
    monkeypatch.setattr(
        "integrations.github.identity.saved_github_username",
        lambda: "octocat",
    )

    cli.identify_saved_github_username()

    assert stub.identified == [{"github_username": "octocat"}]
    assert stub.persistent_properties == {"github_username": "octocat"}


def test_identify_saved_github_username_noop_when_store_empty(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    stub = _StubAnalytics()
    monkeypatch.setattr(cli, "get_analytics", lambda: stub)
    monkeypatch.setattr("integrations.github.identity.saved_github_username", lambda: "")

    cli.identify_saved_github_username()

    assert stub.identified == []


def test_identify_github_username_reports_failures_to_sentry(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    captured_errors: list[BaseException] = []
    expected_error = RuntimeError("analytics unavailable")

    def raise_error() -> _StubAnalytics:
        raise expected_error

    monkeypatch.setattr(cli, "get_analytics", raise_error)
    monkeypatch.setattr(cli, "capture_exception", captured_errors.append)

    cli.identify_github_username("octocat")

    assert captured_errors == [expected_error]


def test_capture_github_login_completed(monkeypatch: pytest.MonkeyPatch) -> None:
    stub = _StubAnalytics()
    monkeypatch.setattr(cli, "get_analytics", lambda: stub)

    cli.capture_github_login_completed("octocat")

    assert stub.events == [
        (Event.GITHUB_LOGIN_COMPLETED, {"github_username": "octocat"}),
    ]


def test_build_cli_invoked_properties_includes_full_command_path() -> None:
    properties = cli.build_cli_invoked_properties(
        entrypoint="opensre",
        command_parts=["remote", "ops", "status"],
        debug=True,
    )

    assert properties == {
        "entrypoint": "opensre",
        "command_path": "opensre remote ops status",
        "command_family": "remote",
        "json_output": False,
        "verbose": False,
        "debug": True,
        "yes": False,
        "interactive": True,
        "subcommand": "ops",
        "command_leaf": "status",
    }


def test_build_cli_invoked_properties_handles_root_invocation() -> None:
    properties = cli.build_cli_invoked_properties(
        entrypoint="opensre",
        command_parts=[],
    )

    assert properties["command_path"] == "opensre"
    assert properties["command_family"] == "root"
    assert "subcommand" not in properties
    assert "command_leaf" not in properties


def test_capture_update_helpers_emit_expected_events(monkeypatch: pytest.MonkeyPatch) -> None:
    stub = _StubAnalytics()
    monkeypatch.setattr(cli, "get_analytics", lambda: stub)

    cli.capture_update_started(check_only=True)
    cli.capture_update_completed(check_only=False, updated=True)
    cli.capture_update_failed(check_only=False, reason="RuntimeError")

    assert stub.events == [
        (Event.UPDATE_STARTED, {"check_only": True}),
        (Event.UPDATE_COMPLETED, {"check_only": False, "updated": True}),
        (Event.UPDATE_FAILED, {"check_only": False, "reason": "RuntimeError"}),
    ]


def test_capture_eval_process_metrics_emit_expected_contract(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    stub = _StubAnalytics()
    monkeypatch.setattr(cli, "get_analytics", lambda: stub)

    cli.capture_eval_process_started(rubric="must cite logs", mode="opensre_llm_judge")
    cli.capture_eval_process_completed(
        duration_ms=740.3,
        overall_pass=True,
        score_0_100=92,
        rubric_item_count=4,
        mode="opensre_llm_judge",
    )
    cli.capture_eval_process_parse_failed(
        failure_type="ValueError",
        mode="opensre_llm_judge",
    )
    cli.capture_eval_process_failed(
        duration_ms=1200.0,
        failure_stage="invoke_judge",
        failure_type="RuntimeError",
        mode="opensre_llm_judge",
    )
    cli.capture_eval_process_skipped(reason="missing_rubric", mode="opensre_llm_judge")

    for event, properties in stub.events:
        assert properties is not None
        required = cli.EVAL_AND_TERMINAL_EVENT_CONTRACT.get(event)
        if required is None:
            continue
        assert required.issubset(properties.keys())


def test_capture_terminal_metrics_emit_expected_contract(monkeypatch: pytest.MonkeyPatch) -> None:
    stub = _StubAnalytics()
    monkeypatch.setattr(cli, "get_analytics", lambda: stub)

    cli.capture_terminal_actions_planned(planned_count=3, has_unhandled_clause=True)
    cli.capture_terminal_actions_executed(
        planned_count=3,
        executed_count=2,
        executed_success_count=1,
    )
    cli.capture_terminal_turn_summarized(
        planned_count=3,
        executed_count=2,
        executed_success_count=1,
        fallback_to_llm=True,
        session_turn_index=8,
        session_fallback_count=3,
        session_action_success_percent=75.0,
        session_fallback_rate_percent=37.5,
    )

    for event, properties in stub.events:
        assert properties is not None
        required = cli.EVAL_AND_TERMINAL_EVENT_CONTRACT.get(event)
        if required is None:
            continue
        assert required.issubset(properties.keys())


def test_eval_and_terminal_kpi_queries_cover_core_metrics() -> None:
    expected_keys = {
        "eval_pass_rate",
        "eval_latency_p50_p95_ms",
        "eval_parse_error_rate",
        "terminal_action_execution_success_rate",
        "terminal_fallback_rate",
    }
    assert expected_keys.issubset(cli.EVAL_AND_TERMINAL_KPI_QUERIES.keys())
    for query in cli.EVAL_AND_TERMINAL_KPI_QUERIES.values():
        assert "FROM events" in query


def test_track_investigation_emits_lifecycle_once(monkeypatch: pytest.MonkeyPatch) -> None:
    stub = _StubAnalytics()
    monkeypatch.setattr(cli, "get_analytics", lambda: stub)

    with cli.track_investigation(
        entrypoint=EntrypointSource.CLI_COMMAND,
        trigger_mode=TriggerMode.FILE,
        input_path="alert.json",
    ):
        pass

    emitted_events = [event for event, _ in stub.events]
    assert emitted_events == [Event.INVESTIGATION_STARTED, Event.INVESTIGATION_COMPLETED]
    _assert_investigation_events_have_source(stub.events)
    started_props = stub.events[0][1] or {}
    completed_props = stub.events[1][1] or {}
    assert started_props["source"] == "test"
    assert started_props["entrypoint_source"] == "cli_command"
    assert started_props["category"] == "test"
    assert started_props["trigger_mode"] == "file"
    assert started_props["is_test"] is True
    assert started_props["investigation_id"] == completed_props["investigation_id"]
    assert started_props["investigation_loop_count"] == 0
    assert completed_props["investigation_loop_count"] == 0


def test_track_investigation_emits_failed_on_exception(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    stub = _StubAnalytics()
    monkeypatch.setattr(cli, "get_analytics", lambda: stub)

    # Wrap the raise in a callable so the ``raise`` lives inside ``_trigger``
    # rather than directly in the test body. ``pytest.raises`` then sees a
    # plain function call as its protected expression, which lets CodeQL
    # ``py/unreachable-statement`` prove the assertions below are reachable
    # (the previous nested-``with`` workaround still tripped the rule).
    def _trigger() -> None:
        with cli.track_investigation(
            entrypoint=EntrypointSource.MCP,
            trigger_mode=TriggerMode.SERVICE_RUNTIME,
        ):
            raise RuntimeError("boom")

    with pytest.raises(RuntimeError, match="boom"):
        _trigger()

    emitted_events = [event for event, _ in stub.events]
    assert emitted_events == [Event.INVESTIGATION_STARTED, Event.INVESTIGATION_FAILED]
    _assert_investigation_events_have_source(stub.events)
    failed_props = stub.events[1][1] or {}
    assert failed_props["failure_type"] == "RuntimeError"
    assert failed_props["failure_message"] == "boom"


def test_capture_investigation_failed_includes_state_loop_metrics(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    stub = _StubAnalytics()
    monkeypatch.setattr(cli, "get_analytics", lambda: stub)

    cli.capture_investigation_failed(
        failure_type="RuntimeError",
        failure_message="boom",
        shared_properties={"investigation_id": "inv-fail"},
        state={"investigation_loop_count": 4, "investigation_iteration_cap": 20},
    )

    failed_props = stub.events[0][1] or {}
    assert failed_props["investigation_loop_count"] == 4
    assert failed_props["investigation_iteration_cap"] == 20


def test_track_investigation_failed_uses_tracker_loop_metrics(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    stub = _StubAnalytics()
    monkeypatch.setattr(cli, "get_analytics", lambda: stub)

    def _trigger() -> None:
        with cli.track_investigation(
            entrypoint=EntrypointSource.CLI_COMMAND,
            trigger_mode=TriggerMode.FILE,
        ) as tracker:
            tracker.record_loop_metrics_from_state(
                {"investigation_loop_count": 6, "investigation_iteration_cap": 20}
            )
            raise RuntimeError("boom")

    with pytest.raises(RuntimeError, match="boom"):
        _trigger()

    failed_props = stub.events[1][1] or {}
    assert failed_props["investigation_loop_count"] == 6
    assert failed_props["investigation_iteration_cap"] == 20


def test_capture_investigation_outcome_and_cancelled(monkeypatch: pytest.MonkeyPatch) -> None:
    stub = _StubAnalytics()
    monkeypatch.setattr(cli, "get_analytics", lambda: stub)

    cli.capture_investigation_outcome(
        investigation_id="inv-123",
        status="failed",
        investigation_target="generic",
        error_excerpt="boom",
        failure_category="unknown",
        state={"investigation_loop_count": 5, "investigation_iteration_cap": 20},
    )
    cli.capture_investigation_cancelled(
        investigation_id="inv-456",
        investigation_target="alert.json",
        state={"investigation_loop_count": 2, "investigation_iteration_cap": 20},
    )

    assert stub.events[0][0] == Event.INVESTIGATION_OUTCOME
    outcome_props = stub.events[0][1] or {}
    assert outcome_props["investigation_id"] == "inv-123"
    assert outcome_props["status"] == "failed"
    assert outcome_props["investigation_target"] == "generic"
    assert outcome_props["error_excerpt"] == "boom"
    assert outcome_props["investigation_loop_count"] == 5
    assert stub.events[1][0] == Event.INVESTIGATION_CANCELLED
    cancelled_props = stub.events[1][1] or {}
    assert cancelled_props["investigation_id"] == "inv-456"
    assert cancelled_props["failure_category"] == "user_cancelled"
    assert cancelled_props["investigation_loop_count"] == 2


def test_track_investigation_records_loop_metrics_on_completion(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    stub = _StubAnalytics()
    monkeypatch.setattr(cli, "get_analytics", lambda: stub)

    with cli.track_investigation(
        entrypoint=EntrypointSource.CLI_COMMAND,
        trigger_mode=TriggerMode.FILE,
    ) as tracker:
        tracker.record_loop_metrics_from_state(
            {"investigation_loop_count": 8, "investigation_iteration_cap": 20}
        )

    completed_props = stub.events[1][1] or {}
    assert completed_props["investigation_loop_count"] == 8
    assert completed_props["investigation_iteration_cap"] == 20


def test_track_investigation_nested_context_dedupes(monkeypatch: pytest.MonkeyPatch) -> None:
    stub = _StubAnalytics()
    monkeypatch.setattr(cli, "get_analytics", lambda: stub)

    with (
        cli.track_investigation(
            entrypoint=EntrypointSource.SDK,
            trigger_mode=TriggerMode.SERVICE_RUNTIME,
        ),
        cli.track_investigation(
            entrypoint=EntrypointSource.CLI_COMMAND,
            trigger_mode=TriggerMode.FILE,
        ),
    ):
        pass

    emitted_events = [event for event, _ in stub.events]
    assert emitted_events == [Event.INVESTIGATION_STARTED, Event.INVESTIGATION_COMPLETED]
    _assert_investigation_events_have_source(stub.events)


def test_capture_diagnosis_category_mismatch(monkeypatch: pytest.MonkeyPatch) -> None:
    stub = _StubAnalytics()
    monkeypatch.setattr(cli, "get_analytics", lambda: stub)

    cli.capture_diagnosis_category_mismatch(
        root_cause_category="dns_resolution_failure",
        mismatch_reason="root cause text signals database (2 keyword hits)",
    )

    assert stub.events == [
        (
            Event.DIAGNOSIS_CATEGORY_MISMATCH,
            {
                "category_text_mismatch": True,
                "root_cause_category": "dns_resolution_failure",
                "mismatch_reason": "root cause text signals database (2 keyword hits)",
            },
        )
    ]