Spaces:
Running
Running
| 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 | |
| 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" | |