retoor
3499f6cda0
- No client-side encryption: plaintext content-addressed remote (SHA-256 names), no backup key, no cryptography dependency (server disk encryption is the trust model). - Filters back user data: images, archives, databases, PDFs accepted; 10 MiB cap; binary sniffing removed; temp names hardened (~$, #..#, .temp). - SQLite zero-error policy: backup-API snapshots + integrity_check, journal folding, locked/corrupt loud skips, verified restores. - Recovery: reindex from manifests, remote adopt, on-demand blob fetch, 5-day retention + thinning, GC, date-guarded remote purge, metrics. - Scheduler with WebDAV quota signal and 70% pressure backstop (floor kept). - 37 tests incl. live-monitor capture safety and DB safety.
110 lines
3.9 KiB
Python
110 lines
3.9 KiB
Python
"""Bone-level proof: temp files and unfinished writes are never versioned.
|
|
|
|
Runs against the real inotify monitor through the HTTP API (same fixture
|
|
shape as test_service). Each test fails if a single temp/unfinished state
|
|
is ever stored.
|
|
"""
|
|
|
|
import base64
|
|
import os
|
|
import time
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from versiond.api import create_app
|
|
from versiond.config import Config, Paths
|
|
|
|
from test_service import CONFIG, history, wait_for
|
|
|
|
|
|
@pytest.fixture
|
|
def env(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("VERSIOND_HOME", str(tmp_path / "home"))
|
|
paths = Paths.resolve()
|
|
paths.ensure()
|
|
paths.config_file.write_text(CONFIG)
|
|
cfg = Config.load(paths)
|
|
work = tmp_path / "work"
|
|
(work / "src").mkdir(parents=True)
|
|
(work / "src" / "app.py").write_text("v0\n")
|
|
with TestClient(create_app(cfg), base_url="http://127.0.0.1:9922") as client:
|
|
client.headers["Authorization"] = f"Bearer {cfg.api_token()}"
|
|
yield client, work
|
|
|
|
|
|
def _add_root(client, work):
|
|
r = client.post("/api/v1/roots", json={"path": str(work)})
|
|
assert r.status_code == 201, r.text
|
|
root = r.json()
|
|
wait_for(lambda: client.get(f"/api/v1/roots/{root['id']}").json()["baseline_state"] == "done")
|
|
return root
|
|
|
|
|
|
def test_temp_names_never_stored(env):
|
|
client, work = env
|
|
_add_root(client, work)
|
|
temps = ["a.tmp", "b.temp", "c.swp", "d.kate-swp", "e~", "f.part",
|
|
"g.crdownload", "~$lock.docx", "#autosave#", "4913", "h.orig"]
|
|
for name in temps:
|
|
(work / "src" / name).write_text("unfinished\n")
|
|
time.sleep(0.8) # past the 0.2s coalescing window: anything stored would show
|
|
for name in temps:
|
|
assert history(client, work / "src" / name) == [], name
|
|
# Same via the push API: every one must be refused, none stored.
|
|
for name in temps:
|
|
body = {"path": str(work / "src" / name),
|
|
"content_b64": base64.b64encode(b"unfinished\n").decode()}
|
|
assert client.post("/api/v1/snapshots", json=body).status_code == 422, name
|
|
|
|
|
|
def test_open_file_not_captured_until_close(env):
|
|
client, work = env
|
|
_add_root(client, work)
|
|
target = work / "src" / "slow.py"
|
|
fh = open(target, "w")
|
|
try:
|
|
fh.write("partial-1\n")
|
|
fh.flush()
|
|
os.fsync(fh.fileno()) # bytes on disk, but writer still holds it open
|
|
time.sleep(0.6)
|
|
assert history(client, target) == [] # no CLOSE_WRITE yet: nothing stored
|
|
fh.write("partial-2\n")
|
|
fh.flush()
|
|
finally:
|
|
fh.close() # only now is the state finished and capturable
|
|
wait_for(lambda: history(client, target))
|
|
versions = history(client, target)
|
|
assert len(versions) == 1
|
|
assert client.get(f"/api/v1/versions/{versions[0]['id']}/content").text == "partial-1\npartial-2\n"
|
|
|
|
|
|
def test_burst_collapses_to_first_and_last(env):
|
|
client, work = env
|
|
_add_root(client, work)
|
|
target = work / "src" / "busy.py"
|
|
target.write_text("v0\n")
|
|
wait_for(lambda: len(history(client, target)) == 1) # first observed state committed
|
|
for i in range(1, 6):
|
|
target.write_text(f"v{i}\n") # burst inside the coalescing window
|
|
wait_for(lambda: len(history(client, target)) == 2, timeout=8)
|
|
time.sleep(1.5) # past max_hold: no third version may appear
|
|
versions = history(client, target)
|
|
assert len(versions) == 2
|
|
bodies = [client.get(f"/api/v1/versions/{v['id']}/content").text for v in reversed(versions)]
|
|
assert bodies == ["v0\n", "v5\n"]
|
|
|
|
|
|
def test_atomic_save_is_one_version(env):
|
|
client, work = env
|
|
_add_root(client, work)
|
|
target = work / "src" / "app.py"
|
|
wait_for(lambda: history(client, target))
|
|
tmp = work / "src" / "app.py.tmp"
|
|
tmp.write_text("atomic\n")
|
|
os.replace(tmp, target)
|
|
wait_for(lambda: len(history(client, target)) == 2)
|
|
time.sleep(0.6)
|
|
assert len(history(client, target)) == 2 # temp state never stored
|
|
assert history(client, work / "src" / "app.py.tmp") == []
|