Spaces:
Running
Running
File size: 3,930 Bytes
62068da | 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 | import json
from contextlib import asynccontextmanager
from fastapi.testclient import TestClient
from status import atomic_write_json, read_status, write_status
from strategies.base import Memory
from web.server import build_app
@asynccontextmanager
async def _lifespan(_app):
yield
def _app(tmp_path, *, web_token=None):
return build_app(
_lifespan,
status_path=tmp_path / "status.json",
commands_path=tmp_path / "commands.json",
key_events_path=tmp_path / "events.log",
history_path=tmp_path / "history.jsonl",
memory_path=tmp_path / "memory.json",
web_token=web_token,
)
def test_atomic_write_json_creates_parent_and_valid_json(tmp_path):
path = tmp_path / "nested" / "snapshot.json"
atomic_write_json(path, {"message": "你好", "items": [1, 2]})
assert json.loads(path.read_text()) == {"message": "你好", "items": [1, 2]}
assert list(path.parent.glob("*.tmp")) == []
def test_write_status_keeps_timestamp_and_failure_contract(tmp_path):
path = tmp_path / "status.json"
assert write_status(path, {"tick": 7}) is True
snapshot = read_status(path)
assert snapshot["tick"] == 7
assert isinstance(snapshot["written_at"], float)
assert write_status(path, {"bad": object()}) is True # default=str remains supported
def test_memory_download_is_directly_restorable(tmp_path):
memory = Memory()
memory.obstacles.add((1, 2))
memory.resources[(3, 4)] = 9
atomic_write_json(tmp_path / "memory.json", memory.to_dict())
with TestClient(_app(tmp_path)) as client:
response = client.get("/api/memory")
assert response.status_code == 200
assert response.headers["cache-control"] == "no-store"
assert response.headers["content-disposition"] == 'attachment; filename="memory.json"'
restored = Memory.from_dict(response.json())
assert restored.obstacles == {(1, 2)}
assert restored.resources == {(3, 4): 9}
def test_memory_download_returns_404_when_missing(tmp_path):
with TestClient(_app(tmp_path)) as client:
response = client.get("/api/memory")
assert response.status_code == 404
assert response.headers["cache-control"] == "no-store"
def test_memory_and_commands_require_configured_bearer_token(tmp_path):
memory = Memory()
atomic_write_json(tmp_path / "memory.json", memory.to_dict())
with TestClient(_app(tmp_path, web_token="secret")) as client:
assert client.get("/api/memory").status_code == 401
assert client.get("/api/status").status_code == 401
assert client.get("/api/key-events").status_code == 401
assert client.get("/api/history").status_code == 401
assert client.post("/api/command", json={"type": "core_auto"}).status_code == 401
headers = {"Authorization": "Bearer secret"}
assert client.get("/api/memory", headers=headers).status_code == 200
response = client.post(
"/api/command", headers=headers,
json={"type": "core_migrate", "direction": "RIGHT"})
assert response.status_code == 200
command = json.loads((tmp_path / "commands.json").read_text())["commands"][0]
assert command["type"] == "core_migrate"
assert command["direction"] == "RIGHT"
def test_core_migration_command_rejects_invalid_direction(tmp_path):
with TestClient(_app(tmp_path)) as client:
response = client.post(
"/api/command",
json={"type": "core_migrate", "direction": "NORTHEAST"})
assert response.status_code == 400
assert not (tmp_path / "commands.json").exists()
def test_memory_download_returns_503_when_invalid(tmp_path):
(tmp_path / "memory.json").write_text("{partial", encoding="utf-8")
with TestClient(_app(tmp_path)) as client:
response = client.get("/api/memory")
assert response.status_code == 503
assert response.headers["cache-control"] == "no-store"
|