qbit-scripting/deploy/test_deploy.py

# Copyright (C) 2026 Kingsrook, LLC.
# SPDX-License-Identifier: Apache-2.0
"""Static rendering and fail-closed snapshot tests; no containers or clusters."""
import copy
import hashlib
import json
from pathlib import Path
import re
import subprocess
import sys
import tempfile
from typing import Any
import unittest

import render_qualification as renderer
import snapshot_execution as snapshot


def fixture() -> dict[str, Any]:
   """Supply synthetic identifiers for rendering tests only."""
   # Synthetic non-secret identifiers; fixtures are never applied.
   return {
      "enabled": True, "namespace": "qqq-qual-unit", "image": "invalid.example/unit@sha256:" + "a" * 64,
      "sandboxRuntimeClass": "unit-sandbox", "podSecurityVersion": "v1.34",
      "tlsSecret": "unit-server", "trustSecret": "unit-clients", "passwordSecret": "unit-passwords",
      "certificateNamespaces": {"b" * 64: "unit-client"}, "clientNamespace": "unit-client",
      "clientPodLabels": {"app": "unit-client"}, "nodeSelector": {"qualification-pool": "unit"},
      "profile": "unit-profile", "toolchain": "unit-toolchain", "evidence": "unit-evidence",
      "requiredEvidence": "unit-evidence", "cpuMillis": 1000, "memoryMiB": 1024, "pidLimit": 256,
      "pidPolicyEvidence": "unit-node-policy", "tmpMiB": 16, "startupSeconds": 120,
      "slots": 1, "connections": 2, "coldOperations": 1, "entries": 8, "codeBytes": 8388608,
      "guestMemoryMiB": 64, "idleMillis": 30000, "fuel": 100000000,
   }


class DeploymentTest(unittest.TestCase):
   """Verify generated boundaries without a deployment or native runtime."""

   def test_disabled_input_and_every_missing_field_refuse(self) -> None:
      """An incomplete descriptor never yields runnable manifests."""
      disabled = json.loads((Path(__file__).parent / "qualification.disabled.json").read_text())
      with self.assertRaises(ValueError):
         renderer.render(disabled)
      for field in renderer.REQUIRED:
         value = fixture()
         value.pop(field)
         with self.subTest(field=field), self.assertRaises(ValueError):
            renderer.render(value)

   def test_fixed_security_network_and_exact_supervisor_config(self) -> None:
      """Keep secrets referenced and network authority narrowly selected."""
      manifest = renderer.render(fixture())
      by_kind = {entry["kind"]: entry for entry in manifest["items"] if entry["kind"] != "NetworkPolicy"}
      pod = by_kind["Deployment"]["spec"]["template"]["spec"]
      container = pod["containers"][0]
      self.assertFalse(pod["automountServiceAccountToken"])
      self.assertFalse(pod["enableServiceLinks"])
      self.assertEqual(65532, pod["securityContext"]["runAsUser"])
      self.assertEqual({"type": "RuntimeDefault"}, pod["securityContext"]["seccompProfile"])
      self.assertEqual(["ALL"], container["securityContext"]["capabilities"]["drop"])
      self.assertTrue(container["securityContext"]["readOnlyRootFilesystem"])
      self.assertFalse(container["securityContext"]["allowPrivilegeEscalation"])
      self.assertNotIn("livenessProbe", container)
      self.assertNotIn("pidsLimit", pod)
      self.assertTrue(all("valueFrom" in item and "value" not in item for item in container["env"]))
      settings = json.loads(by_kind["ConfigMap"]["data"]["config.json"])
      self.assertEqual(snapshot.WASM_SHA, settings["runtimeSha256"])
      self.assertEqual(snapshot.EXTISM_SHA, settings["nativeSha256"])
      self.assertEqual("0.0.0.0", settings["bind"])
      self.assertFalse(any("password" in key.lower() for key in settings))
      policies = [item["spec"] for item in manifest["items"] if item["kind"] == "NetworkPolicy"]
      self.assertEqual([], policies[0]["egress"])
      self.assertEqual([], policies[0]["ingress"])
      peer = policies[1]["ingress"][0]["from"]
      self.assertEqual(1, len(peer))
      self.assertEqual({"namespaceSelector", "podSelector"}, set(peer[0]))
      self.assertEqual([{"protocol": "TCP", "port": 18443}], policies[1]["ingress"][0]["ports"])
      self.assertNotIn("Secret", by_kind)
      self.assertNotIn("RuntimeClass", by_kind)

   def test_malformed_policy_cannot_fall_back(self) -> None:
      """Reject permissive substitutes and mixed JSON types."""
      changes = {"image": "example/unit:latest", "namespace": "production", "sandboxRuntimeClass": "runc",
               "certificateNamespaces": {"*": "unit"}, "clientPodLabels": {}, "nodeSelector": {},
               "pidLimit": 0, "cpuMillis": True, "memoryMiB": 32, "slots": 65,
               "requiredEvidence": "different", "enabled": "true", "profile": "REPLACE_ME"}
      for field, value in changes.items():
         config = fixture()
         config[field] = value
         with self.subTest(field=field), self.assertRaises(ValueError):
            renderer.render(config)
      config = fixture()
      config["secretValue"] = "not-accepted"
      with self.assertRaises(ValueError):
         renderer.render(config)

   def test_input_is_not_mutated_and_duplicate_json_refused(self) -> None:
      """Preserve caller intent and reject ambiguous repeated keys."""
      config = fixture()
      before = copy.deepcopy(config)
      renderer.render(config)
      self.assertEqual(before, config)
      for hook in (renderer.no_duplicates, snapshot.no_duplicates):
         with self.assertRaises(ValueError):
            json.loads('{"enabled":true,"enabled":false}', object_pairs_hook=hook)

   def test_wire_identity_and_kubernetes_label_grammars_refuse_invalid_values(self) -> None:
      """Refuse values accepted previously but rejected by the real consumers."""
      for value in ("wrong/profile", "x" * 129):
         for key in ("profile", "toolchain", "evidence", "requiredEvidence"):
            config = fixture()
            config[key] = value
            if key in {"evidence", "requiredEvidence"}:
               config["evidence"] = config["requiredEvidence"] = value
            with self.subTest(key=key, value=value), self.assertRaises(ValueError):
               renderer.render(config)
         config = fixture()
         config["certificateNamespaces"] = {"b" * 64: value}
         with self.assertRaises(ValueError):
            renderer.render(config)
      for selectors in ({"app-": "client"}, {"app": "client."}, {"bad..domain/app": "client"},
                        {"-domain/app": "client"}, {"a" * 64 + "/app": "client"}):
         for key in ("clientPodLabels", "nodeSelector"):
            config = fixture()
            config[key] = selectors
            with self.subTest(key=key, selectors=selectors), self.assertRaises(ValueError):
               renderer.render(config)
      config = fixture()
      config["profile"] = "x" * 128
      config["clientPodLabels"] = {"qualified.example/app_name": "client.v1"}
      renderer.render(config)

   def test_complete_positive_fixture_cross_object_contracts(self) -> None:
      """Review the entire rendered fixture and its actual mount/selector linkage."""
      directory = Path(__file__).parent / "fixtures"
      source = json.loads((directory / "synthetic-input.json").read_text(encoding="utf-8"))
      expected = json.loads((directory / "synthetic-manifest.json").read_text(encoding="utf-8"))
      actual = renderer.render(source)
      self.assertEqual(expected, actual)
      self.assertEqual(6, len(actual["items"]))
      identities = [(item["kind"], item["metadata"].get("namespace"), item["metadata"]["name"]) for item in actual["items"]]
      self.assertEqual(len(identities), len(set(identities)))
      by_kind = {item["kind"]: item for item in actual["items"] if item["kind"] != "NetworkPolicy"}
      namespace = by_kind["Namespace"]
      self.assertNotIn("namespace", namespace["metadata"])
      self.assertEqual("restricted", namespace["metadata"]["labels"]["pod-security.kubernetes.io/enforce"])
      for item in actual["items"]:
         if item["kind"] != "Namespace":
            self.assertEqual(namespace["metadata"]["name"], item["metadata"]["namespace"])
      deployment = by_kind["Deployment"]["spec"]
      template = deployment["template"]
      pod = template["spec"]
      container = pod["containers"][0]
      self.assertEqual(deployment["selector"]["matchLabels"], template["metadata"]["labels"])
      self.assertEqual(template["metadata"]["labels"], by_kind["Service"]["spec"]["selector"])
      self.assertEqual("false", template["metadata"]["annotations"]["qqq.kingsrook.com/productionQualified"])
      self.assertEqual("false", template["metadata"]["annotations"]["qqq.kingsrook.com/redistributionCleared"])
      self.assertEqual(source["cpuMillis"], int(container["resources"]["limits"]["cpu"][:-1]))
      self.assertEqual(container["resources"]["requests"]["memory"], container["resources"]["limits"]["memory"])
      self.assertEqual("ClusterIP", by_kind["Service"]["spec"]["type"])
      self.assertNotIn("externalIPs", by_kind["Service"]["spec"])
      self.assertEqual("mtls", by_kind["Service"]["spec"]["ports"][0]["targetPort"])
      self.assertEqual(18443, container["ports"][0]["containerPort"])
      config_map = by_kind["ConfigMap"]
      settings = json.loads(config_map["data"]["config.json"])
      self.assertTrue(config_map["immutable"])
      volumes = {volume["name"]: volume for volume in pod["volumes"]}
      mounts = {mount["name"]: mount for mount in container["volumeMounts"]}
      self.assertEqual(set(volumes), set(mounts))
      self.assertEqual(config_map["metadata"]["name"], volumes["config"]["configMap"]["name"])
      for name, setting, secret_field in (("tls", "keyStore", "tlsSecret"), ("trust", "trustStore", "trustSecret")):
         secret = volumes[name]["secret"]
         self.assertEqual(source[secret_field], secret["secretName"])
         self.assertEqual(0o440, secret["defaultMode"])
         self.assertTrue(mounts[name]["readOnly"])
         self.assertEqual(mounts[name]["mountPath"] + "/" + secret["items"][0]["path"], settings[setting])
      self.assertEqual(0o444, volumes["config"]["configMap"]["defaultMode"])
      self.assertEqual({"medium": "Memory", "sizeLimit": str(source["tmpMiB"]) + "Mi"}, volumes["tmp"]["emptyDir"])
      for forbidden in ("hostPID", "hostIPC", "hostNetwork", "hostUsers", "initContainers", "ephemeralContainers"):
         self.assertNotIn(forbidden, pod)
      self.assertFalse(any("hostPath" in value or "persistentVolumeClaim" in value for value in volumes.values()))
      self.assertEqual(30, pod["terminationGracePeriodSeconds"])
      self.assertNotIn("livenessProbe", container)

   def test_actual_cli_renders_fixture_once_and_refuses_disabled_without_output(self) -> None:
      """Exercise file parsing and exclusive output creation without cluster tools."""
      root = Path(__file__).parent
      with tempfile.TemporaryDirectory() as temporary:
         output = Path(temporary) / "manifest.json"
         command = [sys.executable, str(root / "render_qualification.py"), str(root / "fixtures/synthetic-input.json"), str(output)]
         first = subprocess.run(command, capture_output=True, timeout=10, check=False)
         self.assertEqual(0, first.returncode)
         self.assertEqual(json.loads((root / "fixtures/synthetic-manifest.json").read_text(encoding="utf-8")),
                          json.loads(output.read_text(encoding="utf-8")))
         before = output.read_bytes()
         second = subprocess.run(command, capture_output=True, timeout=10, check=False)
         self.assertNotEqual(0, second.returncode)
         self.assertEqual(before, output.read_bytes())
         disabled_output = Path(temporary) / "disabled.json"
         refused = subprocess.run([sys.executable, str(root / "render_qualification.py"), str(root / "qualification.disabled.json"),
                                   str(disabled_output)], capture_output=True, timeout=10, check=False)
         self.assertNotEqual(0, refused.returncode)
         self.assertFalse(disabled_output.exists())

   def test_rendered_config_matches_actual_closed_java_entrypoint(self) -> None:
      """Catch drift between packaging and the real SupervisorMain schema."""
      java = (Path(__file__).parent.parent / "supervisor/src/main/java/com/kingsrook/scripting/supervisor/SupervisorMain.java").read_text(encoding="utf-8")
      matched = re.search(r"Protocol\.keys\(config, (.*?)\);", java, re.DOTALL)
      self.assertIsNotNone(matched)
      assert matched is not None
      accepted = set(re.findall(r'"([A-Za-z0-9]+)"', matched.group(1)))
      config_map = next(item for item in renderer.render(fixture())["items"] if item["kind"] == "ConfigMap")
      self.assertEqual(accepted, set(json.loads(config_map["data"]["config.json"])))

   def test_context_verifier_refuses_unlisted_file(self) -> None:
      """Even a harmless extra context file cannot silently enter the image."""
      with tempfile.TemporaryDirectory() as temporary:
         root = Path(temporary)
         rows = []
         files = {relative for relative, _ in snapshot.FILES.values()} | {
            "payload/native/libjnidispatch.so", "Containerfile", "metadata/NOTICE-GAPS.md", ".dockerignore",
         }
         for relative in files:
            target = root / relative
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_bytes(b"bounded test bytes")
            rows.append({"path": relative, "sha256": snapshot.digest(target)})
         manifest = {"productionQualified": False, "redistributionCleared": False, "architecture": "linux/amd64",
                     "runtimeSha256": snapshot.WASM_SHA, "nativeSha256": snapshot.EXTISM_SHA, "files": rows}
         (root / "metadata/candidate.json").write_text(json.dumps(manifest), encoding="utf-8")
         (root / "unlisted.txt").write_text("must refuse", encoding="utf-8")
         with self.assertRaisesRegex(ValueError, "Unexpected or missing build-context file"):
            snapshot.verify_snapshot(root)

   def test_snapshot_missing_unknown_and_hash_mismatch_refuse_before_output(self) -> None:
      """Do not create a context after any input-integrity failure."""
      with tempfile.TemporaryDirectory() as temporary:
         output = Path(temporary) / "output"
         for descriptor in ({}, {"certificate": {}}):
            with self.assertRaises(ValueError):
               snapshot.snapshot(descriptor, output)
         source = Path(temporary) / "source"
         source.write_bytes(b"fixed harmless fixture")
         descriptor = {key: {"path": str(source), "sha256": "0" * 64} for key in snapshot.FILES}
         with self.assertRaises(ValueError):
            snapshot.snapshot(descriptor, output)
         self.assertFalse(output.exists())


class PinnedWasmTest(unittest.TestCase):
   """The fixed Wasm pin must always correspond to the recorded source inputs."""

   RUNTIME = Path(__file__).resolve().parent.parent / "wasm-runtime"

   def pin(self) -> dict[str, Any]:
      """Load the recorded pin."""
      return json.loads((self.RUNTIME / "PINNED-WASM.json").read_text(encoding="utf-8"))

   def test_deploy_constant_and_inventories_equal_the_recorded_pin(self) -> None:
      """Every place that names the fixed Wasm hash must agree with PINNED-WASM.json."""
      pin = self.pin()
      self.assertEqual(pin["sha256"], snapshot.WASM_SHA)
      inventory = json.loads((self.RUNTIME / "docs" / "license-inventory.json").read_text(encoding="utf-8"))
      imports = json.loads((self.RUNTIME / "docs" / "wasm-imports.json").read_text(encoding="utf-8"))
      self.assertEqual(pin["sha256"], inventory["artifactSha256"])
      self.assertEqual(pin["sha256"], imports["sha256"])
      self.assertEqual(pin["bytes"], imports["bytes"])
      review = (self.RUNTIME / "docs" / "release-license-review.md").read_text(encoding="utf-8")
      self.assertIn(pin["sha256"], review)

   def test_pin_matches_the_current_source_inputs_and_lockfile(self) -> None:
      """A source or Cargo.lock edit without a reviewed re-pin must fail here, not silently drift."""
      pin = self.pin()
      combined = ""
      for name in pin["sourceInputs"]:
         digest = hashlib.sha256((self.RUNTIME / name).read_bytes()).hexdigest()
         combined += name + "\n" + digest + "\n"
      self.assertEqual(pin["sourceInputsSha256"], hashlib.sha256(combined.encode("utf-8")).hexdigest())
      self.assertEqual(pin["cargoLockSha256"], hashlib.sha256((self.RUNTIME / "Cargo.lock").read_bytes()).hexdigest())

   def test_pin_names_locked_toolchain_and_previous_artifact(self) -> None:
      """The recipe inputs are explicit: no unpinned tool, and the superseded hash stays recorded."""
      pin = self.pin()
      self.assertRegex(pin["sha256"], r"^[0-9a-f]{64}$")
      self.assertNotEqual(pin["sha256"], pin["previous"]["sha256"])
      self.assertEqual("1.97.1", pin["toolchain"]["rust"])
      for key in ("archiveSha256", "clangExeSha256", "libclangDllSha256", "wasiLibcSha256"):
         self.assertRegex(pin["toolchain"]["wasiSdk"][key], r"^[0-9a-f]{64}$")
      self.assertFalse(pin["qualification"].count("productionQualified=true"))
      script = (self.RUNTIME / "Build-Wasm.ps1").read_text(encoding="utf-8")
      for fragment in ("--offline", "--locked", "OutputDir must not already exist", "does not match the pin"):
         self.assertIn(fragment, script)


if __name__ == "__main__":
   unittest.main()