#!/usr/bin/env python3 """ Unit tests for the Marktguru skill & MCP Server. Runs completely offline using standard library unittest. """ import sys import os import json import unittest from datetime import datetime, timezone, timedelta # Add scripts directory to path SCRIPTS_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "scripts")) if SCRIPTS_DIR not in sys.path: sys.path.insert(0, SCRIPTS_DIR) from marktguru_client import ( resolve_supermarket_ids, format_iso_date, parse_offer, list_supported_supermarkets, DEFAULT_ZIP_CODE, _read_cache, _write_cache, _get_cache_dir, ) from marktguru_mcp import process_message, SERVER_NAME, PROTOCOL_VERSION class TestMarktguruClient(unittest.TestCase): def test_default_zip_code(self): self.assertEqual(DEFAULT_ZIP_CODE, "93059") def test_resolve_supermarket_ids_aliases(self): # Aldi maps to Aldi Nord and Aldi Süd aldi_ids = resolve_supermarket_ids(["aldi"]) self.assertIn("127153", aldi_ids) self.assertIn("127139", aldi_ids) # Edeka maps to Edeka and E center edeka_ids = resolve_supermarket_ids(["edeka"]) self.assertIn("127093", edeka_ids) self.assertIn("126699", edeka_ids) # Lidl lidl_ids = resolve_supermarket_ids(["lidl"]) self.assertEqual(lidl_ids, "126679") # Comma-separated string support combined = resolve_supermarket_ids("Lidl, Kaufland") self.assertIn("126679", combined) self.assertIn("126654", combined) # Unknown supermarket is ignored gracefully unknown = resolve_supermarket_ids(["NonExistentMarket123"]) self.assertIsNone(unknown) def test_format_iso_date(self): self.assertEqual(format_iso_date("2026-10-15T12:00:00Z"), "15.10.2026") self.assertEqual(format_iso_date(None), "-") self.assertEqual(format_iso_date(""), "-") self.assertEqual(format_iso_date("invalid-date-format"), "invalid-da") def test_parse_offer_defensive(self): # 1. Complete payload now = datetime.now(timezone.utc) from_date = (now - timedelta(days=1)).isoformat() to_date = (now + timedelta(days=4)).isoformat() raw_complete = { "id": 99999, "price": 1.49, "oldPrice": 2.29, "referencePrice": 5.96, "requiresLoyalityMembership": True, "description": "250g Packung", "product": {"name": "Süßrahmbutter"}, "brand": {"name": "Landliebe"}, "advertisers": [{"name": "EDEKA", "id": "retailers/127093"}], "unit": {"shortName": "kg"}, "volume": 0.25, "validityDates": [{"from": from_date, "to": to_date}] } parsed = parse_offer(raw_complete, "93059") self.assertIsNotNone(parsed) self.assertEqual(parsed["id"], 99999) self.assertEqual(parsed["supermarket"], "EDEKA") self.assertEqual(parsed["price"], 1.49) self.assertEqual(parsed["old_price"], 2.29) self.assertEqual(parsed["discount_percent"], 35) # (2.29 - 1.49) / 2.29 = 34.9% -> 35% self.assertTrue(parsed["loyalty_app_required"]) self.assertTrue(parsed["is_valid_now"]) self.assertGreaterEqual(parsed["days_remaining"], 3) self.assertIn("https://mg2de.b-cdn.net/api/v1/offers/99999", parsed["image_url"]) # 2. Minimal / empty payload raw_empty = {} parsed_min = parse_offer(raw_empty, "93059") self.assertIsNotNone(parsed_min) self.assertEqual(parsed_min["supermarket"], "Unbekannt") self.assertIsNone(parsed_min["price"]) self.assertIsNone(parsed_min["discount_percent"]) # 3. Invalid types (not a dict) self.assertIsNone(parse_offer("not a dict", "93059")) self.assertIsNone(parse_offer(None, "93059")) def test_cache_mechanism(self): test_key = "test_unit_key_123" test_data = {"hello": "world", "timestamp": 12345} _write_cache(test_key, test_data) data, is_fresh = _read_cache(test_key, max_age_seconds=60) self.assertTrue(is_fresh) self.assertEqual(data, test_data) # Stale cache test data_stale, is_fresh_stale = _read_cache(test_key, max_age_seconds=-1) self.assertFalse(is_fresh_stale) self.assertEqual(data_stale, test_data) # still returned as fallback! def test_list_supported_supermarkets(self): catalog = list_supported_supermarkets() self.assertEqual(catalog["default_zip_code"], "93059") self.assertIn("Discounter", catalog["categories"]) self.assertIn("Supermarkt", catalog["categories"]) self.assertIn("aldi", catalog["supported_aliases"]) self.assertIn("edeka", catalog["supported_aliases"]) class TestMarktguruMCP(unittest.TestCase): def test_mcp_initialize(self): req = json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}) resp = process_message(req) self.assertEqual(resp["id"], 1) self.assertEqual(resp["result"]["protocolVersion"], PROTOCOL_VERSION) self.assertEqual(resp["result"]["serverInfo"]["name"], SERVER_NAME) def test_mcp_tools_list(self): req = json.dumps({"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}) resp = process_message(req) tools = [t["name"] for t in resp["result"]["tools"]] self.assertIn("search_supermarket_offers", tools) self.assertIn("compare_supermarket_prices", tools) self.assertIn("list_supported_supermarkets", tools) def test_mcp_tools_call_supermarket_catalog(self): req = json.dumps({ "jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "list_supported_supermarkets", "arguments": {}} }) resp = process_message(req) self.assertFalse(resp["result"]["isError"]) content_text = resp["result"]["content"][0]["text"] data = json.loads(content_text) self.assertEqual(data["default_zip_code"], "93059") def test_mcp_tools_call_missing_query(self): req = json.dumps({ "jsonrpc": "2.0", "id": 4, "method": "tools/call", "params": {"name": "search_supermarket_offers", "arguments": {"query": ""}} }) resp = process_message(req) self.assertTrue(resp["result"]["isError"]) self.assertIn("required", resp["result"]["content"][0]["text"]) def test_mcp_notification_silence(self): req = json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized"}) resp = process_message(req) self.assertIsNone(resp) # Notifications must not produce responses if __name__ == "__main__": unittest.main()