Add evidence-based regression and schema compatibility checks

2026-07-20 14:21:42 -03:00
parent d81286e322
commit 501e6cd589
13 changed files with 1496 additions and 17 deletions
@@ -0,0 +1,186 @@
+"""Tests for conservative ToolStep JSON Schema compatibility."""
+
+from unittest.mock import patch
+
+from cmdforge.preflight import (
+    PreflightReport, _check_toolstep_compatibility, analyze_tool,
+)
+from cmdforge.schema_compat import compare_json_schemas
+from cmdforge.tool import McpStep, PromptStep, Tool, ToolStep
+
+
+class TestSchemaCompatibility:
+    def test_type_mismatch_is_incompatible(self):
+        result = compare_json_schemas(
+            {"type": "string"}, {"type": "integer"}
+        )
+        assert result.state == "incompatible"
+
+    def test_integer_is_accepted_by_number(self):
+        result = compare_json_schemas(
+            {"type": "integer"}, {"type": "number"}
+        )
+        assert result.state == "compatible"
+
+    def test_missing_required_property_is_incompatible(self):
+        result = compare_json_schemas(
+            {"type": "object", "properties": {}},
+            {
+                "type": "object",
+                "properties": {"name": {"type": "string"}},
+                "required": ["name"],
+            },
+        )
+        assert result.state == "incompatible"
+        assert "required" in result.detail
+
+    def test_matching_object_property_is_compatible(self):
+        schema = {
+            "type": "object",
+            "properties": {"name": {"type": "string"}},
+            "required": ["name"],
+        }
+        assert compare_json_schemas(schema, schema).state == "compatible"
+
+    def test_literal_enum_value_is_checked(self):
+        assert compare_json_schemas(
+            {"const": "fast"}, {"type": "string", "enum": ["fast", "slow"]}
+        ).state == "compatible"
+        assert compare_json_schemas(
+            {"const": "invalid"}, {"type": "string", "enum": ["fast"]}
+        ).state == "incompatible"
+
+    def test_unconstrained_producer_does_not_satisfy_consumer_enum(self):
+        result = compare_json_schemas(
+            {"type": "string"}, {"type": "string", "enum": ["fast"]}
+        )
+        assert result.state == "incompatible"
+
+    def test_closed_consumer_rejects_open_producer_object(self):
+        result = compare_json_schemas(
+            {"type": "object", "properties": {}},
+            {"type": "object", "properties": {}, "additionalProperties": False},
+        )
+        assert result.state == "incompatible"
+
+    def test_open_producer_cannot_guarantee_typed_optional_consumer_field(self):
+        result = compare_json_schemas(
+            {"type": "object", "properties": {}},
+            {
+                "type": "object",
+                "properties": {"limit": {"type": "integer"}},
+            },
+        )
+        assert result.state == "incompatible"
+        assert "limit" in result.detail
+
+    def test_unproven_pattern_is_unknown(self):
+        result = compare_json_schemas(
+            {"type": "string"}, {"type": "string", "pattern": "^[a-z]+$"}
+        )
+        assert result.state == "unknown"
+
+
+class TestToolStepCompatibility:
+    @staticmethod
+    def child(input_schema):
+        return Tool(name="child", input_schema=input_schema)
+
+    def analyze(self, parent, child):
+        report = PreflightReport()
+        with patch("cmdforge.tool.load_tool", return_value=child):
+            _check_toolstep_compatibility(parent, report)
+        return report
+
+    def test_compatible_stdin_contract(self):
+        parent = Tool(
+            name="parent",
+            steps=[ToolStep(tool="child", output_var="result")],
+        )
+        child = self.child({
+            "type": "object",
+            "properties": {"input": {"type": "string"}},
+            "required": ["input"],
+        })
+        report = self.analyze(parent, child)
+        assert report.compatibility[0]["state"] == "compatible"
+
+    def test_upstream_serialization_mismatch_is_reported(self):
+        producer = PromptStep(
+            prompt="produce", provider="mock", output_var="value",
+            output_schema={"type": "string"},
+        )
+        parent = Tool(
+            name="parent",
+            steps=[
+                producer,
+                ToolStep(
+                    tool="child", output_var="result", input_template="{value}"
+                ),
+            ],
+        )
+        child = self.child({
+            "type": "object",
+            "properties": {"input": {"type": "integer"}},
+            "required": ["input"],
+        })
+        report = self.analyze(parent, child)
+        finding = report.compatibility[0]
+        assert finding["state"] == "incompatible"
+        assert finding["upstream_schema"] == {"type": "string"}
+        assert any("incompatible" in warning for warning in report.warnings)
+
+    def test_missing_required_argument_is_incompatible(self):
+        parent = Tool(
+            name="parent",
+            steps=[ToolStep(tool="child", output_var="result")],
+        )
+        child = self.child({
+            "type": "object",
+            "properties": {
+                "input": {"type": "string"},
+                "limit": {"type": "integer"},
+            },
+            "required": ["input", "limit"],
+        })
+        report = self.analyze(parent, child)
+        assert report.compatibility[0]["state"] == "incompatible"
+        assert "limit" in report.compatibility[0]["detail"]
+
+    def test_explicit_argument_reflects_string_transport(self):
+        parent = Tool(
+            name="parent",
+            steps=[ToolStep(
+                tool="child", output_var="result", args={"limit": "{input}"}
+            )],
+        )
+        child = self.child({
+            "type": "object",
+            "properties": {
+                "input": {"type": "string"},
+                "limit": {"type": "integer"},
+            },
+            "required": ["input", "limit"],
+        })
+        assert self.analyze(parent, child).compatibility[0]["state"] == "incompatible"
+
+    def test_mcp_steps_are_ignored(self):
+        parent = Tool(
+            name="parent",
+            steps=[McpStep(server="remote", tool="same-name", output_var="result")],
+        )
+        report = PreflightReport()
+        with patch("cmdforge.tool.load_tool") as loader:
+            _check_toolstep_compatibility(parent, report)
+        loader.assert_not_called()
+        assert report.compatibility == []
+
+    def test_server_mode_does_not_resolve_local_tools(self):
+        parent = Tool(
+            name="published",
+            steps=[ToolStep(tool="child", output_var="result")],
+        )
+        with patch("cmdforge.tool.load_tool") as loader:
+            report = analyze_tool(parent, check_local_dependencies=False)
+        loader.assert_not_called()
+        assert report.compatibility == []