test_integration_cart_flow.py•9.4 kB
"""Integration test covering cart lifecycle operations."""
from __future__ import annotations
import pytest
from fastmcp.client import Client
from fastmcp.exceptions import ToolError
from app.server import mcp
from .integration_utils import (
TARGET_BASE_SITE,
gather_product_context,
list_base_site_ids,
reset_occ_client,
structured,
)
@pytest.mark.asyncio
async def test_cart_lifecycle_flow() -> None:
"""Create an anonymous cart, manipulate entries, and verify results."""
reset_occ_client()
client = Client(mcp)
try:
async with client:
candidate_sites = [TARGET_BASE_SITE]
candidate_sites.extend(
site for site in await list_base_site_ids(client) if site not in candidate_sites
)
context = None
base_site_id = None
first_product_code = None
second_product_code = None
for site_uid in candidate_sites:
try:
context_candidate = await gather_product_context(client, site_uid)
except Exception:
continue
try:
cart_probe = await client.call_tool(
"carts.create",
{"base_site_id": site_uid, "user_id": "anonymous"},
)
cart_probe_payload = structured(cart_probe)
probe_cart_guid = (
cart_probe_payload.get("guid") or cart_probe_payload.get("code")
)
if not probe_cart_guid:
continue
await client.call_tool(
"cartEntries.add",
{
"base_site_id": site_uid,
"user_id": "anonymous",
"cart_id": probe_cart_guid,
"entry": {
"product": {"code": context_candidate["products"][0]["code"]},
"quantity": 1,
},
},
)
except ToolError:
continue
finally:
if "probe_cart_guid" in locals():
try:
await client.call_tool(
"carts.delete",
{
"base_site_id": site_uid,
"user_id": "anonymous",
"cart_id": probe_cart_guid,
},
)
except Exception:
pass
context = context_candidate
base_site_id = site_uid
break
if context is None or base_site_id is None:
pytest.skip("No base site allowed cart modifications for anonymous user")
base_site = context["base_site"]
products = context["products"]
first_product_code = products[0]["code"]
second_product_code = products[1]["code"] if len(products) > 1 else first_product_code
print(f"Using base site: {base_site_id}")
cart_result = await client.call_tool(
"carts.create",
{
"base_site_id": base_site_id,
"user_id": "anonymous",
},
)
cart_payload = structured(cart_result)
cart_guid = cart_payload.get("guid") or cart_payload.get("code")
cart_code = cart_payload.get("code")
assert cart_guid, "Expected cart GUID in create response"
print(f"Created cart: code={cart_code}, guid={cart_guid}")
try:
entry_result = await client.call_tool(
"cartEntries.add",
{
"base_site_id": base_site_id,
"user_id": "anonymous",
"cart_id": cart_guid,
"entry": {"product": {"code": first_product_code}, "quantity": 1},
},
)
except ToolError as exc:
pytest.skip(
f"Cart entry operations not permitted in this environment: {exc}"
)
entry_mod = structured(entry_result)
entry_number = entry_mod.get("entry", {}).get("entryNumber")
assert entry_number is not None, "Expected entry number from add response"
entry_number_str = str(entry_number)
print(f"Added entry {entry_number_str} for product {first_product_code}")
cart_snapshot = await client.call_tool(
"carts.get",
{
"base_site_id": base_site_id,
"user_id": "anonymous",
"cart_id": cart_guid,
},
)
cart_data = structured(cart_snapshot)
total_items = cart_data.get("totalItems")
print(f"Cart now has {total_items} item(s)")
second_entry_result = await client.call_tool(
"cartEntries.add",
{
"base_site_id": base_site_id,
"user_id": "anonymous",
"cart_id": cart_guid,
"entry": {"product": {"code": second_product_code}, "quantity": 2},
},
)
second_mod = structured(second_entry_result)
second_entry_number = second_mod.get("entry", {}).get("entryNumber")
assert second_entry_number is not None, "Expected entry number for second add"
second_entry_number_str = str(second_entry_number)
print(
f"Added entry {second_entry_number_str} for product {second_product_code}"
)
update_result = await client.call_tool(
"cartEntries.update",
{
"base_site_id": base_site_id,
"user_id": "anonymous",
"cart_id": cart_guid,
"entry_number": entry_number_str,
"entry": {"product": {"code": first_product_code}, "quantity": 3},
},
)
update_payload = structured(update_result)
updated_quantity = update_payload.get("entry", {}).get("quantity")
if updated_quantity is not None:
try:
updated_quantity = int(updated_quantity)
except (TypeError, ValueError):
pass
assert updated_quantity == 3, "Expected quantity to be updated to 3"
print(f"Updated entry {entry_number_str} quantity to {updated_quantity}")
delete_entry_meta = await client.call_tool(
"cartEntries.delete",
{
"base_site_id": base_site_id,
"user_id": "anonymous",
"cart_id": cart_guid,
"entry_number": second_entry_number_str,
},
)
delete_entry_info = structured(delete_entry_meta)
status_code = delete_entry_info.get("status_code") or delete_entry_info.get("status")
if status_code is not None:
try:
status_code = int(status_code)
except (TypeError, ValueError):
pass
assert status_code in {200, 204}, "Expected successful deletion of entry"
print(f"Deleted entry {second_entry_number_str}; status: {status_code}")
final_cart = await client.call_tool(
"carts.get",
{
"base_site_id": base_site_id,
"user_id": "anonymous",
"cart_id": cart_guid,
},
)
final_cart_data = structured(final_cart)
total_items = final_cart_data.get("totalItems")
if total_items is not None:
try:
total_items = int(total_items)
except (TypeError, ValueError):
pass
assert total_items is None or total_items >= 1, "Expected at least one item remaining in cart"
print(f"Final cart items: {total_items}")
delete_cart_meta = await client.call_tool(
"carts.delete",
{
"base_site_id": base_site_id,
"user_id": "anonymous",
"cart_id": cart_guid,
},
)
delete_cart_info = structured(delete_cart_meta)
delete_cart_status = delete_cart_info.get("status_code") or delete_cart_info.get("status")
if delete_cart_status is not None:
try:
delete_cart_status = int(delete_cart_status)
except (TypeError, ValueError):
pass
assert delete_cart_status in {200, 204}, "Expected cart deletion to succeed"
print(f"Deleted cart: status {delete_cart_status}")
finally:
await client.close()