personal memory agent
0

Configure Feed

Select the types of activity you want to include in your feed.

solstone / tests / test_channel_adapters.py
21 kB 631 lines
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 subprocess 9import sys 10import zipfile 11from datetime import UTC, datetime 12from pathlib import Path 13from typing import Any 14 15import pytest 16 17import scripts.release_build_host as build_rail 18import scripts.release_install_smoke as smoke 19import scripts.release_proof_host as proof_rail 20from scripts.channel_adapters import adapter_common as common 21from scripts.channel_adapters import build_host_macos, proof_host 22from scripts.check_release_preflight import expected_presign_lane_tool_evidence 23from scripts.release_digest import candidate_digest 24from scripts.release_tool_pins import ( 25 HOST_VARIANT_TOOL_KEYS, 26 MACOS_SWIFT_FLATTENED_BANNER, 27 UV_MACOS_FIXTURE_BANNER, 28) 29from tests.helpers.release_wheel_fixtures import ROOT_LAUNCHER_BYTES, record_hash 30 31 32def _completed( 33 stdout: str = "", 34 *, 35 stderr: str = "", 36 returncode: int = 0, 37) -> subprocess.CompletedProcess[str]: 38 return subprocess.CompletedProcess( 39 args=[], 40 returncode=returncode, 41 stdout=stdout, 42 stderr=stderr, 43 ) 44 45 46def _lane( 47 name: str = "lane", 48 *, 49 host: str = "build-host.example", 50 port: int | None = 2222, 51) -> common.LaneConfig: 52 return common.LaneConfig( 53 name=name, 54 mode="ssh", 55 host=host, 56 port=port, 57 user="builder", 58 identity_file="~/.ssh/solstone-channel-adapter", 59 extra_ssh_options=("-o", "BatchMode=yes"), 60 remote_python="python3", 61 remote_work_prefix="/tmp/solstone-channel-adapter", 62 remote_run_wrapper="operator-session-wrapper", 63 tmux_window="adapter:build", 64 unlock_workdir="~/projects/build-worktree", 65 ) 66 67 68def _local_lane() -> common.LaneConfig: 69 return common.LaneConfig( 70 name="proof.linux-x86_64-musl", 71 mode="local", 72 ) 73 74 75def _write_metadata_wheel(path: Path) -> None: 76 distribution, version = path.name.removesuffix(".whl").split("-")[:2] 77 metadata_name = f"{distribution}-{version}.dist-info/METADATA" 78 metadata = f"Name: {distribution.replace('_', '-')}\nVersion: {version}\n" 79 with zipfile.ZipFile(path, "w") as wheel: 80 members = {metadata_name: metadata.encode("utf-8")} 81 if path.name.startswith("solstone-"): 82 members[f"solstone-{version}.dist-info/WHEEL"] = b"Wheel-Version: 1.0\n" 83 for name, content in ROOT_LAUNCHER_BYTES.items(): 84 members[f"solstone-{version}.data/scripts/{name}"] = content 85 record_name = f"solstone-{version}.dist-info/RECORD" 86 record = "\n".join( 87 f"{name},{record_hash(content)},{len(content)}" 88 for name, content in members.items() 89 ) 90 members[record_name] = f"{record}\n{record_name},,".encode("utf-8") 91 for name, content in members.items(): 92 info = zipfile.ZipInfo(name) 93 info.external_attr = ( 94 0o755 << 16 95 if Path(name).name in smoke.ROOT_LAUNCHER_NAMES 96 else 0o644 << 16 97 ) 98 wheel.writestr(info, content) 99 100 101def _file_entry(path: Path) -> dict[str, Any]: 102 data = path.read_bytes() 103 return { 104 "name": path.name, 105 "bytes": len(data), 106 "sha256": hashlib.sha256(data).hexdigest(), 107 } 108 109 110def _tool_stdout(*, uv: str = UV_MACOS_FIXTURE_BANNER) -> str: 111 expected = expected_presign_lane_tool_evidence("macos-arm64") 112 observed = { 113 **expected, 114 "uv": uv, 115 "swift": MACOS_SWIFT_FLATTENED_BANNER, 116 } 117 return ( 118 "\n".join(f"{key}\t{observed[key]}" for key in sorted(observed)) 119 + f"\n{build_host_macos.TOOLCHAIN_TOKEN}\n" 120 ) 121 122 123def _artifact_listing_stdout(artifact_bytes: dict[str, bytes]) -> str: 124 lines = [] 125 for name, data in artifact_bytes.items(): 126 lines.append( 127 "\t".join( 128 ( 129 build_host_macos.ARTIFACT_TOKEN, 130 name, 131 hashlib.sha256(data).hexdigest(), 132 str(len(data)), 133 ) 134 ) 135 ) 136 lines.append(build_host_macos.DIST_TOKEN) 137 return "\n".join(lines) + "\n" 138 139 140def _write_build_request(tmp_path: Path) -> tuple[Path, build_rail.SourceBundle, dict]: 141 bundle = tmp_path / "source.bundle" 142 bundle.write_bytes(b"bundle") 143 sha256, byte_count = common.sha256_size(bundle) 144 source_bundle = build_rail.SourceBundle( 145 path=bundle, 146 source_commit="a" * 40, 147 sha256=sha256, 148 bytes=byte_count, 149 ) 150 channel = build_rail.ExternalBuildHostChannel(["adapter"]) 151 payload = channel._request_payload( 152 cohort_id="cohort", 153 source_bundle=source_bundle, 154 expected_commit=source_bundle.source_commit, 155 ) 156 request_dir = tmp_path / "request" 157 request_dir.mkdir() 158 request_bundle = request_dir / "source.bundle" 159 request_bundle.write_bytes(bundle.read_bytes()) 160 request_path = request_dir / "request.json" 161 request_path.write_text(json.dumps(payload), encoding="utf-8") 162 return request_path, source_bundle, payload 163 164 165def _write_proof_request(tmp_path: Path) -> tuple[Path, dict[str, Any], dict[str, Any]]: 166 request_dir = tmp_path / "proof-request" 167 candidate_dir = request_dir / "candidate" 168 output_dir = request_dir / "output" 169 candidate_dir.mkdir(parents=True) 170 output_dir.mkdir() 171 for name in ( 172 "solstone-1.0.0-py3-none-any.whl", 173 "solstone_core-1.0.0-py3-none-linux_x86_64.whl", 174 ): 175 _write_metadata_wheel(candidate_dir / name) 176 native_members = { 177 name: { 178 "path": f"linux-x86/{name}", 179 "sha256": hashlib.sha256(name.encode("utf-8")).hexdigest(), 180 "bytes": len(name.encode("utf-8")), 181 } 182 for name in smoke.CORE_SCRIPT_NAMES 183 } 184 digest = candidate_digest(candidate_dir) 185 ledger: dict[str, Any] = { 186 "source_commit": "b" * 40, 187 "core_lock_sha256": "c" * 64, 188 "candidate": { 189 "candidate_digest": digest, 190 "files": [ 191 _file_entry(path) 192 for path in sorted(candidate_dir.iterdir(), key=lambda item: item.name) 193 ], 194 }, 195 "native_members": {"linux-x86_64-musl": native_members}, 196 } 197 ledger_bytes = json.dumps(ledger, sort_keys=True).encode("utf-8") 198 ledger_sha = hashlib.sha256(ledger_bytes).hexdigest() 199 (request_dir / "ledger.json").write_bytes(ledger_bytes) 200 install_paths = smoke.target_install_paths_from_ledger( 201 ledger, 202 target="linux-x86_64-musl", 203 candidate_dir=candidate_dir, 204 ) 205 channel = proof_rail.ExternalProofHostChannel( 206 "linux-x86_64-musl", 207 ["adapter"], 208 ) 209 payload = channel._request_payload( 210 cohort_id="cohort", 211 target="linux-x86_64-musl", 212 version="1.0.0", 213 source_commit="b" * 40, 214 core_lock_sha256="c" * 64, 215 candidate_digest=digest, 216 ledger_sha256=ledger_sha, 217 install_paths=install_paths, 218 ) 219 request_path = request_dir / "request.json" 220 request_path.write_text(json.dumps(payload), encoding="utf-8") 221 return request_path, payload, ledger 222 223 224def _write_valid_install_proof( 225 proof_path: Path, 226 *, 227 request_payload: dict[str, Any], 228 ledger: dict[str, Any], 229) -> None: 230 request_dir = proof_path.parents[1] 231 candidate_dir = request_dir / request_payload["paths"]["candidate_dir"] 232 env_root = request_dir / "env" 233 (env_root / "bin").mkdir(parents=True, exist_ok=True) 234 python_path = env_root / "bin" / "python" 235 python_path.write_bytes(b"python") 236 executable_paths = { 237 name: env_root / "bin" / name for name in smoke.INSTALL_SCRIPT_NAMES 238 } 239 for name in smoke.ROOT_LAUNCHER_NAMES: 240 executable_paths[name].write_bytes(ROOT_LAUNCHER_BYTES[name]) 241 for name in smoke.CORE_SCRIPT_NAMES: 242 executable_paths[name].write_text(name, encoding="utf-8") 243 install_paths = smoke.target_install_paths_from_ledger( 244 ledger, 245 target=request_payload["target"], 246 candidate_dir=candidate_dir, 247 ) 248 expected_members, expected_failures = smoke._expected_install_members( 249 ledger, 250 request_payload["target"], 251 candidate_dir=candidate_dir, 252 install_paths=install_paths, 253 ) 254 assert expected_failures == [] 255 proof = smoke.build_install_proof( 256 target=request_payload["target"], 257 version=request_payload["version"], 258 source_commit=request_payload["source_commit"], 259 core_lock_sha256=request_payload["core_lock_sha256"], 260 candidate_digest=request_payload["candidate_digest"], 261 ledger_sha256=request_payload["ledger_sha256"], 262 candidate_dir=candidate_dir, 263 candidate_paths=install_paths, 264 ledger_payload=ledger, 265 observation=smoke.InstallObservation( 266 env_root=env_root, 267 preexisting_distributions=(), 268 install=smoke.CommandResult( 269 argv=( 270 str(python_path), 271 "-m", 272 "pip", 273 "install", 274 "--no-index", 275 "--no-deps", 276 *(str(path) for path in install_paths), 277 ), 278 exit_code=0, 279 stdout="installed", 280 env=smoke.SCRUBBED_COMMAND_ENV, 281 ), 282 installed_distributions=smoke.expected_distribution_entries(install_paths), 283 installed_members=tuple( 284 { 285 "name": name, 286 "path": path, 287 "sha256": expected_members[name]["sha256"], 288 "symlink": False, 289 } 290 for name, path in sorted(executable_paths.items()) 291 ), 292 smoke={ 293 name: smoke.CommandResult( 294 argv=(str(path), "--version"), 295 exit_code=0, 296 stdout=( 297 f"{smoke.CORE_SMOKE_STDOUT[name]} {request_payload['version']}" 298 ), 299 env=smoke.SCRUBBED_COMMAND_ENV, 300 ) 301 for name, path in sorted(executable_paths.items()) 302 }, 303 ), 304 recorded_at=datetime(2026, 7, 20, 12, tzinfo=UTC), 305 ) 306 smoke.write_install_proof( 307 proof_path, 308 proof, 309 target=request_payload["target"], 310 version=request_payload["version"], 311 source_commit=request_payload["source_commit"], 312 core_lock_sha256=request_payload["core_lock_sha256"], 313 candidate_digest=request_payload["candidate_digest"], 314 ledger_sha256=request_payload["ledger_sha256"], 315 candidate_dir=candidate_dir, 316 ledger_payload=ledger, 317 ) 318 319 320def test_build_request_response_round_trip_through_rail_parser( 321 tmp_path: Path, 322 monkeypatch: pytest.MonkeyPatch, 323) -> None: 324 request_path, source_bundle, payload = _write_build_request(tmp_path) 325 expected_files = list(payload["expected_outputs"].values()) 326 artifact_bytes = { 327 name: f"artifact:{name}".encode("utf-8") for name in expected_files 328 } 329 330 def fake_runner(argv, **kwargs): 331 if argv[0] == "scp" and ":" in argv[-2]: 332 name = str(argv[-2]).rsplit("/", 1)[-1] 333 Path(argv[-1]).write_bytes(artifact_bytes[name]) 334 script = kwargs.get("input_text") or "" 335 if "emit python" in script: 336 return _completed(_tool_stdout()) 337 if "git checkout" in script: 338 return _completed(f"{build_host_macos.CHECKOUT_TOKEN}\n") 339 if "for f in" in script: 340 return _completed(_artifact_listing_stdout(artifact_bytes)) 341 return _completed() 342 343 monkeypatch.setattr(common, "run", fake_runner) 344 monkeypatch.chdir(request_path.parent) 345 346 build_host_macos.build_macos(_lane(), request_path) 347 348 response = json.loads((request_path.parent / "response.json").read_text()) 349 build_rail._validate_attestation( 350 response, 351 expected_commit=source_bundle.source_commit, 352 source_bundle=source_bundle, 353 ) 354 evidence = build_rail._validate_macos_tool_evidence(response) 355 wheel_names, record_names = build_rail._names_from_payload(response) 356 assert set(evidence) == set(expected_presign_lane_tool_evidence("macos-arm64")) 357 assert len(wheel_names) == 2 358 assert tuple(record_names) == ( 359 build_rail.MACOS_ROOT_RECORD, 360 build_rail.MACOS_CORE_RECORD, 361 ) 362 363 364def test_build_retrieved_artifact_digest_mismatch_writes_no_response( 365 tmp_path: Path, 366 monkeypatch: pytest.MonkeyPatch, 367) -> None: 368 request_path, _source_bundle, payload = _write_build_request(tmp_path) 369 expected_files = list(payload["expected_outputs"].values()) 370 artifact_bytes = { 371 name: f"artifact:{name}".encode("utf-8") for name in expected_files 372 } 373 truncated_name = expected_files[0] 374 375 def fake_runner(argv, **kwargs): 376 if argv[0] == "scp" and ":" in argv[-2]: 377 name = str(argv[-2]).rsplit("/", 1)[-1] 378 data = artifact_bytes[name] 379 Path(argv[-1]).write_bytes(data[:-1] if name == truncated_name else data) 380 script = kwargs.get("input_text") or "" 381 if "emit python" in script: 382 return _completed(_tool_stdout()) 383 if "git checkout" in script: 384 return _completed(f"{build_host_macos.CHECKOUT_TOKEN}\n") 385 if "for f in" in script: 386 return _completed(_artifact_listing_stdout(artifact_bytes)) 387 return _completed() 388 389 monkeypatch.setattr(common, "run", fake_runner) 390 monkeypatch.chdir(request_path.parent) 391 392 with pytest.raises(SystemExit): 393 build_host_macos.build_macos(_lane(), request_path) 394 395 assert not (request_path.parent / "response.json").exists() 396 397 398def test_proof_request_response_round_trip_through_rail_parser( 399 tmp_path: Path, 400 monkeypatch: pytest.MonkeyPatch, 401) -> None: 402 request_path, request_payload, ledger = _write_proof_request(tmp_path) 403 proof_path = request_path.parent / "output" / "proof.json" 404 405 def fake_runner(argv, **kwargs): 406 if argv == ["uname", "-s"]: 407 return _completed("Linux\n") 408 if argv == ["uname", "-m"]: 409 return _completed("x86_64\n") 410 if argv[:2] == [sys.executable, "-c"]: 411 _write_valid_install_proof( 412 proof_path, 413 request_payload=request_payload, 414 ledger=ledger, 415 ) 416 sha256, byte_count = common.sha256_size(proof_path) 417 return _completed( 418 f'{proof_host.PROOF_TOKEN} {{"bytes": {byte_count}, "sha256": "{sha256}"}}\n' 419 ) 420 raise AssertionError(argv) 421 422 monkeypatch.setattr(proof_host, "run", fake_runner) 423 424 proof_host.prove("linux-x86_64-musl", _local_lane(), request_path) 425 426 response = json.loads((request_path.parent / "response.json").read_text()) 427 channel = proof_rail.ExternalProofHostChannel("linux-x86_64-musl", ["adapter"]) 428 proof_descriptor = channel._validate_response( 429 response, 430 cohort_id="cohort", 431 target="linux-x86_64-musl", 432 candidate_digest=request_payload["candidate_digest"], 433 ledger_sha256=request_payload["ledger_sha256"], 434 ) 435 assert proof_descriptor["path"] == "output/proof.json" 436 proof_failures = smoke.validate_install_proof_bytes( 437 proof_path.read_bytes(), 438 target=request_payload["target"], 439 version=request_payload["version"], 440 source_commit=request_payload["source_commit"], 441 core_lock_sha256=request_payload["core_lock_sha256"], 442 candidate_digest=request_payload["candidate_digest"], 443 ledger_sha256=request_payload["ledger_sha256"], 444 candidate_dir=request_path.parent / request_payload["paths"]["candidate_dir"], 445 ledger_payload=ledger, 446 ) 447 assert proof_failures == [] 448 449 450def test_source_and_retrieved_digest_verification(tmp_path: Path) -> None: 451 proof = tmp_path / "proof.json" 452 proof.write_bytes(b"proof") 453 sha256, byte_count = common.sha256_size(proof) 454 455 common.verify_retrieved_file( 456 proof, 457 expected_sha256=sha256, 458 expected_bytes=byte_count, 459 label="proof.json", 460 ) 461 with pytest.raises(SystemExit): 462 common.verify_retrieved_file( 463 proof, 464 expected_sha256="0" * 64, 465 expected_bytes=byte_count, 466 label="proof.json", 467 ) 468 469 470def test_macos_tool_evidence_derives_from_rail_pins( 471 monkeypatch: pytest.MonkeyPatch, 472) -> None: 473 monkeypatch.setattr( 474 common, "run", lambda _argv, **_kwargs: _completed(_tool_stdout()) 475 ) 476 477 evidence = build_host_macos._derive_tool_evidence(_lane()) 478 expected = expected_presign_lane_tool_evidence("macos-arm64") 479 480 assert set(evidence) == set(expected) 481 assert evidence["uv"] == UV_MACOS_FIXTURE_BANNER 482 assert evidence["swift"] == MACOS_SWIFT_FLATTENED_BANNER 483 for key in set(expected) - set(HOST_VARIANT_TOOL_KEYS): 484 assert evidence[key] == expected[key] 485 486 487def test_host_variant_banner_mutation_writes_no_evidence_and_exits_nonzero( 488 tmp_path: Path, 489 monkeypatch: pytest.MonkeyPatch, 490) -> None: 491 request_path, _source_bundle, _payload = _write_build_request(tmp_path) 492 mutated = "uv 0.11.5 (aarch64-apple-darwin)" 493 494 monkeypatch.setattr( 495 common, 496 "run", 497 lambda _argv, **kwargs: ( 498 _completed(_tool_stdout(uv=mutated)) 499 if "emit python" in (kwargs.get("input_text") or "") 500 else _completed(f"{build_host_macos.CHECKOUT_TOKEN}\n") 501 ), 502 ) 503 monkeypatch.chdir(request_path.parent) 504 505 with pytest.raises(SystemExit): 506 build_host_macos.build_macos(_lane(), request_path) 507 508 assert not (request_path.parent / "response.json").exists() 509 510 511def test_public_evidence_failure_writes_no_response( 512 tmp_path: Path, 513 monkeypatch: pytest.MonkeyPatch, 514) -> None: 515 request_path, _source_bundle, _payload = _write_build_request(tmp_path) 516 private_shape = "uv 0.11.4 (builder.local)" 517 518 monkeypatch.setattr( 519 common, 520 "run", 521 lambda _argv, **kwargs: ( 522 _completed(_tool_stdout(uv=private_shape)) 523 if "emit python" in (kwargs.get("input_text") or "") 524 else _completed(f"{build_host_macos.CHECKOUT_TOKEN}\n") 525 ), 526 ) 527 monkeypatch.chdir(request_path.parent) 528 529 with pytest.raises(SystemExit): 530 build_host_macos.build_macos(_lane(), request_path) 531 532 assert not (request_path.parent / "response.json").exists() 533 534 535def test_sentinel_requires_exit_zero_and_token() -> None: 536 common.require_success_token(_completed("TOKEN\n"), "TOKEN", "label") 537 with pytest.raises(SystemExit): 538 common.require_success_token(_completed("", returncode=0), "TOKEN", "label") 539 with pytest.raises(SystemExit): 540 common.require_success_token( 541 _completed("TOKEN\n", returncode=1), "TOKEN", "label" 542 ) 543 544 545def test_ssh_argv_from_structured_config() -> None: 546 flag = "-" + "p" 547 548 argv = common.build_ssh_argv(_lane(), ["bash", "-s"]) 549 550 assert argv == [ 551 "ssh", 552 "-o", 553 "BatchMode=yes", 554 flag, 555 "2222", 556 "-i", 557 "~/.ssh/solstone-channel-adapter", 558 "builder@build-host.example", 559 "bash", 560 "-s", 561 ] 562 563 564def test_scp_argv_from_structured_config() -> None: 565 flag = "-" + "P" 566 567 argv = common.build_scp_argv(_lane(), "local", "remote", direction="to") 568 569 assert argv == [ 570 "scp", 571 "-q", 572 "-o", 573 "BatchMode=yes", 574 flag, 575 "2222", 576 "-i", 577 "~/.ssh/solstone-channel-adapter", 578 "local", 579 "builder@build-host.example:remote", 580 ] 581 582 583def test_cleanup_argument_handling(monkeypatch: pytest.MonkeyPatch) -> None: 584 calls: list[tuple[list[str], str]] = [] 585 586 def fake_runner(argv, **kwargs): 587 calls.append((list(argv), kwargs.get("input_text") or "")) 588 return _completed() 589 590 monkeypatch.setattr(common, "run", fake_runner) 591 592 build_host_macos.cleanup(_lane(), "cohort", "f" * 64) 593 proof_host.cleanup( 594 "macos-arm64", _lane(name="proof.macos-arm64"), "cohort", "e" * 64 595 ) 596 597 assert len(calls) == 2 598 assert all("cohort" in script for _argv, script in calls) 599 600 601def test_config_validation_fails_before_side_effects( 602 tmp_path: Path, 603 monkeypatch: pytest.MonkeyPatch, 604) -> None: 605 config = tmp_path / "channel-adapters.json" 606 config.write_text( 607 '{"schema_version": 1, "build": {}, "proof": {}}', encoding="utf-8" 608 ) 609 calls: list[list[str]] = [] 610 611 def fake_runner(argv, **_kwargs): 612 calls.append(list(argv)) 613 return _completed() 614 615 monkeypatch.setenv(common.CONFIG_ENV, str(config)) 616 monkeypatch.setattr(common, "run", fake_runner) 617 618 with pytest.raises(SystemExit): 619 build_host_macos.main(["build-macos", "request.json"]) 620 621 assert calls == [] 622 623 624def test_target_env_keys_coupling() -> None: 625 config_targets = { 626 "linux-x86_64-musl", 627 "linux-aarch64-musl", 628 "macos-arm64", 629 } 630 631 assert config_targets == set(proof_rail.TARGET_ENV_KEYS)