insights/tools/insightscli-commands/insightscli_commands/tests/test_commands.py

"""Tests for Insights-specific insightscli commands and hooks."""

from __future__ import annotations

import json
import time

import pytest
from unittest.mock import patch

from insightscli_commands.telemetry_props import (
    _AGENT_ENV_MARKERS,
    _DEVBOX_ENV_MARKERS,
    _detect_agent,
    _detect_environment,
    _infer_process_manager,
    _is_insights_dev,
    _repo_commit,
    _repo_commit_properties,
)


class TestProcessManagerInference:
    @pytest.mark.parametrize(
        ("env_value", "argv", "expected"),
        [
            (None, ["insightscli", "start"], "phrocs"),
            (None, ["insightscli", "start", "--mprocs"], "mprocs"),
            ("/usr/local/bin/phrocs", ["insightscli", "start", "--mprocs"], "phrocs"),
        ],
        ids=["default", "mprocs_flag", "env_override"],
    )
    def test_infer_process_manager(self, monkeypatch, env_value, argv, expected) -> None:
        if env_value is None:
            monkeypatch.delenv("INSIGHTSCLI_PROCESS_MANAGER", raising=False)
        else:
            monkeypatch.setenv("INSIGHTSCLI_PROCESS_MANAGER", env_value)
        monkeypatch.setattr("sys.argv", argv)

        assert _infer_process_manager("start") == expected


class TestDetectEnvironment:
    @pytest.fixture(autouse=True)
    def _neutral_environment(self, monkeypatch, tmp_path):
        # is_ci is the seam the code consults -- mock it instead of coupling
        # to the framework's internal list of CI env vars.
        monkeypatch.setattr("insightscli_commands.telemetry_props.is_ci", lambda: False)
        monkeypatch.setattr("insightscli_commands.telemetry_props._BOX_MARKER", tmp_path / "absent")
        for var in ("INSIGHTSCLI_ENVIRONMENT", *_DEVBOX_ENV_MARKERS):
            monkeypatch.delenv(var, raising=False)

    @pytest.mark.parametrize(
        ("env_vars", "ci", "box", "expected"),
        [
            ({}, False, False, "local"),
            ({}, True, False, "ci"),
            ({"CODER": "true"}, False, False, "devbox"),
            ({"CODER_WORKSPACE_NAME": "raul-devbox"}, False, False, "devbox"),
            ({}, False, True, "box"),
            ({"CODER": "true"}, False, True, "devbox"),
            ({"INSIGHTSCLI_ENVIRONMENT": "sandbox"}, False, False, "sandbox"),
            ({"INSIGHTSCLI_ENVIRONMENT": " Sandbox "}, False, False, "sandbox"),
            ({"INSIGHTSCLI_ENVIRONMENT": "sandbox"}, True, False, "sandbox"),
            ({"CODER": "true"}, True, False, "ci"),
        ],
        ids=[
            "local",
            "ci",
            "coder",
            "coder_workspace",
            "box",
            "devbox_beats_box",
            "declared",
            "declared_normalized",
            "declared_beats_ci",
            "ci_beats_devbox",
        ],
    )
    def test_classification(self, monkeypatch, tmp_path, env_vars, ci, box, expected) -> None:
        if ci:
            monkeypatch.setattr("insightscli_commands.telemetry_props.is_ci", lambda: True)
        if box:
            monkeypatch.setattr("insightscli_commands.telemetry_props._BOX_MARKER", tmp_path)
        for key, value in env_vars.items():
            monkeypatch.setenv(key, value)
        assert _detect_environment() == expected


class TestDetectAgent:
    @pytest.fixture(autouse=True)
    def _no_ambient_agent(self, monkeypatch):
        monkeypatch.delenv("INSIGHTSCLI_AGENT", raising=False)
        for var, _ in _AGENT_ENV_MARKERS:
            monkeypatch.delenv(var, raising=False)

    @pytest.mark.parametrize(
        ("env_vars", "expected"),
        [
            ({}, None),
            ({"CLAUDECODE": "1"}, "claude-code"),
            ({"CODEX_SANDBOX": "seatbelt"}, "codex"),
            ({"INSIGHTS_CODE_VERSION": "1.2.3"}, "insights-code"),
            ({"INSIGHTS_CODE_VERSION": "1.2.3", "CLAUDECODE": "1"}, "insights-code"),
            ({"INSIGHTSCLI_AGENT": " Goose "}, "goose"),
            ({"INSIGHTSCLI_AGENT": "goose", "CLAUDECODE": "1"}, "goose"),
        ],
        ids=[
            "human",
            "claude_code",
            "codex",
            "insights_code",
            "insights_code_beats_claude",
            "declared_normalized",
            "declared_beats_sniffed",
        ],
    )
    def test_detection(self, monkeypatch, env_vars, expected) -> None:
        for key, value in env_vars.items():
            monkeypatch.setenv(key, value)
        assert _detect_agent() == expected


class TestRepoCommitProperties:
    @pytest.fixture(autouse=True)
    def _fresh_cache(self):
        _repo_commit.cache_clear()
        yield
        _repo_commit.cache_clear()

    def test_returns_sha_and_date_in_a_git_checkout(self) -> None:
        props = _repo_commit_properties()
        assert set(props) == {"repo_sha", "repo_commit_date"}
        # %h honors core.abbrev, which git permits down to 4
        assert len(props["repo_sha"]) >= 4
        assert props["repo_commit_date"][:2] == "20"  # ISO date

    def test_returns_empty_when_git_fails(self) -> None:
        with patch("insightscli_commands.telemetry_props.subprocess.run") as mock_run:
            mock_run.return_value.returncode = 128
            assert _repo_commit_properties() == {}


class TestIsInsightsDev:
    @pytest.fixture()
    def _config_dir(self, tmp_path, monkeypatch):
        config_file = tmp_path / "insightscli_telemetry.json"
        monkeypatch.setattr("insightscli.telemetry.get_config_path", lambda: config_file)
        return config_file

    def _write_config(self, path, **kwargs):
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(kwargs))

    def test_fresh_cache_skips_api(self, _config_dir):
        self._write_config(_config_dir, is_insights_org_member=True, org_check_timestamp=time.time())
        assert _is_insights_dev() is True

    @patch("insightscli_commands.telemetry_props._check_github_org_membership", return_value=True)
    def test_cache_miss_calls_gh_and_persists(self, _mock_gh, _config_dir):
        assert _is_insights_dev() is True
        assert json.loads(_config_dir.read_text())["is_insights_org_member"] is True

    @patch("insightscli_commands.telemetry_props._check_email_domain", return_value=True)
    @patch("insightscli_commands.telemetry_props._check_github_org_membership", return_value=None)
    def test_gh_unavailable_falls_back_to_email_and_caches(self, _mock_gh, mock_email, _config_dir):
        assert _is_insights_dev() is True
        mock_email.assert_called_once()
        assert json.loads(_config_dir.read_text())["is_insights_org_member"] is True