AI-Pricing-Agent/tests/test_apify_batch.py

"""Batching + raw-item cache for the Apify client (no network).

These lock in the two things that make competitive data affordable:
  1. N ASINs cost ONE actor run, not N runs.
  2. A cache hit costs zero actor runs.
"""
from __future__ import annotations

import json

import pytest

from pricing_agent.schemas import BuyBoxStatus
from pricing_agent.tools.amazon.apify import (
    ApifyAmazonClient,
    RawItemCache,
    _match_requested_asin,
)

OUR_SELLER = "A3AQP8TDYVYCGL"


def _item(asin: str, price: float, seller_id: str = "A1RIVAL") -> dict:
    return {
        "asin": asin,
        "price": {"value": price, "currency": "$"},
        "seller": {"name": "Rival", "id": seller_id},
        "offers": [
            {
                "price": {"value": price, "currency": "$"},
                "condition": "New",
                "seller": {"name": "Rival", "id": seller_id},
            }
        ],
    }


class _SpyClient(ApifyAmazonClient):
    """Counts actor runs instead of hitting the network."""

    def __init__(self, **kw):
        super().__init__(token="x", **kw)
        self.runs: list[list[str]] = []

    def _run_actor(self, asins):  # type: ignore[override]
        self.runs.append(list(asins))
        return {a: _item(a, 10.0 + i) for i, a in enumerate(asins)}


def test_many_asins_cost_one_actor_run():
    c = _SpyClient(seller_id=OUR_SELLER, batch_size=20)
    snaps = c.get_competitive_many(["B001", "B002", "B003"])

    assert len(c.runs) == 1, "3 ASINs must not cost 3 actor runs"
    assert c.runs[0] == ["B001", "B002", "B003"]
    assert set(snaps) == {"B001", "B002", "B003"}
    assert snaps["B001"].buy_box_price == 10.0


def test_batch_size_chunks_runs():
    c = _SpyClient(seller_id=OUR_SELLER, batch_size=2)
    c.get_competitive_many(["B001", "B002", "B003", "B004", "B005"])
    assert [len(r) for r in c.runs] == [2, 2, 1]


def test_duplicate_asins_scraped_once():
    c = _SpyClient(seller_id=OUR_SELLER)
    c.get_competitive_many(["B001", "B001", "B002"])
    assert c.runs == [["B001", "B002"]]


def test_cache_hit_costs_no_actor_run(tmp_path):
    cache = RawItemCache(tmp_path / "c.json", ttl_seconds=3600)
    c1 = _SpyClient(seller_id=OUR_SELLER, cache=cache)
    c1.get_competitive_many(["B001", "B002"])
    assert len(c1.runs) == 1

    # Fresh client, same cache file -> served entirely from disk.
    c2 = _SpyClient(seller_id=OUR_SELLER, cache=RawItemCache(tmp_path / "c.json", 3600))
    snaps = c2.get_competitive_many(["B001", "B002"])
    assert c2.runs == [], "cached ASINs must not trigger an actor run"
    assert snaps["B001"].buy_box_price == 10.0


def test_cache_only_scrapes_the_misses(tmp_path):
    path = tmp_path / "c.json"
    c1 = _SpyClient(seller_id=OUR_SELLER, cache=RawItemCache(path, 3600))
    c1.get_competitive_many(["B001"])

    c2 = _SpyClient(seller_id=OUR_SELLER, cache=RawItemCache(path, 3600))
    c2.get_competitive_many(["B001", "B002"])
    assert c2.runs == [["B002"]], "must scrape only the cache miss"


def test_expired_cache_entry_is_refetched(tmp_path):
    path = tmp_path / "c.json"
    c1 = _SpyClient(seller_id=OUR_SELLER, cache=RawItemCache(path, 3600))
    c1.get_competitive_many(["B001"])

    # ttl=0 disables the cache entirely; a negative-age TTL would be nonsense.
    c2 = _SpyClient(seller_id=OUR_SELLER, cache=RawItemCache(path, 0))
    c2.get_competitive_many(["B001"])
    assert c2.runs == [["B001"]]


def test_corrupt_cache_file_does_not_break_pricing(tmp_path):
    path = tmp_path / "c.json"
    path.write_text("{not json", encoding="utf-8")
    cache = RawItemCache(path, 3600)
    assert cache.get("B001") is None
    c = _SpyClient(seller_id=OUR_SELLER, cache=cache)
    snaps = c.get_competitive_many(["B001"])
    assert snaps["B001"].buy_box_price == 10.0


def test_cache_stores_raw_payload_so_mapper_fixes_apply(tmp_path):
    """Caching the raw item (not the snapshot) means a mapper fix applies immediately.

    Here the cached payload is one of OUR offers; a client configured with our seller id
    must map it to WON with no band, without re-scraping.
    """
    path = tmp_path / "c.json"
    cache = RawItemCache(path, 3600)
    cache.put_many({"B001": _item("B001", 6.99, seller_id=OUR_SELLER)})

    c = _SpyClient(seller_id=OUR_SELLER, cache=RawItemCache(path, 3600))
    snap = c.get_competitive_many(["B001"])["B001"]
    assert c.runs == []
    assert snap.buy_box_status == BuyBoxStatus.WON
    assert snap.competitive_low is None  # our own offer is not a competitor


def test_actor_failure_degrades_to_unknown_for_every_asin():
    class _Boom(ApifyAmazonClient):
        def _run_actor(self, asins):  # type: ignore[override]
            raise RuntimeError("actor exploded")

    c = _Boom(token="x", seller_id=OUR_SELLER)
    snaps = c.get_competitive_many(["B001", "B002"])
    assert all(s.buy_box_status == BuyBoxStatus.UNKNOWN for s in snaps.values())
    assert "actor exploded" in snaps["B001"].reason


@pytest.mark.parametrize(
    "item",
    [
        {"asin": "B001"},
        {"asin": "BXXXX", "originalAsin": "B001"},
        {"asin": "BXXXX", "url": "https://www.amazon.com/dp/B001"},
        {"asin": "BXXXX", "input": {"url": "https://www.amazon.com/dp/B001"}},
    ],
)
def test_redirected_variant_still_maps_back_to_requested_asin(item):
    """Amazon can redirect /dp/<asin> to a parent listing — don't lose the SKU."""
    assert _match_requested_asin(item, {"B001", "B002"}) == "B001"


def test_unrelated_item_is_not_force_matched():
    assert _match_requested_asin({"asin": "BZZZZ"}, {"B001"}) is None