mirror of
https://github.com/superdesigndev/treg.git
synced 2026-10-02 03:24:35 +08:00
Jason ZhouandClaude Opus 5.5
2de39612db
* fix(call): replays report zero cost; stored answers readable by call id
An idempotent replay echoed the first call's charge in X-Treg-Cost-Micro,
so a client summing that header counted one call twice (our own MCP tool
said "nothing was charged" beside a non-zero cost_usd). A replay now sends
X-Treg-Cost-Micro: 0 and the original charge in X-Treg-Original-Cost-Micro;
the CLI charge line reads the new header and falls back for older servers.
GET /calls/{id}/result took only the numeric audit row id, which a caller
holding the X-Treg-Call-Id it was given could not use without a second
lookup. It now accepts either. The endpoint was also absent from every
agent-facing doc, so a team that paid for answers its own code dropped had
no way to learn they could fetch them back; llms.txt, skill.md and
integrate.md now say so. Fragment updated: interface/api.md.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
* docs(context): replay cost header and call-id result lookup in money + archive
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---------
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1101 lines
51 KiB
Python
1101 lines
51 KiB
Python
"""The CLI is a thin client — unit-test parsing + the identity-config round-trip (no network)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import stat
|
|
|
|
import pytest
|
|
|
|
from treg import cli
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolate_cli_config(tmp_path, monkeypatch):
|
|
"""Never let a CLI test touch the real ~/.treg/config.json. Some commands persist config
|
|
(e.g. _clear_active_if_targeted -> _save_config), so an un-isolated test would wipe the
|
|
developer's own login mid-suite. Redirect CONFIG_PATH to a tmp file for every test here."""
|
|
monkeypatch.setattr(cli, "CONFIG_PATH", tmp_path / "config.json")
|
|
|
|
|
|
def test_resources_list_uses_the_unified_server_endpoint(monkeypatch, capsys):
|
|
calls = []
|
|
|
|
class Response:
|
|
status_code = 200
|
|
headers = {}
|
|
text = ""
|
|
|
|
def __init__(self, body): self.body = body
|
|
def json(self): return self.body
|
|
|
|
class Client:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): return False
|
|
def get(self, path, params=None):
|
|
calls.append((path, params))
|
|
if path == "/orgs":
|
|
return Response([{"org_id": 7, "slug": "team", "active": True}])
|
|
return Response([{"provider": "fishaudio", "kind": "voice",
|
|
"upstream_id": "voice-1", "display_name": "Narrator"}])
|
|
|
|
monkeypatch.setattr(cli, "_client", lambda cfg: Client())
|
|
args = type("Args", (), {
|
|
"provider": "fishaudio", "kind": "voice", "include_deleted": False,
|
|
})()
|
|
cli.cmd_resources_list(args, {"base_url": "http://registry", "active_org": "team"})
|
|
assert calls == [
|
|
("/orgs", None),
|
|
("/orgs/7/provider-resources", {"provider": "fishaudio", "kind": "voice"}),
|
|
]
|
|
assert '"upstream_id": "voice-1"' in capsys.readouterr().out
|
|
|
|
|
|
def test_call_named_and_single_url():
|
|
p = cli.build_parser()
|
|
a = p.parse_args(["call", "echo", "v1/x", "--method", "POST"])
|
|
assert a.target == "echo" and a.path == "v1/x" and a.method == "POST"
|
|
b = p.parse_args(["call", "https://api.intercom.io/me"])
|
|
assert b.target == "https://api.intercom.io/me" and b.path == ""
|
|
|
|
|
|
def test_host_subcommand_is_registered_not_mistaken_for_system_binary():
|
|
"""Issue #557: `treg host` must parse as the host subcommand, not fall through to /usr/bin/host.
|
|
When `host` was missing from an older release, `_looks_like_a_program` matched it to the system
|
|
`host` binary (DNS lookup tool), causing `treg host face.jpg` to run `/usr/bin/host face.jpg`
|
|
via `treg with` — banner, then a confusing DNS error. The fix: keep `host` as a registered
|
|
subcommand so the bare-word fallback never applies."""
|
|
p = cli.build_parser()
|
|
subcommands = cli._subcommands(p)
|
|
assert "host" in subcommands, "`host` must be a registered subcommand"
|
|
assert p.parse_args(["host", "face.jpg"]).fn is cli.cmd_host
|
|
# The bare-word fallback (`treg claude` → `treg with claude`) must NOT treat `host` as a program
|
|
# even though /usr/bin/host exists on most Unix systems.
|
|
assert cli._looks_like_a_program(["host", "face.jpg"], subcommands) is False
|
|
|
|
|
|
def test_legacy_multiorg_config_migrates(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(cli, "CONFIG_PATH", tmp_path / "config.json")
|
|
(tmp_path / "config.json").write_text(json.dumps({
|
|
"base_url": "https://treg.to", "active_org": "team-a",
|
|
"orgs": {"team-a": {"token": "OLD", "org_id": 3}}}))
|
|
cfg = cli._load_config()
|
|
assert cfg["token"] == "OLD" and cfg["active_org"] == "team-a" and cfg["identity"] is False
|
|
|
|
|
|
def test_load_config_defaults_to_production_not_localhost(tmp_path, monkeypatch):
|
|
"""A fresh CLI (no ~/.treg/config.json) must default to https://treg.to, NOT localhost.
|
|
The first-run bug was: install.sh's `treg config --base-url` failed silently, so login
|
|
opened Chrome to http://localhost:18790/login — nothing listening, connection refused."""
|
|
monkeypatch.setattr(cli, "CONFIG_PATH", tmp_path / "missing" / "config.json")
|
|
cfg = cli._load_config()
|
|
assert cfg["base_url"] == cli.PRODUCTION_BASE_URL
|
|
assert "localhost" not in cfg["base_url"]
|
|
assert "127.0.0.1" not in cfg["base_url"]
|
|
|
|
|
|
def test_is_loopback_url_detects_localhost_variants():
|
|
"""_is_loopback_url recognises localhost, 127.x.x.x, and [::1] so the login flow can warn
|
|
when pointed at a local server that isn't running."""
|
|
assert cli._is_loopback_url("http://localhost:18790") is True
|
|
assert cli._is_loopback_url("http://127.0.0.1:8000") is True
|
|
assert cli._is_loopback_url("http://127.0.0.42/path") is True
|
|
assert cli._is_loopback_url("https://[::1]/") is True # IPv6 needs brackets in URLs
|
|
assert cli._is_loopback_url("https://[::1]:8000/") is True
|
|
assert cli._is_loopback_url("https://treg.to") is False
|
|
assert cli._is_loopback_url("https://api.example.com:443/v1") is False
|
|
assert cli._is_loopback_url("") is False # empty URL shouldn't crash
|
|
|
|
|
|
def test_login_exits_with_helpful_message_when_localhost_unreachable(monkeypatch):
|
|
"""When base_url is localhost and the server can't be reached, login must fail early with
|
|
a message that tells the user how to fix it (point at production), not silently open a
|
|
dead browser page."""
|
|
import httpx
|
|
|
|
def fake_post(url, **kw):
|
|
raise httpx.ConnectError("Connection refused")
|
|
|
|
monkeypatch.setattr(httpx, "post", fake_post)
|
|
args = type("A", (), {"token": None, "email": None})()
|
|
cfg = {"base_url": "http://localhost:18790", "token": None}
|
|
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
cli.cmd_login(args, cfg)
|
|
|
|
error_msg = str(exc_info.value)
|
|
assert "localhost" in error_msg.lower() or "Cannot reach" in error_msg
|
|
assert "treg config --base-url" in error_msg
|
|
assert cli.PRODUCTION_BASE_URL in error_msg
|
|
|
|
|
|
def test_token_org_claim_reads_a_team_pinned_token():
|
|
"""`_token_org_claim` decodes the org slug a team-pinned identity token carries — without the
|
|
signature check (the server still authorizes), and returning None on anything else."""
|
|
import base64
|
|
payload = base64.urlsafe_b64encode(
|
|
json.dumps({"uid": 7, "exp": 9999999999, "tv": 0, "org": "team-b"}).encode()
|
|
).decode().rstrip("=")
|
|
assert cli._token_org_claim(f"{payload}.not-a-real-signature") == "team-b"
|
|
assert cli._token_org_claim("tok-opaque-per-org") is None # opaque membership token
|
|
assert cli._token_org_claim(None) is None
|
|
assert cli._token_org_claim("") is None
|
|
|
|
|
|
def test_login_token_keeps_a_typed_default_as_a_human_identity(monkeypatch):
|
|
import base64
|
|
|
|
payload = base64.urlsafe_b64encode(json.dumps({"scp": "team"}).encode()).decode().rstrip("=")
|
|
|
|
class Response:
|
|
status_code = 200
|
|
def json(self): return {"email": "me@example.com"}
|
|
|
|
class Client:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): return False
|
|
def get(self, path): return Response()
|
|
|
|
monkeypatch.setattr(cli, "_client", lambda cfg: Client())
|
|
monkeypatch.setattr(cli, "_pick_active_org", lambda cfg: None)
|
|
cfg = {"base_url": "http://x"}
|
|
cli.cmd_login(type("A", (), {"token": f"{payload}.sig", "email": None})(), cfg)
|
|
assert cfg["identity"] is True
|
|
|
|
|
|
def test_pick_active_org_prefers_the_tokens_baked_org(monkeypatch):
|
|
"""Against an older server that marks nothing active for a team-pinned token, `_pick_active_org`
|
|
must land on the token's own org — not the first membership, which for a multi-team user is an
|
|
arbitrary other team (`treg login --token <superdesign key>` used to activate `oauth-test`)."""
|
|
import base64
|
|
|
|
class _Resp:
|
|
status_code = 200
|
|
def json(self):
|
|
return [{"slug": "first-team", "active": False}, {"slug": "second-team", "active": False}]
|
|
|
|
class _Client:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *a): return False
|
|
def get(self, path): return _Resp()
|
|
|
|
monkeypatch.setattr(cli, "_client", lambda cfg: _Client())
|
|
payload = base64.urlsafe_b64encode(
|
|
json.dumps({"uid": 7, "exp": 9999999999, "tv": 0, "org": "second-team"}).encode()
|
|
).decode().rstrip("=")
|
|
cfg = {"base_url": "http://x", "token": f"{payload}.sig", "active_org": None}
|
|
cli._pick_active_org(cfg)
|
|
assert cfg["active_org"] == "second-team"
|
|
# an opaque token (no claim) still falls back to the first membership
|
|
cfg = {"base_url": "http://x", "token": "tok-opaque", "active_org": None}
|
|
cli._pick_active_org(cfg)
|
|
assert cfg["active_org"] == "first-team"
|
|
|
|
|
|
def test_client_sends_token_and_active_org(monkeypatch):
|
|
monkeypatch.setattr(cli, "_ORG_OVERRIDE", None)
|
|
cfg = {"base_url": "http://x", "token": "TK", "active_org": "team-a"}
|
|
with cli._client(cfg) as c:
|
|
assert c.headers["X-Treg-Token"] == "TK" and c.headers["X-Treg-Org"] == "team-a"
|
|
|
|
|
|
def test_org_override_beats_active(monkeypatch):
|
|
monkeypatch.setattr(cli, "_ORG_OVERRIDE", "team-b")
|
|
cfg = {"base_url": "http://x", "token": "TK", "active_org": "team-a"}
|
|
assert cli._effective_org(cfg) == "team-b"
|
|
with cli._client(cfg) as c:
|
|
assert c.headers["X-Treg-Org"] == "team-b"
|
|
|
|
|
|
def test_org_use_saves_team_and_active_default_together(monkeypatch):
|
|
class Response:
|
|
status_code = 200
|
|
|
|
def __init__(self, body):
|
|
self.body = body
|
|
|
|
def json(self):
|
|
return self.body
|
|
|
|
class Client:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): return False
|
|
def get(self, path, **kwargs):
|
|
if path == "/orgs":
|
|
return Response([
|
|
{"slug": "team-one", "active": True},
|
|
{"slug": "team-two", "active": False},
|
|
])
|
|
assert path == "/auth/cli-token"
|
|
assert kwargs["headers"] == {"X-Treg-Org": "team-two"}
|
|
return Response({
|
|
"org": "team-two", "token": "team-two-default",
|
|
"default_key_state": "active",
|
|
})
|
|
|
|
monkeypatch.setattr(cli, "_client", lambda cfg, auth=True: Client())
|
|
cfg = {"base_url": "http://x", "token": "team-one-default",
|
|
"active_org": "team-one", "identity": True}
|
|
cli.cmd_org_use(type("A", (), {"slug": "team-two"})(), cfg)
|
|
assert cfg["active_org"] == "team-two" and cfg["token"] == "team-two-default"
|
|
saved = cli._load_config()
|
|
assert saved["active_org"] == "team-two" and saved["token"] == "team-two-default"
|
|
|
|
|
|
def test_org_use_keeps_previous_team_when_default_is_disabled(monkeypatch):
|
|
class Response:
|
|
status_code = 200
|
|
|
|
def __init__(self, body):
|
|
self.body = body
|
|
|
|
def json(self):
|
|
return self.body
|
|
|
|
class Client:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): return False
|
|
def get(self, path, **kwargs):
|
|
if path == "/orgs":
|
|
return Response([
|
|
{"slug": "team-one", "active": True},
|
|
{"slug": "team-two", "active": False},
|
|
])
|
|
return Response({
|
|
"org": "team-two", "token": "disabled-default",
|
|
"default_key_state": "disabled",
|
|
})
|
|
|
|
monkeypatch.setattr(cli, "_client", lambda cfg, auth=True: Client())
|
|
cfg = {"base_url": "http://x", "token": "team-one-default",
|
|
"active_org": "team-one", "identity": True}
|
|
cli._save_config(cfg)
|
|
before = dict(cfg)
|
|
with pytest.raises(SystemExit, match="Default key is disabled.*Active team unchanged"):
|
|
cli.cmd_org_use(type("A", (), {"slug": "team-two"})(), cfg)
|
|
assert cfg == before
|
|
saved = cli._load_config()
|
|
assert saved["active_org"] == before["active_org"]
|
|
assert saved["token"] == before["token"]
|
|
|
|
|
|
def test_org_use_does_not_move_an_opaque_key_to_another_team(monkeypatch):
|
|
class Response:
|
|
status_code = 200
|
|
|
|
def json(self):
|
|
return [{"slug": "team-one"}, {"slug": "team-two"}]
|
|
|
|
class Client:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): return False
|
|
def get(self, path, **kwargs): return Response()
|
|
|
|
monkeypatch.setattr(cli, "_client", lambda cfg, auth=True: Client())
|
|
cfg = {"base_url": "http://x", "token": "opaque-agent-key",
|
|
"active_org": "team-one", "identity": False}
|
|
cli._save_config(cfg)
|
|
with pytest.raises(SystemExit, match="cannot switch teams.*Active team unchanged"):
|
|
cli.cmd_org_use(type("A", (), {"slug": "team-two"})(), cfg)
|
|
saved = cli._load_config()
|
|
assert saved["active_org"] == "team-one"
|
|
assert saved["token"] == "opaque-agent-key"
|
|
|
|
|
|
def test_pop_org_flag():
|
|
a = ["tool", "ls", "--org", "team-b"]
|
|
assert cli._pop_org_flag(a) == "team-b" and a == ["tool", "ls"]
|
|
b = ["tool", "ls", "--org=team-c"]
|
|
assert cli._pop_org_flag(b) == "team-c" and b == ["tool", "ls"]
|
|
assert cli._pop_org_flag(["x"]) is None
|
|
|
|
|
|
def test_admin_client_prefers_admin_token():
|
|
cfg = {"base_url": "http://x", "admin_token": "ENV", "token": "USER"}
|
|
with cli._admin_client(cfg) as c:
|
|
assert c.headers["X-Treg-Token"] == "ENV"
|
|
del cfg["admin_token"]
|
|
with cli._admin_client(cfg) as c:
|
|
assert c.headers["X-Treg-Token"] == "USER"
|
|
|
|
|
|
def test_org_delete_requires_matching_slug():
|
|
cfg = {"base_url": "http://x", "token": "T", "active_org": "team-a"}
|
|
args = type("A", (), {"slug": "wrong"})()
|
|
with pytest.raises(SystemExit):
|
|
cli.cmd_org_delete(args, cfg)
|
|
|
|
|
|
# ---- bug-hunt regressions -----------------------------------------------------------------
|
|
def test_corrupt_config_does_not_brick_the_cli(tmp_path, monkeypatch):
|
|
"""A half-written / hand-broken config must load as empty, not JSONDecodeError on every run."""
|
|
monkeypatch.setattr(cli, "CONFIG_PATH", tmp_path / "config.json")
|
|
(tmp_path / "config.json").write_text("{ this is not valid json")
|
|
cfg = cli._load_config() # must not raise
|
|
assert cfg["token"] is None and cfg["base_url"].startswith("http")
|
|
|
|
|
|
def test_save_config_is_atomic(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(cli, "CONFIG_PATH", tmp_path / "config.json")
|
|
cli._save_config({"base_url": "http://x", "token": "T"})
|
|
assert not (tmp_path / "config.json.tmp").exists() # temp renamed away, no litter
|
|
assert cli._load_config()["token"] == "T"
|
|
assert stat.S_IMODE((tmp_path / "config.json").stat().st_mode) == 0o600
|
|
|
|
|
|
def test_bootstrap_token_is_never_installed_into_mcp():
|
|
import base64
|
|
payload = base64.urlsafe_b64encode(json.dumps({"scp": "bootstrap"}).encode()).decode().rstrip("=")
|
|
with pytest.raises(SystemExit, match="temporary login token"):
|
|
cli.cmd_mcp_install(object(), {"token": f"{payload}.sig"})
|
|
|
|
|
|
def test_team_create_and_join_replace_the_saved_bootstrap(monkeypatch):
|
|
class Response:
|
|
status_code = 200
|
|
text = "{}"
|
|
def __init__(self, body): self.body = body
|
|
def json(self): return self.body
|
|
|
|
class Client:
|
|
def __init__(self, response): self.response = response
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): return False
|
|
def post(self, path, json): return self.response
|
|
|
|
monkeypatch.setattr(cli, "_show", lambda response: None)
|
|
created = Response({"token": "team-default", "org": "acme"})
|
|
monkeypatch.setattr(cli, "_client", lambda cfg: Client(created))
|
|
cfg = {"token": "bootstrap", "identity": True}
|
|
cli.cmd_org_create(type("A", (), {"name": "Acme"})(), cfg)
|
|
assert cfg == {"token": "team-default", "identity": True, "active_org": "acme"}
|
|
|
|
joined = Response({"token": "joined-default", "org": "joined"})
|
|
monkeypatch.setattr(cli, "_client", lambda cfg, auth=False: Client(joined))
|
|
cli.cmd_org_join(type("A", (), {"code": "inv", "email": "me@example.com"})(), cfg)
|
|
assert cfg["token"] == "joined-default" and cfg["active_org"] == "joined"
|
|
|
|
|
|
class _FakeResp:
|
|
status_code = 200
|
|
def json(self): return {}
|
|
text = "{}"
|
|
|
|
|
|
class _FakeClient:
|
|
def __init__(self): self.calls = []
|
|
def __enter__(self): return self
|
|
def __exit__(self, *a): return False
|
|
def request(self, method, url, params=None, content=None, headers=None):
|
|
self.calls.append((method, url, params, content, headers))
|
|
return _FakeResp()
|
|
|
|
|
|
def test_call_preserves_duplicate_query_keys(monkeypatch):
|
|
fake = _FakeClient()
|
|
monkeypatch.setattr(cli, "_client", lambda cfg: fake)
|
|
monkeypatch.setattr(cli, "_show", lambda r: None)
|
|
args = cli.build_parser().parse_args(["call", "echo", "--query", "tag=a", "--query", "tag=b"])
|
|
cli.cmd_call(args, {"base_url": "http://x"})
|
|
_, _, params, _, _ = fake.calls[0]
|
|
assert params == [("tag", "a"), ("tag", "b")] # both survive; a dict would drop tag=a
|
|
|
|
|
|
def test_call_sends_explicit_authorization_method_as_treg_control_header(monkeypatch):
|
|
fake = _FakeClient()
|
|
monkeypatch.setattr(cli, "_client", lambda cfg: fake)
|
|
monkeypatch.setattr(cli, "_show", lambda r: None)
|
|
args = cli.build_parser().parse_args([
|
|
"call", "future-provider.endpoint",
|
|
"--authorization-method", "delegated-admin",
|
|
"--query", "page_id=PAGE-1",
|
|
])
|
|
cli.cmd_call(args, {"base_url": "http://x"})
|
|
assert fake.calls[0][4]["X-Treg-Authorization-Method"] == "delegated-admin"
|
|
|
|
|
|
def _args(**kw):
|
|
return type("A", (), kw)()
|
|
|
|
|
|
# Bad local input is a clean exit with a message, never a traceback.
|
|
@pytest.mark.parametrize("run", [
|
|
pytest.param(lambda: cli.cmd_call(cli.build_parser().parse_args(["call", "echo", "--query", "flag"]),
|
|
{"base_url": "http://x"}), id="call-query-without-equals"),
|
|
pytest.param(lambda: cli._parse_bind("secret=abc"), id="bind-non-int-secret"),
|
|
pytest.param(lambda: cli._load_json_arg("{bad", "binding"), id="bad-json-arg"),
|
|
pytest.param(lambda: cli._pop_org_flag(["tool", "ls", "--org"]), id="org-flag-missing-value"),
|
|
pytest.param(lambda: cli.cmd_oauth_connect(
|
|
_args(client_secret="/nonexistent/x.json", name="g", scopes=[], provider=None, capability=None),
|
|
{"base_url": "http://x"}), id="oauth-connect-missing-file"),
|
|
# Registry mode needs --provider; BYO needs --client-secret. Neither is a usage error.
|
|
pytest.param(lambda: cli.cmd_oauth_connect(
|
|
_args(client_secret=None, name=None, scopes=[], provider=None, capability=None),
|
|
{"base_url": "http://x"}), id="oauth-connect-without-provider-or-client-secret"),
|
|
pytest.param(lambda: cli.cmd_skill_push(_args(file="/nonexistent/skill.json"), {"base_url": "http://x"}),
|
|
id="skill-push-missing-file"),
|
|
])
|
|
def test_bad_local_input_exits_cleanly(monkeypatch, run):
|
|
monkeypatch.setattr(cli, "_client", lambda cfg: _FakeClient())
|
|
with pytest.raises(SystemExit):
|
|
run()
|
|
|
|
|
|
def test_oauth_connect_prints_provider_guidance_from_the_api(monkeypatch, capsys):
|
|
class Response:
|
|
status_code = 200
|
|
|
|
def json(self):
|
|
return {
|
|
"state": "STATE",
|
|
"redirect_uri": "https://registry.example/oauth/callback",
|
|
"consent_url": "https://provider.example/authorize",
|
|
"connect_guidance": "Use the linked workspace administrator grant.",
|
|
}
|
|
|
|
class StatusResponse:
|
|
def json(self):
|
|
return {"status": "done", "secret_id": 7, "name": "future-provider"}
|
|
|
|
class Client:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): return False
|
|
def post(self, path, json): return Response()
|
|
def get(self, path): return StatusResponse()
|
|
|
|
monkeypatch.setattr(cli, "_client", lambda cfg: Client())
|
|
monkeypatch.setattr(cli.time, "sleep", lambda seconds: None)
|
|
args = cli.build_parser().parse_args([
|
|
"connections", "connect", "--provider", "future-provider",
|
|
])
|
|
cli.cmd_oauth_connect(args, {"base_url": "http://x"})
|
|
output = capsys.readouterr().out
|
|
assert "Use the linked workspace administrator grant." in output
|
|
|
|
|
|
def test_clear_active_only_when_targeted(monkeypatch):
|
|
# a one-shot --org override on a DIFFERENT org must not wipe the stored active org
|
|
monkeypatch.setattr(cli, "_ORG_OVERRIDE", "beta")
|
|
cfg = {"active_org": "alpha"}
|
|
cli._clear_active_if_targeted(cfg)
|
|
assert cfg["active_org"] == "alpha" # untouched
|
|
# acting on the stored active org clears it
|
|
monkeypatch.setattr(cli, "_ORG_OVERRIDE", None)
|
|
cli._clear_active_if_targeted(cfg)
|
|
assert cfg["active_org"] is None
|
|
|
|
|
|
def test_find_env_upwards_locates_project_env(tmp_path):
|
|
"""A skills dir sits UNDER a project whose .env is at the root — the walk-up must find it so
|
|
env-credentialed skills (render/vercel) aren't gapped 'needs env var … not found'."""
|
|
(tmp_path / ".env").write_text("RENDER_API_KEY=x\n")
|
|
skills = tmp_path / ".claude" / "skills"
|
|
skills.mkdir(parents=True)
|
|
found = cli._find_env_upwards(str(skills))
|
|
assert found == str(tmp_path / ".env")
|
|
# a path nested deeper under the project still resolves to the same root .env
|
|
assert cli._find_env_upwards(str(skills / "render")) == str(tmp_path / ".env")
|
|
|
|
|
|
def test_secret_add_env_var_parses_and_strips_quotes(tmp_path, monkeypatch):
|
|
"""`secret add --env-var` reads ONE var from an .env via treg's parser — a quoted value
|
|
(AGENTMAIL_API_KEY="am_…") is stored WITHOUT the quotes (the bug agents hit hand-extracting)."""
|
|
(tmp_path / ".env").write_text('AGENTMAIL_API_KEY="am_us_pod_QUOTED"\nOTHER=nope\n')
|
|
posted = {}
|
|
|
|
class _FakeResp:
|
|
status_code = 200
|
|
def json(self): return {"id": 1}
|
|
|
|
class _FakeClient:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *a): return False
|
|
def post(self, path, json):
|
|
posted.update(json)
|
|
return _FakeResp()
|
|
|
|
monkeypatch.setattr(cli, "_client", lambda cfg: _FakeClient())
|
|
args = type("A", (), {"name": "agentmail-key", "env_var": "AGENTMAIL_API_KEY",
|
|
"env_file": str(tmp_path / ".env"), "dir": None, "file": None,
|
|
"value": None, "kind": "env"})()
|
|
cli.cmd_secret_add(args, {"base_url": "http://x", "token": "t"})
|
|
assert posted["value"] == "am_us_pod_QUOTED" # no surrounding quotes
|
|
assert posted["name"] == "agentmail-key" and posted["kind"] == "env"
|
|
|
|
|
|
def test_onboard_setup_import_args_are_complete():
|
|
"""`_run_setup` builds upload args from build_parser() so it can't drift out of sync with new
|
|
upload flags (regression: a hand-built Namespace missing `no_oauth` crashed onboarding Set up)."""
|
|
a = cli.build_parser().parse_args(["upload"])
|
|
for attr in ("no_oauth", "llm", "llm_token", "llm_model", "llm_base_url",
|
|
"dry_run", "all", "select", "replace", "env_file", "skills_dir", "mode"):
|
|
assert hasattr(a, attr), f"upload args missing {attr}"
|
|
|
|
|
|
def test_scan_upload_import_verbs():
|
|
"""`treg scan` is the read-only preview (forced dry_run, no prompts); `treg upload` is the real
|
|
thing; `treg import` stays a working alias of upload (old docs/scripts must not break)."""
|
|
p = cli.build_parser()
|
|
s = p.parse_args(["scan"])
|
|
assert s.dry_run and s.as_scan and s.all and s.no_oauth and s.fn is cli.cmd_import
|
|
u = p.parse_args(["upload"])
|
|
assert not u.dry_run and not u.as_scan and u.fn is cli.cmd_import
|
|
i = p.parse_args(["import", "--dry-run"]) # alias keeps full flag surface
|
|
assert i.dry_run and not i.as_scan and i.cmd == "import" and i.fn is cli.cmd_import
|
|
|
|
|
|
def test_onboard_setup_source_picks_scan_dirs(tmp_path, monkeypatch):
|
|
"""`treg onboard --path setup --source global` scans the machine-wide agent skill folders
|
|
(~/.claude/skills, …); `--source local` keeps the project-only scan; `--source both` unions them."""
|
|
gdir = tmp_path / "home" / ".claude" / "skills"
|
|
(gdir / "globskill").mkdir(parents=True)
|
|
(gdir / "globskill" / "SKILL.md").write_text("---\nname: globskill\n---\nhi")
|
|
proj = tmp_path / "proj"
|
|
local = proj / ".claude" / "skills" / "localskill"
|
|
local.mkdir(parents=True)
|
|
(local / "SKILL.md").write_text("---\nname: localskill\n---\nhi")
|
|
monkeypatch.chdir(proj)
|
|
|
|
from treg import agents as ag
|
|
monkeypatch.setattr(ag, "detect_installed", lambda: ["claude-code"])
|
|
monkeypatch.setattr(ag, "global_dir", lambda a: gdir)
|
|
monkeypatch.setattr(cli, "_onboard_active_org", lambda cfg: {"name": "T", "slug": "t", "role": "admin"})
|
|
scanned: list[list[str]] = []
|
|
monkeypatch.setattr(cli, "_import_skills", lambda args, cfg, dirs, env: scanned.append([str(d) for d in dirs]))
|
|
monkeypatch.setattr(cli, "_client", lambda cfg: (_ for _ in ()).throw(RuntimeError("no server")))
|
|
|
|
def run(source):
|
|
scanned.clear()
|
|
args = cli.build_parser().parse_args(["onboard", "--path", "setup", "--source", source])
|
|
cli._run_setup({"base_url": "http://x", "token": "t"}, args)
|
|
return scanned[0] if scanned else []
|
|
|
|
assert run("global") == [str(gdir)]
|
|
got_local = run("local")
|
|
assert str(gdir) not in got_local and any(d.endswith("skills") for d in got_local)
|
|
got_both = run("both")
|
|
assert str(gdir) in got_both and got_both != [str(gdir)]
|
|
|
|
|
|
def test_only_resolvable_gaps():
|
|
def mk(gaps):
|
|
return type("D", (), {"gaps": gaps})()
|
|
|
|
assert cli._only_resolvable_gaps(mk([])) # no gaps → checkable
|
|
assert cli._only_resolvable_gaps(mk(["needs env var STRIPE_KEY — not found in the env"])) # env-var → fixable
|
|
assert not cli._only_resolvable_gaps(mk(["treg.json secret file missing: token.json"])) # file gap → not
|
|
|
|
|
|
def test_prompt_missing_skill_creds_fills_values_and_clears_gaps(monkeypatch):
|
|
"""A skill needing an env var absent from .env prompts for it; on a value, the var lands in
|
|
`values` and the gap clears so the skill registers. Blank input leaves it skipped."""
|
|
answers = iter(["dev-token-123", ""]) # first var answered, second left blank
|
|
monkeypatch.setattr(cli.getpass, "getpass", lambda prompt="": next(answers))
|
|
good = type("D", (), {"gaps": ["needs env var GOOGLE_ADS_DEVELOPER_TOKEN — not found in the env"]})()
|
|
skip = type("D", (), {"gaps": ["needs env var INTERCOM_TOKEN — not found in the env"]})()
|
|
values = {}
|
|
cli._prompt_missing_skill_creds([good, skip], values)
|
|
assert values == {"GOOGLE_ADS_DEVELOPER_TOKEN": "dev-token-123"} # only the answered one
|
|
assert good.gaps == [] # resolved → will register
|
|
assert skip.gaps and "INTERCOM_TOKEN" in skip.gaps[0] # blank → still gapped, still skipped
|
|
|
|
|
|
def test_load_catalog_prefers_newer_bundled_over_older_server(tmp_path, monkeypatch):
|
|
"""A CLI updated ahead of its server must NOT regress to the server's older catalog (which lacks new
|
|
CLIs + auth_mechanism/detect). _load_catalog uses whichever is newer by CATALOG_VERSION."""
|
|
from treg import providers as prov
|
|
monkeypatch.setattr(cli, "CONFIG_PATH", tmp_path / "config.json")
|
|
|
|
class _Resp:
|
|
status_code = 200
|
|
text = '{"version": 1, "providers": [{"provider": "OldOnly"}]}'
|
|
def json(self): return {"version": 1, "providers": [{"provider": "OldOnly"}]}
|
|
|
|
class _C:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *a): return False
|
|
def get(self, path): return _Resp()
|
|
|
|
monkeypatch.setattr(cli, "_client", lambda cfg, auth=False: _C())
|
|
cat = cli._load_catalog({"base_url": "http://old-server"})
|
|
assert cat is prov.CATALOG # bundled (v8) wins over the server's v1
|
|
|
|
# …but a server that's newer/equal wins (it can grow without a CLI release)
|
|
class _RespNew(_Resp):
|
|
text = '{"version": 999, "providers": [{"provider": "NewServer"}]}'
|
|
def json(self): return {"version": 999, "providers": [{"provider": "NewServer"}]}
|
|
monkeypatch.setattr(cli, "_client", lambda cfg, auth=False: type("X", (_C,), {"get": lambda s, p: _RespNew()})())
|
|
cat2 = cli._load_catalog({"base_url": "http://new-server"})
|
|
assert cat2 == [{"provider": "NewServer"}]
|
|
|
|
|
|
# ---- shared-key output redaction (the streaming scrubber) ---------------------------------
|
|
@pytest.mark.parametrize("secrets,chunks,expected", [
|
|
([b"SEKRET"], [b"before SEK", b"RET after"], b"before *** after"), # across a chunk boundary
|
|
([b"KEY"], [b"hello world"], b"hello world"), # no secret in the stream
|
|
([], [b"anything at all"], b"anything at all"), # no secrets at all
|
|
])
|
|
def test_stream_redactor(secrets, chunks, expected):
|
|
r = cli._StreamRedactor(secrets)
|
|
out = b"".join(r.feed(c) for c in chunks) + r.flush()
|
|
assert out == expected
|
|
for secret in secrets:
|
|
assert secret not in out
|
|
|
|
|
|
def test_traversable_by_others_root_yes_missing_no():
|
|
# world-traversable → True; a missing path (OSError) → False (conservative)
|
|
assert cli._traversable_by_others("/") is True
|
|
assert cli._traversable_by_others("/no/such/path/zzz") is False
|
|
|
|
|
|
def test_call_content_type_flag_and_json_sniff(monkeypatch):
|
|
"""`call` sends a Content-Type: explicit --content-type wins; else a JSON body sniffs to
|
|
application/json (npm publish et al. reject an untyped body); a non-JSON body sends none."""
|
|
p = cli.build_parser()
|
|
assert p.parse_args(["call", "t", "p", "--content-type", "text/plain"]).content_type == "text/plain"
|
|
|
|
fake = _FakeClient()
|
|
monkeypatch.setattr(cli, "_client", lambda cfg: fake)
|
|
monkeypatch.setattr(cli, "_show", lambda r: None)
|
|
cfg = {"base_url": "http://x", "token": "T"}
|
|
|
|
def sent_headers(*extra) -> dict:
|
|
cli.cmd_call(p.parse_args(["call", "t", "p", "--method", "PUT", *extra]), cfg)
|
|
return fake.calls[-1][4]
|
|
|
|
assert sent_headers("--data", '{"ok":1}') == {"content-type": "application/json"} # sniffed from JSON body
|
|
assert sent_headers("--data", "plain text") == {} # non-JSON body: no guess
|
|
assert sent_headers("--data", "plain", "--content-type", "text/csv") == {"content-type": "text/csv"} # flag wins
|
|
|
|
|
|
def test_hidden_aliases_still_parse_and_route():
|
|
"""Every removed/renamed top-level command keeps working for existing scripts + agent
|
|
instructions — it is only absent from --help."""
|
|
p = cli.build_parser()
|
|
for argv, fn in [
|
|
(["add", "stripe", "--base-url", "https://api.stripe.com", "--secret", "STRIPE_KEY"], cli.cmd_add),
|
|
(["oauth", "providers"], cli.cmd_oauth_providers),
|
|
(["oauth", "connect", "--provider", "gsc"], cli.cmd_oauth_connect),
|
|
(["run", "stripe", "--", "get", "/v1/balance"], cli.cmd_run),
|
|
(["runs"], cli.cmd_runs),
|
|
(["calls"], cli.cmd_calls),
|
|
(["shell", "start"], cli.cmd_shell_start),
|
|
(["shell", "stop"], cli.cmd_shell_stop),
|
|
(["setup-local-run"], cli.cmd_setup_local_run),
|
|
(["import"], cli.cmd_import),
|
|
]:
|
|
assert p.parse_args(argv).fn is fn, argv
|
|
|
|
|
|
def test_alias_flags_match_their_canonical_command():
|
|
"""An alias is the SAME parser, not a stub: its flags must still parse."""
|
|
p = cli.build_parser()
|
|
assert p.parse_args(["run", "--server", "sk", "--", "x"]).server is True
|
|
assert p.parse_args(["cli", "run", "--server", "sk", "--", "x"]).server is True
|
|
assert p.parse_args(["shell", "start", "--ttl", "60"]).ttl == 60
|
|
assert p.parse_args(["cli", "shell", "start", "--ttl", "60"]).ttl == 60
|
|
assert p.parse_args(["setup-local-run", "--run-proof", "P"]).run_proof == "P"
|
|
assert p.parse_args(["cli", "setup", "--run-proof", "P"]).run_proof == "P"
|
|
assert p.parse_args(["runs", "--limit", "5"]).limit == 5
|
|
assert p.parse_args(["calls", "--limit", "5"]).limit == 5
|
|
|
|
|
|
def test_audit_merges_calls_and_runs(monkeypatch, capsys):
|
|
"""The default `audit` view interleaves both logs by time, and drops the `local_run`
|
|
CallRecords that /runs already reports as its local rows (else each shows up twice)."""
|
|
class _Resp:
|
|
status_code = 200
|
|
def __init__(self, body): self._b = body
|
|
def json(self): return self._b
|
|
|
|
bodies = {
|
|
"/calls": [
|
|
{"id": 2, "user_email": "a@x.dev", "tool_name": "stripe", "method": "GET",
|
|
"path": "v1/charges", "status_code": 200, "kind": "call", "created_at": "2026-07-28T10:00:00"},
|
|
{"id": 1, "user_email": "a@x.dev", "tool_name": "gh", "method": "GRANT",
|
|
"path": "pr list", "status_code": 200, "kind": "local_run", "created_at": "2026-07-28T09:00:00"},
|
|
],
|
|
"/runs": [
|
|
{"id": "l1", "user_email": "a@x.dev", "tool": "gh", "argv": ["pr", "list"],
|
|
"exit_code": None, "where": "local", "created_at": "2026-07-28T09:00:00"},
|
|
],
|
|
}
|
|
|
|
class _C:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *a): return False
|
|
def get(self, url, params=None): return _Resp(bodies[url])
|
|
|
|
monkeypatch.setattr(cli, "_client", lambda cfg: _C())
|
|
args = cli.build_parser().parse_args(["audit"])
|
|
cli.cmd_audit(args, {"base_url": "http://x"})
|
|
rows = json.loads(capsys.readouterr().out)
|
|
assert [r["kind"] for r in rows] == ["call", "run"] # newest first
|
|
assert rows[0]["detail"] == "GET v1/charges" and rows[0]["where"] == "proxy"
|
|
assert rows[1]["detail"] == "pr list" and rows[1]["where"] == "local"
|
|
|
|
|
|
def test_audit_filters_delegate_to_the_single_source_views(monkeypatch):
|
|
seen = []
|
|
monkeypatch.setattr(cli, "cmd_calls", lambda a, c: seen.append("calls"))
|
|
monkeypatch.setattr(cli, "cmd_runs", lambda a, c: seen.append("runs"))
|
|
p = cli.build_parser()
|
|
cli.cmd_audit(p.parse_args(["audit", "--calls"]), {})
|
|
cli.cmd_audit(p.parse_args(["audit", "--runs"]), {})
|
|
assert seen == ["calls", "runs"]
|
|
with pytest.raises(SystemExit): # the two filters are mutually exclusive
|
|
p.parse_args(["audit", "--calls", "--runs"])
|
|
|
|
|
|
# ---- catalog search / get (the discover -> inspect pair) ----------------------------------
|
|
def test_catalog_verbs_parse_without_displacing_a_platform_slug():
|
|
"""`search`/`get` are positional verbs, not subparsers — so `treg catalog tiktok` still browses
|
|
a shelf, and the multi-word query needs no quoting."""
|
|
p = cli.build_parser()
|
|
a = p.parse_args(["catalog", "search", "tiktok", "comments", "--limit", "5"])
|
|
assert a.fn is cli.cmd_catalog and a.platform == "search" and a.rest == ["tiktok", "comments"] and a.limit == 5
|
|
g = p.parse_args(["catalog", "get", "tikhub.tiktok.video.comments"])
|
|
assert g.platform == "get" and g.rest == ["tikhub.tiktok.video.comments"]
|
|
assert p.parse_args(["catalog", "tiktok"]).rest == []
|
|
assert p.parse_args(["catalog"]).platform is None
|
|
|
|
|
|
class _CatalogResp:
|
|
def __init__(self, body, status=200): self._b, self.status_code = body, status
|
|
def json(self): return self._b
|
|
|
|
|
|
def _stub_catalog_client(monkeypatch, routes: dict):
|
|
class _C:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *a): return False
|
|
def get(self, url, params=None): return routes[url]
|
|
monkeypatch.setattr(cli, "_client", lambda cfg, auth=True: _C())
|
|
|
|
|
|
def test_catalog_get_renders_params_siblings_and_the_command(monkeypatch, capsys):
|
|
body = {
|
|
"endpoint": {
|
|
"id": "justoneapi.tiktok.video.comments", "provider": "justoneapi", "provider_display": "JustOneAPI",
|
|
"summary": "A post's comments", "method": "GET", "path": "/api/tiktok/get-post-comment/v1",
|
|
"scope": "any_account", "tier": "core",
|
|
"cost": {"type": "per_success", "value": 0.1, "currency": "CNY", "usd": 0.014, "note": "billed on success"},
|
|
"verified": "2026-07-28", "docs_url": "https://docs.example/comments", "has_example": True,
|
|
"input": {"queryParams": {
|
|
"awemeId": {"type": "string", "required": True, "note": "the post id", "example": "76662"},
|
|
"cursor": {"type": "string", "required": False, "note": "'0' for the first page"}}},
|
|
"capability": "tiktok.video.comments", "capability_description": "List a video's comments",
|
|
"platform": "tiktok", "platform_label": "TikTok"},
|
|
"provider": {"service": "justoneapi", "display_name": "JustOneAPI",
|
|
"pricing_url": "https://justoneapi.com/", "limits": "60 req/min"},
|
|
"siblings": [{"id": "tikhub.tiktok.video.comments", "provider": "tikhub", "provider_display": "TikHub",
|
|
"summary": "s", "method": "GET", "path": "/p", "scope": "any_account", "tier": "core",
|
|
"cost": {"type": "per_call", "value": 0.001, "currency": "USD", "usd": 0.001},
|
|
"verified": "2026-07-28", "docs_url": "", "has_example": True, "input": None}],
|
|
"call_template": "treg call justoneapi /api/tiktok/get-post-comment/v1 --query awemeId=76662",
|
|
"example_response": {"code": 0, "data": {"comments": [{"text": "hi"}]}},
|
|
"hints": [],
|
|
}
|
|
_stub_catalog_client(monkeypatch, {"/catalog/endpoints/justoneapi.tiktok.video.comments": _CatalogResp(body)})
|
|
cli.cmd_catalog(cli.build_parser().parse_args(
|
|
["catalog", "get", "justoneapi.tiktok.video.comments"]), {"base_url": "http://x"})
|
|
out = capsys.readouterr().out
|
|
assert "$0.014/success" in out and "(CNY 0.1)" in out # usd to compare, original to verify
|
|
provider = body["provider"]
|
|
assert provider["limits"] in out and provider["pricing_url"] in out
|
|
assert "tikhub.tiktok.video.comments" in out # the sibling, for comparison
|
|
assert "awemeId" in out and "the post id" in out and "e.g. 76662" in out
|
|
assert "treg call justoneapi /api/tiktok/get-post-comment/v1 --query awemeId=76662" in out
|
|
assert '"comments"' in out # the example response, inline
|
|
|
|
|
|
def test_a_credit_price_reads_as_dollars_with_the_credits_behind_it():
|
|
"""A bare credit count is not a price — a reader cannot compare "1 credit" to "$0.001". When
|
|
the server priced the credit, dollars lead; only an unpriced credit shows alone."""
|
|
priced = {"type": "per_call", "value": 1, "currency": "credit", "usd": 0.00188}
|
|
assert cli._cost_label(priced) == "$0.00188/call (1 credit)"
|
|
assert cli._cost_usd(priced) == "$0.00188/call" # narrow comparison column
|
|
|
|
unpriced = {"type": "per_success", "value": 3, "currency": "credit", "usd": None}
|
|
assert cli._cost_label(unpriced) == "3 credits/success" # native, labelled as credits
|
|
# a duration-priced video table is quoted per second, not as a whole-call total range
|
|
per_second = {"type": "per_success", "usd": 13.87, "usd_min": 0.4744,
|
|
"rate_usd_min": 0.1186, "rate_usd": 0.462, "rate_unit": "s"}
|
|
assert cli._cost_usd(per_second) == "$0.119-$0.462/s"
|
|
assert cli._cost_usd(unpriced) == "3 credits/success"
|
|
assert "$" not in cli._cost_usd(unpriced), "no rate, no invented dollar figure"
|
|
|
|
|
|
def test_catalog_get_needs_an_id_and_404s_helpfully(monkeypatch, capsys):
|
|
p = cli.build_parser()
|
|
with pytest.raises(SystemExit):
|
|
cli.cmd_catalog(p.parse_args(["catalog", "get"]), {"base_url": "http://x"})
|
|
_stub_catalog_client(monkeypatch, {"/catalog/endpoints/nope.x": _CatalogResp({}, status=404)})
|
|
with pytest.raises(SystemExit):
|
|
cli.cmd_catalog(p.parse_args(["catalog", "get", "nope.x"]), {"base_url": "http://x"})
|
|
assert "find one with: treg catalog search" in capsys.readouterr().out
|
|
|
|
|
|
# ---- onboarding path 1: the catalog ---------------------------------------------------------
|
|
class _CatResp:
|
|
def __init__(self, payload, status=200, text=None):
|
|
self._payload, self.status_code = payload, status
|
|
self.text = text if text is not None else json.dumps(payload)
|
|
|
|
def json(self):
|
|
return self._payload
|
|
|
|
|
|
class _CatClient:
|
|
"""Stands in for `_client(cfg)`: answers the three endpoints the catalog path uses."""
|
|
|
|
def __init__(self, access: dict, call_status: int = 200):
|
|
self.access, self.call_status = access, call_status
|
|
self.seen: list[str] = []
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *a):
|
|
return False
|
|
|
|
def get(self, path, **kw):
|
|
self.seen.append(path)
|
|
if path.startswith("/catalog/search"):
|
|
return _CatResp({"results": [{"id": "tikhub.tiktok.user.profile", "provider": "tikhub",
|
|
"summary": "Public TikTok profile"}]})
|
|
if path.endswith("/access"):
|
|
return _CatResp(self.access)
|
|
if path.endswith("/balance"):
|
|
return _CatResp({"balance_micro": 999_000, "blocks": [], "entries": []})
|
|
return _CatResp({}, 404)
|
|
|
|
def request(self, method, path, **kw):
|
|
self.seen.append(f"{method} {path}")
|
|
return _CatResp({"ok": True}, self.call_status, text='{"ok": true}')
|
|
|
|
|
|
def _onboard_catalog(monkeypatch, access, call_status=200):
|
|
client = _CatClient(access, call_status)
|
|
monkeypatch.setattr(cli, "_client", lambda cfg: client)
|
|
monkeypatch.setattr(cli, "_active_org_id", lambda cfg, c: 1)
|
|
args = cli.build_parser().parse_args(["onboard", "--path", "catalog", "--yes"])
|
|
cli._run_catalog({"base_url": "http://x", "token": "t"}, args)
|
|
return client
|
|
|
|
|
|
def test_onboard_catalog_calls_when_treg_serves_it(monkeypatch, capsys):
|
|
"""The happy path: treg's own key covers it, so the price is stated and the call is made."""
|
|
c = _onboard_catalog(monkeypatch, {"tier": "platform", "estimated_cost_usd": 0.001})
|
|
out = capsys.readouterr().out
|
|
assert "on treg's key" in out and "$0.001" in out
|
|
assert "GET /call/tikhub.tiktok.user.profile" in c.seen # it really called
|
|
assert "treg balance" in out # and showed what it cost
|
|
|
|
|
|
def test_onboard_catalog_says_not_metered_for_your_own_key(monkeypatch, capsys):
|
|
c = _onboard_catalog(monkeypatch, {"tier": "credential"})
|
|
out = capsys.readouterr().out
|
|
assert "OWN credential" in out and "not metered" in out
|
|
assert any(s.startswith("GET /call/") for s in c.seen)
|
|
assert "treg balance" not in out # nothing was spent, so don't show a balance
|
|
|
|
|
|
def test_onboard_catalog_dead_end_names_the_one_command_that_fixes_it(monkeypatch, capsys):
|
|
"""Before platform providers are switched on (TREG_PLATFORM_PROVIDERS empty) this is what every
|
|
new user sees — so it must not read like a failure, and must never pretend to call."""
|
|
c = _onboard_catalog(monkeypatch, {"tier": "none", "detail": "no tikhub credential in this org yet"})
|
|
out = capsys.readouterr().out
|
|
assert "treg connections connect --provider tikhub" in out
|
|
assert not any(s.startswith("GET /call/") for s in c.seen) # nothing was called
|
|
|
|
|
|
def test_show_prints_failure_diagnostics_on_stderr_and_keeps_stdout_the_body(capsys):
|
|
"""A failed call must be filable: stdout stays the exact upstream body (whatever parses it), and
|
|
stderr gets one line with the HTTP status, whose answer it is, and the call id. A runner that
|
|
saved only stdout recorded 115 Moz quota 403s as a bare "cli_error" (2026-09-04)."""
|
|
import httpx
|
|
body = b'{"error":"The account does not have enough quota remaining for current period."}'
|
|
relayed = httpx.Response(403, content=body, headers={
|
|
"content-type": "application/json", "X-Treg-Call-Id": "c2075d8d79eb4436aafc310e81081c1b"})
|
|
with pytest.raises(SystemExit):
|
|
cli._show(relayed)
|
|
out, err = capsys.readouterr()
|
|
assert json.loads(out) == json.loads(body)
|
|
assert "HTTP 403" in err and "provider answered" in err and "c2075d8d79eb4436aafc310e81081c1b" in err
|
|
assert "charged" not in err # no cost header → no claim about money
|
|
|
|
refused = httpx.Response(402, content=b'{"detail":{"error":"route_max_cost"}}', headers={
|
|
"content-type": "application/json", "X-Treg-Error": "1"})
|
|
with pytest.raises(SystemExit):
|
|
cli._show(refused)
|
|
_, err = capsys.readouterr()
|
|
assert "HTTP 402" in err and "treg refused" in err
|
|
|
|
ok = httpx.Response(200, content=b'{"ok":true}', headers={"content-type": "application/json"})
|
|
cli._show(ok)
|
|
out, err = capsys.readouterr()
|
|
assert json.loads(out) == {"ok": True} and err == ""
|
|
|
|
|
|
def test_show_prints_the_charge_and_call_id_for_a_metered_success(capsys):
|
|
"""A metered 2xx gets one stderr line with the settled charge and the call id; stdout is the exact
|
|
body. No cost header (own key, or a non-call response) → nothing extra. A replay says so."""
|
|
import httpx
|
|
metered = httpx.Response(200, content=b'{"results":[{"x":1}],"next_token":"abc"}', headers={
|
|
"content-type": "application/json", "X-Treg-Cost-Micro": "6667",
|
|
"X-Treg-Call-Id": "125117d50cd3470cb25ba11f3d9789ad"})
|
|
cli._show(metered)
|
|
out, err = capsys.readouterr()
|
|
assert json.loads(out) == {"results": [{"x": 1}], "next_token": "abc"}
|
|
assert err.strip() == "treg: charged $0.006667 · call id 125117d50cd3470cb25ba11f3d9789ad"
|
|
|
|
own_key = httpx.Response(200, content=b'{"ok":true}', headers={
|
|
"content-type": "application/json", "X-Treg-Call-Id": "deadbeef"})
|
|
cli._show(own_key)
|
|
_, err = capsys.readouterr()
|
|
assert err == ""
|
|
|
|
replay = httpx.Response(200, content=b'{}', headers={
|
|
"content-type": "application/json", "X-Treg-Cost-Micro": "0", "X-Treg-Original-Cost-Micro": "6667",
|
|
"X-Treg-Idempotent-Replay": "true", "X-Treg-Call-Id": "c1"})
|
|
cli._show(replay)
|
|
_, err = capsys.readouterr()
|
|
assert "$0.006667 by the original call" in err and "nothing new charged" in err
|
|
|
|
async_submission = httpx.Response(200, content=b'{"run_id":"r1"}', headers={
|
|
"content-type": "application/json", "X-Treg-Cost-Micro": "2400000",
|
|
"X-Treg-Async": "{}", "X-Treg-Call-Id": "async1"})
|
|
cli._show(async_submission)
|
|
_, err = capsys.readouterr()
|
|
assert err.strip() == "treg: reserved up to $2.4 for async settlement · call id async1"
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------
|
|
# The hub family (phase 4)
|
|
|
|
def test_hub_init_writes_the_four_files(tmp_path, monkeypatch):
|
|
from treg import cli
|
|
import argparse
|
|
args = argparse.Namespace(name="leads-db", script=True, dir=str(tmp_path / "leads"))
|
|
cli.cmd_hub_init(args, {})
|
|
folder = tmp_path / "leads"
|
|
assert {p.name for p in folder.iterdir()} == {"recipe.json", "run.js", "check.json", "README.md"}
|
|
import json
|
|
m = json.loads((folder / "recipe.json").read_text())
|
|
assert m["name"] == "leads-db" and m["script"] == "run.js" and "required" not in json.dumps(m)
|
|
body = cli._hub_read_folder(str(folder))
|
|
assert set(body) == {"manifest", "check", "readme", "script"}
|
|
steps = argparse.Namespace(name="flow", script=False, dir=str(tmp_path / "flow"))
|
|
cli.cmd_hub_init(steps, {})
|
|
assert not (tmp_path / "flow" / "run.js").exists() and "steps" in json.loads((tmp_path / "flow" / "recipe.json").read_text())
|
|
|
|
|
|
def test_hub_is_in_the_help_and_parses():
|
|
from treg import cli
|
|
assert any(name == "hub" for _, rows in cli.HELP_GROUPS for name, _ in rows)
|
|
parser = cli.build_parser() if hasattr(cli, "build_parser") else None
|
|
if parser is not None:
|
|
a = parser.parse_args(["hub", "run", ".", "--input", "domain=figma.com", "--input", "limit=3"])
|
|
assert a.hub_cmd == "run" and a.input == ["domain=figma.com", "limit=3"]
|
|
e = parser.parse_args(["hub", "earnings", "acme.leads-db", "--days", "30", "--csv"])
|
|
assert e.hub_cmd == "earnings" and e.days == 30 and e.csv
|
|
pr = parser.parse_args(["hub", "price", "acme.leads-db", "0.02"])
|
|
assert pr.hub_cmd == "price" and pr.price_usd == 0.02
|
|
rt = parser.parse_args(["hub", "retire", "acme.leads-db"])
|
|
assert rt.hub_cmd == "retire" and rt.tool_id == "acme.leads-db"
|
|
|
|
|
|
|
|
def test_catalog_get_prints_a_hub_tool_contract(capsys):
|
|
from treg import cli
|
|
e = {"id": "acme.leads-db", "kind": "hub", "version": 2, "provider": "acme", "made_of": 2, "recipe": "script",
|
|
"price_line": "seller $0.01 + steps", "cost": {"usd": 0.01}, "status": "live",
|
|
"check": {"status": "passed", "checked_at": "2026-09-09T20:11:00"},
|
|
"inputs": {"domain": {"type": "string", "example": "figma.com", "note": "one domain"},
|
|
"limit": {"type": "int", "default": 20, "max": 100}},
|
|
"output": {"fields": ["rows", "count"]},
|
|
"call_template": {"cli": "treg call acme.leads-db --data '{\"domain\": \"figma.com\"}'", "http": "POST https://treg.to/call/acme.leads-db"},
|
|
"page": "https://treg.to/hub/acme.leads-db"}
|
|
cli._catalog_get_hub(e)
|
|
out = capsys.readouterr().out
|
|
assert "seller $0.01 + steps" in out and "domain" in out and "(required)" in out and "rows, count" in out
|
|
assert "treg call acme.leads-db" in out and "https://treg.to/hub/acme.leads-db" in out
|
|
|
|
|
|
|
|
def test_a_hub_refusal_names_the_missing_tool_in_its_fix_commands(capsys):
|
|
"""From the case study walk (2026-09-09): the refusal must name the tool the manifest wants,
|
|
one command per line, in the order to run them."""
|
|
from treg import cli
|
|
|
|
class R:
|
|
status_code = 422
|
|
headers = {"content-type": "application/json"}
|
|
def json(self):
|
|
return {"detail": {"error": "manifest_invalid", "field": "uses[0]",
|
|
"rule": "'supabase' is not one of your team's tools (register it first: treg tool add)"}}
|
|
import pytest as _pytest
|
|
with _pytest.raises(SystemExit):
|
|
cli._hub_report(R(), json_out=False)
|
|
out = capsys.readouterr().out
|
|
assert "uses[0]" in out
|
|
assert "treg secret add SUPABASE_KEY --value" in out
|
|
assert "treg tool add supabase --base-url" in out
|
|
lines = [l for l in out.splitlines() if "treg " in l]
|
|
assert len(lines) >= 3 and all(l.count("treg ") == 1 for l in lines) # one command per line
|
|
def test_host_prints_the_url_alone_and_the_full_response_under_json(monkeypatch, tmp_path, capsys):
|
|
"""`$(treg host face.jpg)` must yield the bare URL; `--json` is the GLOBAL flag main() pops from
|
|
argv, so cmd_host reads _JSON_OVERRIDE like every other command (a per-subcommand flag is dead)."""
|
|
class Response:
|
|
status_code = 201
|
|
def json(self):
|
|
return {"url": "http://x/m/tok", "token": "tok", "content_type": "image/png",
|
|
"size": 3, "expires_at": "2026-09-21T00:00:00Z"}
|
|
|
|
class Client:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): return False
|
|
def post(self, path, **kwargs):
|
|
assert path == "/media" and kwargs["content"] == b"png" and kwargs["headers"]["content-type"] == "image/png"
|
|
return Response()
|
|
|
|
monkeypatch.setattr(cli, "_client", lambda cfg, **k: Client())
|
|
monkeypatch.setattr(cli, "_load_config", lambda: {"base_url": "http://x", "token": "t"})
|
|
f = tmp_path / "face.png"
|
|
f.write_bytes(b"png")
|
|
cli.main(["host", str(f)])
|
|
assert capsys.readouterr().out == "http://x/m/tok\n"
|
|
cli.main(["host", str(f), "--json"])
|
|
assert json.loads(capsys.readouterr().out)["token"] == "tok"
|
|
|
|
|
|
def test_org_rename_follows_slug_change_locally(monkeypatch, tmp_path):
|
|
"""`treg org rename --slug` rewrites active_org; the pinned token is untouched because the server
|
|
keeps the old slug as an alias."""
|
|
monkeypatch.setattr(cli, "CONFIG_PATH", tmp_path / "config.json")
|
|
saved = {}
|
|
monkeypatch.setattr(cli, "_save_config", lambda cfg: saved.update(cfg))
|
|
monkeypatch.setattr(cli, "_active_org_id", lambda cfg, c: 7)
|
|
calls = []
|
|
|
|
class Response:
|
|
status_code = 200
|
|
def json(self):
|
|
return {"org_id": 7, "org": "team-b", "previous_slug": "team-a", "name": "Team B"}
|
|
|
|
class Client:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *a): return False
|
|
def patch(self, path, json=None):
|
|
calls.append((path, json))
|
|
return Response()
|
|
|
|
monkeypatch.setattr(cli, "_client", lambda cfg: Client())
|
|
cfg = {"base_url": "http://x", "token": "TK", "active_org": "team-a"}
|
|
cli.cmd_org_rename(type("A", (), {"name": "Team B", "slug": "team-b"})(), cfg)
|
|
assert calls == [("/orgs/7", {"name": "Team B", "slug": "team-b"})]
|
|
assert cfg["active_org"] == "team-b" and cfg["token"] == "TK" and saved["active_org"] == "team-b"
|