test_integration_live.py•4.15 kB
"""Live integration test covering discovery and product flows."""
from __future__ import annotations
import pytest
from fastmcp.client import Client
from app.server import mcp
from .integration_utils import (
TARGET_BASE_SITE,
gather_product_context,
reset_occ_client,
structured,
)
@pytest.mark.asyncio
async def test_mcp_live_discovery_flow() -> None:
"""Exercise the discovery endpoints against the live OCC service."""
reset_occ_client()
client = Client(mcp)
try:
async with client:
context = await gather_product_context(client, TARGET_BASE_SITE)
base_site = context["base_site"]
base_site_id = base_site["uid"]
print(f"Selected base site: {base_site_id} ({base_site.get('name')})")
store_entry = context.get("store_entry")
if store_entry:
store_name = store_entry.get("displayName") or store_entry.get("name")
print(f"Store: {store_name}")
store_detail = await client.call_tool(
"stores.get",
{
"base_site_id": base_site_id,
"base_store_uid": "powertools",
},
)
store_payload = structured(store_detail)
assert store_payload.get("pointsOfService"), "Expected base store to list points of service"
catalog_entry = context["catalog_entry"]
catalog_id = context["catalog_id"]
print(f"Catalog: {catalog_id} ({catalog_entry.get('name')})")
catalog_detail = await client.call_tool(
"catalogs.get",
{"base_site_id": base_site_id, "catalog_id": catalog_id},
)
assert structured(catalog_detail), "Expected catalog details payload"
version_id = context["version_id"]
print(f"Catalog version: {version_id}")
version_payload = context["version_payload"]
assert version_payload, "Expected catalog version response"
category_entry = context["category_entry"]
category_id = context["category_id"]
print(f"Category: {category_id} ({category_entry.get('name')})")
products = context["products"]
sample_products = [
product.get("name") or product.get("code") for product in products[:3]
]
print(f"Sample products: {sample_products}")
product_code = products[0]["code"]
product_detail = await client.call_tool(
"products.get",
{
"base_site_id": base_site_id,
"product_code": product_code,
},
)
product_payload = structured(product_detail)
assert (
product_payload.get("code") == product_code
), "Product detail call should echo product code"
print(f"Product: {product_payload.get('name')} ({product_code})")
stock_result = await client.call_tool(
"products.stock",
{
"base_site_id": base_site_id,
"product_code": product_code,
"location": "New York",
},
)
stock_payload = structured(stock_result)
assert "stockData" in stock_payload or stock_payload, "Expected stock payload"
stock_status = stock_payload.get("stockData", {}).get("stockLevelStatus")
print(f"Stock level status: {stock_status}")
stock_count = await client.call_tool(
"products.stockCount",
{
"base_site_id": base_site_id,
"product_code": product_code,
"location": "New York",
},
)
stock_meta = structured(stock_count)
assert "total_stock_levels" in stock_meta, "Expected stock count metadata"
print(f"Total stock locations: {stock_meta.get('total_stock_levels')}")
finally:
await client.close()