zcbot/tests/test_software_contracts.py

from __future__ import annotations

import json
import tempfile
import unittest
from pathlib import Path
from shutil import copy2

from core.software_contracts import (
    CONTRACT_ROOT,
    DEFAULT_CAPABILITIES,
    _ContractRegistry,
    get_contract,
    node_supports_request,
    version_at_least,
)


def _request(plot_type: str) -> dict:
    roles = {"input": "sample", "x": "x", "y": "y"}
    if plot_type in {"contour", "surface_3d", "ternary", "heatmap"}:
        roles["z"] = "z"
    return {
        "schema_version": 2,
        "inputs": [{
            "key": "sample",
            "artifact_id": "f4186347-65cc-4f07-9c26-bf11992beef8",
        }],
        "operation": {"plot": {"type": plot_type, "series": [roles]}},
        "outputs": [{"key": "figure_png", "type": "figure", "format": "png"}],
    }


class SoftwareContractTests(unittest.TestCase):
    def test_origin_contract_drives_schema_outputs_and_summary(self) -> None:
        contract = get_contract("origin.plot@v2")
        request = _request("line")
        normalized, digest = contract.normalize_request(request)
        self.assertEqual(normalized, request)
        self.assertEqual(len(digest), 64)
        self.assertEqual(contract.output_namespace, "origin")
        self.assertEqual(
            set(contract.expected_outputs(request)),
            {"figure_png", "plot_spec", "provenance"},
        )
        self.assertEqual(DEFAULT_CAPABILITIES, ("origin.plot@v2",))

    def test_adapter_version_and_features_gate_additive_requests(self) -> None:
        contract = get_contract("origin.plot@v2")
        legacy_runtime = {
            "available_slots": 1,
            "origin": {"health": "ready", "adapter_version": "0.3.0"},
        }
        self.assertFalse(node_supports_request(contract, _request("line"), legacy_runtime))
        self.assertFalse(node_supports_request(contract, _request("heatmap"), legacy_runtime))
        current_runtime = {
            "capability_runtime": {
                "origin.plot@v2": {
                    "health": "ready",
                    "available_slots": 1,
                    "adapter_version": "0.5.0",
                    "features": ["line", "heatmap"],
                }
            }
        }
        self.assertTrue(node_supports_request(contract, _request("heatmap"), current_runtime))
        self.assertFalse(node_supports_request(contract, _request("bar"), current_runtime))

        multi_panel = _request("line")
        multi_panel["operation"]["plot"] = {
            "type": "multi_panel",
            "layout": {"rows": 1, "columns": 1},
            "panels": [{
                "key": "main",
                "series": [{
                    "input": "sample", "x": "x", "y": "y", "kind": "line",
                }],
            }],
        }
        normalized, _ = contract.normalize_request(multi_panel)
        self.assertEqual(normalized, multi_panel)
        current_runtime["capability_runtime"]["origin.plot@v2"].update({
            "adapter_version": "0.6.0",
            "features": ["multi_panel"],
        })
        self.assertFalse(node_supports_request(contract, multi_panel, current_runtime))
        current_runtime["capability_runtime"]["origin.plot@v2"].update({
            "adapter_version": "0.8.1",
        })
        self.assertTrue(node_supports_request(contract, multi_panel, current_runtime))

        recipe = json.loads(json.dumps(multi_panel))
        recipe["operation"]["plot"].update({
            "type": "recipe",
            "recipe_version": 1,
        })
        normalized, _ = contract.normalize_request(recipe)
        self.assertEqual(normalized, recipe)
        current_runtime["capability_runtime"]["origin.plot@v2"].update({
            "adapter_version": "0.8.1",
            "features": ["recipe"],
        })
        self.assertFalse(node_supports_request(contract, recipe, current_runtime))
        current_runtime["capability_runtime"]["origin.plot@v2"].update({
            "adapter_version": "0.9.0",
        })
        self.assertTrue(node_supports_request(contract, recipe, current_runtime))

        recipe["operation"]["plot"].pop("recipe_version")
        with self.assertRaises(ValueError):
            contract.normalize_request(recipe)

        bubble = _request("line")
        bubble["operation"]["plot"] = {
            "type": "bubble",
            "series": [{
                "input": "sample", "x": "x", "y": "y", "size": "particle_size",
            }],
        }
        normalized, _ = contract.normalize_request(bubble)
        self.assertEqual(normalized, bubble)
        self.assertFalse(node_supports_request(contract, bubble, current_runtime))
        current_runtime["capability_runtime"]["origin.plot@v2"].update({
            "adapter_version": "0.7.0",
            "features": ["bubble"],
        })
        self.assertTrue(node_supports_request(contract, bubble, current_runtime))

        area = _request("line")
        area["operation"]["plot"]["type"] = "area"
        normalized, _ = contract.normalize_request(area)
        self.assertEqual(normalized, area)
        self.assertFalse(node_supports_request(contract, area, current_runtime))
        current_runtime["capability_runtime"]["origin.plot@v2"].update({
            "adapter_version": "0.8.0",
            "features": ["area"],
        })
        self.assertTrue(node_supports_request(contract, area, current_runtime))

    def test_semantic_version_comparison_is_numeric(self) -> None:
        self.assertTrue(version_at_least("0.10.0", "0.4.0"))
        self.assertFalse(version_at_least("0.3.9", "0.4.0"))

    def test_registry_hot_reloads_and_retains_last_valid_snapshot(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            target = root / "origin.plot.v2.json"
            copy2(CONTRACT_ROOT / target.name, target)
            registry = _ContractRegistry(root)
            original = registry.contracts()["origin.plot@v2"]

            value = json.loads(target.read_text(encoding="utf-8"))
            value["display_name"] = "OriginPro hot reload"
            target.write_text(json.dumps(value), encoding="utf-8")
            refreshed = registry.contracts()["origin.plot@v2"]
            self.assertEqual(refreshed.display_name, "OriginPro hot reload")

            target.write_text("{invalid", encoding="utf-8")
            with self.assertLogs("core.software_contracts", level="ERROR"):
                retained = registry.contracts()["origin.plot@v2"]
            self.assertEqual(retained.display_name, "OriginPro hot reload")
            self.assertNotEqual(original.display_name, retained.display_name)


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