tests/test_multitenant.py

from __future__ import annotations

import pytest
from sqlalchemy import select

from app import embeddings as emb_mod
from app.models import KnowledgeChunk, KnowledgeSource, Tenant
from app.rag import index as index_mod
from app.tenancy import create_tenant, get_tenant_by_slug, tenant_shopify_client


def _fake_vec(text: str) -> list[float]:
    t = text.lower()
    return [1.0 if "envio" in t or "envío" in t or "envíos" in t else 0.0,
            1.0 if "garant" in t else 0.0]


@pytest.fixture(autouse=True)
def fake_embeddings(monkeypatch):
    async def fake_embed_texts(texts, *, kind="passage"):
        return [_fake_vec(t) for t in texts]

    async def fake_embed_query(text):
        return _fake_vec(text)

    monkeypatch.setattr(emb_mod, "embed_texts", fake_embed_texts)
    monkeypatch.setattr(emb_mod, "embed_query", fake_embed_query)


async def _seed_source(db, tenant_id, tmp_path, name, body):
    p = tmp_path / f"{name}.txt"
    p.write_text(body, encoding="utf-8")
    src = KnowledgeSource(kind="file", name=name, location=str(p), tenant_id=tenant_id)
    db.add(src)
    await db.flush()
    await index_mod.index_source(db, src)
    return src


async def test_knowledge_search_is_isolated_per_tenant(db_session, tmp_path):
    a = Tenant(slug="a")
    b = Tenant(slug="b")
    db_session.add_all([a, b])
    await db_session.flush()

    await _seed_source(db_session, a.id, tmp_path, "a_env", "Envíos gratis en la tienda A")
    await _seed_source(db_session, b.id, tmp_path, "b_env", "Envíos también en la tienda B")

    # chunks were tagged with their tenant
    chunks = (await db_session.execute(select(KnowledgeChunk))).scalars().all()
    assert {c.tenant_id for c in chunks} == {a.id, b.id}

    # searching tenant A only returns A's chunks, never B's
    res_a = await index_mod.search(db_session, "envío", k=5, tenant_id=a.id)
    assert res_a
    assert all(c.tenant_id == a.id for c, _ in res_a)

    res_b = await index_mod.search(db_session, "envío", k=5, tenant_id=b.id)
    assert res_b
    assert all(c.tenant_id == b.id for c, _ in res_b)


async def test_create_tenant_encrypts_secret(db_session):
    t = await create_tenant(
        db_session,
        "toorx",
        brand_name="TOORX",
        shopify_shop="toorx.myshopify.com",
        shopify_client_id="cid",
        shopify_client_secret="shpss_supersecret",
    )
    # secret stored encrypted (not plaintext)
    assert t.shopify_client_secret_enc
    assert "shpss_supersecret" not in t.shopify_client_secret_enc

    # a Shopify client is built for a tenant with full creds
    assert tenant_shopify_client(t) is not None

    # a tenant without creds gets no client (info-only)
    plain = await create_tenant(db_session, "infoonly", brand_name="X")
    assert tenant_shopify_client(plain) is None


async def test_order_lockout_is_per_tenant(db_session):
    """A spammed email in tenant A must not lock that email in tenant B."""
    from app.models import ChatSession, Tenant
    from app.tools import order_tool
    from app.tools.registry import ToolContext

    a = Tenant(slug="ta")
    b = Tenant(slug="tb")
    db_session.add_all([a, b])
    await db_session.flush()

    class FakeShopify:
        async def execute(self, query, variables=None):
            return {"orders": {"edges": []}}  # no match -> failed attempts

    from app.ratelimit import get_order_limiter

    get_order_limiter().clear()

    async def session_for(tid):
        s = ChatSession(shop="x", tenant_id=tid)
        db_session.add(s)
        await db_session.flush()
        return s

    # Exhaust attempts for email in tenant A
    for _ in range(8):
        ctx_a = ToolContext(db=db_session, session=await session_for(a.id), tenant_id=a.id, shopify=FakeShopify())
        await order_tool.run({"email": "spam@x.com", "order_number": "9999"}, ctx_a)

    # Same email in tenant B is NOT locked (independent counter)
    ctx_b = ToolContext(db=db_session, session=await session_for(b.id), tenant_id=b.id, shopify=FakeShopify())
    out_b = await order_tool.run({"email": "spam@x.com", "order_number": "9999"}, ctx_b)
    assert out_b["status"] == "not_found"  # not "locked"


async def test_get_tenant_by_slug(db_session):
    db_session.add(Tenant(slug="toorx", brand_name="T"))
    await db_session.flush()
    assert (await get_tenant_by_slug(db_session, "toorx")).slug == "toorx"
    assert await get_tenant_by_slug(db_session, "missing") is None


async def test_resolution_keys_unique_clean_409(db_session):
    """shopify_shop / whatsapp_phone_id map to exactly ONE tenant: a duplicate is a
    clean 409 at write time, never two rows that would mix two client accounts."""
    from fastapi import HTTPException

    from app.tenancy import update_tenant

    a = await create_tenant(db_session, "shop-a", shopify_shop="a.myshopify.com")
    b = await create_tenant(db_session, "shop-b", shopify_shop="b.myshopify.com")
    with pytest.raises(HTTPException) as e1:  # new tenant grabbing A's shop
        await create_tenant(db_session, "shop-a-dup", shopify_shop="a.myshopify.com")
    assert e1.value.status_code == 409
    with pytest.raises(HTTPException) as e2:  # B editing onto A's shop
        await update_tenant(db_session, b, shopify_shop="a.myshopify.com")
    assert e2.value.status_code == 409
    await update_tenant(db_session, a, shopify_shop="a.myshopify.com")  # own value -> ok
    await update_tenant(db_session, a, whatsapp_phone_id="PHONE1")
    with pytest.raises(HTTPException) as e3:  # B grabbing A's WhatsApp number
        await update_tenant(db_session, b, whatsapp_phone_id="PHONE1")
    assert e3.value.status_code == 409
    await update_tenant(db_session, b, whatsapp_phone_id="")  # empty never collides