solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4from __future__ import annotations
5
6import hashlib
7import io
8import os
9import tarfile
10from dataclasses import dataclass
11from pathlib import Path
12from typing import Any
13
14import httpx
15import pytest
16
17from solstone.think.providers import rfdetr_install
18
19ENGINE_BYTES = b"rf-detr cli fixture bytes"
20MODEL_BYTES = b"rf-detr model fixture bytes"
21
22
23@dataclass(frozen=True)
24class _EngineFixture:
25 tarball_bytes: bytes
26 tarball_sha256: str
27 binary_sha256: str
28
29
30def _sha256(data: bytes) -> str:
31 return hashlib.sha256(data).hexdigest()
32
33
34@pytest.fixture
35def engine_tarball(tmp_path: Path) -> _EngineFixture:
36 src = (
37 tmp_path
38 / "engine-src"
39 / f"rfdetr-cli-{rfdetr_install.ENGINE_REF}-linux-cpu-x64"
40 )
41 src.mkdir(parents=True)
42 (src / rfdetr_install.ENGINE_BINARY_NAME).write_bytes(ENGINE_BYTES)
43 (src / "LICENSE").write_text("fixture license\n", encoding="utf-8")
44 (src / "PROVENANCE.txt").write_text("fixture provenance\n", encoding="utf-8")
45 tarball = tmp_path / rfdetr_install.RFDETR_SPEC.engine.tarball_name
46 with tarfile.open(tarball, "w:gz") as archive:
47 archive.add(src, arcname=src.name)
48 tarball_bytes = tarball.read_bytes()
49 return _EngineFixture(
50 tarball_bytes=tarball_bytes,
51 tarball_sha256=_sha256(tarball_bytes),
52 binary_sha256=_sha256(ENGINE_BYTES),
53 )
54
55
56def _fixture_spec(
57 engine: _EngineFixture,
58 *,
59 model_bytes: bytes = MODEL_BYTES,
60 tarball_sha256: str | None = None,
61 binary_sha256: str | None = None,
62 model_sha256: str | None = None,
63 model_size_bytes: int | None = None,
64) -> rfdetr_install.RfdetrSpec:
65 pinned = rfdetr_install.RFDETR_SPEC
66 return rfdetr_install.RfdetrSpec(
67 engine=rfdetr_install.RfdetrEngineSpec(
68 ref=pinned.engine.ref,
69 release_tag=pinned.engine.release_tag,
70 tarball_name=pinned.engine.tarball_name,
71 tarball_sha256=tarball_sha256 or engine.tarball_sha256,
72 binary_name=pinned.engine.binary_name,
73 binary_sha256=binary_sha256 or engine.binary_sha256,
74 ),
75 model=rfdetr_install.RfdetrModelSpec(
76 repo=pinned.model.repo,
77 revision=pinned.model.revision,
78 filename=pinned.model.filename,
79 sha256=model_sha256 or _sha256(model_bytes),
80 size_bytes=(
81 len(model_bytes) if model_size_bytes is None else model_size_bytes
82 ),
83 ),
84 )
85
86
87def _payloads(
88 spec: rfdetr_install.RfdetrSpec,
89 engine: _EngineFixture,
90 *,
91 model_bytes: bytes = MODEL_BYTES,
92) -> dict[str, bytes]:
93 return {
94 spec.engine.tarball_name: engine.tarball_bytes,
95 spec.model.filename: model_bytes,
96 }
97
98
99def _fake_download_factory(
100 payloads: dict[str, bytes],
101 calls: list[str] | None = None,
102):
103 def fake_download(
104 _url: str,
105 dest: Path,
106 file_spec: rfdetr_install.RfdetrFileSpec,
107 **_kwargs: Any,
108 ) -> None:
109 if calls is not None:
110 calls.append(file_spec.path)
111 dest.parent.mkdir(parents=True, exist_ok=True)
112 dest.write_bytes(payloads[file_spec.path])
113
114 return fake_download
115
116
117class _FakeResponse:
118 def __init__(self, payload: bytes) -> None:
119 self._payload = payload
120
121 def raise_for_status(self) -> None:
122 return
123
124 def iter_bytes(self):
125 midpoint = max(1, len(self._payload) // 2)
126 yield self._payload[:midpoint]
127 yield self._payload[midpoint:]
128
129
130class _FakeStream:
131 def __init__(self, response: _FakeResponse) -> None:
132 self.response = response
133
134 def __enter__(self) -> _FakeResponse:
135 return self.response
136
137 def __exit__(self, *_args: object) -> bool:
138 return False
139
140
141def _patch_stream_map(
142 monkeypatch: pytest.MonkeyPatch, payloads: dict[str, bytes]
143) -> None:
144 def fake_stream(_method: str, url: str, **_kwargs: Any) -> _FakeStream:
145 for needle, payload in payloads.items():
146 if needle in url:
147 return _FakeStream(_FakeResponse(payload))
148 raise AssertionError(f"unexpected download URL: {url}")
149
150 monkeypatch.setattr(httpx, "stream", fake_stream)
151
152
153def _tmp_path(path: Path) -> Path:
154 return path.with_name(f"{path.name}.tmp")
155
156
157def _assert_cleaned(spec: rfdetr_install.RfdetrSpec, journal_path: Path) -> None:
158 binary = rfdetr_install.binary_path(spec=spec, journal_path=journal_path)
159 model = rfdetr_install.model_path(spec=spec, journal_path=journal_path)
160 tarball = rfdetr_install._engine_tarball_path(spec, journal_path)
161 assert not binary.exists()
162 assert not _tmp_path(binary).exists()
163 assert not model.exists()
164 assert not _tmp_path(model).exists()
165 assert not tarball.exists()
166 assert not _tmp_path(tarball).exists()
167 assert not rfdetr_install._engine_extract_dir(spec, journal_path).exists()
168 assert not rfdetr_install.sidecar_path(journal_path=journal_path).exists()
169
170
171def _mark_supported(monkeypatch: pytest.MonkeyPatch) -> None:
172 monkeypatch.setattr(rfdetr_install, "_platform_info", lambda: ("linux", "x86_64"))
173
174
175def test_sidecar_round_trip() -> None:
176 record = rfdetr_install.RfdetrInstallRecord(
177 status="installed",
178 engine_ref=rfdetr_install.RFDETR_SPEC.engine.ref,
179 engine_sha256=rfdetr_install.RFDETR_SPEC.engine.binary_sha256,
180 model_repo=rfdetr_install.RFDETR_SPEC.model.repo,
181 model_revision=rfdetr_install.RFDETR_SPEC.model.revision,
182 model_file=rfdetr_install.RFDETR_SPEC.model.filename,
183 model_sha256=rfdetr_install.RFDETR_SPEC.model.sha256,
184 )
185 assert rfdetr_install.RfdetrInstallRecord.from_json(record.to_json()) == record
186
187 unavailable = rfdetr_install.RfdetrInstallRecord(status="platform_unavailable")
188 assert (
189 rfdetr_install.RfdetrInstallRecord.from_json(unavailable.to_json())
190 == unavailable
191 )
192
193
194def test_install_writes_files_and_sidecar(
195 tmp_path: Path, monkeypatch: pytest.MonkeyPatch, engine_tarball: _EngineFixture
196) -> None:
197 _mark_supported(monkeypatch)
198 spec = _fixture_spec(engine_tarball)
199 monkeypatch.setattr(
200 rfdetr_install,
201 "_download_file",
202 _fake_download_factory(_payloads(spec, engine_tarball)),
203 )
204
205 record = rfdetr_install.install_rfdetr(spec=spec, journal_path=tmp_path)
206
207 binary = rfdetr_install.binary_path(spec=spec, journal_path=tmp_path)
208 model = rfdetr_install.model_path(spec=spec, journal_path=tmp_path)
209 sidecar = rfdetr_install.sidecar_path(journal_path=tmp_path)
210 assert binary.read_bytes() == ENGINE_BYTES
211 assert os.access(binary, os.X_OK)
212 assert model.read_bytes() == MODEL_BYTES
213 assert sidecar.is_file()
214 assert record.status == "installed"
215 assert record == rfdetr_install.RfdetrInstallRecord.from_json(
216 sidecar.read_text(encoding="utf-8")
217 )
218
219
220def test_present_valid_install_is_noop(
221 tmp_path: Path, monkeypatch: pytest.MonkeyPatch, engine_tarball: _EngineFixture
222) -> None:
223 _mark_supported(monkeypatch)
224 spec = _fixture_spec(engine_tarball)
225 monkeypatch.setattr(
226 rfdetr_install,
227 "_download_file",
228 _fake_download_factory(_payloads(spec, engine_tarball)),
229 )
230 rfdetr_install.install_rfdetr(spec=spec, journal_path=tmp_path)
231 monkeypatch.setattr(
232 rfdetr_install,
233 "_download_file",
234 lambda *_args, **_kwargs: pytest.fail("download should not start"),
235 )
236
237 record = rfdetr_install.install_rfdetr(spec=spec, journal_path=tmp_path)
238
239 assert record.engine_ref == spec.engine.ref
240 assert record.model_file == spec.model.filename
241
242
243def test_check_valid_uses_no_network(
244 tmp_path: Path, monkeypatch: pytest.MonkeyPatch, engine_tarball: _EngineFixture
245) -> None:
246 _mark_supported(monkeypatch)
247 spec = _fixture_spec(engine_tarball)
248 monkeypatch.setattr(
249 rfdetr_install,
250 "_download_file",
251 _fake_download_factory(_payloads(spec, engine_tarball)),
252 )
253 rfdetr_install.install_rfdetr(spec=spec, journal_path=tmp_path)
254 monkeypatch.setattr(
255 rfdetr_install,
256 "_download_file",
257 lambda *_args, **_kwargs: pytest.fail("download should not start"),
258 )
259
260 assert (
261 rfdetr_install.check_rfdetr_model(spec=spec, journal_path=tmp_path).engine_ref
262 == spec.engine.ref
263 )
264
265
266def test_check_missing_raises_without_download(
267 tmp_path: Path, monkeypatch: pytest.MonkeyPatch, engine_tarball: _EngineFixture
268) -> None:
269 _mark_supported(monkeypatch)
270 spec = _fixture_spec(engine_tarball)
271 monkeypatch.setattr(
272 rfdetr_install,
273 "_download_file",
274 lambda *_args, **_kwargs: pytest.fail("download should not start"),
275 )
276
277 with pytest.raises(rfdetr_install.RfdetrInstallError) as exc_info:
278 rfdetr_install.check_rfdetr_model(spec=spec, journal_path=tmp_path)
279
280 assert exc_info.value.reason_code == "sidecar_missing"
281
282
283def test_force_refetches_when_present(
284 tmp_path: Path, monkeypatch: pytest.MonkeyPatch, engine_tarball: _EngineFixture
285) -> None:
286 _mark_supported(monkeypatch)
287 spec = _fixture_spec(engine_tarball)
288 monkeypatch.setattr(
289 rfdetr_install,
290 "_download_file",
291 _fake_download_factory(_payloads(spec, engine_tarball)),
292 )
293 rfdetr_install.install_rfdetr(spec=spec, journal_path=tmp_path)
294 calls: list[str] = []
295 monkeypatch.setattr(
296 rfdetr_install,
297 "_download_file",
298 _fake_download_factory(_payloads(spec, engine_tarball), calls),
299 )
300
301 rfdetr_install.install_rfdetr(force=True, spec=spec, journal_path=tmp_path)
302
303 assert calls == [spec.engine.tarball_name, spec.model.filename]
304
305
306def test_model_sha256_mismatch_cleans_partial_install(
307 tmp_path: Path, monkeypatch: pytest.MonkeyPatch, engine_tarball: _EngineFixture
308) -> None:
309 _mark_supported(monkeypatch)
310 spec = _fixture_spec(engine_tarball)
311 bad_model = bytes([MODEL_BYTES[0] ^ 1]) + MODEL_BYTES[1:]
312 _patch_stream_map(
313 monkeypatch,
314 _payloads(spec, engine_tarball, model_bytes=bad_model),
315 )
316
317 with pytest.raises(rfdetr_install.RfdetrInstallError) as exc_info:
318 rfdetr_install.install_rfdetr(spec=spec, journal_path=tmp_path)
319
320 assert exc_info.value.reason_code == "sha256_mismatch"
321 _assert_cleaned(spec, tmp_path)
322
323
324def test_tarball_sha256_mismatch_cleans_partial_install(
325 tmp_path: Path, monkeypatch: pytest.MonkeyPatch, engine_tarball: _EngineFixture
326) -> None:
327 _mark_supported(monkeypatch)
328 spec = _fixture_spec(engine_tarball, tarball_sha256="0" * 64)
329 _patch_stream_map(monkeypatch, _payloads(spec, engine_tarball))
330
331 with pytest.raises(rfdetr_install.RfdetrInstallError) as exc_info:
332 rfdetr_install.install_rfdetr(spec=spec, journal_path=tmp_path)
333
334 assert exc_info.value.reason_code == "sha256_mismatch"
335 _assert_cleaned(spec, tmp_path)
336
337
338def test_inner_binary_sha256_mismatch_cleans_partial_install(
339 tmp_path: Path, monkeypatch: pytest.MonkeyPatch, engine_tarball: _EngineFixture
340) -> None:
341 _mark_supported(monkeypatch)
342 spec = _fixture_spec(engine_tarball, binary_sha256="0" * 64)
343 _patch_stream_map(monkeypatch, _payloads(spec, engine_tarball))
344
345 with pytest.raises(rfdetr_install.RfdetrInstallError) as exc_info:
346 rfdetr_install.install_rfdetr(spec=spec, journal_path=tmp_path)
347
348 assert exc_info.value.reason_code == "sha256_mismatch"
349 _assert_cleaned(spec, tmp_path)
350
351
352def test_model_download_verifies_before_rename(
353 tmp_path: Path, monkeypatch: pytest.MonkeyPatch, engine_tarball: _EngineFixture
354) -> None:
355 spec = _fixture_spec(engine_tarball)
356 file_spec = rfdetr_install.RfdetrFileSpec(
357 spec.model.filename,
358 spec.model.sha256,
359 spec.model.size_bytes,
360 )
361 bad_model = bytes([MODEL_BYTES[0] ^ 1]) + MODEL_BYTES[1:]
362 _patch_stream_map(monkeypatch, {spec.model.filename: bad_model})
363 dest = tmp_path / spec.model.filename
364
365 with pytest.raises(rfdetr_install.RfdetrInstallError) as exc_info:
366 rfdetr_install._download_file(
367 f"https://example.test/{spec.model.filename}", dest, file_spec
368 )
369
370 assert exc_info.value.reason_code == "sha256_mismatch"
371 assert not dest.exists()
372 assert not _tmp_path(dest).exists()
373
374
375def test_extract_rejects_path_traversal(tmp_path: Path) -> None:
376 tarball = tmp_path / "bad.tar.gz"
377 with tarfile.open(tarball, "w:gz") as archive:
378 member = tarfile.TarInfo("../evil")
379 payload = b"bad"
380 member.size = len(payload)
381 archive.addfile(member, io.BytesIO(payload))
382
383 with pytest.raises(rfdetr_install.RfdetrInstallError) as exc_info:
384 rfdetr_install._safe_extract_tarball(tarball, tmp_path / "extract")
385
386 assert exc_info.value.reason_code == "archive_path_traversal"
387
388
389def test_platform_unavailable_writes_marker_no_download(
390 tmp_path: Path, monkeypatch: pytest.MonkeyPatch, engine_tarball: _EngineFixture
391) -> None:
392 monkeypatch.setattr(rfdetr_install, "_platform_info", lambda: ("linux", "aarch64"))
393 spec = _fixture_spec(engine_tarball)
394 monkeypatch.setattr(
395 rfdetr_install,
396 "_download_file",
397 lambda *_args, **_kwargs: pytest.fail("download should not start"),
398 )
399
400 record = rfdetr_install.install_rfdetr(spec=spec, journal_path=tmp_path)
401
402 assert record.status == "platform_unavailable"
403 sidecar = rfdetr_install.sidecar_path(journal_path=tmp_path)
404 assert sidecar.is_file()
405 assert (
406 rfdetr_install.RfdetrInstallRecord.from_json(
407 sidecar.read_text(encoding="utf-8")
408 ).status
409 == "platform_unavailable"
410 )
411 assert not rfdetr_install.binary_path(spec=spec, journal_path=tmp_path).exists()
412 assert not rfdetr_install.model_path(spec=spec, journal_path=tmp_path).exists()
413 assert (
414 rfdetr_install.check_rfdetr_model(spec=spec, journal_path=tmp_path).status
415 == "platform_unavailable"
416 )
417
418
419def test_rfdetr_paths_three_states(
420 tmp_path: Path, monkeypatch: pytest.MonkeyPatch, engine_tarball: _EngineFixture
421) -> None:
422 _mark_supported(monkeypatch)
423 spec = _fixture_spec(engine_tarball)
424 empty = tmp_path / "empty"
425 assert (
426 rfdetr_install.rfdetr_paths(spec=spec, journal_path=empty).status
427 == "not_installed"
428 )
429 monkeypatch.setattr(
430 rfdetr_install,
431 "_download_file",
432 _fake_download_factory(_payloads(spec, engine_tarball)),
433 )
434 rfdetr_install.install_rfdetr(spec=spec, journal_path=tmp_path)
435
436 installed = rfdetr_install.rfdetr_paths(spec=spec, journal_path=tmp_path)
437 assert installed.status == "installed"
438 assert installed.binary_path == rfdetr_install.binary_path(
439 spec=spec, journal_path=tmp_path
440 )
441 assert installed.model_path == rfdetr_install.model_path(
442 spec=spec, journal_path=tmp_path
443 )
444
445 monkeypatch.setattr(rfdetr_install, "_platform_info", lambda: ("linux", "aarch64"))
446 unavailable = rfdetr_install.rfdetr_paths(spec=spec, journal_path=tmp_path)
447 assert unavailable.status == "platform_unavailable"
448 assert unavailable.binary_path is None
449 assert unavailable.model_path is None
450
451
452def test_invalid_sidecar_shape_fails_check(
453 tmp_path: Path, monkeypatch: pytest.MonkeyPatch, engine_tarball: _EngineFixture
454) -> None:
455 _mark_supported(monkeypatch)
456 spec = _fixture_spec(engine_tarball)
457 sidecar = rfdetr_install.sidecar_path(journal_path=tmp_path)
458 sidecar.parent.mkdir(parents=True, exist_ok=True)
459 sidecar.write_text("{}\n", encoding="utf-8")
460
461 with pytest.raises(rfdetr_install.RfdetrInstallError) as exc_info:
462 rfdetr_install.check_rfdetr_model(spec=spec, journal_path=tmp_path)
463
464 assert exc_info.value.reason_code == "sidecar_invalid"