148 lines
6.8 KiB
Python
148 lines
6.8 KiB
Python
import importlib
|
|
import os
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path):
|
|
os.environ["DB_PATH"] = str(tmp_path / "test.db")
|
|
os.environ["SYNC_TOKEN"] = "geheim"
|
|
os.environ["FILES_DIR"] = str(tmp_path / "files")
|
|
from app import config, db, files, main
|
|
importlib.reload(config)
|
|
importlib.reload(db)
|
|
importlib.reload(files)
|
|
importlib.reload(main)
|
|
with TestClient(main.app) as c:
|
|
yield c
|
|
|
|
|
|
AUTH = {"Authorization": "Bearer geheim"}
|
|
|
|
|
|
def sync(client, since=0, **collections):
|
|
body = {"since": since, **collections}
|
|
r = client.post("/api/sync", json=body, headers=AUTH)
|
|
assert r.status_code == 200, r.text
|
|
return r.json()
|
|
|
|
|
|
def test_health_ohne_token(client):
|
|
assert client.get("/api/health").json() == {"ok": True}
|
|
|
|
|
|
def test_falscher_token(client):
|
|
assert client.post("/api/sync", json={}).status_code == 401
|
|
assert client.post("/api/sync", json={}, headers={"Authorization": "Bearer falsch"}).status_code == 401
|
|
assert client.get("/api/events").status_code == 401
|
|
|
|
|
|
def test_push_und_pull(client):
|
|
a = sync(client, events=[{"id": "e1", "updated": "2026-09-19T10:00:00.000Z", "title": "Zahnarzt"}])
|
|
assert a["cursor"] == 1
|
|
assert [e["id"] for e in a["events"]] == ["e1"]
|
|
|
|
# Anderes Gerät ohne Stand bekommt alles
|
|
b = sync(client)
|
|
assert [e["title"] for e in b["events"]] == ["Zahnarzt"]
|
|
assert b["cursor"] == 1
|
|
|
|
# Mit aktuellem Stand kommt nichts Neues
|
|
c = sync(client, since=1)
|
|
assert c["events"] == [] and c["cursor"] == 1
|
|
|
|
|
|
def test_neuester_gewinnt(client):
|
|
sync(client, todos=[{"id": "t1", "updated": "2026-09-19T10:00:00.000Z", "title": "alt"}])
|
|
sync(client, todos=[{"id": "t1", "updated": "2026-09-19T09:00:00.000Z", "title": "älter"}])
|
|
assert client.get("/api/todos", headers=AUTH).json()[0]["title"] == "alt"
|
|
r = sync(client, since=0, todos=[{"id": "t1", "updated": "2026-09-19T11:00:00.000Z", "title": "neu"}])
|
|
assert r["todos"][0]["title"] == "neu"
|
|
assert r["cursor"] == 2
|
|
|
|
|
|
def test_abgelehnte_aenderung_bekommt_serverstand(client):
|
|
sync(client, todos=[{"id": "t1", "updated": "2026-09-19T10:00:00.000Z", "title": "server"}])
|
|
r = sync(client, since=1, todos=[{"id": "t1", "updated": "2026-09-19T09:00:00.000Z", "title": "client"}])
|
|
assert [t["title"] for t in r["todos"]] == ["server"]
|
|
assert r["cursor"] == 1
|
|
|
|
|
|
def test_tombstone(client):
|
|
sync(client, notes=[{"id": "Plan.md", "updated": "2026-09-19T10:00:00.000Z", "content": "x"}])
|
|
r = sync(client, since=1, notes=[{"id": "Plan.md", "updated": "2026-09-19T10:05:00.000Z", "deleted": True}])
|
|
assert r["notes"][0]["deleted"] is True
|
|
assert client.get("/api/notes", headers=AUTH).json() == []
|
|
# Ein verspäteter älterer Stand belebt die Notiz nicht wieder
|
|
sync(client, notes=[{"id": "Plan.md", "updated": "2026-09-19T10:01:00.000Z", "content": "y"}])
|
|
assert client.get("/api/notes", headers=AUTH).json() == []
|
|
|
|
|
|
def test_ordner_laufen_mit(client):
|
|
r = sync(client, folders=[{"id": "Projekte/Alt", "updated": "2026-09-19T10:00:00.000Z", "deleted": False}])
|
|
assert r["folders"][0]["id"] == "Projekte/Alt"
|
|
r = sync(client, since=1, folders=[{"id": "Projekte/Alt", "updated": "2026-09-19T10:05:00.000Z", "deleted": True}])
|
|
assert r["folders"][0]["deleted"] is True
|
|
|
|
|
|
def test_seit_cursor_liefert_nur_neues(client):
|
|
sync(client, events=[{"id": "e1", "updated": "2026-09-19T10:00:00.000Z", "title": "a"}])
|
|
sync(client, events=[{"id": "e2", "updated": "2026-09-19T10:00:00.000Z", "title": "b"}])
|
|
r = sync(client, since=1)
|
|
assert [e["id"] for e in r["events"]] == ["e2"]
|
|
|
|
|
|
def test_dateien_nur_auf_anfrage(client):
|
|
sync(client, files=[{"id": "-home-x/s1.jsonl", "updated": "2026-09-22T10:00:00.000Z", "content": "zeile\n"}])
|
|
# Handy und Extension fragen ohne files: nichts davon in der Antwort
|
|
r = sync(client, events=[], todos=[], notes=[], folders=[])
|
|
assert "files" not in r and r["events"] == []
|
|
assert "files" not in sync(client)
|
|
# Nur Dateien angefragt: nur der Steckbrief zurück, kein Inhalt
|
|
r = sync(client, files=[])
|
|
assert r["files"] == [{"id": "-home-x/s1.jsonl", "updated": "2026-09-22T10:00:00.000Z", "size": 6}] and "events" not in r
|
|
assert r["cursor"] == 1
|
|
|
|
|
|
def test_dateien_inhalt_auf_platte_und_anhaengen(client, tmp_path):
|
|
import hashlib
|
|
sync(client, files=[{"id": "p/s.jsonl", "updated": "2026-09-22T10:00:00.000Z", "content": "a\n"}])
|
|
assert (tmp_path / "files" / "p" / "s.jsonl").read_bytes() == b"a\n"
|
|
# Anhang mit passendem Stand
|
|
base = {"size": 2, "hash": hashlib.sha256(b"a\n").hexdigest()}
|
|
r = sync(client, files=[{"id": "p/s.jsonl", "updated": "2026-09-22T11:00:00.000Z", "base": base, "append": "b\n"}])
|
|
assert (tmp_path / "files" / "p" / "s.jsonl").read_bytes() == b"a\nb\n"
|
|
assert r["files"][-1]["size"] == 4
|
|
# Anhang mit falschem Stand: abgelehnt, Server-Stand samt Hash kommt zurück
|
|
r = sync(client, since=99, files=[{"id": "p/s.jsonl", "updated": "2026-09-22T12:00:00.000Z", "base": base, "append": "c\n"}])
|
|
assert r["files"] == [{"id": "p/s.jsonl", "updated": "2026-09-22T11:00:00.000Z", "size": 4, "hash": hashlib.sha256(b"a\nb\n").hexdigest()}]
|
|
assert (tmp_path / "files" / "p" / "s.jsonl").read_bytes() == b"a\nb\n"
|
|
# Inhalt holen: ganz, oder nur den Rest ab bekanntem Stand
|
|
r = client.get("/api/files/p/s.jsonl", headers=AUTH)
|
|
assert r.content == b"a\nb\n" and r.headers["X-From"] == "0" and r.headers["X-Size"] == "4"
|
|
r = client.get("/api/files/p/s.jsonl", params={"from": 2, "hash": base["hash"]}, headers=AUTH)
|
|
assert r.content == b"b\n" and r.headers["X-From"] == "2"
|
|
r = client.get("/api/files/p/s.jsonl", params={"from": 2, "hash": "falsch"}, headers=AUTH)
|
|
assert r.content == b"a\nb\n" and r.headers["X-From"] == "0"
|
|
# Löschen räumt die Platte, Tombstone bleibt, Inhalt weg
|
|
sync(client, files=[{"id": "p/s.jsonl", "updated": "2026-09-22T13:00:00.000Z", "deleted": True}])
|
|
assert not (tmp_path / "files" / "p" / "s.jsonl").exists()
|
|
assert client.get("/api/files/p/s.jsonl", headers=AUTH).status_code == 404
|
|
assert sync(client, files=[])["files"][-1]["deleted"] is True
|
|
|
|
|
|
def test_dateien_aus_datenbank_auf_platte(client, tmp_path):
|
|
import json
|
|
from app import db, files
|
|
# Alter Stand: Inhalt in der Datenbank
|
|
with db.connect(write=True) as conn:
|
|
conn.execute("INSERT INTO items VALUES ('files', 'alt/s.jsonl', '2026-09-01T00:00:00.000Z', 0, ?, 1)",
|
|
(json.dumps({"id": "alt/s.jsonl", "updated": "2026-09-01T00:00:00.000Z", "content": "alt\n"}),))
|
|
with db.connect(write=True) as conn:
|
|
assert files.migrate(conn) == 1
|
|
assert files.migrate(conn) == 0
|
|
assert (tmp_path / "files" / "alt" / "s.jsonl").read_bytes() == b"alt\n"
|
|
assert sync(client, files=[])["files"] == [{"id": "alt/s.jsonl", "updated": "2026-09-01T00:00:00.000Z", "size": 4}]
|