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 json
8import shutil
9import tarfile
10from collections.abc import Iterator
11from contextlib import contextmanager
12from pathlib import Path
13
14import pytest
15
16from solstone.think.journal_io import LockTimeout
17from solstone.think.providers import nvattest_install
18
19
20def test_linux_x86_64_archive_pin_is_exact() -> None:
21 spec = nvattest_install.NVATTEST_ARCHIVES["linux-x86_64"]
22 expected_url = (
23 "https://updates.solstone.app/providers/nvattest/"
24 "libnvat-linux-x86_64-1.2.2-sol.1-archive.tar.xz"
25 )
26 legacy_sha = "3f10da6fca794b7e3025c6645447947ec8bc45bcfde5b5b1d23241c7115630db"
27
28 assert spec.version == "1.2.2-sol.1"
29 assert spec.url == expected_url
30 assert (
31 spec.sha256
32 == "60ef75d1873e7129f03ea80d107d92b2ef216d2a8815958617b30d9c721d474a"
33 )
34 source = Path(nvattest_install.__file__).read_text(encoding="utf-8")
35 assert "developer.download.nvidia.com" not in source
36 assert legacy_sha not in source
37
38
39def test_install_nvattest_reinstalls_partial_cache_without_sidecar(
40 tmp_path: Path,
41 monkeypatch: pytest.MonkeyPatch,
42) -> None:
43 root = nvattest_install.cache_root(tmp_path)
44 (root / "bin").mkdir(parents=True)
45 (root / "bin" / "nvattest").write_text("stale\n", encoding="utf-8")
46 (root / "lib").mkdir()
47
48 spec = _fixture_spec(tmp_path)
49 calls: list[Path] = []
50
51 def fake_download(_url: str, dest: Path, _expected_sha256: str) -> None:
52 calls.append(dest)
53 dest.parent.mkdir(parents=True, exist_ok=True)
54 shutil.copy2(tmp_path / spec.archive_name, dest)
55
56 monkeypatch.setattr(nvattest_install, "_download_file", fake_download)
57
58 installed = nvattest_install.install_nvattest(spec=spec, journal_path=tmp_path)
59
60 assert installed == root
61 assert len(calls) == 1
62 assert (root / "bin" / "nvattest").read_text(encoding="utf-8") == "new\n"
63 assert (root / "lib" / "libnvat.so").is_symlink()
64 assert (root / "lib" / "libnvat.so").readlink() == Path("libnvat.so.1")
65 sidecar = json.loads(
66 (root / nvattest_install.SIDECAR_NAME).read_text(encoding="utf-8")
67 )
68 assert sidecar["archive_sha256"] == spec.sha256
69 assert sidecar["version"] == spec.version
70
71
72def test_install_nvattest_reinstalls_stale_sidecar(
73 tmp_path: Path,
74 monkeypatch: pytest.MonkeyPatch,
75) -> None:
76 root = nvattest_install.cache_root(tmp_path)
77 (root / "bin").mkdir(parents=True)
78 (root / "bin" / "nvattest").write_text("old\n", encoding="utf-8")
79 (root / "lib").mkdir()
80 (root / nvattest_install.SIDECAR_NAME).write_text(
81 json.dumps({"archive_sha256": "old", "version": "0.0.0"}) + "\n",
82 encoding="utf-8",
83 )
84
85 spec = _fixture_spec(tmp_path)
86 calls: list[Path] = []
87
88 def fake_download(_url: str, dest: Path, _expected_sha256: str) -> None:
89 calls.append(dest)
90 dest.parent.mkdir(parents=True, exist_ok=True)
91 shutil.copy2(tmp_path / spec.archive_name, dest)
92
93 monkeypatch.setattr(nvattest_install, "_download_file", fake_download)
94
95 nvattest_install.install_nvattest(spec=spec, journal_path=tmp_path)
96
97 assert len(calls) == 1
98 assert (root / "bin" / "nvattest").read_text(encoding="utf-8") == "new\n"
99
100
101def test_install_nvattest_valid_sidecar_is_noop(
102 tmp_path: Path,
103 monkeypatch: pytest.MonkeyPatch,
104) -> None:
105 root = nvattest_install.cache_root(tmp_path)
106 (root / "bin").mkdir(parents=True)
107 (root / "bin" / "nvattest").write_text("installed\n", encoding="utf-8")
108 (root / "lib").mkdir()
109 (root / "share" / "ca").mkdir(parents=True)
110 (root / "share" / "ca" / "ca-bundle.pem").write_text("ca\n", encoding="utf-8")
111 spec = nvattest_install.NvattestArchiveSpec(
112 version="1.0.0",
113 url="https://example.invalid/nvattest.tar.gz",
114 archive_name="nvattest.tar.gz",
115 sha256="abc123",
116 )
117 (root / nvattest_install.SIDECAR_NAME).write_text(
118 json.dumps({"archive_sha256": spec.sha256, "version": spec.version}) + "\n",
119 encoding="utf-8",
120 )
121
122 monkeypatch.setattr(
123 nvattest_install,
124 "_download_file",
125 lambda *_args, **_kwargs: pytest.fail("download should not run"),
126 )
127
128 assert nvattest_install.install_nvattest(spec=spec, journal_path=tmp_path) == root
129
130
131def test_install_nvattest_upgrades_old_nvidia_sidecar(
132 tmp_path: Path,
133 monkeypatch: pytest.MonkeyPatch,
134) -> None:
135 root = nvattest_install.cache_root(tmp_path)
136 (root / "bin").mkdir(parents=True)
137 (root / "bin" / "nvattest").write_text("old\n", encoding="utf-8")
138 (root / "lib").mkdir()
139 (root / "share" / "ca").mkdir(parents=True)
140 (root / "share" / "ca" / "ca-bundle.pem").write_text("old-ca\n", encoding="utf-8")
141 (root / nvattest_install.SIDECAR_NAME).write_text(
142 json.dumps(
143 {
144 "archive_sha256": (
145 "3f10da6fca794b7e3025c6645447947ec8bc45bcfde5b5b1d23241c7115630db"
146 ),
147 "version": "1.2.2",
148 }
149 )
150 + "\n",
151 encoding="utf-8",
152 )
153 spec = _fixture_spec(tmp_path, version="1.2.2-sol.1")
154 calls: list[Path] = []
155
156 def fake_download(_url: str, dest: Path, _expected_sha256: str) -> None:
157 calls.append(dest)
158 dest.parent.mkdir(parents=True, exist_ok=True)
159 shutil.copy2(tmp_path / spec.archive_name, dest)
160
161 monkeypatch.setattr(nvattest_install, "_download_file", fake_download)
162
163 nvattest_install.install_nvattest(spec=spec, journal_path=tmp_path)
164
165 assert len(calls) == 1
166 assert (root / "bin" / "nvattest").read_text(encoding="utf-8") == "new\n"
167 sidecar = json.loads(
168 (root / nvattest_install.SIDECAR_NAME).read_text(encoding="utf-8")
169 )
170 assert sidecar["version"] == "1.2.2-sol.1"
171
172
173def test_install_nvattest_rejects_hash_mismatch_without_partial_tree(
174 tmp_path: Path,
175 monkeypatch: pytest.MonkeyPatch,
176) -> None:
177 class FakeResponse:
178 def __enter__(self):
179 return self
180
181 def __exit__(self, *_exc_info) -> bool:
182 return False
183
184 def raise_for_status(self) -> None:
185 return None
186
187 def iter_bytes(self):
188 yield b"not the pinned archive"
189
190 def fake_stream(*_args, **_kwargs):
191 return FakeResponse()
192
193 monkeypatch.setattr("httpx.stream", fake_stream)
194 spec = nvattest_install.NvattestArchiveSpec(
195 version="1.2.2-sol.1",
196 url="https://example.invalid/nvattest.tar.xz",
197 archive_name="nvattest.tar.xz",
198 sha256="0" * 64,
199 )
200
201 with pytest.raises(nvattest_install.NvattestInstallError) as exc_info:
202 nvattest_install.install_nvattest(spec=spec, journal_path=tmp_path)
203
204 assert exc_info.value.reason_code == "sha256_mismatch"
205 root = nvattest_install.cache_root(tmp_path)
206 assert not (root / "bin").exists()
207 assert not (root / "lib").exists()
208 assert not (root / "share").exists()
209 assert not (root / ".downloads" / "nvattest.tar.xz").exists()
210 assert not (root / ".downloads" / "nvattest.tar.xz.tmp").exists()
211
212
213def test_ensure_nvattest_unsupported_platform_does_not_touch_cache(
214 tmp_path: Path,
215 monkeypatch: pytest.MonkeyPatch,
216) -> None:
217 monkeypatch.setattr(nvattest_install, "nvattest_archive_key", lambda: None)
218
219 result = nvattest_install.ensure_nvattest_installed(journal_path=tmp_path)
220
221 assert result.status == "platform_unsupported"
222 assert result.reason_code == "platform_unsupported"
223 assert not nvattest_install.cache_root(tmp_path).exists()
224
225
226def test_ensure_nvattest_lock_timeout_is_in_flight(
227 tmp_path: Path,
228 monkeypatch: pytest.MonkeyPatch,
229) -> None:
230 @contextmanager
231 def fake_hold_lock(path: Path, *, timeout: float, **_kwargs) -> Iterator[None]:
232 raise LockTimeout(path, timeout)
233 yield
234
235 monkeypatch.setattr(nvattest_install, "hold_lock", fake_hold_lock)
236
237 result = nvattest_install.ensure_nvattest_installed(journal_path=tmp_path)
238
239 assert result.status == "install_in_flight"
240 assert result.reason_code == "install-in-progress"
241
242
243def test_ensure_nvattest_override_skips_cache_download(
244 tmp_path: Path,
245 monkeypatch: pytest.MonkeyPatch,
246) -> None:
247 override = tmp_path / "override"
248 monkeypatch.setenv(nvattest_install.SPP_NVATTEST_DIR_ENV, str(override))
249 monkeypatch.setattr(
250 nvattest_install,
251 "_download_file",
252 lambda *_args, **_kwargs: pytest.fail("download should not run"),
253 )
254
255 result = nvattest_install.ensure_nvattest_installed(journal_path=tmp_path)
256
257 assert result.status == "already_installed"
258 assert result.nvattest_dir == override
259 assert not nvattest_install.cache_root(tmp_path).exists()
260
261
262def test_install_nvattest_accepts_wrapped_archive(
263 tmp_path: Path,
264 monkeypatch: pytest.MonkeyPatch,
265) -> None:
266 spec = _fixture_spec(tmp_path, wrapped=True)
267
268 def fake_download(_url: str, dest: Path, _expected_sha256: str) -> None:
269 dest.parent.mkdir(parents=True, exist_ok=True)
270 shutil.copy2(tmp_path / spec.archive_name, dest)
271
272 monkeypatch.setattr(nvattest_install, "_download_file", fake_download)
273
274 installed = nvattest_install.install_nvattest(spec=spec, journal_path=tmp_path)
275
276 assert (installed / "bin" / "nvattest").read_text(encoding="utf-8") == "new\n"
277
278
279def _fixture_spec(
280 tmp_path: Path,
281 *,
282 version: str = "9.9.9",
283 wrapped: bool = False,
284) -> nvattest_install.NvattestArchiveSpec:
285 archive_name = "nvattest-fixture.tar.gz"
286 source = tmp_path / "source" / "nvattest-fixture"
287 (source / "bin").mkdir(parents=True)
288 (source / "bin" / "nvattest").write_text("new\n", encoding="utf-8")
289 (source / "lib").mkdir()
290 (source / "lib" / "libnvat.so.1").write_text("library\n", encoding="utf-8")
291 (source / "lib" / "libnvat.so").symlink_to("libnvat.so.1")
292 (source / "LICENSE").write_text("license\n", encoding="utf-8")
293 (source / "share" / "ca").mkdir(parents=True)
294 (source / "share" / "ca" / "ca-bundle.pem").write_text("ca\n", encoding="utf-8")
295 (source / "share" / "THIRD_PARTY_NOTICES.md").write_text(
296 "notices\n",
297 encoding="utf-8",
298 )
299
300 archive_path = tmp_path / archive_name
301 with tarfile.open(archive_path, "w:gz") as archive:
302 if wrapped:
303 archive.add(source, arcname=source.name)
304 else:
305 for child in source.iterdir():
306 archive.add(child, arcname=child.name)
307 return nvattest_install.NvattestArchiveSpec(
308 version=version,
309 url="https://example.invalid/nvattest-fixture.tar.gz",
310 archive_name=archive_name,
311 sha256=hashlib.sha256(archive_path.read_bytes()).hexdigest(),
312 )