333 lines
12 KiB
Python
333 lines
12 KiB
Python
"""The reference client in client/dsh_sync_client.py.
|
|
|
|
Run against a stub HTTP server rather than the real app so these stay fast and
|
|
can inject failures (corrupt artifacts, 409s, dropped connections) that are
|
|
awkward to provoke through the live API.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import sys
|
|
import tarfile
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from pathlib import Path
|
|
from threading import Thread
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "client"))
|
|
from dsh_sync_client import DshSync, SyncError, _parse_value, semver_tuple # noqa: E402
|
|
|
|
|
|
def _tar(plugin_id="demo", version="1.0.0", files=("manifest.json", "files/main.py")):
|
|
buf = io.BytesIO()
|
|
with tarfile.open(fileobj=buf, mode="w:gz") as tf:
|
|
manifest = json.dumps({"id": plugin_id, "version": version}).encode()
|
|
for name in files:
|
|
blob = manifest if name.endswith("manifest.json") else b"print('hi')\n"
|
|
info = tarfile.TarInfo(name)
|
|
info.size = len(blob)
|
|
tf.addfile(info, io.BytesIO(blob))
|
|
return buf.getvalue()
|
|
|
|
|
|
class StubServer:
|
|
"""Minimal dsh-sync surface; `routes` maps (method, path) -> callable."""
|
|
|
|
def __init__(self):
|
|
self.routes: dict[tuple[str, str], callable] = {}
|
|
self.requests: list[tuple[str, str, dict]] = []
|
|
outer = self
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *a):
|
|
pass
|
|
|
|
def _go(self, method):
|
|
body = b""
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
if length:
|
|
body = self.rfile.read(length)
|
|
parsed = json.loads(body) if body else None
|
|
outer.requests.append((method, self.path, {"body": parsed,
|
|
"headers": dict(self.headers)}))
|
|
fn = outer.routes.get((method, self.path.split("?")[0]))
|
|
if fn is None:
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
self.wfile.write(b'{"detail":"no route"}')
|
|
return
|
|
status, payload, headers = fn(self.path, parsed, dict(self.headers))
|
|
blob = payload if isinstance(payload, bytes) else json.dumps(payload).encode()
|
|
self.send_response(status)
|
|
for k, v in (headers or {}).items():
|
|
self.send_header(k, v)
|
|
self.send_header("Content-Length", str(len(blob)))
|
|
self.end_headers()
|
|
self.wfile.write(blob)
|
|
|
|
def do_GET(self):
|
|
self._go("GET")
|
|
|
|
def do_POST(self):
|
|
self._go("POST")
|
|
|
|
def do_PUT(self):
|
|
self._go("PUT")
|
|
|
|
self.httpd = HTTPServer(("127.0.0.1", 0), Handler)
|
|
self.thread = Thread(target=self.httpd.serve_forever, daemon=True)
|
|
self.thread.start()
|
|
|
|
@property
|
|
def url(self) -> str:
|
|
return f"http://127.0.0.1:{self.httpd.server_port}"
|
|
|
|
def stop(self):
|
|
self.httpd.shutdown()
|
|
self.httpd.server_close()
|
|
|
|
|
|
@pytest.fixture()
|
|
def server():
|
|
s = StubServer()
|
|
yield s
|
|
s.stop()
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(server, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("DSH_STATE_DIR", str(tmp_path / "state"))
|
|
c = DshSync(tmp_path / "state")
|
|
c.state.update(base_url=server.url, token="dsh_test", user={"id": 1, "username": "u"})
|
|
c._save_state()
|
|
return c
|
|
|
|
|
|
# ------------------------------------------------------------------- helpers
|
|
|
|
|
|
def test_semver_tuple_orders_numerically():
|
|
assert semver_tuple("1.10.0") > semver_tuple("1.9.0")
|
|
assert semver_tuple("2.0.0-beta") > semver_tuple("1.99.99")
|
|
assert semver_tuple("1") == (1, 0, 0)
|
|
|
|
|
|
def test_parse_value_prefers_json_then_falls_back_to_text():
|
|
assert _parse_value("0.7") == 0.7
|
|
assert _parse_value('{"a":1}') == {"a": 1}
|
|
assert _parse_value("dark") == "dark"
|
|
|
|
|
|
# --------------------------------------------------------------------- login
|
|
|
|
|
|
def test_login_stores_token_and_user(server, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("DSH_STATE_DIR", str(tmp_path / "s"))
|
|
server.routes[("POST", "/v1/tokens")] = lambda p, b, h: (
|
|
200, {"token": "dsh_abc", "user": {"id": 5, "username": "alice"}}, {})
|
|
c = DshSync(tmp_path / "s")
|
|
out = c.login(server.url, "invite", "alice")
|
|
assert out["token"] == "dsh_abc"
|
|
assert c.state["token"] == "dsh_abc"
|
|
assert DshSync(tmp_path / "s").state["token"] == "dsh_abc" # persisted
|
|
|
|
|
|
def test_login_reports_bad_invite(server, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("DSH_STATE_DIR", str(tmp_path / "s"))
|
|
server.routes[("POST", "/v1/tokens")] = lambda p, b, h: (403, {"detail": "bad"}, {})
|
|
with pytest.raises(SyncError, match="login failed"):
|
|
DshSync(tmp_path / "s").login(server.url, "wrong", "alice")
|
|
|
|
|
|
def test_request_without_auth_is_refused(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("DSH_STATE_DIR", str(tmp_path / "s"))
|
|
with pytest.raises(SyncError, match="not logged in"):
|
|
DshSync(tmp_path / "s").pull()
|
|
|
|
|
|
# ---------------------------------------------------------------------- pull
|
|
|
|
|
|
def test_pull_returns_values_and_caches_etag(server, client):
|
|
server.routes[("GET", "/v1/settings")] = lambda p, b, h: (
|
|
200,
|
|
{"settings": {"model.temperature": {"value": 0.7, "version": 3, "scope": "user"}}},
|
|
{"ETag": '"abc"'},
|
|
)
|
|
assert client.pull() == {"model.temperature": 0.7}
|
|
assert client.state["settings_etag"] == '"abc"'
|
|
assert client.state["settings_versions"]["model.temperature"]["version"] == 3
|
|
|
|
|
|
def test_pull_sends_if_none_match_on_second_call(server, client):
|
|
calls = []
|
|
|
|
def handler(p, b, h):
|
|
calls.append(h.get("If-None-Match"))
|
|
if len(calls) == 1:
|
|
return 200, {"settings": {"k": {"value": 1, "version": 1, "scope": "user"}}}, {"ETag": '"e1"'}
|
|
return 304, b"", {"ETag": '"e1"'}
|
|
|
|
server.routes[("GET", "/v1/settings")] = handler
|
|
client.pull()
|
|
client.pull()
|
|
assert calls[0] is None
|
|
assert calls[1] == '"e1"'
|
|
|
|
|
|
def test_pull_falls_back_to_cache_when_offline(server, client):
|
|
server.routes[("GET", "/v1/settings")] = lambda p, b, h: (
|
|
200, {"settings": {"k": {"value": "cached", "version": 1, "scope": "user"}}}, {})
|
|
client.pull()
|
|
server.stop() # server goes away
|
|
assert client.pull() == {"k": "cached"}
|
|
assert client.state["stale"] is True
|
|
|
|
|
|
def test_pull_without_cache_and_server_down_raises(server, client):
|
|
server.stop()
|
|
with pytest.raises(SyncError, match="unreachable"):
|
|
client.pull()
|
|
|
|
|
|
# ---------------------------------------------------------------------- push
|
|
|
|
|
|
def test_push_sends_base_version_from_cache(server, client):
|
|
server.routes[("GET", "/v1/settings")] = lambda p, b, h: (
|
|
200, {"settings": {"k": {"value": 1, "version": 4, "scope": "user"}}}, {})
|
|
client.pull()
|
|
sent = {}
|
|
|
|
def put(path, body, headers):
|
|
sent.update(body)
|
|
return 200, {"key": "k", "version": 5}, {}
|
|
|
|
server.routes[("PUT", "/v1/settings/user/k")] = put
|
|
client.push("k", 2)
|
|
assert sent["base_version"] == 4
|
|
assert sent["value"] == 2
|
|
|
|
|
|
def test_push_without_cache_omits_base_version(server, client):
|
|
sent = {}
|
|
|
|
def put(path, body, headers):
|
|
sent.update(body)
|
|
return 200, {"key": "new", "version": 1}, {}
|
|
|
|
server.routes[("PUT", "/v1/settings/user/new")] = put
|
|
client.push("new", "v")
|
|
assert "base_version" not in sent
|
|
|
|
|
|
def test_push_surfaces_conflict_as_readable_error(server, client):
|
|
server.routes[("PUT", "/v1/settings/user/k")] = lambda p, b, h: (
|
|
409, {"detail": {"message": "version conflict"}}, {})
|
|
with pytest.raises(SyncError, match="conflict on 'k'"):
|
|
client.push("k", 1)
|
|
|
|
|
|
# ------------------------------------------------------------------- install
|
|
|
|
|
|
def _install_routes(server, blob, *, sha=None, size=None, version="1.0.0"):
|
|
digest = sha if sha is not None else hashlib.sha256(blob).hexdigest()
|
|
server.routes[("GET", "/v1/plugins")] = lambda p, b, h: (
|
|
200, {"plugins": [{"id": "demo", "latest": {
|
|
"version": version, "sha256": digest,
|
|
"size_bytes": size if size is not None else len(blob),
|
|
"min_harness_version": "0.0.0"}, "versions": []}]}, {})
|
|
server.routes[("GET", f"/v1/plugins/demo/{version}/download")] = lambda p, b, h: (
|
|
200, blob, {"X-Sha256": digest})
|
|
|
|
|
|
def test_install_verifies_and_unpacks(server, client, tmp_path):
|
|
_install_routes(server, _tar())
|
|
target = client.install("demo", target_root=tmp_path / "plugins")
|
|
assert (target / "manifest.json").is_file()
|
|
assert (target / "files" / "main.py").is_file()
|
|
|
|
|
|
def test_install_rejects_sha256_mismatch(server, client, tmp_path):
|
|
_install_routes(server, _tar(), sha="0" * 64)
|
|
with pytest.raises(SyncError, match="sha256 mismatch"):
|
|
client.install("demo", target_root=tmp_path / "plugins")
|
|
|
|
|
|
def test_install_rejects_size_mismatch(server, client, tmp_path):
|
|
_install_routes(server, _tar(), size=999999)
|
|
with pytest.raises(SyncError, match="size mismatch"):
|
|
client.install("demo", target_root=tmp_path / "plugins")
|
|
|
|
|
|
def test_install_unknown_plugin(server, client, tmp_path):
|
|
server.routes[("GET", "/v1/plugins")] = lambda p, b, h: (200, {"plugins": []}, {})
|
|
with pytest.raises(SyncError, match="not found"):
|
|
client.install("nope", target_root=tmp_path / "plugins")
|
|
|
|
|
|
def test_install_rejects_path_traversal(server, client, tmp_path):
|
|
_install_routes(server, _tar(files=("manifest.json", "../../evil.py")))
|
|
with pytest.raises(SyncError, match="unsafe path"):
|
|
client.install("demo", target_root=tmp_path / "plugins")
|
|
|
|
|
|
def test_install_replaces_existing_atomically(server, client, tmp_path):
|
|
root = tmp_path / "plugins"
|
|
stale = root / "demo"
|
|
stale.mkdir(parents=True)
|
|
(stale / "old.txt").write_text("stale", encoding="utf-8")
|
|
|
|
_install_routes(server, _tar())
|
|
client.install("demo", target_root=root)
|
|
assert not (root / "demo" / "old.txt").exists()
|
|
assert (root / "demo" / "manifest.json").is_file()
|
|
assert not list(root.glob(".*")) # temp/bak dirs cleaned up
|
|
|
|
|
|
def test_install_honours_min_harness_version(server, client, tmp_path):
|
|
server.routes[("GET", "/v1/plugins")] = lambda p, b, h: (
|
|
200, {"plugins": [{"id": "demo", "versions": [], "latest": {
|
|
"version": "9.0.0", "sha256": "x", "size_bytes": 1,
|
|
"min_harness_version": "5.0.0"}}]}, {})
|
|
with pytest.raises(SyncError, match="requires harness"):
|
|
client.install("demo", harness_version="1.0.0", target_root=tmp_path / "p")
|
|
|
|
|
|
# -------------------------------------------------------------------- status
|
|
|
|
|
|
def test_status_reports_only_safe_fields(client):
|
|
s = client.status()
|
|
assert set(s) == {"base_url", "user", "stale", "last_pull", "settings_etag"}
|
|
assert "token" not in s
|
|
|
|
|
|
def test_wrapped_archive_has_its_wrapper_stripped(server, client, tmp_path):
|
|
"""`plugin-x/manifest.json` + `plugin-x/files/y` must land flat."""
|
|
buf = io.BytesIO()
|
|
with tarfile.open(fileobj=buf, mode="w:gz") as tf:
|
|
for name, blob in (("demo-1.0.0/manifest.json", b'{"id":"demo"}'),
|
|
("demo-1.0.0/files/main.py", b"print(1)\n")):
|
|
info = tarfile.TarInfo(name)
|
|
info.size = len(blob)
|
|
tf.addfile(info, io.BytesIO(blob))
|
|
_install_routes(server, buf.getvalue())
|
|
target = client.install("demo", target_root=tmp_path / "plugins")
|
|
assert (target / "manifest.json").is_file()
|
|
assert (target / "files" / "main.py").is_file(), "wrapper must not eat files/"
|
|
|
|
|
|
def test_push_succeeds_even_when_refresh_fails(server, client, tmp_path):
|
|
"""The write landed; a broken refresh must not look like a failed push."""
|
|
server.routes[("PUT", "/v1/settings/user/k")] = lambda p, b, h: (
|
|
200, {"key": "k", "version": 1}, {})
|
|
# No GET /v1/settings route -> refresh 404s.
|
|
out = client.push("k", "v")
|
|
assert out["version"] == 1
|
|
assert client.state["stale"] is True
|