Bind Python client qualification to exact assertions

scripts/test_pre1_qualification_case.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 from __future__ import annotations
 
+import copy
 import importlib.util
 import unittest
 from pathlib import Path
@@ -23,6 +24,18 @@ def test_description_is_complete_and_content_free(self) -> None:
         self.assertEqual(value["component"], module.COMPONENT)
         self.assertEqual(value["scenarios"], sorted(module.SCENARIO_COMMANDS))
         self.assertTrue(value["commands_sha256"].startswith("sha256:"))
+        self.assertTrue(value["semantic_binding"]["exact_assertion_per_scenario"])
+        self.assertEqual(
+            set(value["semantic_binding"]["cell_dimensions_consumed"]),
+            {
+                "runtime",
+                "target",
+                "directory_flavor",
+                "mode",
+                "cell_id",
+                "scenario_id",
+            },
+        )
         self.assertTrue(value["non_authorizing"])
 
     def test_cell_parser_rejects_wrong_component_and_boundary(self) -> None:
@@ -44,9 +57,115 @@ def test_cell_parser_rejects_wrong_component_and_boundary(self) -> None:
     def test_referenced_test_files_exist(self) -> None:
         commands = [module.SUPPORT_COMMAND, *module.SCENARIO_COMMANDS.values()]
         for command in commands:
-            for token in command:
-                if token.startswith(("tests/", "scripts/")) and "." in Path(token).name:
-                    self.assertTrue((ROOT / token).is_file(), token)
+            node_id = command[4]
+            assertion = node_id.rsplit("::", 1)[-1]
+            source = ROOT / node_id.split("::", 1)[0]
+            marker = f"def {assertion}("
+            self.assertTrue(source.is_file(), source)
+            self.assertIn(marker, source.read_text())
+
+    def test_every_scenario_has_one_unique_exact_assertion(self) -> None:
+        self.assertEqual(set(module.SCENARIO_CASES), set(module.SCENARIO_COMMANDS))
+        assertions = [row["assertion"] for row in module.SCENARIO_CASES.values()]
+        self.assertEqual(len(assertions), len(set(assertions)))
+        self.assertNotIn(module.SUPPORT_CASE["assertion"], assertions)
+
+    def test_semantic_context_negative_controls_change_the_binding(self) -> None:
+        base = (
+            "client-python|cpython-3.11|macos-arm64|rust|restricted",
+            "rate-limit",
+            "sha256:" + "a" * 64,
+            "sha256:" + "b" * 64,
+            "sha256:" + "c" * 64,
+            "sha256:" + "d" * 64,
+        )
+        expected = module.canonical_sha256(module.semantic_execution_context(*base))
+        mutations = [
+            (base[0].replace("cpython-3.11", "cpython-3.14"), *base[1:]),
+            (base[0].replace("macos-arm64", "windows-x86_64"), *base[1:]),
+            (base[0].replace("|rust|", "|php|"), *base[1:]),
+            (base[0].replace("restricted", "public"), *base[1:]),
+            (base[0], "disk-full", *base[2:]),
+            (*base[:2], "sha256:" + "e" * 64, *base[3:]),
+            (*base[:3], "sha256:" + "e" * 64, *base[4:]),
+            (*base[:4], "sha256:" + "e" * 64, base[5]),
+            (*base[:5], "sha256:" + "e" * 64),
+        ]
+        for mutation in mutations:
+            with self.subTest(mutation=mutation[:2]):
+                try:
+                    observed = module.canonical_sha256(
+                        module.semantic_execution_context(*mutation)
+                    )
+                except ValueError:
+                    continue
+                self.assertNotEqual(observed, expected)
+
+    def test_environment_manifest_requires_offline_package_smoke_and_digest(self) -> None:
+        candidate = "sha256:" + "a" * 64
+        artifacts = "sha256:" + "b" * 64
+        runtime_map = "sha256:" + "c" * 64
+        value = {
+            "schema": "iicp.pre1-qualification-environment.v1",
+            "status": "READY",
+            "target": "macos-arm64",
+            "bindings": {
+                "candidate_manifest_sha256": candidate,
+                "artifact_materialization_sha256": artifacts,
+                "runtime_map_sha256": runtime_map,
+                "runner_inventory_sha256": "sha256:" + "d" * 64,
+            },
+            "network": {},
+            "source_state": {},
+            "runtimes": {
+                "cpython-3.11": {
+                    "lock_inputs_sha256": "sha256:" + "e" * 64,
+                    "dependency_cache_sha256": "sha256:" + "f" * 64,
+                    "online_prepare_status": "PASS",
+                    "offline_install_status": "PASS",
+                    "package_artifact_smoke_status": "PASS",
+                    "egress_disabled_during_offline": True,
+                    "empty_volatile_cache_at_start": True,
+                }
+            },
+            "content_free": True,
+            "secrets_present": False,
+            "non_authorizing": True,
+            "environment_sha256": None,
+        }
+        value["environment_sha256"] = module.canonical_sha256(value)
+        self.assertEqual(
+            module._validate_environment_manifest(
+                value,
+                target="macos-arm64",
+                runtime="cpython-3.11",
+                candidate_digest=candidate,
+                materialization_digest=artifacts,
+                runtime_map_digest=runtime_map,
+            ),
+            value["environment_sha256"],
+        )
+        for mutation in (
+            ("offline_install_status", "FAIL"),
+            ("package_artifact_smoke_status", "FAIL"),
+            ("egress_disabled_during_offline", False),
+        ):
+            changed = copy.deepcopy(value)
+            changed["runtimes"]["cpython-3.11"][mutation[0]] = mutation[1]
+            changed["environment_sha256"] = None
+            changed["environment_sha256"] = module.canonical_sha256(changed)
+            with self.assertRaises(ValueError):
+                module._validate_environment_manifest(
+                    changed,
+                    target="macos-arm64",
+                    runtime="cpython-3.11",
+                    candidate_digest=candidate,
+                    materialization_digest=artifacts,
+                    runtime_map_digest=runtime_map,
+                )
+
+    def test_python_runtime_is_exactly_bound_to_cell(self) -> None:
+        self.assertEqual(module.expected_runtime_version("cpython-3.11", {}), "3.11")
 
 
 if __name__ == "__main__":

tests/test_dispatch_ticket_trust_crypto.py

@@ -32,6 +32,43 @@ def _decision(vector: dict, keys: dict[str, dict], signature_valid: bool) -> str
     return "accept_anchored"
 
 
+def _assert_fixture_decision(vector_id: str, expected: str) -> None:
+    fixture = json.loads(
+        (Path(__file__).parents[1] / "parity" / "dispatch-ticket-trust-v2-crypto.json").read_text()
+    )
+    domain = _decode(fixture["domain_separator_b64url"])
+    keys = {key["key_id"]: key for key in fixture["keys"]}
+    vector = next(value for value in fixture["vectors"] if value["id"] == vector_id)
+    public_key = Ed25519PublicKey.from_public_bytes(
+        _decode(keys[vector["claims"]["key_id"]]["public_key_b64url"])
+    )
+    try:
+        public_key.verify(
+            _decode(vector["signature_b64url"]),
+            domain + _canonical(vector["claims"]),
+        )
+        signature_valid = True
+    except InvalidSignature:
+        signature_valid = False
+    assert _decision(vector, keys, signature_valid) == expected
+
+
+def test_expired_dispatch_ticket_key_fails_closed() -> None:
+    _assert_fixture_decision("expired_key_refused", "reject_key_expired")
+
+
+def test_replayed_dispatch_ticket_fails_closed() -> None:
+    _assert_fixture_decision("local_replay_refused", "reject_local_replay")
+
+
+def test_revoked_dispatch_ticket_key_fails_closed_after_rotation() -> None:
+    _assert_fixture_decision("revoked_key_refused", "reject_key_revoked")
+
+
+def test_tampered_dispatch_ticket_signature_fails_closed() -> None:
+    _assert_fixture_decision("tampered_claim_refused", "reject_signature")
+
+
 def test_dispatch_ticket_v2_signed_vectors_are_portable() -> None:
     fixture = json.loads((Path(__file__).parents[1] / "parity" / "dispatch-ticket-trust-v2-crypto.json").read_text())
     domain = _decode(fixture["domain_separator_b64url"])

tests/test_endpoint_security.py

@@ -82,6 +82,27 @@ async def test_private_provider_requires_opt_in_and_uses_pinned_transport(monkey
         thread.join(timeout=2)
 
 
+@pytest.mark.asyncio
+async def test_tls_handshake_failure_is_transient_and_bounded() -> None:
+    server = ThreadingHTTPServer(("127.0.0.1", 0), _ProviderHandler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        with pytest.raises(IicpError) as failure:
+            await post_json(
+                f"https://127.0.0.1:{server.server_port}/task",
+                {},
+                timeout_ms=2_000,
+                tls_verify=True,
+            )
+        assert failure.value.code == "IICP-E004"
+        assert failure.value.retryable is True
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=2)
+
+
 def test_shared_fixture_matches_python_policy() -> None:
     fixture = json.loads((Path(__file__).parent / "fixtures" / "endpoint-security-v1.json").read_text())
     for vector in fixture["address_vectors"]:

tests/test_pre1_release_boundaries.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+import sys
+import tomllib
+from pathlib import Path
+
+import iicp_client
+
+ROOT = Path(__file__).parents[1]
+PROJECT = tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8"))["project"]
+QUALITY_RUNNER = (ROOT / "scripts" / "run_sdk_quality.py").read_text(encoding="utf-8")
+RELEASE_WORKFLOW = (ROOT / ".github" / "workflows" / "release.yml").read_text(encoding="utf-8")
+
+
+def test_minimum_python_version_is_declared_and_candidate_remains_pre1() -> None:
+    assert PROJECT["requires-python"] == ">=3.11"
+    assert sys.version_info >= (3, 11)
+    assert PROJECT["version"].split(".", 1)[0] == "0"
+
+
+def test_package_version_self_report_matches_candidate_contract() -> None:
+    assert PROJECT["name"] == "iicp-client"
+    assert iicp_client.__version__ == PROJECT["version"]
+
+
+def test_offline_candidate_contract_pins_locked_release_inputs() -> None:
+    lock = (ROOT / "uv.lock").read_text(encoding="utf-8")
+    assert 'name = "iicp-client"' in lock
+    assert f'version = "{PROJECT["version"]}"' in lock
+    assert '"uv", "run", "--isolated", "--python", runtime, "--locked"' in QUALITY_RUNNER
+    assert "python -m build" in RELEASE_WORKFLOW
+    assert "python -m venv /tmp/iicp-release-smoke" in RELEASE_WORKFLOW
+    assert "pip install dist/*.whl" in RELEASE_WORKFLOW

tests/test_pre1_runtime_boundaries.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from iicp_client.identity import NodeIdentity, load_node, node_path, save_node
+
+
+def test_malformed_node_configuration_fails_closed(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("IICP_HOME", str(tmp_path))
+    path = node_path("malformed")
+    path.write_text("{not-json", encoding="utf-8")
+    with pytest.raises(json.JSONDecodeError):
+        load_node("malformed")
+
+
+def test_missing_node_configuration_is_explicitly_absent(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("IICP_HOME", str(tmp_path))
+    assert load_node("missing") is None
+
+
+def test_permission_denied_config_write_leaves_no_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("IICP_HOME", str(tmp_path))
+    node = NodeIdentity.generate(
+        operator_id="operator-test",
+        name="permission-denied",
+        backend_url="http://127.0.0.1:11434",
+        model="test-model",
+    )
+    destination = node_path(node.name)
+    original = Path.write_text
+
+    def refuse(path: Path, *_args: object, **_kwargs: object) -> int:
+        if path == destination:
+            raise PermissionError("simulated permission denied")
+        return original(path, *_args, **_kwargs)
+
+    monkeypatch.setattr(Path, "write_text", refuse)
+    with pytest.raises(PermissionError, match="permission denied"):
+        save_node(node)
+    assert not destination.exists()

tests/test_restricted_directory.py

@@ -35,6 +35,12 @@ def test_context_and_decision_fail_closed(tmp_path):
         SecretRef("file", str(secret)).resolve()
 
 
+def test_missing_restricted_credential_fails_before_network(monkeypatch):
+    monkeypatch.delenv("IICP_TEST_MISSING_RESTRICTED_MEMBER", raising=False)
+    with pytest.raises(IicpError, match="credential is unavailable"):
+        SecretRef("environment", "IICP_TEST_MISSING_RESTRICTED_MEMBER").resolve()
+
+
 @pytest.mark.asyncio
 async def test_restricted_discovery_sends_membership_and_requires_decision(monkeypatch):
     seen: dict[str, str] = {}
@@ -52,6 +58,36 @@ async def handler(request: httpx.Request) -> httpx.Response:
     assert seen["x-iicp-subject-id"] == "client-a"
 
 
+@pytest.mark.asyncio
+async def test_restricted_directory_failure_does_not_fall_back(monkeypatch):
+    calls = []
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        calls.append(str(request.url))
+        raise httpx.ConnectError("isolated directory unavailable", request=request)
+
+    original = httpx.AsyncClient
+    monkeypatch.setattr(
+        httpx,
+        "AsyncClient",
+        lambda *args, **kwargs: original(
+            transport=httpx.MockTransport(handler),
+            **{key: value for key, value in kwargs.items() if key != "transport"},
+        ),
+    )
+    client = IicpClient(
+        ClientConfig(
+            directory_url="https://directory.test",
+            route_discovery_mode="ticketed",
+            restricted_directory=context(),
+        )
+    )
+    with pytest.raises(IicpError, match="Network error"):
+        await client.discover_async("urn:iicp:intent:llm:chat:v1")
+    assert len(calls) == 1
+    assert calls[0].startswith("https://directory.test/v1/discover?")
+
+
 def test_restricted_mode_refuses_legacy_fallback():
     with pytest.raises(ValueError, match="legacy"):
         IicpClient(ClientConfig(route_discovery_mode="legacy", restricted_directory=context()))