Ryan Malloy
dab9c6848e
5 new MCP tools for chip/module discovery, comparison, and procurement planning using Espressif's public product catalog API. Data is lazily fetched and cached in-memory with a 24h TTL. Tools: esp_product_search, esp_product_info, esp_chip_compare, esp_product_recommend, esp_product_availability Resource: esp://products/catalog
515 lines
18 KiB
Python
515 lines
18 KiB
Python
"""
|
|
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
|