"""WebDAV backup against an in-memory WebDAV server.""" import base64 import json import time import httpx import pytest from fastapi.testclient import TestClient from versiond.api import create_app from versiond.config import Config, Paths from versiond.crypto import KeyRing from versiond.store import _codec from test_service import CONFIG, history, wait_for class FakeDAV: """Just enough RFC 4918 for the client: PROPFIND, MKCOL, PUT, GET, DELETE.""" def __init__(self, user="u1", password="secret"): self.auth = "Basic " + base64.b64encode(f"{user}:{password}".encode()).decode() self.dirs = {"/"} self.files: dict[str, bytes] = {} self.down = False self.requests = 0 @staticmethod def parent(path): return path.rsplit("/", 1)[0] or "/" def handler(self, request: httpx.Request) -> httpx.Response: self.requests += 1 if self.down: raise httpx.ConnectError("connection refused", request=request) if request.headers.get("authorization") != self.auth: return httpx.Response(401) path = request.url.path.rstrip("/") or "/" method = request.method if method == "PROPFIND": return httpx.Response(207 if path in self.dirs or path in self.files else 404) if method == "MKCOL": if path in self.dirs: return httpx.Response(405) if self.parent(path) not in self.dirs: return httpx.Response(409) self.dirs.add(path) return httpx.Response(201) if method == "PUT": if self.parent(path) not in self.dirs: return httpx.Response(409) if request.headers.get("if-none-match") == "*" and path in self.files: return httpx.Response(412) self.files[path] = request.read() return httpx.Response(201) if method == "GET": return httpx.Response(200, content=self.files[path]) if path in self.files else httpx.Response(404) if method == "DELETE": self.files.pop(path, None) return httpx.Response(204) return httpx.Response(405) @pytest.fixture def remote_env(tmp_path, monkeypatch): monkeypatch.setenv("VERSIOND_HOME", str(tmp_path / "home")) paths = Paths.resolve() paths.ensure() paths.config_file.write_text(CONFIG + "\n[upload]\nmanifest_interval_seconds = 0.2\n") cfg = Config.load(paths) dav = FakeDAV() project = tmp_path / "proj" project.mkdir() (project / "main.py").write_text("print('hello')\n") (project / ".env").write_text("SECRET=1\n") app = create_app(cfg, remote_transport=httpx.MockTransport(dav.handler)) with TestClient(app, base_url="http://127.0.0.1:9922") as client: client.headers["Authorization"] = f"Bearer {cfg.api_token()}" app.state.services.uploader.offline_sleep = 0.3 yield client, dav, project, cfg REMOTE = {"url": "https://dav.example", "username": "u1", "password": "secret", "base_path": "/versioned/"} def progress(client): return client.get("/api/v1/progress").json() def test_backup_to_webdav(remote_env): client, dav, project, cfg = remote_env bad = client.put("/api/v1/config/remote", json={**REMOTE, "password": "wrong"}) assert bad.status_code == 400 and "remote-auth" in bad.text r = client.put("/api/v1/config/remote", json=REMOTE) assert r.status_code == 200, r.text remote = r.json() directory = remote["directory"] assert directory.startswith("/versioned/") and remote["adopted"] is False assert "password" not in remote and remote["password_set"] is True owner = json.loads(dav.files[directory + "/meta/owner.json"]) assert owner["key_id"] == remote["key_id"] assert "secret" not in cfg.paths.config_file.read_text() # password only in credentials client.post("/api/v1/roots", json={"path": str(project)}) wait_for(lambda: progress(client)["upload"]["versions_local"] == 0 and progress(client)["upload"]["versions_durable"] == 2, timeout=10) keys = KeyRing.load_or_create(cfg.key_file) blob_paths = [p for p in dav.files if p.startswith(directory + "/blobs/")] contents = {_codec.decompress(keys.decrypt(dav.files[p])) for p in blob_paths} assert contents == {b"print('hello')\n", b"SECRET=1\n"} assert all(b"SECRET" not in dav.files[p] for p in blob_paths) # encrypted at rest manifests = [p for p in dav.files if p.startswith(directory + "/manifests/")] records = [json.loads(line) for p in manifests for line in _codec.decompress(keys.decrypt(dav.files[p])).decode().splitlines()] assert {r["path"] for r in records if r["type"] == "version"} == {str(project / "main.py"), str(project / ".env")} # reconfiguring the same machine adopts its own directory again = client.put("/api/v1/config/remote", json={**REMOTE, "password": None}) assert again.json()["directory"] == directory and again.json()["adopted"] is True # offline: changes wait locally, then upload when the server is back dav.down = True (project / "main.py").write_text("print('offline edit')\n") wait_for(lambda: progress(client)["upload"]["state"] == "offline", timeout=10) assert progress(client)["upload"]["versions_local"] == 1 dav.down = False wait_for(lambda: progress(client)["upload"]["versions_local"] == 0, timeout=15) assert progress(client)["upload"]["state"] == "online" def test_stats_progress_dashboard(remote_env): client, _, project, _ = remote_env root = client.post("/api/v1/roots", json={"path": str(project)}).json() wait_for(lambda: client.get(f"/api/v1/roots/{root['id']}").json()["baseline_state"] == "done") (project / "main.py").write_text("v2\n") wait_for(lambda: len(history(client, project / "main.py")) == 2) stats = client.get("/api/v1/stats").json() assert stats["files"]["tracked"] == 2 assert stats["versions"]["total"] == 3 assert stats["versions"]["by_source"] == {"scan": 2, "monitor": 1} assert sum(stats["versions"]["per_hour_last_24h"]) == 3 assert stats["top_files_7d"][0]["path"] == str(project / "main.py") p = progress(client) assert p["upload"]["state"] == "unconfigured" assert p["scans"]["last"][0]["percent"] == 100.0 assert p["monitor"]["counters"]["capture:committed"] == 1 page = client.get("/dashboard", headers={"Authorization": ""}) assert page.status_code == 200 and "versiond" in page.text with client.stream("GET", "/api/v1/progress/stream", params={"interval": 0.25, "max_events": 1}) as stream: for line in stream.iter_lines(): if line.startswith("data: "): assert "upload" in json.loads(line[6:]) break