Spaces:
Sleeping
Sleeping
File size: 4,416 Bytes
116524e | 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 | from __future__ import annotations
from pathlib import Path
import pytest
from click.testing import CliRunner
from ace.cli import cloud
from ace.cli.client import KaybaAPIError, KaybaClient
class _FakeTraceClient:
def __init__(self, traces=None):
self._traces = traces or []
self.upload_calls: list[list[dict[str, str]]] = []
def list_traces(self):
return {"traces": self._traces}
def upload_traces(self, traces):
self.upload_calls.append(traces)
return {
"count": len(traces),
"traces": [
{"id": f"trace-{idx}", "filename": trace["filename"]}
for idx, trace in enumerate(traces, start=1)
],
}
class _FakeResponse:
def __init__(self, status_code: int, text: str, json_data=None):
self.status_code = status_code
self.text = text
self._json_data = json_data
def json(self):
if self._json_data is None:
raise ValueError("no json")
return self._json_data
class _FakeSession:
def __init__(self, response: _FakeResponse):
self.response = response
self.headers = {}
def request(self, method: str, url: str, json=None, params=None):
return self.response
def test_detect_file_type_handles_jsonl_and_markdown():
assert cloud._detect_file_type("trace.jsonl") == "json"
assert cloud._detect_file_type("trace.markdown") == "md"
def test_traces_list_empty_state_explains_manual_upload(monkeypatch):
runner = CliRunner()
fake_client = _FakeTraceClient()
monkeypatch.setattr(cloud, "_client", lambda api_key, base_url: fake_client)
result = runner.invoke(cloud.traces_list, [])
assert result.exit_code == 0
assert "Kayba does not auto-import local agent transcripts yet." in result.output
assert "~/.claude/projects/<project>/*.jsonl" in result.output
def test_traces_upload_skips_oversized_files(tmp_path: Path, monkeypatch):
runner = CliRunner()
fake_client = _FakeTraceClient()
monkeypatch.setattr(cloud, "_client", lambda api_key, base_url: fake_client)
small = tmp_path / "small.jsonl"
small.write_text('{"ok": true}\n', encoding="utf-8")
large = tmp_path / "large.txt"
large.write_text("x" * (cloud.MAX_TRACE_CHARS + 1), encoding="utf-8")
result = runner.invoke(cloud.traces_upload, [str(small), str(large)])
assert result.exit_code == 0
assert "Uploaded 1 trace(s)." in result.output
assert "Skipping large.txt" in result.output
assert len(fake_client.upload_calls) == 1
assert fake_client.upload_calls[0][0]["filename"] == "small.jsonl"
assert fake_client.upload_calls[0][0]["fileType"] == "json"
def test_prompts_install_replaces_managed_block(tmp_path: Path):
runner = CliRunner()
prompt_file = tmp_path / "prompt.md"
target_file = tmp_path / "CLAUDE.md"
prompt_file.write_text("First prompt", encoding="utf-8")
first = runner.invoke(
cloud.prompts_install,
["--input", str(prompt_file), "--target", "claude-code", "--file", str(target_file)],
)
assert first.exit_code == 0
first_text = target_file.read_text(encoding="utf-8")
assert "First prompt" in first_text
assert first_text.count(cloud.PROMPT_BLOCK_START) == 1
prompt_file.write_text("Second prompt", encoding="utf-8")
second = runner.invoke(
cloud.prompts_install,
["--input", str(prompt_file), "--target", "claude-code", "--file", str(target_file)],
)
assert second.exit_code == 0
second_text = target_file.read_text(encoding="utf-8")
assert "Second prompt" in second_text
assert "First prompt" not in second_text
assert second_text.count(cloud.PROMPT_BLOCK_START) == 1
def test_client_formats_non_json_http_errors():
client = KaybaClient(api_key="test-key", base_url="https://example.com")
client.session = _FakeSession(
_FakeResponse(502, "<html><body>gateway exploded in a surprisingly long way</body></html>")
)
with pytest.raises(KaybaAPIError) as exc:
client._request("GET", "/traces")
assert exc.value.code == "HTTP_ERROR"
assert "HTTP 502 from Kayba API:" in exc.value.message
assert "<html><body>gateway exploded" in exc.value.message
|