tests/tools/test_tools.py

from __future__ import annotations

from app.models import ChatSession, KnowledgeChunk, KnowledgeSource
from app.tools import registry
from app.tools.registry import ToolContext


class FakeShopify:
    def __init__(self, orders_payload=None, products_payload=None):
        self.orders_payload = orders_payload or {"orders": {"edges": []}}
        self.products_payload = products_payload or {"products": {"edges": []}}

    async def execute(self, query, variables=None):
        return self.orders_payload if "orders(" in query.lower() else self.products_payload


async def _session(db):
    s = ChatSession(shop="x")
    db.add(s)
    await db.flush()
    return s


def test_specs_expose_tools():
    names = {s.name for s in registry.specs()}
    assert {
        "search_knowledge",
        "search_products",
        "create_cart_link",
        "watch_stock",
        "lookup_order",
        "reorder",
        "cancel_order",
        "change_shipping_address",
        "start_return",
        "escalate_to_human",
    } <= names


async def test_cart_link_with_discount(db_session):
    ctx = ToolContext(db=db_session, session=await _session(db_session), shop="s.myshopify.com")
    out = await registry.dispatch(
        "create_cart_link", {"items": [{"variant_id": "9"}], "discount": "VERANO10"}, ctx
    )
    assert out["checkout_url"] == "https://s.myshopify.com/cart/9:1?discount=VERANO10"


async def test_watch_stock_saves(db_session):
    from sqlalchemy import select

    from app.models import StockWatch

    ctx = ToolContext(db=db_session, session=await _session(db_session), tenant_id=1)
    miss = await registry.dispatch("watch_stock", {}, ctx)
    assert miss["status"] == "need_info"

    out = await registry.dispatch(
        "watch_stock", {"email": "c@x.com", "product_title": "Cinta X100"}, ctx
    )
    assert out["status"] == "saved"
    rows = (await db_session.execute(select(StockWatch))).scalars().all()
    assert rows[0].email == "c@x.com" and rows[0].product_title == "Cinta X100"


async def test_dispatch_knowledge_returns_context(db_session, monkeypatch):
    from app import embeddings as emb

    async def fake_query(q):
        return [1.0, 0.0]

    async def fake_texts(texts, *, kind="passage"):
        return [[1.0, 0.0] for _ in texts]

    monkeypatch.setattr(emb, "embed_query", fake_query)
    monkeypatch.setattr(emb, "embed_texts", fake_texts)

    src = KnowledgeSource(kind="url", name="FAQ", location="https://e")
    db_session.add(src)
    await db_session.flush()
    db_session.add(
        KnowledgeChunk(
            source_id=src.id, ordinal=0, text="Envíos gratis desde 199€", embedding=[1.0, 0.0],
            meta={"source_name": "FAQ"},
        )
    )
    await db_session.flush()

    ctx = ToolContext(db=db_session, session=await _session(db_session))
    out = await registry.dispatch("search_knowledge", {"query": "envíos"}, ctx)
    assert "Envíos gratis" in out["context"]
    assert out["sources"] == ["FAQ"]


async def test_dispatch_products_surfaces_cards_and_add_url(db_session):
    payload = {
        "products": {
            "edges": [
                {
                    "node": {
                        "title": "X100",
                        "onlineStoreUrl": "https://shop/x100",
                        "description": "d",
                        "featuredImage": {"url": "https://img/x.jpg"},
                        "variants": {"edges": [{"node": {"id": "gid://shopify/ProductVariant/555", "title": "Default", "price": "10.0", "availableForSale": True}}]},
                    }
                }
            ]
        }
    }
    ctx = ToolContext(
        db=db_session, session=await _session(db_session),
        shopify=FakeShopify(products_payload=payload), shop="shop.myshopify.com",
    )
    out = await registry.dispatch("search_products", {"query": "x"}, ctx)
    assert out["status"] == "ok"
    assert out["products"][0]["title"] == "X100"
    # surfaced as a visual card with image + one-tap add-to-cart permalink
    assert len(ctx.cards) == 1
    assert ctx.cards[0]["image"] == "https://img/x.jpg"
    assert ctx.cards[0]["add_url"] == "https://shop.myshopify.com/cart/555:1"


async def test_create_cart_link(db_session):
    ctx = ToolContext(db=db_session, session=await _session(db_session), shop="shop.myshopify.com")
    out = await registry.dispatch(
        "create_cart_link",
        {"items": [{"variant_id": "555", "quantity": 2}, {"variant_id": "gid://shopify/ProductVariant/777"}]},
        ctx,
    )
    assert out["status"] == "ok"
    assert out["checkout_url"] == "https://shop.myshopify.com/cart/555:2,777:1"


async def test_order_tool_needs_email_first(db_session):
    ctx = ToolContext(db=db_session, session=await _session(db_session), shopify=FakeShopify())
    out = await registry.dispatch("lookup_order", {}, ctx)
    assert out["status"] == "need_info"
    assert out["need"] == ["email"]


async def test_order_tool_email_only_asks_for_second_identifier(db_session):
    ctx = ToolContext(db=db_session, session=await _session(db_session), shopify=FakeShopify())
    out = await registry.dispatch("lookup_order", {"email": "a@b.c"}, ctx)
    assert out["status"] == "need_info"
    assert out["need"] == ["second_identifier"]
    assert "postal_code" in out["accepts"]  # flexible: not only the order number


async def test_order_tool_verified_with_postal_code(db_session):
    payload = {
        "orders": {
            "edges": [
                {
                    "node": {
                        "id": "gid://shopify/Order/1",
                        "name": "#1001",
                        "email": "a@b.c",
                        "customer": {"email": "a@b.c", "firstName": "Ana", "lastName": "G"},
                        "shippingAddress": {"zip": "08480", "city": "X"},
                        "displayFulfillmentStatus": "IN_PROGRESS",
                        "fulfillments": [
                            {"displayStatus": "IN_TRANSIT", "trackingInfo": [{"company": "GLS", "number": "T1", "url": "u"}]}
                        ],
                    }
                }
            ]
        }
    }
    ctx = ToolContext(
        db=db_session, session=await _session(db_session), shopify=FakeShopify(orders_payload=payload)
    )
    # customer gives email + postal code (NOT the order number) → still verifies
    out = await registry.dispatch("lookup_order", {"email": "a@b.c", "postal_code": "08480"}, ctx)
    assert out["status"] == "verified"
    assert out["tracking"]["tracking"][0]["number"] == "T1"
    assert "address" not in out["tracking"]


async def test_order_tool_verified_returns_tracking_no_pii(db_session):
    payload = {
        "orders": {
            "edges": [
                {
                    "node": {
                        "id": "gid://shopify/Order/1",
                        "name": "#1001",
                        "email": "a@b.c",
                        "displayFinancialStatus": "PAID",
                        "displayFulfillmentStatus": "IN_PROGRESS",
                        "customer": {"email": "a@b.c"},
                        "fulfillments": [
                            {
                                "displayStatus": "IN_TRANSIT",
                                "estimatedDeliveryAt": "2026-06-10",
                                "deliveredAt": None,
                                "inTransitAt": "2026-06-05",
                                "trackingInfo": [{"company": "GLS", "number": "T1", "url": "http://t/T1"}],
                            }
                        ],
                    }
                }
            ]
        }
    }
    ctx = ToolContext(
        db=db_session, session=await _session(db_session), shopify=FakeShopify(orders_payload=payload)
    )
    out = await registry.dispatch(
        "lookup_order", {"email": "a@b.c", "order_number": "1001"}, ctx
    )
    assert out["status"] == "verified"
    assert out["tracking"]["tracking"][0]["number"] == "T1"
    assert "address" not in out["tracking"]


async def test_order_tool_not_found_is_generic(db_session):
    ctx = ToolContext(db=db_session, session=await _session(db_session), shopify=FakeShopify())
    out = await registry.dispatch(
        "lookup_order", {"email": "a@b.c", "order_number": "9999"}, ctx
    )
    assert out["status"] == "not_found"  # no field-specific leak


async def test_order_tool_invalid_email_is_generic(db_session):
    ctx = ToolContext(db=db_session, session=await _session(db_session), shopify=FakeShopify())
    out = await registry.dispatch(
        "lookup_order", {"email": 'x" OR id:>0', "order_number": "1001"}, ctx
    )
    assert out["status"] == "not_found"  # injection attempt → generic, no crash


async def test_order_tool_rate_limited_across_sessions(db_session, monkeypatch):
    # Rotating the session must NOT reset the per-email throttle.
    from app.ratelimit import get_order_limiter

    limiter = get_order_limiter()
    monkeypatch.setattr(limiter, "max_hits", 3)

    last = None
    for _ in range(4):
        ctx = ToolContext(
            db=db_session, session=await _session(db_session), shopify=FakeShopify()
        )
        last = await registry.dispatch(
            "lookup_order", {"email": "spam@x.com", "order_number": "9999"}, ctx
        )
    assert last["status"] == "locked"  # 4th attempt across fresh sessions is throttled


async def test_escalate_tool_sends(db_session):
    captured = {}

    async def fake_sender(to, subject, body):
        captured.update(to=to, subject=subject, body=body)
        return True

    ctx = ToolContext(
        db=db_session,
        session=await _session(db_session),
        support_email="support@store.com",
        escalation_sender=fake_sender,
    )
    out = await registry.dispatch(
        "escalate_to_human", {"email": "c@x.com", "question": "¿devoluciones?"}, ctx
    )
    assert out["status"] == "sent"
    assert captured["to"] == "support@store.com"
    assert "c@x.com" in captured["body"]
    assert "¿devoluciones?" in captured["body"]