source/space/tests/test_app.py

from __future__ import annotations

import hashlib
import importlib.util
import json
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch

APP_PATH = Path(__file__).resolve().parents[1] / "app.py"
SPEC = importlib.util.spec_from_file_location("barunaction_space_app", APP_PATH)
assert SPEC is not None and SPEC.loader is not None
app = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(app)


class _Record:
    def __init__(self, payload: dict[str, object]) -> None:
        self.payload = payload

    def to_dict(self) -> dict[str, object]:
        return self.payload


class _Compiler:
    def infer(self, **_: object) -> SimpleNamespace:
        action = {
            "calls": [{"args": {"query": "Cubbon Park"}, "tool": "show_map"}],
            "decision": "CALL",
            "mode": "SINGLE",
        }
        policy = {
            "authorization_required": True,
            "confirmation_required": False,
            "execution_permitted": False,
            "proposed_call_count": 1,
            "reason_codes": ["external_authorization_required", "model_output_is_proposal_only"],
            "side_effecting_tools": [],
        }
        return SimpleNamespace(
            action=_Record(action),
            candidate_id="candidate-v2",
            checkpoint_format="float",
            error=None,
            generated_tokens=31,
            ok=True,
            policy=_Record(policy),
            prompt_sha256="a" * 64,
            prompt_tokens=121,
            raw_output=json.dumps(action, separators=(",", ":")),
        )


class PublicSpaceTests(unittest.TestCase):
    def test_success_is_a_validated_nonexecuting_proposal(self) -> None:
        with patch.object(app, "_get_compiler", return_value=_Compiler()):
            status, action, safety, raw, provenance = app.compile_action(
                "Show me Cubbon Park",
                app.DEFAULT_TOOLS_JSON,
                "{}",
                app.DEFAULT_NOW,
            )

        self.assertIn("Validated Action IR proposal", status)
        self.assertIn("Nothing was executed", status)
        self.assertEqual(action["calls"][0]["tool"], "show_map")
        self.assertFalse(safety["execution_permitted"])
        self.assertFalse(safety["external_side_effects"])
        self.assertFalse(safety["space_executes_tools"])
        self.assertIn('"show_map"', raw)
        self.assertTrue(provenance["checkpoint_verified"])
        self.assertEqual(provenance["revision"], "candidate-v2")

    def test_invalid_json_fails_before_model_load(self) -> None:
        with patch.object(app, "_get_compiler") as loader:
            status, action, safety, raw, provenance = app.compile_action(
                "Turn on the flashlight",
                '[{"name":"one","name":"two"}]',
                "{}",
                app.DEFAULT_NOW,
            )

        loader.assert_not_called()
        self.assertIn("invalid_json", status)
        self.assertIsNone(action)
        self.assertFalse(safety["execution_permitted"])
        self.assertEqual(raw, "")
        self.assertFalse(provenance["checkpoint_verified"])

    def test_load_failure_does_not_expose_exception_text(self) -> None:
        with patch.object(
            app,
            "_get_compiler",
            side_effect=RuntimeError("private-token-value"),
        ):
            result = app.compile_action(
                "Open Wi-Fi settings",
                app.DEFAULT_TOOLS_JSON,
                "{}",
                app.DEFAULT_NOW,
            )

        combined = json.dumps(result)
        self.assertIn("model_unavailable", combined)
        self.assertNotIn("private-token-value", combined)
        self.assertFalse(result[2]["execution_permitted"])

    def test_checkpoint_verifier_is_fail_closed(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            model_file = root / "model.safetensors"
            manifest_file = root / "checkpoint_manifest.json"
            model_file.write_bytes(b"model")
            manifest_file.write_bytes(b"manifest")
            expected = hashlib.sha256(b"model").hexdigest()
            manifest_expected = hashlib.sha256(b"manifest").hexdigest()

            with (
                patch.object(app, "EXPECTED_CHECKPOINT_SHA256", {"model.safetensors": expected}),
                patch.object(app, "CHECKPOINT_MANIFEST_SHA256", manifest_expected),
            ):
                app._verify_checkpoint_files(root)
                model_file.write_bytes(b"tampered")
                with self.assertRaisesRegex(RuntimeError, "digest mismatch"):
                    app._verify_checkpoint_files(root)


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