Gabriel Vidal
3a0ef08dcf
- liveness falls back to psutil where there is no /proc (macOS) - GET /feed + /feed/file: the hub pulls transcripts it has no mount for - POST /pair trades a one-time code for the worker's bearer; /unpair drops it - /health reports worker identity + permission mode - worker/install-macos.sh: venv, launchd agent bound to the Tailscale IP, prints the pairing string (addr/code/claude login) - Dockerfile copies every sidecar module (claude_cli was missing, so the in-container runner could not import) Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
102 lines
3.3 KiB
Python
102 lines
3.3 KiB
Python
"""The transcript feed: listing, byte-offset reads, shrink, traversal."""
|
|
|
|
import os
|
|
import pathlib
|
|
|
|
import pytest
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.testclient import TestClient
|
|
|
|
import feed
|
|
|
|
TOKEN = "t0ken"
|
|
|
|
|
|
def _auth(h):
|
|
if h != f"Bearer {TOKEN}":
|
|
raise HTTPException(401, "unauthorized")
|
|
|
|
|
|
@pytest.fixture()
|
|
def root(tmp_path):
|
|
return tmp_path / "projects"
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(root):
|
|
app = FastAPI()
|
|
app.include_router(feed.make_router(_auth, lambda: root))
|
|
return TestClient(app, headers={"Authorization": f"Bearer {TOKEN}"})
|
|
|
|
|
|
def _write(root, rel, text):
|
|
p = root / rel
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
p.write_text(text)
|
|
return p
|
|
|
|
|
|
def test_lists_transcripts_subagents_and_meta_only(root, client):
|
|
_write(root, "-Users-g-proj/a.jsonl", "{}\n")
|
|
_write(root, "-Users-g-proj/a/subagents/agent-1.jsonl", "{}\n")
|
|
_write(root, "-Users-g-proj/a/subagents/agent-1.meta.json", "{}")
|
|
_write(root, "-Users-g-proj/a/tool-results/x.txt", "secret")
|
|
rels = sorted(f["rel"] for f in client.get("/feed").json()["files"])
|
|
assert rels == ["-Users-g-proj/a.jsonl",
|
|
"-Users-g-proj/a/subagents/agent-1.jsonl",
|
|
"-Users-g-proj/a/subagents/agent-1.meta.json"]
|
|
|
|
|
|
def test_since_filters_on_mtime(root, client):
|
|
old = _write(root, "p/old.jsonl", "{}\n")
|
|
os.utime(old, (1000, 1000))
|
|
_write(root, "p/new.jsonl", "{}\n")
|
|
rels = [f["rel"] for f in client.get("/feed?since=2000").json()["files"]]
|
|
assert rels == ["p/new.jsonl"]
|
|
|
|
|
|
def test_offset_read_then_append(root, client):
|
|
p = _write(root, "p/s.jsonl", '{"a":1}\n')
|
|
r = client.get("/feed/file", params={"rel": "p/s.jsonl"})
|
|
assert r.content == b'{"a":1}\n' and r.headers["x-feed-size"] == "8"
|
|
with open(p, "a") as f:
|
|
f.write('{"b":2}\n')
|
|
r = client.get("/feed/file", params={"rel": "p/s.jsonl", "offset": 8})
|
|
assert r.content == b'{"b":2}\n' and r.headers["x-feed-size"] == "16"
|
|
|
|
|
|
def test_truncate_shows_in_size(root, client):
|
|
p = _write(root, "p/s.jsonl", "x" * 100)
|
|
p.write_text("y" * 10)
|
|
r = client.get("/feed/file", params={"rel": "p/s.jsonl", "offset": 50})
|
|
assert r.content == b"" and r.headers["x-feed-size"] == "10"
|
|
|
|
|
|
def test_limit_bounds_the_chunk(root, client):
|
|
_write(root, "p/s.jsonl", "abcdef")
|
|
r = client.get("/feed/file", params={"rel": "p/s.jsonl", "offset": 1,
|
|
"limit": 2})
|
|
assert r.content == b"bc"
|
|
|
|
|
|
@pytest.mark.parametrize("rel", [
|
|
"../outside.jsonl", "/etc/passwd", "p/../../outside.jsonl",
|
|
"p/s.txt", "p//s.jsonl", "", "p/link.jsonl"])
|
|
def test_traversal_and_foreign_files_are_404(root, client, rel):
|
|
_write(root, "p/s.jsonl", "{}")
|
|
outside = _write(root.parent, "outside.jsonl", "{}")
|
|
(root / "p" / "link.jsonl").symlink_to(outside)
|
|
assert client.get("/feed/file", params={"rel": rel}).status_code == 404
|
|
|
|
|
|
def test_needs_the_bearer(root):
|
|
app = FastAPI()
|
|
app.include_router(feed.make_router(_auth, lambda: root))
|
|
c = TestClient(app)
|
|
assert c.get("/feed").status_code == 401
|
|
assert c.get("/feed/file", params={"rel": "p/s.jsonl"}).status_code == 401
|
|
|
|
|
|
def test_missing_root_is_an_empty_feed(tmp_path, client):
|
|
assert client.get("/feed").json()["files"] == []
|