implement server mvp: fastapi app, org formatter, sqlite store, tests, dockerfile
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
118
server/tests/test_api.py
Normal file
118
server/tests/test_api.py
Normal file
@@ -0,0 +1,118 @@
|
||||
import os
|
||||
import tempfile
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
# Set env vars before importing app so store/paths are overridden in tests
|
||||
TOKEN = "test-token-abc"
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def env_setup(tmp_path, monkeypatch):
|
||||
db = str(tmp_path / "capture.sqlite3")
|
||||
org = str(tmp_path / "synq.org")
|
||||
monkeypatch.setenv("PHONE_CAPTURE_TOKEN", TOKEN)
|
||||
monkeypatch.setenv("PHONE_CAPTURE_DB_PATH", db)
|
||||
monkeypatch.setenv("PHONE_CAPTURE_ORG_PATH", org)
|
||||
# reset singleton store so each test gets a fresh db
|
||||
import app.main as m
|
||||
m._store = None
|
||||
yield
|
||||
m._store = None
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def client():
|
||||
from app.main import app
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def auth_headers():
|
||||
return {"Authorization": f"Bearer {TOKEN}"}
|
||||
|
||||
|
||||
VALID_PAYLOAD = {
|
||||
"id": "phone-20260517-143122-a8f2",
|
||||
"created_at": "2026-05-17T14:31:22-04:00",
|
||||
"kind": "todo",
|
||||
"body": "buy printer paper",
|
||||
"tags": ["home", "errands"],
|
||||
"device": "android",
|
||||
}
|
||||
|
||||
|
||||
class TestHealth:
|
||||
def test_health_returns_200(self, client):
|
||||
r = client.get("/health")
|
||||
assert r.status_code == 200
|
||||
data = r.json()
|
||||
assert data["ok"] is True
|
||||
assert data["service"] == "synq"
|
||||
assert data["version"] == "0.1.0"
|
||||
|
||||
def test_health_no_auth_required(self, client):
|
||||
r = client.get("/health")
|
||||
assert r.status_code == 200
|
||||
|
||||
|
||||
class TestCapture:
|
||||
def test_valid_capture_accepted(self, client, tmp_path):
|
||||
r = client.post("/capture", json=VALID_PAYLOAD, headers=auth_headers())
|
||||
assert r.status_code == 200
|
||||
data = r.json()
|
||||
assert data["ok"] is True
|
||||
assert data["status"] == "accepted"
|
||||
assert data["id"] == VALID_PAYLOAD["id"]
|
||||
|
||||
def test_valid_capture_appended_to_org(self, client, monkeypatch, tmp_path):
|
||||
org = str(tmp_path / "synq.org")
|
||||
monkeypatch.setenv("PHONE_CAPTURE_ORG_PATH", org)
|
||||
client.post("/capture", json=VALID_PAYLOAD, headers=auth_headers())
|
||||
content = open(org, encoding="utf-8").read()
|
||||
assert "buy printer paper" in content
|
||||
assert ":ID: phone-20260517-143122-a8f2" in content
|
||||
|
||||
def test_duplicate_capture_returns_already_seen(self, client):
|
||||
client.post("/capture", json=VALID_PAYLOAD, headers=auth_headers())
|
||||
r = client.post("/capture", json=VALID_PAYLOAD, headers=auth_headers())
|
||||
assert r.status_code == 200
|
||||
assert r.json()["status"] == "already_seen"
|
||||
|
||||
def test_duplicate_not_appended_twice(self, client, monkeypatch, tmp_path):
|
||||
org = str(tmp_path / "synq.org")
|
||||
monkeypatch.setenv("PHONE_CAPTURE_ORG_PATH", org)
|
||||
client.post("/capture", json=VALID_PAYLOAD, headers=auth_headers())
|
||||
client.post("/capture", json=VALID_PAYLOAD, headers=auth_headers())
|
||||
content = open(org, encoding="utf-8").read()
|
||||
assert content.count(VALID_PAYLOAD["id"]) == 1
|
||||
|
||||
def test_missing_token_rejected(self, client):
|
||||
r = client.post("/capture", json=VALID_PAYLOAD)
|
||||
assert r.status_code == 401
|
||||
assert r.json()["detail"] == "unauthorized"
|
||||
|
||||
def test_wrong_token_rejected(self, client):
|
||||
r = client.post("/capture", json=VALID_PAYLOAD, headers={"Authorization": "Bearer wrong"})
|
||||
assert r.status_code == 401
|
||||
|
||||
def test_empty_body_rejected(self, client):
|
||||
payload = {**VALID_PAYLOAD, "body": " "}
|
||||
r = client.post("/capture", json=payload, headers=auth_headers())
|
||||
assert r.status_code == 400
|
||||
assert "body" in r.json()["detail"].lower()
|
||||
|
||||
def test_missing_body_field_rejected(self, client):
|
||||
payload = {k: v for k, v in VALID_PAYLOAD.items() if k != "body"}
|
||||
r = client.post("/capture", json=payload, headers=auth_headers())
|
||||
assert r.status_code in (400, 422)
|
||||
|
||||
def test_invalid_kind_rejected(self, client):
|
||||
payload = {**VALID_PAYLOAD, "kind": "journal"}
|
||||
r = client.post("/capture", json=payload, headers=auth_headers())
|
||||
assert r.status_code in (400, 422)
|
||||
|
||||
def test_note_kind_accepted(self, client):
|
||||
payload = {**VALID_PAYLOAD, "id": "phone-20260517-143200-note1", "kind": "note"}
|
||||
r = client.post("/capture", json=payload, headers=auth_headers())
|
||||
assert r.status_code == 200
|
||||
assert r.json()["status"] == "accepted"
|
||||
Reference in New Issue
Block a user