File size: 7,002 Bytes
36dada9
 
 
f2a771d
36dada9
 
 
 
 
 
 
 
 
 
 
 
 
 
f2a771d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
36dada9
 
 
 
f2a771d
36dada9
 
 
 
 
 
 
 
 
f2a771d
36dada9
 
 
f2a771d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
a0c38ef
 
 
 
f2a771d
 
 
 
 
 
 
 
 
 
 
 
 
 
a0c38ef
 
f2a771d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
import json

import TerraFin.agent.cli as agent_cli
from TerraFin.agent.definitions import DEFAULT_HOSTED_AGENT_NAME


class _FakeClient:
    def __init__(self, *, transport: str = "auto", base_url: str | None = None, timeout: float = 10.0) -> None:
        self.transport = transport
        self.base_url = base_url
        self.timeout = timeout

    def resolve(self, query: str):
        return {"query": query, "transport": self.transport}

    def market_snapshot(self, name: str, *, depth: str = "auto", view: str = "daily"):
        raise RuntimeError(f"boom:{name}:{depth}:{view}")

    def runtime_agents(self):
        return {"agents": [{"name": DEFAULT_HOSTED_AGENT_NAME}]}

    def runtime_create_session(
        self,
        agent_name: str,
        *,
        session_id: str | None = None,
        system_prompt: str | None = None,
        metadata: dict | None = None,
    ):
        _ = metadata
        return {
            "sessionId": session_id or "runtime:test",
            "agentName": agent_name,
            "systemPrompt": system_prompt,
        }


def test_cli_emits_json_for_success(monkeypatch, capsys) -> None:
    monkeypatch.setattr(agent_cli, "TerraFinAgentClient", _FakeClient)

    exit_code = agent_cli.main(["--transport", "python", "resolve", "AAPL"])

    captured = capsys.readouterr()
    assert exit_code == 0
    assert json.loads(captured.out) == {"query": "AAPL", "transport": "python"}


def test_cli_returns_nonzero_for_errors(monkeypatch, capsys) -> None:
    monkeypatch.setattr(agent_cli, "TerraFinAgentClient", _FakeClient)

    exit_code = agent_cli.main(["snapshot", "AAPL", "--depth", "full", "--view", "weekly"])

    captured = capsys.readouterr()
    assert exit_code == 1
    assert captured.err.strip() == "boom:AAPL:full:weekly"


def test_cli_supports_runtime_create_session(monkeypatch, capsys) -> None:
    monkeypatch.setattr(agent_cli, "TerraFinAgentClient", _FakeClient)

    exit_code = agent_cli.main(
        [
            "runtime-create-session",
            DEFAULT_HOSTED_AGENT_NAME,
            "--session-id",
            "runtime:cli",
        ]
    )

    captured = capsys.readouterr()
    assert exit_code == 0
    assert json.loads(captured.out) == {
        "sessionId": "runtime:cli",
        "agentName": DEFAULT_HOSTED_AGENT_NAME,
        "systemPrompt": None,
    }


def test_cli_models_list_all_reports_featured_models(monkeypatch, tmp_path) -> None:
    monkeypatch.setenv("TERRAFIN_AGENT_MODELS_PATH", str(tmp_path / "agent-models.json"))
    # Block .env autoload + drop the var so a developer's local
    # TERRAFIN_OPENAI_MODEL doesn't leak into the registry-default assertion.
    monkeypatch.setenv("TERRAFIN_DISABLE_DOTENV", "1")
    monkeypatch.delenv("TERRAFIN_OPENAI_MODEL", raising=False)

    payload = agent_cli._models_list_payload(include_models=True)
    assert payload["current"]["modelRef"] == "openai/gpt-4.1-mini"
    assert any(provider["providerId"] == "github-copilot" for provider in payload["providers"])
    assert any(
        model["modelRef"] == "github-copilot/gpt-4o"
        for provider in payload["providers"]
        if provider["providerId"] == "github-copilot"
        for model in provider["models"]
    )


def test_cli_models_list_all_human_output_is_table(monkeypatch, capsys, tmp_path) -> None:
    monkeypatch.setenv("TERRAFIN_AGENT_MODELS_PATH", str(tmp_path / "agent-models.json"))
    monkeypatch.setenv("TERRAFIN_DISABLE_DOTENV", "1")
    monkeypatch.delenv("TERRAFIN_OPENAI_MODEL", raising=False)

    exit_code = agent_cli.main(["models", "list", "--all"])

    captured = capsys.readouterr()
    assert exit_code == 0
    assert captured.out.startswith("Current: openai/gpt-4.1-mini")
    assert "Model" in captured.out
    assert "Auth" in captured.out
    assert "Local" in captured.out
    assert "openai/gpt-4.1-mini" in captured.out
    assert "github-copilot/gpt-4o" in captured.out
    assert not captured.out.lstrip().startswith("{")


def test_cli_models_use_persists_default_model(monkeypatch, capsys, tmp_path) -> None:
    monkeypatch.setenv("TERRAFIN_AGENT_MODELS_PATH", str(tmp_path / "agent-models.json"))

    exit_code = agent_cli.main(["models", "use", "google/gemini-3.1-pro-preview"])

    captured = capsys.readouterr()
    saved = json.loads((tmp_path / "agent-models.json").read_text(encoding="utf-8"))
    assert exit_code == 0
    assert "Saved default model: google/gemini-3.1-pro-preview" in captured.out
    assert saved["defaultModelRef"] == "google/gemini-3.1-pro-preview"


def test_cli_models_auth_login_github_copilot_saves_token_and_default(monkeypatch, capsys, tmp_path) -> None:
    monkeypatch.setenv("TERRAFIN_AGENT_MODELS_PATH", str(tmp_path / "agent-models.json"))

    exit_code = agent_cli.main(
        [
            "models",
            "auth",
            "login-github-copilot",
            "--token",
            "ghu_saved_token",
            "--set-default",
            "--yes",
        ]
    )

    captured = capsys.readouterr()
    saved = json.loads((tmp_path / "agent-models.json").read_text(encoding="utf-8"))
    assert exit_code == 0
    assert "Saved GitHub Copilot credentials (token)." in captured.out
    assert "Default model: github-copilot/gpt-4o" in captured.out
    assert saved["auth"]["github-copilot"]["authMode"] == "token"
    assert saved["auth"]["github-copilot"]["githubToken"] == "ghu_saved_token"
    assert saved["defaultModelRef"] == "github-copilot/gpt-4o"


def test_cli_models_auth_login_github_copilot_device_flow(monkeypatch, capsys, tmp_path) -> None:
    monkeypatch.setenv("TERRAFIN_AGENT_MODELS_PATH", str(tmp_path / "agent-models.json"))
    monkeypatch.setattr(agent_cli, "request_github_copilot_device_code", lambda: type(
        "DeviceCode",
        (),
        {
            "device_code": "device-code-123",
            "user_code": "ABCD-EFGH",
            "interval_seconds": 5,
            "expires_in_seconds": 900,
            "authorization_url": "https://github.com/login/device?user_code=ABCD-EFGH",
        },
    )())
    monkeypatch.setattr(
        agent_cli,
        "poll_github_copilot_device_access_token",
        lambda *, device_code, interval_seconds, expires_in_seconds: "gho_device_token",
    )
    monkeypatch.setattr(agent_cli.sys.stdin, "isatty", lambda: True)

    exit_code = agent_cli.main(
        [
            "models",
            "auth",
            "login-github-copilot",
            "--set-default",
            "--yes",
        ]
    )

    captured = capsys.readouterr()
    saved = json.loads((tmp_path / "agent-models.json").read_text(encoding="utf-8"))
    assert exit_code == 0
    assert "Saved GitHub Copilot credentials (device)." in captured.out
    assert "Default model: github-copilot/gpt-4o" in captured.out
    assert "Authorize GitHub Copilot" in captured.err
    assert saved["auth"]["github-copilot"]["authMode"] == "device"
    assert saved["auth"]["github-copilot"]["githubToken"] == "gho_device_token"