tests/test_catalog_query_customer_phrasing.py

import re
from types import SimpleNamespace

import pytest

from services.catalog_query import Row, _ROW_CACHE, _execute_spec, load_rows, parse


def _row(title, price, availability="available"):
    return Row(
        title=title,
        price=price,
        availability=availability,
        source=f"https://example.test/{title.lower().replace(' ', '-')}",
        hay=title.lower(),
    )


def _collection_row(title, price, collection, availability="available"):
    row = _row(title, price, availability)
    normalized = re.sub(r"[^a-z0-9]+", " ", collection.lower()).strip()
    row.colls = f"|{normalized}|"
    row.colls_disp = collection
    return row


class _FakeCollection:
    name = "langchain"
    id = "same-collection-id"

    def __init__(self, title, price):
        self._title = title
        self._price = price

    def count(self):
        return 1

    def get(self, limit, include):
        return {
            "documents": [f"Product: {self._title}\nPrice: Rs.{self._price}\nAvail: InStock"],
            "metadatas": [
                {
                    "source": f"https://example.test/products/{self._title.lower().replace(' ', '-')}",
                    "chunk_kind": "product",
                    "content_type": "product",
                    "product_title": self._title,
                    "price": self._price,
                    "availability": "available",
                }
            ],
        }


def test_load_rows_cache_is_tenant_isolated_for_same_collection_name_and_count():
    _ROW_CACHE.clear()
    db_a = SimpleNamespace(_db_name="tenant_a", _collection=_FakeCollection("Tenant A Product", 100))
    db_b = SimpleNamespace(_db_name="tenant_b", _collection=_FakeCollection("Tenant B Product", 200))

    rows_a = load_rows(db_a)
    rows_b = load_rows(db_b)

    assert rows_a[0].title == "Tenant A Product"
    assert rows_b[0].title == "Tenant B Product"


def test_load_rows_preserves_collection_names_containing_pipe():
    class Collection:
        name = "langchain"
        id = "pipe-collection"

        def count(self):
            return 1

        def get(self, limit, include):
            return {
                "documents": ["Product: Clearance Bag\nPrice: Rs.1,750\nAvailability: InStock"],
                "metadatas": [{
                    "source": "https://example.test/products/clearance-bag",
                    "chunk_kind": "product",
                    "product_title": "Clearance Bag",
                    "price": 1750,
                    "availability": "available",
                    "collections": '["Under Rs. 1,750 | Clearance Sale", "Bags for Women"]',
                }],
            }

    _ROW_CACHE.clear()
    rows = load_rows(SimpleNamespace(_db_name="pipe_tenant", _collection=Collection()))

    assert rows[0].colls == "|under rs 1 750 clearance sale|bags for women|"


@pytest.mark.parametrize(
    ("question", "aggregation", "expected"),
    [
        ("Ignoring sold-out items, which product costs the least?", "min", "Available Cheap - Rs.20"),
        ("Among available products only, what costs the most?", "max", "Available Expensive - Rs.981,000"),
        ("Show me the priciest product I can actually buy right now.", "max", "Available Expensive - Rs.981,000"),
    ],
)
def test_customer_extreme_phrasings_use_full_available_catalog(question, aggregation, expected):
    rows = [
        _row("Sold Out Cheap", 5, "out of stock"),
        _row("Available Cheap", 20),
        _row("Available Expensive", 981000),
        _row("Sold Out Expensive", 2000000, "sold out"),
    ]

    spec = parse(question)
    answer, _ = _execute_spec(spec, rows, None, 5)

    assert spec.agg == aggregation
    assert spec.anchors == []
    assert expected in answer
    assert "Sold Out" not in answer


def test_count_split_preserves_hand_in_hand_bags_collection():
    rows = [
        _collection_row("Black Handbag", 2500, "Hand Bags"),
        _collection_row("Tan Handbag", 2200, "Hand Bags", "sold out"),
        _collection_row("Canvas Tote", 1800, "Bags for Women"),
    ]

    spec = parse("Hand Bags mein kitne mojood kitne nahi")
    answer, _ = _execute_spec(spec, rows, None, 10)

    assert spec.agg == "count_split"
    assert spec.anchors == ["hand", "bags"]
    assert "Of 2 hand bags products: 1 available, 1 out of stock" in answer


def test_price_named_collection_is_not_filtered_by_its_own_title():
    collection = "Under Rs. 1,750 | Clearance Sale"
    rows = [
        _collection_row("Mustard Bag", 1499, collection),
        _collection_row("Tan Bag", 1499, collection),
        _collection_row("Beige Bag", 1750, collection),
        _collection_row("Pink Bag", 1750, collection),
        _collection_row("Blue Bag", 1499, collection, "sold out"),
        _collection_row("Silver Bag", 1750, collection, "sold out"),
    ]

    question = "how many Under Rs. 1,750 | Clearance Sale do you have?"
    spec = parse(question)
    answer, _ = _execute_spec(spec, rows, None, 10, raw_q=question)

    assert spec.agg == "count"
    assert "We have 6" in answer


@pytest.mark.parametrize(
    "question",
    [
        "Do you stock the Talkng Flash Card Lernning Toy?",
        "im looking for a talking flash crad learning toy, do u have it?",
        "Got any Talking Flash Cards Learnng Toy?",
    ],
)
def test_typoed_inventory_questions_resolve_the_named_product(question):
    rows = [
        _row("Talking Flash Card Learning Toy", 1350),
        _row("Talking Flash Cards Early Educational Device 224 Cards Rechargeable", 1500),
    ]

    spec = parse(question)
    answer, _ = _execute_spec(spec, rows, None, 5)

    assert spec.agg == "exists"
    assert "Talking Flash Card Learning Toy" in answer
    assert "Rs.1,350" in answer
    assert "Early Educational Device" not in answer


@pytest.mark.parametrize(
    "question",
    [
        "do u have Rasha For Women By Al?",   # subset of the long SEO title
        "do you have Rsaha For Women By Al?",  # + a typo in the distinctive word
    ],
)
def test_subset_of_seo_title_resolves_named_product(question):
    """A customer types part of a long SEO title (and maybe a typo); the title's
    extra trailing words must not sink the match into a false 'not carried'."""
    rows = [
        _row("Rasha For Women By Al Rehab EDP", 990),
        _row("Sultan For Men & Women Attar By Al Haramain", 2499),
        _row("CR7 For Men By Cristiano Ronaldo EDT", 3500),
    ]

    spec = parse(question)
    answer, _ = _execute_spec(spec, rows, None, 5)

    assert spec.agg == "exists"
    assert "Rasha For Women By Al Rehab EDP" in answer
    assert "don't" not in answer.lower() and "not carry" not in answer.lower()
    # generic-word siblings must NOT be dumped as matches
    assert "Sultan" not in answer and "CR7" not in answer


def test_typoed_distinctive_token_beats_generic_sibling():
    """A typo on the DISTINCTIVE word ("mgahribi"→"Maghribi") must resolve to that
    brand's products, not fall through to a sibling sharing only generic words
    ("by al"). Applies to the price/list path, not just exists."""
    rows = [
        _row("Rasha For Women By Al Rehab EDP", 990),
        _row("Rawdha By Ahmed Al Maghribi", 10750),
        _row("Zeleny For Unisex By Ahmed Al Maghribi", 4999),
        _row("Sultan For Men & Women By Al Haramain", 2499),
    ]
    spec = parse("how much is the mgahribi by al?")
    answer, _ = _execute_spec(spec, rows, None, 5)
    assert "Maghribi" in answer
    assert "Rasha" not in answer and "Sultan" not in answer


def test_absent_typoed_name_does_not_dump_siblings():
    """A query for a product that simply isn't carried ("kwaai") must not dump
    unrelated products — say it's absent (exists) or hand off to RAG (price)."""
    rows = [
        _row("Today To Do Spiral Gold Rings Notepad", 695),
        _row("Revolution Gel Pen Set", 450),
    ]
    spec = parse("do you have kwaai and whats the price?")
    answer, _ = _execute_spec(spec, rows, None, 5)
    assert "Today To Do" not in answer and "Revolution" not in answer


@pytest.mark.parametrize(
    "question",
    [
        "and do you also stock the Wish For Women By Chopard?",
        "do you have the Wish For Women By Chopard too?",
    ],
)
def test_followup_adverb_does_not_pollute_anchor(question):
    """"also"/"too" in a follow-up ("do you ALSO stock X") must not fold into the
    product name — that produced "also wish ..." → a false absence that then broke
    a pronoun follow-up with no antecedent to bind."""
    rows = [
        _row("Wish For Women By Chopard EDP", 9250),
        _row("Ajayeb Dubai By Lattafa For Men", 3000),
    ]
    spec = parse(question)
    answer, _ = _execute_spec(spec, rows, None, 5)
    assert spec.agg == "exists"
    assert "also" not in spec.anchors and "too" not in spec.anchors
    assert "Wish For Women By Chopard EDP" in answer
    assert "don't" not in answer.lower()