CmdForge/tests/test_schema_compat.py

"""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 == []