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 os
7import subprocess
8from collections.abc import Sequence
9from datetime import UTC, datetime
10from pathlib import Path
11
12import pytest
13
14import scripts.check_rust_release_manifest as checker
15import scripts.release_advisory_policy as policy
16
17DB_COMMIT = "a" * 40
18DB_ARCHIVE = "b" * 64
19SNAPSHOT = "advisory-db-1234567890abcdef"
20POLICY_TIME = datetime(2026, 7, 20, 12, 0, tzinfo=UTC)
21FETCH_TIME = datetime(2026, 7, 20, 11, 30, tzinfo=UTC)
22COMMIT_TIME = "2026-07-17T15:52:38Z"
23
24MALFORMED_DB_COMMIT_CASES = (
25 ("short-39", "a" * 39),
26 ("short-63", "a" * 63),
27 ("long-41", "a" * 41),
28 ("long-65", "a" * 65),
29 ("uppercase", "A" * 40),
30 ("non-hex", "g" * 40),
31 ("empty", ""),
32 ("whitespace", " " + "a" * 40),
33 ("extra-line", "a" * 40 + "\nunexpected\n"),
34)
35MALFORMED_ARCHIVE_DIGEST_CASES = (
36 ("short", "b" * 63),
37 ("uppercase", "B" * 64),
38 ("non-hex", "g" * 64),
39 ("empty", ""),
40 ("extra-line", "b" * 64 + "\nunexpected"),
41)
42MALFORMED_DB_COMMITS = tuple(
43 pytest.param(value, id=name) for name, value in MALFORMED_DB_COMMIT_CASES
44)
45MALFORMED_ARCHIVE_DIGESTS = tuple(
46 pytest.param(value, id=name) for name, value in MALFORMED_ARCHIVE_DIGEST_CASES
47)
48
49
50class FakeRunner:
51 def __init__(
52 self,
53 snapshot: Path,
54 *,
55 status: str | Sequence[str] = "",
56 fail_check: bool = False,
57 commit_stdout: str = DB_COMMIT + "\n",
58 commit_timestamp: str = COMMIT_TIME + "\n",
59 top_level: Path | None = None,
60 scanned_path: Path | None = None,
61 debug_stderr: str | None = None,
62 ) -> None:
63 self.snapshot = snapshot
64 self.status_outputs = list(status) if not isinstance(status, str) else [status]
65 self.fail_check = fail_check
66 self.commit_stdout = commit_stdout
67 self.commit_timestamp = commit_timestamp
68 self.top_level = top_level
69 self.scanned_path = scanned_path or snapshot
70 self.debug_stderr = debug_stderr
71 self.events: list[str] = []
72 self.config_bytes: bytes | None = None
73 self.cargo_cwds: list[Path | None] = []
74 self.cargo_argvs: list[list[str]] = []
75
76 def __call__(self, argv, **kwargs) -> subprocess.CompletedProcess[str]:
77 command = list(argv)
78 if command[0] == "cargo-deny":
79 if command[-2:] == ["fetch", "db"]:
80 raise AssertionError("cargo-deny fetch subcommand should be gone")
81 if command[-2:] == ["check", "advisories"]:
82 self.events.append("check")
83 self.cargo_cwds.append(kwargs.get("cwd"))
84 self.cargo_argvs.append(command)
85 self.config_bytes = Path(
86 command[command.index("--config") + 1]
87 ).read_bytes()
88 if self.fail_check:
89 return subprocess.CompletedProcess(command, 1, "", "denied")
90 stderr = self.debug_stderr
91 if stderr is None:
92 stderr = (
93 "2026-07-21 14:40:36 [DEBUG] "
94 f"Opening advisory database at '{self.scanned_path}'\n"
95 )
96 return subprocess.CompletedProcess(command, 0, "", stderr)
97 if command[:2] == ["git", "-C"]:
98 git_root = Path(command[2])
99 subcommand = command[3:]
100 if subcommand == ["rev-parse", "--show-toplevel"]:
101 self.events.append("show-toplevel")
102 top_level = self.top_level or git_root
103 return subprocess.CompletedProcess(command, 0, f"{top_level}\n", "")
104 if subcommand[:1] == ["status"]:
105 self.events.append("status")
106 output = self.status_outputs.pop(0)
107 if not self.status_outputs:
108 self.status_outputs.append(output)
109 return subprocess.CompletedProcess(command, 0, output, "")
110 if subcommand[:2] == ["rev-parse", "--verify"]:
111 self.events.append("rev-parse")
112 return subprocess.CompletedProcess(command, 0, self.commit_stdout, "")
113 if subcommand == ["show", "-s", "--format=%cI", "HEAD"]:
114 self.events.append("commit-timestamp")
115 return subprocess.CompletedProcess(
116 command, 0, self.commit_timestamp, ""
117 )
118 raise AssertionError(f"unexpected command: {command}")
119
120
121def _repo(tmp_path: Path, deny_text: str | None = None) -> Path:
122 root = tmp_path / "repo"
123 (root / "core").mkdir(parents=True, exist_ok=True)
124 (root / "core" / "deny.toml").write_text(
125 deny_text or '[licenses]\nallow = ["MIT"]\n',
126 encoding="utf-8",
127 )
128 return root
129
130
131def _clock(events: list[str] | None = None, value: datetime = POLICY_TIME):
132 def now() -> datetime:
133 if events is not None:
134 events.append("policy-clock")
135 return value
136
137 return now
138
139
140def _write_snapshot(
141 db_root: Path,
142 *,
143 name: str = SNAPSHOT,
144 advisory_count: int = 1,
145 fetch_time: datetime | None = FETCH_TIME,
146) -> Path:
147 snapshot = db_root / name
148 (snapshot / ".git").mkdir(parents=True)
149 for index in range(advisory_count):
150 advisory = (
151 snapshot / "crates" / f"probe{index}" / f"RUSTSEC-2026-{index:04d}.md"
152 )
153 advisory.parent.mkdir(parents=True, exist_ok=True)
154 advisory.write_text(
155 "```toml\n"
156 "[advisory]\n"
157 f'id = "RUSTSEC-2026-{index:04d}"\n'
158 f'package = "probe{index}"\n'
159 'date = "2026-01-01"\n'
160 'url = "https://example.invalid/RUSTSEC-2026-0001"\n'
161 'categories = ["unmaintained"]\n'
162 "keywords = []\n\n"
163 "[versions]\n"
164 "patched = []\n"
165 "```\n",
166 encoding="utf-8",
167 )
168 if fetch_time is not None:
169 fetch_head = snapshot / ".git" / "FETCH_HEAD"
170 fetch_head.write_text("", encoding="utf-8")
171 timestamp = fetch_time.timestamp()
172 os.utime(fetch_head, (timestamp, timestamp))
173 return snapshot
174
175
176def _db_root(
177 tmp_path: Path,
178 *,
179 advisory_count: int = 1,
180 fetch_time: datetime | None = FETCH_TIME,
181) -> tuple[Path, Path]:
182 root = tmp_path / "db-root"
183 root.mkdir()
184 (root / "db.lock").write_text("", encoding="utf-8")
185 return root, _write_snapshot(
186 root,
187 advisory_count=advisory_count,
188 fetch_time=fetch_time,
189 )
190
191
192def _prepare(
193 tmp_path: Path,
194 *,
195 runner: FakeRunner | None = None,
196 db_root: Path | None = None,
197 snapshot: Path | None = None,
198 clock=None,
199 archive: str = DB_ARCHIVE,
200 cleanup_unlink=policy._unlink_path,
201 cleanup_rmdir=policy._remove_dir,
202) -> tuple[policy.PolicyRun, FakeRunner, Path]:
203 repo = _repo(tmp_path)
204 if db_root is None or snapshot is None:
205 db_root, snapshot = _db_root(tmp_path)
206 runner = runner or FakeRunner(snapshot)
207 result = policy.prepare_policy_run(
208 repo,
209 advisory_source_id="internal",
210 db_urls=("ssh://example.test/db.git",),
211 db_root=db_root,
212 runner=runner,
213 temp_path_factory=lambda label: tmp_path / label,
214 clock=clock or _clock(runner.events),
215 archive_hasher=lambda observed: (
216 runner.events.append(f"archive:{observed.name}") or archive
217 ),
218 cleanup_unlink=cleanup_unlink,
219 cleanup_rmdir=cleanup_rmdir,
220 )
221 return result, runner, repo
222
223
224def test_materialized_config_and_advisory_check_argv(tmp_path: Path) -> None:
225 result, runner, repo = _prepare(tmp_path)
226
227 assert result.result == "pass"
228 assert result.db_snapshot_basename == SNAPSHOT
229 assert result.advisory_count == 1
230 assert result.advisory_acquired_at == "2026-07-20T11:30:00Z"
231 assert result.db_commit_timestamp == COMMIT_TIME
232 assert runner.config_bytes is not None
233 text = runner.config_bytes.decode("utf-8")
234 assert text.startswith('[licenses]\nallow = ["MIT"]\n\n[advisories]\n')
235 assert 'db-urls = ["ssh://example.test/db.git"]' in text
236 assert "git-fetch-with-cli = true" in text
237 assert 'maximum-db-staleness = "PT24H"' in text
238 assert "24 hours" not in text
239
240 argv = runner.cargo_argvs[0]
241 assert argv == [
242 "cargo-deny",
243 "--config",
244 argv[2],
245 "--manifest-path",
246 str(repo / "core" / "Cargo.toml"),
247 "-L",
248 "debug",
249 "--locked",
250 "--offline",
251 "check",
252 "advisories",
253 ]
254 assert "fetch" not in runner.events
255 assert runner.events == [
256 "show-toplevel",
257 "status",
258 "rev-parse",
259 f"archive:{SNAPSHOT}",
260 "commit-timestamp",
261 "check",
262 "status",
263 "policy-clock",
264 ]
265 assert runner.cargo_cwds == [repo]
266
267
268def test_core_deny_toml_advisories_table_fails_loudly(tmp_path: Path) -> None:
269 db_root, _snapshot = _db_root(tmp_path)
270 with pytest.raises(policy.ReleasePolicyError) as exc:
271 policy.prepare_policy_run(
272 _repo(tmp_path, '[advisories]\ndb-path = "x"\n'),
273 advisory_source_id="internal-feed",
274 db_urls=("ssh://example.test/advisory-db.git",),
275 db_root=db_root,
276 runner=FakeRunner(db_root / SNAPSHOT),
277 temp_path_factory=lambda label: tmp_path / label,
278 )
279
280 assert exc.value.failures[0].error == "core deny.toml already defines advisories"
281
282
283@pytest.mark.parametrize(
284 "source_id,db_urls,error",
285 [
286 ("", ("ssh://example.test/db.git",), "advisory source id is not a public slug"),
287 ("internal", (), "advisory db source is empty"),
288 (
289 "internal",
290 ("https://github.com/RustSec/advisory-db",),
291 "advisory db url points at GitHub",
292 ),
293 (
294 "internal",
295 ("ssh://mirror.github.com/RustSec/db",),
296 "advisory db url points at GitHub",
297 ),
298 ],
299)
300def test_empty_and_github_advisory_sources_are_rejected(
301 tmp_path: Path,
302 source_id: str,
303 db_urls: tuple[str, ...],
304 error: str,
305) -> None:
306 with pytest.raises(policy.ReleasePolicyError) as exc:
307 policy.prepare_policy_run(
308 _repo(tmp_path),
309 advisory_source_id=source_id,
310 db_urls=db_urls,
311 db_root=tmp_path / "db",
312 runner=FakeRunner(tmp_path / "db" / SNAPSHOT),
313 temp_path_factory=lambda label: tmp_path / label,
314 )
315
316 assert any(failure.error == error for failure in exc.value.failures)
317
318
319def test_snapshot_count_must_be_exactly_one(tmp_path: Path) -> None:
320 db_root = tmp_path / "empty-db"
321 db_root.mkdir()
322
323 with pytest.raises(policy.ReleasePolicyError) as exc:
324 policy.prepare_policy_run(
325 _repo(tmp_path),
326 advisory_source_id="internal",
327 db_urls=("ssh://example.test/db.git",),
328 db_root=db_root,
329 runner=FakeRunner(db_root / SNAPSHOT),
330 temp_path_factory=lambda label: tmp_path / label,
331 )
332 assert exc.value.failures[0].error == "advisory db snapshot count is invalid"
333 assert exc.value.failures[0].actual == "0"
334
335 db_root = tmp_path / "multi-db"
336 db_root.mkdir()
337 first = _write_snapshot(db_root, name="advisory-db-one")
338 _write_snapshot(db_root, name="advisory-db-two")
339 with pytest.raises(policy.ReleasePolicyError) as exc:
340 policy.prepare_policy_run(
341 _repo(tmp_path),
342 advisory_source_id="internal",
343 db_urls=("ssh://example.test/db.git",),
344 db_root=db_root,
345 runner=FakeRunner(first),
346 temp_path_factory=lambda label: tmp_path / f"multi-{label}",
347 )
348 assert exc.value.failures[0].error == "advisory db snapshot count is invalid"
349 assert exc.value.failures[0].actual == "2"
350
351
352def test_non_top_level_snapshot_fails_walk_up_check(tmp_path: Path) -> None:
353 db_root, snapshot = _db_root(tmp_path)
354 runner = FakeRunner(snapshot, top_level=tmp_path)
355
356 with pytest.raises(policy.ReleasePolicyError) as exc:
357 _prepare(tmp_path, db_root=db_root, snapshot=snapshot, runner=runner)
358
359 assert (
360 exc.value.failures[0].error
361 == "advisory db snapshot is not an isolated git checkout"
362 )
363 assert "check" not in runner.events
364
365
366def test_scanned_advisory_db_must_match_measured_snapshot(tmp_path: Path) -> None:
367 db_root, snapshot = _db_root(tmp_path)
368 runner = FakeRunner(snapshot, scanned_path=tmp_path / "other-db")
369
370 with pytest.raises(policy.ReleasePolicyError) as exc:
371 _prepare(tmp_path, db_root=db_root, snapshot=snapshot, runner=runner)
372
373 assert (
374 exc.value.failures[0].error
375 == "cargo-deny scanned a different advisory database"
376 )
377
378
379def test_missing_debug_line_fails_closed(tmp_path: Path) -> None:
380 db_root, snapshot = _db_root(tmp_path)
381 runner = FakeRunner(snapshot, debug_stderr="debug without database path\n")
382
383 with pytest.raises(policy.ReleasePolicyError) as exc:
384 _prepare(tmp_path, db_root=db_root, snapshot=snapshot, runner=runner)
385
386 assert (
387 exc.value.failures[0].error
388 == "cargo-deny advisory database debug line is missing"
389 )
390
391
392def test_absent_fetch_head_fails_closed(tmp_path: Path) -> None:
393 db_root, snapshot = _db_root(tmp_path, fetch_time=None)
394 runner = FakeRunner(snapshot)
395
396 with pytest.raises(policy.ReleasePolicyError) as exc:
397 _prepare(tmp_path, db_root=db_root, snapshot=snapshot, runner=runner)
398
399 assert exc.value.failures[0].error == "advisory db FETCH_HEAD is missing"
400
401
402def test_stale_fetch_head_fails(tmp_path: Path) -> None:
403 db_root, snapshot = _db_root(
404 tmp_path,
405 fetch_time=datetime(2026, 7, 18, 11, 59, 59, tzinfo=UTC),
406 )
407
408 with pytest.raises(policy.ReleasePolicyError) as exc:
409 _prepare(tmp_path, db_root=db_root, snapshot=snapshot)
410
411 assert exc.value.failures[0].error == "advisory fetch time is stale"
412
413
414def test_over_age_content_fails(tmp_path: Path) -> None:
415 db_root, snapshot = _db_root(tmp_path)
416 runner = FakeRunner(snapshot, commit_timestamp="2026-07-05T11:59:59Z\n")
417
418 with pytest.raises(policy.ReleasePolicyError) as exc:
419 _prepare(tmp_path, db_root=db_root, snapshot=snapshot, runner=runner)
420
421 assert exc.value.failures[0].error == "advisory db content is stale"
422
423
424def test_zero_advisory_count_fails(tmp_path: Path) -> None:
425 db_root, snapshot = _db_root(tmp_path, advisory_count=0)
426 runner = FakeRunner(snapshot)
427
428 with pytest.raises(policy.ReleasePolicyError) as exc:
429 _prepare(tmp_path, db_root=db_root, snapshot=snapshot, runner=runner)
430
431 assert exc.value.failures[0].error == "advisory db snapshot contains no advisories"
432 assert "check" not in runner.events
433
434
435def test_post_run_dirty_snapshot_fails(tmp_path: Path) -> None:
436 db_root, snapshot = _db_root(tmp_path)
437 runner = FakeRunner(snapshot, status=("", "?? late-file\n"))
438
439 with pytest.raises(policy.ReleasePolicyError) as exc:
440 _prepare(tmp_path, db_root=db_root, snapshot=snapshot, runner=runner)
441
442 assert (
443 exc.value.failures[0].error
444 == "advisory db snapshot has uncommitted or ignored material"
445 )
446 assert runner.events.count("status") == 2
447
448
449def test_fresh_fetch_and_four_day_content_pass(tmp_path: Path) -> None:
450 db_root, snapshot = _db_root(tmp_path)
451 runner = FakeRunner(snapshot, commit_timestamp="2026-07-16T12:00:00Z\n")
452
453 result, _runner, _repo_path = _prepare(
454 tmp_path,
455 db_root=db_root,
456 snapshot=snapshot,
457 runner=runner,
458 )
459
460 assert result.advisory_acquired_at == "2026-07-20T11:30:00Z"
461 assert result.db_commit_timestamp == "2026-07-16T12:00:00Z"
462
463
464def test_prepare_policy_run_accepts_sha256_db_commit(tmp_path: Path) -> None:
465 db_root, snapshot = _db_root(tmp_path)
466 result, _runner, _repo_path = _prepare(
467 tmp_path,
468 db_root=db_root,
469 snapshot=snapshot,
470 runner=FakeRunner(snapshot, commit_stdout="a" * 64 + "\n"),
471 )
472
473 assert result.db_commit == "a" * 64
474
475
476@pytest.mark.parametrize("commit_stdout", MALFORMED_DB_COMMITS)
477def test_prepare_policy_run_rejects_malformed_db_commit_observation(
478 tmp_path: Path,
479 commit_stdout: str,
480) -> None:
481 db_root, snapshot = _db_root(tmp_path)
482 runner = FakeRunner(snapshot, commit_stdout=commit_stdout)
483
484 with pytest.raises(policy.ReleasePolicyError) as exc:
485 _prepare(tmp_path, db_root=db_root, snapshot=snapshot, runner=runner)
486
487 assert exc.value.failures[0].error == "advisory_snapshot.db_commit is invalid"
488 assert exc.value.failures[0].expected == "exactly 40 or 64 lowercase hex characters"
489 assert "check" not in runner.events
490 if commit_stdout.startswith("A"):
491 assert exc.value.failures[0].actual == commit_stdout.removesuffix("\n")
492
493
494@pytest.mark.parametrize("digest", MALFORMED_ARCHIVE_DIGESTS)
495def test_prepare_policy_run_rejects_malformed_archive_digest_observation(
496 tmp_path: Path,
497 digest: str,
498) -> None:
499 db_root, snapshot = _db_root(tmp_path)
500 runner = FakeRunner(snapshot)
501
502 with pytest.raises(policy.ReleasePolicyError) as exc:
503 _prepare(
504 tmp_path,
505 db_root=db_root,
506 snapshot=snapshot,
507 runner=runner,
508 archive=digest,
509 )
510
511 assert (
512 exc.value.failures[0].error == "advisory_snapshot.db_archive_sha256 is invalid"
513 )
514 assert exc.value.failures[0].expected == "exactly 64 lowercase hex characters"
515 assert "check" not in runner.events
516 if digest.startswith("B"):
517 assert exc.value.failures[0].actual == digest
518
519
520def test_policy_run_constructor_accepts_sha256_db_commit() -> None:
521 result = policy.PolicyRun(
522 advisory_source_id="internal",
523 db_snapshot_basename=SNAPSHOT,
524 db_commit="a" * 64,
525 db_archive_sha256=DB_ARCHIVE,
526 advisory_count=1,
527 advisory_acquired_at="2026-07-20T11:30:00Z",
528 db_commit_timestamp=COMMIT_TIME,
529 policy_checked_at="2026-07-20T12:00:00Z",
530 result="pass",
531 )
532
533 assert result.db_commit == "a" * 64
534
535
536@pytest.mark.parametrize(
537 ("field", "value", "error"),
538 [
539 *(
540 pytest.param(
541 "db_commit",
542 value,
543 "policy_run.db_commit is invalid",
544 id=f"commit-{name}",
545 )
546 for name, value in MALFORMED_DB_COMMIT_CASES
547 ),
548 *(
549 pytest.param(
550 "db_archive_sha256",
551 value,
552 "policy_run.db_archive_sha256 is invalid",
553 id=f"archive-{name}",
554 )
555 for name, value in MALFORMED_ARCHIVE_DIGEST_CASES
556 ),
557 pytest.param(
558 "db_snapshot_basename",
559 "../db",
560 "policy_run.db_snapshot_basename is invalid",
561 id="snapshot-path",
562 ),
563 pytest.param(
564 "db_snapshot_basename",
565 "",
566 "policy_run.db_snapshot_basename is invalid",
567 id="snapshot-empty",
568 ),
569 pytest.param(
570 "advisory_count",
571 0,
572 "policy_run.advisory_count is invalid",
573 id="count-zero",
574 ),
575 pytest.param(
576 "advisory_count",
577 True,
578 "policy_run.advisory_count is invalid",
579 id="count-bool",
580 ),
581 pytest.param(
582 "db_commit_timestamp",
583 "2026-07-19T12:00:00-06:00",
584 "policy_run.db_commit_timestamp is invalid",
585 id="commit-time-not-normalized",
586 ),
587 ],
588)
589def test_policy_run_constructor_rejects_malformed_receipt_identity(
590 field: str,
591 value: object,
592 error: str,
593) -> None:
594 kwargs = {
595 "advisory_source_id": "internal",
596 "db_snapshot_basename": SNAPSHOT,
597 "db_commit": DB_COMMIT,
598 "db_archive_sha256": DB_ARCHIVE,
599 "advisory_count": 1,
600 "advisory_acquired_at": "2026-07-20T11:30:00Z",
601 "db_commit_timestamp": COMMIT_TIME,
602 "policy_checked_at": "2026-07-20T12:00:00Z",
603 "result": "pass",
604 }
605 kwargs[field] = value
606
607 with pytest.raises(policy.ReleasePolicyError) as exc:
608 policy.PolicyRun(**kwargs)
609
610 assert exc.value.failures[0].error == error
611 if isinstance(value, str) and value and value[0].isupper():
612 assert exc.value.failures[0].actual == value
613
614
615def _cleanup_failure(_path: Path) -> None:
616 raise OSError(5, "cleanup failed", "/private/tmp/release-advisory-secret")
617
618
619@pytest.mark.parametrize("primary_failure", (False, True), ids=("success", "primary"))
620@pytest.mark.parametrize(
621 ("cleanup_name", "cleanup_kwargs"),
622 [
623 ("unlink", {"cleanup_unlink": _cleanup_failure}),
624 ("rmdir", {"cleanup_rmdir": _cleanup_failure}),
625 ],
626)
627def test_cleanup_failures_surface_without_masking_primary_errors(
628 tmp_path: Path,
629 cleanup_name: str,
630 cleanup_kwargs: dict,
631 primary_failure: bool,
632) -> None:
633 db_root, snapshot = _db_root(tmp_path)
634 runner = FakeRunner(snapshot, status="?? dirty\n" if primary_failure else "")
635
636 with pytest.raises(policy.ReleasePolicyError) as exc:
637 _prepare(
638 tmp_path,
639 db_root=db_root,
640 snapshot=snapshot,
641 runner=runner,
642 **cleanup_kwargs,
643 )
644
645 errors = [failure.error for failure in exc.value.failures]
646 if primary_failure:
647 assert "advisory db snapshot has uncommitted or ignored material" in errors
648 else:
649 assert errors == [
650 "release advisory cleanup failed during materialized config removal"
651 ]
652 assert any(error.startswith("release advisory cleanup failed") for error in errors)
653 assert (
654 checker.validate_public_evidence_text(
655 "release advisory cleanup", str(exc.value)
656 )
657 == []
658 )
659
660
661def _caller_db_inventory(root: Path) -> list[tuple[str, str, bytes]]:
662 items: list[tuple[str, str, bytes]] = []
663 for path in root.rglob("*"):
664 relative = path.relative_to(root).as_posix()
665 if path.is_dir():
666 items.append((relative, "dir", b""))
667 elif path.is_file():
668 items.append((relative, "file", path.read_bytes()))
669 return sorted(items)
670
671
672@pytest.mark.parametrize(
673 ("primary_failure", "cleanup_kwargs"),
674 [
675 (False, {}),
676 (True, {}),
677 (False, {"cleanup_unlink": _cleanup_failure}),
678 ],
679 ids=("success", "primary-failure", "cleanup-failure"),
680)
681def test_caller_owned_db_root_is_preserved(
682 tmp_path: Path,
683 primary_failure: bool,
684 cleanup_kwargs: dict,
685) -> None:
686 db_root, snapshot = _db_root(tmp_path)
687 before = _caller_db_inventory(db_root)
688 runner = FakeRunner(snapshot, status="?? dirty\n" if primary_failure else "")
689
690 if primary_failure or cleanup_kwargs:
691 with pytest.raises(policy.ReleasePolicyError):
692 _prepare(
693 tmp_path,
694 db_root=db_root,
695 snapshot=snapshot,
696 runner=runner,
697 **cleanup_kwargs,
698 )
699 else:
700 _prepare(tmp_path, db_root=db_root, snapshot=snapshot, runner=runner)
701
702 assert _caller_db_inventory(db_root) == before
703
704
705def test_policy_temps_are_cleaned_without_removing_caller_db(tmp_path: Path) -> None:
706 db_root, snapshot = _db_root(tmp_path)
707 temp_root = tmp_path / "policy-temp"
708 _prepare(
709 tmp_path,
710 db_root=db_root,
711 snapshot=snapshot,
712 runner=FakeRunner(snapshot),
713 clock=_clock([]),
714 )
715 assert db_root.is_dir()
716 assert not (tmp_path / "advisory-policy").exists()
717
718 with pytest.raises(policy.ReleasePolicyError):
719 policy.prepare_policy_run(
720 _repo(tmp_path),
721 advisory_source_id="internal",
722 db_urls=("ssh://example.test/db.git",),
723 db_root=db_root,
724 runner=FakeRunner(snapshot, status="?? dirty\n"),
725 temp_path_factory=lambda _label: temp_root,
726 clock=_clock([]),
727 archive_hasher=lambda _db: DB_ARCHIVE,
728 )
729 assert db_root.is_dir()
730 assert not temp_root.exists()