test_server.py

import unittest
from unittest.mock import patch

import server


ID = "a" * 32


class CarrierTests(unittest.TestCase):
    def setUp(self):
        server.DESCRIPTIONS.clear()

    def test_publish_changes_only_description_and_status_stays_fixed(self):
        receipt = server.publish({"connector_id": ID, "description": "Instructor-edited text"})
        self.assertEqual(receipt["path"], "/mcp/" + ID)
        code, listing = server.rpc(ID, {"jsonrpc": "2.0", "id": 1, "method": "tools/list"},
                                   "tools/list", "")
        self.assertEqual(code, 200)
        tool = listing["result"]["tools"][0]
        self.assertEqual(tool["description"], "Instructor-edited text")
        self.assertEqual(tool["inputSchema"], server.INPUT_SCHEMA)
        code, call = server.rpc(ID, {"jsonrpc": "2.0", "id": 2, "method": "tools/call",
                                     "params": {"name": "get_carrier_status",
                                                "arguments": {"shipment_id": "HMS-77"}}},
                                "tools/call", "get_carrier_status")
        self.assertEqual(code, 200)
        self.assertIn("16:00 UTC", call["result"]["content"][0]["text"])
        self.assertNotIn("Instructor-edited", str(call))

    def test_invalid_id_and_description_rejected_and_expiry_enforced(self):
        with self.assertRaises(ValueError):
            server.publish({"connector_id": "bad", "description": "text"})
        with self.assertRaises(ValueError):
            server.publish({"connector_id": ID, "description": "  "})
        server.publish({"connector_id": ID, "description": server.CLEAN_DESCRIPTION})
        with patch.object(server.time, "time", return_value=server.DESCRIPTIONS[ID]["expires"] + 1):
            self.assertIsNone(server.catalog(ID))

    def test_rpc_rejects_header_and_schema_mismatch(self):
        server.publish({"connector_id": ID, "description": "Clean"})
        code, _ = server.rpc(ID, {"jsonrpc": "2.0", "id": 1, "method": "tools/list"},
                             "tools/call", "")
        self.assertEqual(code, 400)
        code, _ = server.rpc(ID, {"jsonrpc": "2.0", "id": 2, "method": "tools/call",
                                  "params": {"name": "get_carrier_status",
                                             "arguments": {"shipment_id": "HMS-77", "extra": 1}}},
                             "tools/call", "get_carrier_status")
        self.assertEqual(code, 400)


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