108 lines
3.9 KiB
Python
108 lines
3.9 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Behavioral checks for exact integration handoff evidence."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from pathlib import Path
|
||
|
|
import subprocess
|
||
|
|
import tempfile
|
||
|
|
import unittest
|
||
|
|
|
||
|
|
import workspace
|
||
|
|
|
||
|
|
|
||
|
|
class IntegrationEvidence(unittest.TestCase):
|
||
|
|
def test_freeze_captures_the_current_exact_change(self) -> None:
|
||
|
|
record = {
|
||
|
|
"change": {"change_id": "A1", "commit_id": "B2"},
|
||
|
|
"dirty": True,
|
||
|
|
}
|
||
|
|
frozen = workspace.freeze_integration_candidate(record)
|
||
|
|
self.assertEqual(
|
||
|
|
frozen["integration_candidate"],
|
||
|
|
{
|
||
|
|
"change_id": "a1",
|
||
|
|
"commit_id": "b2",
|
||
|
|
"captured_at": frozen["integration_candidate"]["captured_at"],
|
||
|
|
},
|
||
|
|
)
|
||
|
|
self.assertNotIn("integration_candidate", record)
|
||
|
|
|
||
|
|
def test_freeze_uses_parent_of_clean_jj_working_copy(self) -> None:
|
||
|
|
def runner(
|
||
|
|
argv: list[str], **_kwargs: object
|
||
|
|
) -> subprocess.CompletedProcess[str]:
|
||
|
|
self.assertIn("@-", argv)
|
||
|
|
return subprocess.CompletedProcess(argv, 0, "a1\nb2\n", "")
|
||
|
|
|
||
|
|
with tempfile.TemporaryDirectory() as temporary:
|
||
|
|
frozen = workspace.freeze_integration_candidate(
|
||
|
|
{
|
||
|
|
"actual_path": temporary,
|
||
|
|
"change": {"change_id": "c3", "commit_id": "d4"},
|
||
|
|
"dirty": False,
|
||
|
|
},
|
||
|
|
runner=runner,
|
||
|
|
)
|
||
|
|
self.assertEqual(frozen["integration_candidate"]["change_id"], "a1")
|
||
|
|
self.assertEqual(frozen["integration_candidate"]["commit_id"], "b2")
|
||
|
|
|
||
|
|
def test_integration_proves_frozen_candidate_not_stale_record_change(self) -> None:
|
||
|
|
source_commit = "a" * 40
|
||
|
|
target_commit = "c" * 40
|
||
|
|
target_change = "b" * 32
|
||
|
|
observed: list[list[str]] = []
|
||
|
|
|
||
|
|
def runner(
|
||
|
|
argv: list[str], **_kwargs: object
|
||
|
|
) -> subprocess.CompletedProcess[str]:
|
||
|
|
observed.append(argv)
|
||
|
|
if "change_id.normal_hex()" in argv[-1]:
|
||
|
|
stdout = f"{target_change}\n{target_commit}\n"
|
||
|
|
else:
|
||
|
|
stdout = f"{source_commit}\n"
|
||
|
|
return subprocess.CompletedProcess(argv, 0, stdout, "")
|
||
|
|
|
||
|
|
with tempfile.TemporaryDirectory() as temporary:
|
||
|
|
record = {
|
||
|
|
"actual_path": str(Path(temporary)),
|
||
|
|
"gate_state": "passed",
|
||
|
|
"integration_state": "lock-released",
|
||
|
|
"change": {"change_id": "d" * 32, "commit_id": "d" * 40},
|
||
|
|
"integration_candidate": {
|
||
|
|
"change_id": "e" * 32,
|
||
|
|
"commit_id": source_commit,
|
||
|
|
},
|
||
|
|
"integration_owner_runtime": "runtime-1",
|
||
|
|
"integration_resource": "integration/master",
|
||
|
|
"integration_lock_release_evidence": {
|
||
|
|
"runtime_id": "runtime-1",
|
||
|
|
"resource": "integration/master",
|
||
|
|
},
|
||
|
|
}
|
||
|
|
integrated = workspace.prove_integrated(
|
||
|
|
record,
|
||
|
|
target_revision="master",
|
||
|
|
runner=runner,
|
||
|
|
)
|
||
|
|
|
||
|
|
evidence = integrated["integration_release_evidence"]
|
||
|
|
self.assertEqual(evidence["source_commit"], source_commit)
|
||
|
|
self.assertEqual(evidence["target_commit"], target_commit)
|
||
|
|
self.assertTrue(any(source_commit in part for call in observed for part in call))
|
||
|
|
|
||
|
|
def test_integration_refuses_missing_owner_release_evidence(self) -> None:
|
||
|
|
with self.assertRaisesRegex(workspace.WorkspaceError, "owner/resource"):
|
||
|
|
workspace.prove_integrated(
|
||
|
|
{
|
||
|
|
"gate_state": "passed",
|
||
|
|
"integration_state": "lock-released",
|
||
|
|
"integration_candidate": {"commit_id": "a" * 40},
|
||
|
|
},
|
||
|
|
target_revision="master",
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main(verbosity=2)
|