""" Tests for the Product Catalog component. Uses mocked HTTP responses to avoid hitting the real Espressif API. """ import time from unittest.mock import AsyncMock, patch import httpx import pytest from fastmcp import FastMCP from mcesptool.components.product_catalog import ( ProductCatalog, _has_capability, _normalize_status, _parse_lead_weeks, _parse_memory_field, _parse_sram_kb, _parse_temp_max, ) from mcesptool.config import ESPToolServerConfig # ── Fixtures ──────────────────────────────────────────────────────────── SAMPLE_PRODUCTS = [ { "id": 1, "name": "ESP32-S3", "nameNew": "ESP32-S3", "type": "SoC", "seriesName": "ESP32-S3", "wifi": "802.11 b/g/n", "wifi6": "", "bluetooth": "BLE 5.0", "threadZigbee": "", "flash": "0", "psram": "0", "sram": "512", "rom": "384", "gpio": "45", "pins": "56", "freq": "240", "operatingTemp": "-40~85C", "voltageRange": "3.0~3.6V", "dimensions": "7x7 mm", "status": "Mass Production", "antenna": "", "mpn": "ESP32-S3", "sizeType": "QFN56", "releaseTime": "2020", "idfSupports": "v4.4+", "spq": "490", "moq": "4900", "leadTime": "8-10 weeks", "eccnCode": "5A992.c", "ccatszCode": "", "hsCode": "8542.31", "preFirmware": "", "replacedByName": "", }, { "id": 2, "name": "ESP32-S3-WROOM-1-N16R8", "nameNew": "ESP32-S3-WROOM-1-N16R8", "type": "Module", "seriesName": "ESP32-S3", "wifi": "802.11 b/g/n", "wifi6": "", "bluetooth": "BLE 5.0", "threadZigbee": "", "flash": "16, Quad", "psram": "8, Octal", "sram": "512", "rom": "384", "gpio": "36", "pins": "44", "freq": "240", "operatingTemp": "-40~85C", "voltageRange": "3.0~3.6V", "dimensions": "18x25.5x3.1 mm", "status": "Mass Production", "antenna": "PCB", "mpn": "ESP32-S3-WROOM-1-N16R8", "sizeType": "", "releaseTime": "2021", "idfSupports": "v4.4+", "spq": "400", "moq": "2000", "leadTime": "6-8 weeks", "eccnCode": "5A992.c", "ccatszCode": "", "hsCode": "8542.31", "preFirmware": "", "replacedByName": "", }, { "id": 3, "name": "ESP32-C6-MINI-1-N4", "nameNew": "ESP32-C6-MINI-1-N4", "type": "Module", "seriesName": "ESP32-C6", "wifi": "802.11 b/g/n/ax", "wifi6": "Yes", "bluetooth": "BLE 5.3", "threadZigbee": "Available", "flash": "4, Quad", "psram": "", "sram": "512", "rom": "320", "gpio": "22", "pins": "53", "freq": "160", "operatingTemp": "-40~105C", "voltageRange": "3.0~3.6V", "dimensions": "13.2x16.6x2.4 mm", "status": "Mass Production", "antenna": "PCB", "mpn": "ESP32-C6-MINI-1-N4", "sizeType": "", "releaseTime": "2023", "idfSupports": "v5.1+", "spq": "500", "moq": "2500", "leadTime": "4-6 weeks", "eccnCode": "5A992.c", "ccatszCode": "", "hsCode": "8542.31", "preFirmware": "", "replacedByName": "", }, { "id": 4, "name": "ESP32-WROOM-32E", "nameNew": "ESP32-WROOM-32E", "type": "Module", "seriesName": "ESP32", "wifi": "802.11 b/g/n", "wifi6": "", "bluetooth": "BT/BLE 4.2", "threadZigbee": "N/A", "flash": "4, Quad", "psram": "", "sram": "520", "rom": "448", "gpio": "34", "pins": "38", "freq": "240", "operatingTemp": "-40~85C", "voltageRange": "3.0~3.6V", "dimensions": "18x20x3.2 mm", "status": "NRND", "antenna": "PCB", "mpn": "ESP32-WROOM-32E", "sizeType": "", "releaseTime": "2019", "idfSupports": "v4.0+", "spq": "500", "moq": "5000", "leadTime": "12-14 weeks", "eccnCode": "5A992.c", "ccatszCode": "", "hsCode": "8542.31", "preFirmware": "", "replacedByName": "", }, ] def _mock_response() -> httpx.Response: """Build a mock httpx response with sample product data.""" return httpx.Response( status_code=200, json={"options": {}, "results": SAMPLE_PRODUCTS}, request=httpx.Request("GET", "https://products.espressif.com/api/user/products"), ) @pytest.fixture def app(): return FastMCP("test-esp") @pytest.fixture def config(): with patch.object(ESPToolServerConfig, "_check_tool_availability", return_value=True): return ESPToolServerConfig.from_environment() @pytest.fixture def catalog(app, config): cat = ProductCatalog(app, config) # Pre-load the cache so tests don't hit the network cat._products = SAMPLE_PRODUCTS cat._last_fetch = time.time() return cat @pytest.fixture def context(): """Minimal mock MCP context.""" return AsyncMock(spec=[]) # ── Helper function unit tests ────────────────────────────────────────── class TestParsers: def test_parse_memory_basic(self): assert _parse_memory_field("16, Quad") == (16, "Quad") assert _parse_memory_field("8, Octal") == (8, "Octal") assert _parse_memory_field("4") == (4, "") def test_parse_memory_bare_int(self): assert _parse_memory_field(8) == (8, "") assert _parse_memory_field(0) == (0, "") def test_parse_memory_empty(self): assert _parse_memory_field("") == (0, "") assert _parse_memory_field("N/A") == (0, "") assert _parse_memory_field("NA") == (0, "") assert _parse_memory_field(None) == (0, "") def test_parse_sram(self): assert _parse_sram_kb("512") == 512 assert _parse_sram_kb("520") == 520 assert _parse_sram_kb("") == 0 assert _parse_sram_kb(512) == 512 assert _parse_sram_kb(None) == 0 def test_has_capability(self): assert _has_capability("802.11 b/g/n") is True assert _has_capability("BLE 5.0") is True assert _has_capability("Available") is True assert _has_capability("") is False assert _has_capability("N/A") is False assert _has_capability("NA") is False def test_normalize_status(self): assert _normalize_status("Mass Production") == "Mass Production" assert _normalize_status("mass production") == "Mass Production" assert _normalize_status("NRND") == "NRND" assert _normalize_status("nrnd") == "NRND" def test_parse_lead_weeks(self): assert _parse_lead_weeks("8-10 weeks") == 8 assert _parse_lead_weeks("4-6 weeks") == 4 assert _parse_lead_weeks("") is None assert _parse_lead_weeks("TBD") is None def test_parse_temp_max(self): assert _parse_temp_max("-40~85C") == 85 assert _parse_temp_max("-40~105C") == 105 assert _parse_temp_max("") is None # ── Tool integration tests ────────────────────────────────────────────── @pytest.mark.asyncio class TestProductSearch: async def test_search_no_filters(self, catalog, context): result = await catalog._product_search_impl(context) assert result["count"] == 4 assert len(result["products"]) == 4 async def test_search_by_series(self, catalog, context): result = await catalog._product_search_impl(context, series="ESP32-S3") assert result["count"] == 2 names = [p["name"] for p in result["products"]] assert "ESP32-S3" in names assert "ESP32-S3-WROOM-1-N16R8" in names async def test_search_by_type(self, catalog, context): result = await catalog._product_search_impl(context, product_type="SoC") assert result["count"] == 1 assert result["products"][0]["name"] == "ESP32-S3" async def test_search_by_wifi(self, catalog, context): result = await catalog._product_search_impl(context, wifi=True) assert result["count"] == 4 # all sample products have wifi async def test_search_by_thread_zigbee(self, catalog, context): result = await catalog._product_search_impl(context, thread_zigbee=True) assert result["count"] == 1 assert result["products"][0]["name"] == "ESP32-C6-MINI-1-N4" async def test_search_by_min_flash(self, catalog, context): result = await catalog._product_search_impl(context, min_flash_mb=8) assert result["count"] == 1 assert result["products"][0]["name"] == "ESP32-S3-WROOM-1-N16R8" async def test_search_by_min_psram(self, catalog, context): result = await catalog._product_search_impl(context, min_psram_mb=4) assert result["count"] == 1 assert result["products"][0]["flash_mb"] == 16 async def test_search_by_status(self, catalog, context): result = await catalog._product_search_impl(context, status="NRND") assert result["count"] == 1 assert result["products"][0]["name"] == "ESP32-WROOM-32E" async def test_search_by_antenna(self, catalog, context): result = await catalog._product_search_impl(context, antenna="PCB") assert result["count"] == 3 async def test_search_by_max_temp(self, catalog, context): result = await catalog._product_search_impl(context, max_temp_c=105) assert result["count"] == 1 assert result["products"][0]["name"] == "ESP32-C6-MINI-1-N4" async def test_search_keyword(self, catalog, context): result = await catalog._product_search_impl(context, keyword="WROOM") assert result["count"] >= 1 names = [p["name"] for p in result["products"]] assert any("WROOM" in n for n in names) async def test_search_combined_filters(self, catalog, context): result = await catalog._product_search_impl( context, series="ESP32-S3", product_type="Module", min_flash_mb=8 ) assert result["count"] == 1 assert result["products"][0]["name"] == "ESP32-S3-WROOM-1-N16R8" async def test_search_no_results(self, catalog, context): result = await catalog._product_search_impl(context, min_flash_mb=128) assert result["count"] == 0 assert result["products"] == [] @pytest.mark.asyncio class TestProductInfo: async def test_info_exact_name(self, catalog, context): result = await catalog._product_info_impl(context, "ESP32-S3-WROOM-1-N16R8") assert result["name"] == "ESP32-S3-WROOM-1-N16R8" assert result["flash_mb"] == 16 assert result["psram_mb"] == 8 assert result["series"] == "ESP32-S3" async def test_info_fuzzy_name(self, catalog, context): result = await catalog._product_info_impl(context, "s3 wroom n16r8") assert result["name"] == "ESP32-S3-WROOM-1-N16R8" async def test_info_not_found(self, catalog, context): result = await catalog._product_info_impl(context, "ESP99-NONEXISTENT") assert "error" in result @pytest.mark.asyncio class TestChipCompare: async def test_compare_two(self, catalog, context): result = await catalog._chip_compare_impl( context, ["ESP32-S3-WROOM-1-N16R8", "ESP32-C6-MINI-1-N4"] ) assert len(result["products_compared"]) == 2 assert "differences" in result assert "common" in result # These two differ in series, flash, etc. assert "series" in result["differences"] async def test_compare_too_few(self, catalog, context): result = await catalog._chip_compare_impl(context, ["ESP32-S3"]) assert "error" in result async def test_compare_too_many(self, catalog, context): result = await catalog._chip_compare_impl( context, ["a", "b", "c", "d", "e"] ) assert "error" in result async def test_compare_not_found(self, catalog, context): result = await catalog._chip_compare_impl( context, ["ESP32-S3", "NONEXISTENT-CHIP"] ) assert "error" in result @pytest.mark.asyncio class TestProductRecommend: async def test_recommend_ble_sensor(self, catalog, context): result = await catalog._product_recommend_impl( context, "battery-powered BLE sensor", None, True ) assert result["count"] >= 1 # All recommendations should be modules (prefer_module=True) and have BLE for p in result["recommendations"]: assert p["type"] == "Module" assert p["bluetooth"] is True async def test_recommend_thread(self, catalog, context): result = await catalog._product_recommend_impl( context, "Thread/Matter smart home device", None, True ) assert result["count"] >= 1 for p in result["recommendations"]: assert p["thread_zigbee"] is True async def test_recommend_with_psram(self, catalog, context): result = await catalog._product_recommend_impl( context, "wifi camera needs psram", None, True ) for p in result["recommendations"]: assert p["psram_mb"] > 0 async def test_recommend_industrial(self, catalog, context): result = await catalog._product_recommend_impl( context, "industrial IoT gateway", "industrial temp range", True ) # Only ESP32-C6-MINI-1-N4 has 105C rating in our sample data assert result["count"] >= 1 async def test_recommend_fallback_to_soc(self, catalog, context): """When no modules match, falls back to including SoCs.""" result = await catalog._product_recommend_impl( context, "wifi device", "needs 32MB flash", True ) # No modules have 32MB flash — falls back to SoC search too assert isinstance(result["count"], int) @pytest.mark.asyncio class TestProductAvailability: async def test_availability_mass_production(self, catalog, context): result = await catalog._product_availability_impl( context, None, "Mass Production", None, None ) assert result["count"] == 3 # NRND one excluded async def test_availability_by_series(self, catalog, context): result = await catalog._product_availability_impl( context, "ESP32-S3", "Mass Production", None, None ) assert result["count"] == 2 async def test_availability_lead_time(self, catalog, context): result = await catalog._product_availability_impl( context, None, "Mass Production", 6, None ) # ESP32-C6-MINI-1-N4 (4 weeks) and ESP32-S3-WROOM-1-N16R8 (6 weeks) both pass <= 6 assert result["count"] == 2 names = [p["name"] for p in result["products"]] assert "ESP32-C6-MINI-1-N4" in names assert "ESP32-S3-WROOM-1-N16R8" in names async def test_availability_max_moq(self, catalog, context): result = await catalog._product_availability_impl( context, None, "Mass Production", None, 2500 ) # ESP32-S3-WROOM-1-N16R8 (2000) and ESP32-C6-MINI-1-N4 (2500) assert result["count"] == 2 # ── Cache behavior tests ──────────────────────────────────────────────── @pytest.mark.asyncio class TestCaching: async def test_cache_hit(self, catalog): """Cache should be used when data is fresh.""" original_fetch_time = catalog._last_fetch products = await catalog._ensure_data() assert len(products) == 4 assert catalog._last_fetch == original_fetch_time # no new fetch async def test_cache_expired(self, catalog): """Cache should refresh when TTL expired.""" catalog._last_fetch = time.time() - 90000 # older than 24h mock_resp = _mock_response() with patch("mcesptool.components.product_catalog.httpx.AsyncClient") as mock_client_cls: mock_client = AsyncMock() mock_client.get.return_value = mock_resp mock_client.__aenter__ = AsyncMock(return_value=mock_client) mock_client.__aexit__ = AsyncMock(return_value=False) mock_client_cls.return_value = mock_client products = await catalog._ensure_data() assert len(products) == 4 mock_client.get.assert_called_once() async def test_force_refresh(self, catalog): """Force refresh bypasses cache.""" mock_resp = _mock_response() with patch("mcesptool.components.product_catalog.httpx.AsyncClient") as mock_client_cls: mock_client = AsyncMock() mock_client.get.return_value = mock_resp mock_client.__aenter__ = AsyncMock(return_value=mock_client) mock_client.__aexit__ = AsyncMock(return_value=False) mock_client_cls.return_value = mock_client products = await catalog._ensure_data(force_refresh=True) assert len(products) == 4 mock_client.get.assert_called_once() # ── Fuzzy matching tests ──────────────────────────────────────────────── class TestFuzzyMatch: def test_exact_match(self, catalog): results = catalog._fuzzy_find(SAMPLE_PRODUCTS, "ESP32-S3") assert len(results) >= 1 assert results[0]["name"] == "ESP32-S3" def test_substring_match(self, catalog): results = catalog._fuzzy_find(SAMPLE_PRODUCTS, "WROOM") assert len(results) >= 1 names = [p["name"] for p in results] assert any("WROOM" in n for n in names) def test_no_match(self, catalog): results = catalog._fuzzy_find(SAMPLE_PRODUCTS, "XYZNONEXISTENT", threshold=90) assert len(results) == 0