From f84022d248263a5012d16b96827cf99f6806b216 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 13:37:59 +0000 Subject: [PATCH 1/3] test: cover the whole API in the integration tests without more requests The integration suites only reached search_items, get_items, get_variations and get_browse_nodes, and each one wrote its own assertions, so the sync and the async clients were not held to the same contract and feeds and reports were never checked against the real API. Both suites now share tests/integration_support.py, which holds the credentials, the snapshot of the results and every assertion, and each suite only builds its client and makes the calls. The calls carry as much as they can instead of being repeated: - The search is also the discovery call, and asks with an availability, a condition and a sort so the request is validated by Amazon as well. - A single get_items mixes a URL, a duplicate and an unknown identifier, so one request checks the parsing of URLs, the removal of duplicates, the order of the answer, its partial errors and the placeholders added by include_unavailable. - Variations and browse nodes reuse the parent ASIN and the node identifiers of the search, and are skipped instead of asked for when the search found none, so no request is spent on a call known to fail. - Feeds and reports are listed and downloaded, reporting an account without access to them as a skip instead of failing the suite. - Releasing the connections, dropping the token and using the async client outside its context manager are folded into the calls above, so they cost no request of their own. The async suite also stops making its calls while the module is imported, which ran them even when the tests were deselected and turned a failure into a collection error, and it no longer spends an extra search on checking its context manager. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_017ZLhDcp9SziFvYduax9evq --- .../aio/integration_test.py | 571 +++++------------- tests/amazon_creatorsapi/integration_test.py | 567 ++++------------- tests/integration_support.py | 540 +++++++++++++++++ 3 files changed, 812 insertions(+), 866 deletions(-) create mode 100644 tests/integration_support.py diff --git a/tests/amazon_creatorsapi/aio/integration_test.py b/tests/amazon_creatorsapi/aio/integration_test.py index 10cf019..8f8f710 100644 --- a/tests/amazon_creatorsapi/aio/integration_test.py +++ b/tests/amazon_creatorsapi/aio/integration_test.py @@ -1,469 +1,178 @@ -"""Integration tests for AsyncAmazonCreatorsApi class.""" +"""Integration tests for the async client of the Amazon Creators API. + +The suite makes every call once in setUpClass and asserts the results from +tests that never reach the network. See tests.integration_support for the +request budget and the assertions shared with the sync suite. + +The async client parses the responses by hand instead of relying on the SDK, +so it is checked against the same real payloads as the sync one. +""" from __future__ import annotations import asyncio import contextlib -import os -from pathlib import Path from typing import TYPE_CHECKING -from unittest import IsolatedAsyncioTestCase, skipUnless - -from dotenv import load_dotenv +from unittest import SkipTest, skipUnless from amazon_creatorsapi.aio import AsyncAmazonCreatorsApi -from amazon_creatorsapi.errors import ItemsNotFoundError +from amazon_creatorsapi.errors import ( + AccessDeniedError, + ItemsNotFoundError, + ResourceNotFoundError, +) +from creatorsapi_python_sdk.models.availability import Availability +from creatorsapi_python_sdk.models.condition import Condition +from creatorsapi_python_sdk.models.sort_by import SortBy +from tests.integration_support import ( + SEARCH_ITEM_COUNT, + SEARCH_KEYWORDS, + SKIP_NO_CREDENTIALS, + ApiSnapshot, + Credentials, + IntegrationAssertions, + build_item_ids, + get_expected_item_ids, + get_found_items, + has_credentials, + load_credentials, + pick_browse_node_ids, + pick_item_asins, + pick_variation_asin, +) if TYPE_CHECKING: - from creatorsapi_python_sdk.models.browse_node import BrowseNode - from creatorsapi_python_sdk.models.item import Item - from creatorsapi_python_sdk.models.search_result import SearchResult - from creatorsapi_python_sdk.models.variations_result import VariationsResult - -# Load environment variables from .env file -load_dotenv(Path(__file__).parents[2] / ".env") - - -def get_api_credentials() -> tuple[ - str | None, str | None, str | None, str | None, str | None, str | None -]: - """Get API credentials from environment variables. - - Returns: - Tuple of credentials, with None for missing values. - """ - return ( - os.environ.get("CREDENTIAL_ID"), - os.environ.get("CREDENTIAL_SECRET"), - os.environ.get("API_VERSION"), - os.environ.get("AFFILIATE_TAG"), - os.environ.get("MARKETPLACE"), - os.environ.get("COUNTRY_CODE"), + from creatorsapi_python_sdk.models.feed import Feed + from creatorsapi_python_sdk.models.report_metadata import ReportMetadata + + +def build_client(credentials: Credentials) -> AsyncAmazonCreatorsApi: + """Build the client used to collect the snapshot of the API.""" + return AsyncAmazonCreatorsApi( + credential_id=credentials.credential_id, + credential_secret=credentials.credential_secret, + version=credentials.version, + tag=credentials.tag, + marketplace=credentials.marketplace, + country=credentials.country, + throttling=1, ) -def has_api_credentials() -> bool: - """Check if all API credentials are available.""" - ( - credential_id, - credential_secret, - api_version, - affiliate_tag, - marketplace, - country_code, - ) = get_api_credentials() - - # Need critical credentials - if not all([credential_id, credential_secret, api_version, affiliate_tag]): - return False - - # Need at least marketplace or country_code - return bool(marketplace or country_code) +async def list_feeds(api: AsyncAmazonCreatorsApi) -> list[Feed] | None: + """Return the feeds of the account, or None when it has no access. + Feeds belong to a program that not every account is enrolled in, so a + rejection is reported as no feeds instead of failing the whole suite. + """ + try: + return await api.list_feeds() + except (AccessDeniedError, ResourceNotFoundError): + return None -def _has_valid_offer(item: Item) -> bool: - """Check if an item has a valid offer with price and availability.""" - if item.offers_v2 is None or not item.offers_v2.listings: - return False - listing = item.offers_v2.listings[0] +async def get_feed_url( + api: AsyncAmazonCreatorsApi, + feeds: list[Feed] | None, +) -> str | None: + """Return the download URL of the first feed, if the account has any.""" + if not feeds: + return None - has_price = ( - listing.price is not None - and listing.price.money is not None - and listing.price.money.amount is not None - ) + return await api.get_feed(feeds[0].feed_name, feeds[0].feed_type) - is_available = ( - listing.availability is None - or listing.availability.type is None - or listing.availability.type != "OutOfStock" - ) - return has_price and is_available +async def list_reports(api: AsyncAmazonCreatorsApi) -> list[ReportMetadata] | None: + """Return the reports of the account, or None when it has no access.""" + try: + return await api.list_reports() + except (AccessDeniedError, ResourceNotFoundError): + return None -def _find_item_with_offers(items: list[Item], search_result: SearchResult) -> Item: - """Find an item with offers, price, and in stock.""" - return next( - (item for item in items if _has_valid_offer(item)), - items[0] if items else search_result.items[0], # type: ignore[index] - ) +async def get_report_url( + api: AsyncAmazonCreatorsApi, + reports: list[ReportMetadata] | None, +) -> str | None: + """Return the download URL of the first report, if the account has any.""" + if not reports: + return None + return await api.get_report(reports[0].filename, reports[0].report_type) -def _find_variation_asin(items: list[Item], search_result: SearchResult) -> str | None: - """Find ASIN to use for variations lookup.""" - item_with_variations = next( - (item for item in items if item.parent_asin), - None, - ) - if item_with_variations: - return item_with_variations.parent_asin - - if search_result.items: - return search_result.items[0].asin - - return None - - -async def _run_api_setup() -> dict[str, object]: - """Run all API calls once and return cached data.""" - ( - credential_id, - credential_secret, - api_version, - affiliate_tag, - marketplace, - country_code, - ) = get_api_credentials() - - api = AsyncAmazonCreatorsApi( - credential_id=credential_id, # type: ignore[arg-type] - credential_secret=credential_secret, # type: ignore[arg-type] - version=api_version, # type: ignore[arg-type] - tag=affiliate_tag, # type: ignore[arg-type] - marketplace=marketplace, - country=country_code, # type: ignore[arg-type] - throttling=1, - ) +async def collect_snapshot(credentials: Credentials) -> ApiSnapshot: + """Make every API call of the suite once and cache their results. - data: dict[str, object] = {} + The catalog calls run inside the context manager, which keeps a pool of + connections open, and the ones for feeds and reports run outside of it, + where the client opens a connection for every request. Both modes are + covered without spending a request on either of them. + """ + api = build_client(credentials) async with api: - # 1. Search items - search_result = await api.search_items(keywords="laptop") - items = search_result.items or [] - - # 2. Find item with offers - item_with_offers = _find_item_with_offers(items, search_result) - - # 3. Get items by ASIN - get_items_result: list[Item] = [] - if item_with_offers.asin: - get_items_result = await api.get_items([item_with_offers.asin]) - - # 4. Get variations - variations_result: VariationsResult | None = None - target_asin = _find_variation_asin(items, search_result) - if target_asin: - with contextlib.suppress(ItemsNotFoundError): - variations_result = await api.get_variations(target_asin) - - # 5. Get browse nodes - browse_nodes_result: list[BrowseNode] = [] - item_with_browse_nodes = next( - ( - item - for item in items - if item.browse_node_info and item.browse_node_info.browse_nodes - ), - None, + search_result = await api.search_items( + keywords=SEARCH_KEYWORDS, + item_count=SEARCH_ITEM_COUNT, + availability=Availability.AVAILABLE, + condition=Condition.NEW, + sort_by=SortBy.FEATURED, ) - if item_with_browse_nodes: - browse_node_info = item_with_browse_nodes.browse_node_info - if browse_node_info and browse_node_info.browse_nodes: - browse_node_id = browse_node_info.browse_nodes[0].id - if browse_node_id: - browse_nodes_result = await api.get_browse_nodes([browse_node_id]) - - # Store in data dict - data["affiliate_tag"] = affiliate_tag - data["search_result"] = search_result - data["item_with_offers"] = item_with_offers - data["get_items_result"] = get_items_result - data["variations_result"] = variations_result - data["browse_nodes_result"] = browse_nodes_result - - return data + found_items = get_found_items(search_result) + asins = pick_item_asins(found_items) + items = await api.get_items( + build_item_ids(asins, api.marketplace), + include_unavailable=True, + ) -# Module-level cache - run setup once when module is loaded (only if credentials exist) -_cached_data: dict[str, object] = {} -if has_api_credentials(): - _cached_data = asyncio.run(_run_api_setup()) + variations = None + variation_asin = pick_variation_asin(found_items) + if variation_asin: + with contextlib.suppress(ItemsNotFoundError): + variations = await api.get_variations(variation_asin) + browse_node_ids = pick_browse_node_ids(found_items) + browse_nodes = ( + await api.get_browse_nodes(browse_node_ids) if browse_node_ids else None + ) -@skipUnless(has_api_credentials(), "Needs Amazon Creators API credentials") -class AsyncIntegrationTest(IsolatedAsyncioTestCase): - """Integration tests that make real async API calls to Amazon Creators API. + feeds = await list_feeds(api) + feed_url = await get_feed_url(api, feeds) + + # Dropping the token must make the next call authenticate again + api._token_manager.clear_token() + reports = await list_reports(api) + report_url = await get_report_url(api, reports) + + return ApiSnapshot( + tag=credentials.tag, + search_result=search_result, + requested_ids=get_expected_item_ids(asins), + items=items, + variations=variations, + browse_node_ids=browse_node_ids, + browse_nodes=browse_nodes, + feeds=feeds, + feed_url=feed_url, + reports=reports, + report_url=report_url, + ) - All API results are cached at module level to minimize the number of - requests. This reduces costs and avoids rate limiting. - """ - NO_VARIATIONS_FOUND_MSG = "No variations found" - - def setUp(self) -> None: - """Set up that runs before each test - loads cached data.""" - self.affiliate_tag: str = _cached_data["affiliate_tag"] # type: ignore[assignment] - self.search_result: SearchResult = _cached_data["search_result"] # type: ignore[assignment] - self.item_with_offers: Item = _cached_data["item_with_offers"] # type: ignore[assignment] - self.get_items_result: list[Item] = _cached_data["get_items_result"] # type: ignore[assignment] - self.variations_result: VariationsResult | None = _cached_data[ - "variations_result" - ] # type: ignore[assignment] - self.browse_nodes_result: list[BrowseNode] = _cached_data["browse_nodes_result"] # type: ignore[assignment] - - async def test_search_items_returns_expected_count(self) -> None: - """Test that search returns no more items than the default page size.""" - # API defaults to a page size of 10, but may return fewer items - items = self.search_result.items - if items: - self.assertLessEqual(len(items), 10) - - async def test_search_items_includes_affiliate_tag(self) -> None: - """Test that search results include the affiliate tag in URLs.""" - if self.search_result.items: - searched_item = self.search_result.items[0] - if searched_item.detail_page_url: - self.assertIn(self.affiliate_tag, searched_item.detail_page_url) - - async def test_search_items_returns_offers_v2(self) -> None: - """Test that search results include OffersV2 data.""" - items = self.search_result.items - self.assertIsNotNone(items) - if items: - self.assertGreater(len(items), 0) - - item = self.item_with_offers - self.assertIsNotNone(item.offers_v2) - - async def test_offers_v2_listing_has_price_info(self) -> None: - """Test that OffersV2 listings include price information.""" - item = self.item_with_offers - self.assertIsNotNone(item.offers_v2) - - if item.offers_v2 and item.offers_v2.listings: - listing = item.offers_v2.listings[0] - self.assertIsNotNone(listing.price) - - if listing.price and listing.price.money: - self.assertIsNotNone(listing.price.money.amount) - self.assertIsNotNone(listing.price.money.currency) - self.assertIsNotNone(listing.price.money.display_amount) - - async def test_offers_v2_listing_has_merchant_info(self) -> None: - """Test that OffersV2 listings include merchant information.""" - item = self.item_with_offers - self.assertIsNotNone(item.offers_v2) - - if item.offers_v2 and item.offers_v2.listings: - listing = item.offers_v2.listings[0] - if listing.merchant_info: - self.assertIsNotNone(listing.merchant_info.name) - - async def test_offers_v2_listing_has_condition(self) -> None: - """Test that OffersV2 listings include condition information.""" - item = self.item_with_offers - self.assertIsNotNone(item.offers_v2) - - if item.offers_v2 and item.offers_v2.listings: - listing = item.offers_v2.listings[0] - if listing.condition: - self.assertIsNotNone(listing.condition.value) - - async def test_offers_v2_listing_has_availability(self) -> None: - """Test that OffersV2 listings include availability information.""" - item = self.item_with_offers - self.assertIsNotNone(item.offers_v2) - - if item.offers_v2 and item.offers_v2.listings: - listing = item.offers_v2.listings[0] - if listing.availability: - self.assertIsNotNone(listing.availability.type) - - async def test_get_items_returns_single_result(self) -> None: - """Test that get_items returns exactly one item when given one ASIN.""" - result = self.get_items_result - self.assertEqual(1, len(result)) - - async def test_get_items_includes_affiliate_tag(self) -> None: - """Test that get_items results include the affiliate tag in URLs.""" - if self.get_items_result: - detail_url = self.get_items_result[0].detail_page_url - if detail_url: - self.assertIn(self.affiliate_tag, detail_url) - - async def test_get_items_returns_offers_v2(self) -> None: - """Test that get_items returns OffersV2 data with price details.""" - if self.get_items_result: - item = self.get_items_result[0] - self.assertIsNotNone(item.offers_v2) - - if item.offers_v2 and item.offers_v2.listings: - listing = item.offers_v2.listings[0] - self.assertIsNotNone(listing) - - if listing.price and listing.price.money: - self.assertIsNotNone(listing.price.money.amount) - self.assertIsNotNone(listing.price.money.display_amount) - - async def test_get_variations_returns_items(self) -> None: - """Test that get_variations returns a list of variation items.""" - if self.variations_result: - self.assertIsNotNone(self.variations_result) - items = self.variations_result.items - self.assertIsNotNone(items) - if items: - self.assertGreater(len(items), 0) - else: - self.skipTest(self.NO_VARIATIONS_FOUND_MSG) - - async def test_get_variations_returns_variation_summary(self) -> None: - """Test that get_variations returns variation summary.""" - if self.variations_result: - summary = self.variations_result.variation_summary - self.assertIsNotNone(summary) - if summary and summary.variation_count is not None: - self.assertGreater(summary.variation_count, 0) - else: - self.skipTest(self.NO_VARIATIONS_FOUND_MSG) - - async def test_get_variations_items_include_affiliate_tag(self) -> None: - """Test that variation items include the affiliate tag in URLs.""" - if self.variations_result: - items = self.variations_result.items - if items: - item = items[0] - if item.detail_page_url: - self.assertIn(self.affiliate_tag, item.detail_page_url) - else: - self.skipTest(self.NO_VARIATIONS_FOUND_MSG) - - async def test_get_browse_nodes_returns_results(self) -> None: - """Test that get_browse_nodes returns browse node information.""" - self.assertGreater(len(self.browse_nodes_result), 0) - - async def test_get_browse_nodes_returns_node_info(self) -> None: - """Test that browse nodes contain expected information.""" - if self.browse_nodes_result: - node = self.browse_nodes_result[0] - self.assertIsNotNone(node.id) - self.assertIsNotNone(node.display_name) - - async def test_offers_v2_listing_has_is_buy_box_winner(self) -> None: - """Test that OffersV2 listings include is_buy_box_winner attribute.""" - item = self.item_with_offers - self.assertIsNotNone(item.offers_v2) - - if item.offers_v2 and item.offers_v2.listings: - listing = item.offers_v2.listings[0] - self.assertIsInstance(listing.is_buy_box_winner, bool) - - async def test_offers_v2_listing_has_type(self) -> None: - """Test that OffersV2 listings include offer type.""" - item = self.item_with_offers - self.assertIsNotNone(item.offers_v2) - - if item.offers_v2 and item.offers_v2.listings: - listing = item.offers_v2.listings[0] - if listing.type: - self.assertIsNotNone(listing.type) - - async def test_offers_v2_price_has_savings_when_available(self) -> None: - """Test that OffersV2 price includes savings info when available.""" - item = self.item_with_offers - self.assertIsNotNone(item.offers_v2) - - if item.offers_v2 and item.offers_v2.listings: - listing = item.offers_v2.listings[0] - if listing.price and listing.price.savings: - savings = listing.price.savings - if savings.money: - self.assertIsNotNone(savings.money.amount) - if savings.percentage is not None: - self.assertIsInstance(savings.percentage, (int, float)) - - async def test_search_items_returns_item_info(self) -> None: - """Test that search results include item info with title.""" - if self.search_result.items: - item = self.search_result.items[0] - self.assertIsNotNone(item.item_info) - if item.item_info: - self.assertIsNotNone(item.item_info.title) - if item.item_info.title: - self.assertIsNotNone(item.item_info.title.display_value) - title_display = item.item_info.title.display_value - self.assertIsNotNone(title_display) - self.assertIsInstance(title_display, str) - if title_display: - self.assertGreater(len(title_display), 0) - - async def test_search_items_returns_valid_asin(self) -> None: - """Test that search results return valid ASIN format.""" - if self.search_result.items: - item = self.search_result.items[0] - self.assertIsNotNone(item.asin) - if item.asin: - self.assertEqual(len(item.asin), 10) - self.assertTrue(item.asin.isalnum()) - - async def test_search_items_returns_images(self) -> None: - """Test that search results include product images.""" - if self.search_result.items: - item = self.search_result.items[0] - self.assertIsNotNone(item.images) - if item.images and item.images.primary: - self.assertIsNotNone(item.images.primary.large) - if item.images.primary.large: - large_url = item.images.primary.large.url - self.assertIsNotNone(large_url) - if large_url: - self.assertTrue(large_url.startswith("http")) - - async def test_get_items_returns_item_info(self) -> None: - """Test that get_items returns item info with title.""" - if self.get_items_result: - item = self.get_items_result[0] - self.assertIsNotNone(item.item_info) - if item.item_info and item.item_info.title: - self.assertIsNotNone(item.item_info.title.display_value) - - async def test_get_variations_returns_offers_v2(self) -> None: - """Test that get_variations returns OffersV2 data for variation items.""" - if self.variations_result and self.variations_result.items: - item_with_offers = next( - (item for item in self.variations_result.items if item.offers_v2), - None, - ) - if item_with_offers and item_with_offers.offers_v2: - self.assertIsNotNone(item_with_offers.offers_v2) - if item_with_offers.offers_v2.listings: - listing = item_with_offers.offers_v2.listings[0] - self.assertIsNotNone(listing) - else: - self.skipTest(self.NO_VARIATIONS_FOUND_MSG) - - async def test_context_manager_works_correctly(self) -> None: - """Test that async context manager works for connection pooling.""" - ( - credential_id, - credential_secret, - api_version, - affiliate_tag, - marketplace, - country_code, - ) = get_api_credentials() - - api = AsyncAmazonCreatorsApi( - credential_id=credential_id, # type: ignore[arg-type] - credential_secret=credential_secret, # type: ignore[arg-type] - version=api_version, # type: ignore[arg-type] - tag=affiliate_tag, # type: ignore[arg-type] - marketplace=marketplace, - country=country_code, # type: ignore[arg-type] - throttling=1, - ) +@skipUnless(has_credentials(), SKIP_NO_CREDENTIALS) +class AsyncIntegrationTest(IntegrationAssertions): + """Run the shared assertions against the async client.""" - async with api: - # Make a simple API call inside context manager - result = await api.search_items(keywords="book", item_count=1) - self.assertIsNotNone(result) - self.assertIsNotNone(result.items) + __test__ = True + @classmethod + def setUpClass(cls) -> None: + """Collect the snapshot of the API once for the whole suite.""" + credentials = load_credentials() -if __name__ == "__main__": - import unittest + if credentials is None: + raise SkipTest(SKIP_NO_CREDENTIALS) - unittest.main() + cls.snapshot = asyncio.run(collect_snapshot(credentials)) diff --git a/tests/amazon_creatorsapi/integration_test.py b/tests/amazon_creatorsapi/integration_test.py index 266d9b6..bca2b54 100644 --- a/tests/amazon_creatorsapi/integration_test.py +++ b/tests/amazon_creatorsapi/integration_test.py @@ -1,473 +1,170 @@ -"""Integration tests for Amazon Creators API.""" +"""Integration tests for the sync client of the Amazon Creators API. + +The suite makes every call once in setUpClass and asserts the results from +tests that never reach the network. See tests.integration_support for the +request budget and the assertions shared with the async suite. +""" from __future__ import annotations import contextlib -import os -from pathlib import Path -from typing import TYPE_CHECKING, ClassVar, cast -from unittest import TestCase, skipUnless - -from dotenv import load_dotenv +from typing import TYPE_CHECKING +from unittest import SkipTest, skipUnless from amazon_creatorsapi import AmazonCreatorsApi -from amazon_creatorsapi.errors import ItemsNotFoundError +from amazon_creatorsapi.errors import ( + AccessDeniedError, + ItemsNotFoundError, + ResourceNotFoundError, +) +from creatorsapi_python_sdk.models.availability import Availability +from creatorsapi_python_sdk.models.condition import Condition +from creatorsapi_python_sdk.models.sort_by import SortBy +from tests.integration_support import ( + SEARCH_ITEM_COUNT, + SEARCH_KEYWORDS, + SKIP_NO_CREDENTIALS, + ApiSnapshot, + Credentials, + IntegrationAssertions, + build_item_ids, + get_expected_item_ids, + get_found_items, + has_credentials, + load_credentials, + pick_browse_node_ids, + pick_item_asins, + pick_variation_asin, +) if TYPE_CHECKING: - from creatorsapi_python_sdk.models.browse_node import BrowseNode - from creatorsapi_python_sdk.models.item import Item - from creatorsapi_python_sdk.models.search_result import SearchResult - from creatorsapi_python_sdk.models.variations_result import VariationsResult - -# Load environment variables from .env file -load_dotenv(Path(__file__).parents[2] / ".env") + from creatorsapi_python_sdk.models.feed import Feed + from creatorsapi_python_sdk.models.report_metadata import ReportMetadata + + +def build_client(credentials: Credentials) -> AmazonCreatorsApi: + """Build the client used to collect the snapshot of the API.""" + return AmazonCreatorsApi( + credential_id=credentials.credential_id, + credential_secret=credentials.credential_secret, + version=credentials.version, + tag=credentials.tag, + marketplace=credentials.marketplace, + country=credentials.country, + throttling=1, + ) -def get_api_credentials() -> tuple[str, str, str, str, str, str]: - """Get API credentials from environment variables. +def list_feeds(api: AmazonCreatorsApi) -> list[Feed] | None: + """Return the feeds of the account, or None when it has no access. - Raises: - ValueError: If any required credential is missing. + Feeds belong to a program that not every account is enrolled in, so a + rejection is reported as no feeds instead of failing the whole suite. """ - credentials = { - "CREDENTIAL_ID": os.environ.get("CREDENTIAL_ID"), - "CREDENTIAL_SECRET": os.environ.get("CREDENTIAL_SECRET"), - "API_VERSION": os.environ.get("API_VERSION"), - "AFFILIATE_TAG": os.environ.get("AFFILIATE_TAG"), - } - - # We need either MARKETPLACE or COUNTRY_CODE - marketplace = os.environ.get("MARKETPLACE") - country_code = os.environ.get("COUNTRY_CODE") - - if not marketplace and not country_code: - msg = "Missing MARKETPLACE or COUNTRY_CODE environment variable" - raise ValueError(msg) - - missing = [key for key, value in credentials.items() if value is None] - if missing: - msg = f"Missing environment variables: {', '.join(missing)}" - raise ValueError(msg) - - return ( - cast("str", credentials["CREDENTIAL_ID"]), - cast("str", credentials["CREDENTIAL_SECRET"]), - cast("str", credentials["API_VERSION"]), - cast("str", credentials["AFFILIATE_TAG"]), - cast("str", marketplace), - cast("str", country_code), - ) - - -def has_api_credentials() -> bool: - """Check if all API credentials are available.""" try: - get_api_credentials() - except ValueError: - return False - return True + return api.list_feeds() + except (AccessDeniedError, ResourceNotFoundError): + return None -@skipUnless(has_api_credentials(), "Needs Amazon Creators API credentials") -class IntegrationTest(TestCase): - """Integration tests that make real API calls to Amazon creators API. +def get_feed_url(api: AmazonCreatorsApi, feeds: list[Feed] | None) -> str | None: + """Return the download URL of the first feed, if the account has any.""" + if not feeds: + return None - All API results are cached at class level to minimize the number of - requests. This reduces costs and avoids rate limiting. - """ + return api.get_feed(feeds[0].feed_name, feeds[0].feed_type) - api: ClassVar[AmazonCreatorsApi] - affiliate_tag: ClassVar[str] - search_result: ClassVar[SearchResult] - item_with_offers: ClassVar[Item] - get_items_result: ClassVar[list[Item]] - variations_result: ClassVar[VariationsResult] - browse_nodes_result: ClassVar[list[BrowseNode]] - NO_VARIATIONS_FOUND_MSG = "No variations found" +def list_reports(api: AmazonCreatorsApi) -> list[ReportMetadata] | None: + """Return the reports of the account, or None when it has no access.""" + try: + return api.list_reports() + except (AccessDeniedError, ResourceNotFoundError): + return None - @classmethod - def _find_item_with_offers(cls, items: list[Item]) -> Item: - """Find an item with offers, price, and in stock. - - Returns an item that: - - Has offers_v2 with listings - - Has a valid price (price.money.amount is not None) - - Is not out of stock (availability.type != OutOfStock) - """ - return next( - (item for item in items if cls._has_valid_offer(item)), - items[0] if items else cls.search_result.items[0], # type: ignore[index] - ) - @classmethod - def _has_valid_offer(cls, item: Item) -> bool: - """Check if an item has a valid offer with price and availability. +def get_report_url( + api: AmazonCreatorsApi, + reports: list[ReportMetadata] | None, +) -> str | None: + """Return the download URL of the first report, if the account has any.""" + if not reports: + return None - Args: - item: The item to check. + return api.get_report(reports[0].filename, reports[0].report_type) - Returns: - True if the item has a valid offer with price and is in stock. - """ - if item.offers_v2 is None or not item.offers_v2.listings: - return False - listing = item.offers_v2.listings[0] +def collect_snapshot(credentials: Credentials) -> ApiSnapshot: + """Make every API call of the suite once and cache their results. - # Check that the listing has a valid price - has_price = ( - listing.price is not None - and listing.price.money is not None - and listing.price.money.amount is not None + The search is also the discovery call: the items it returns provide the + ASINs, the parent ASIN and the browse node identifiers of the calls that + follow, so none of them needs a request of its own. + """ + with build_client(credentials) as api: + search_result = api.search_items( + keywords=SEARCH_KEYWORDS, + item_count=SEARCH_ITEM_COUNT, + availability=Availability.AVAILABLE, + condition=Condition.NEW, + sort_by=SortBy.FEATURED, ) + found_items = get_found_items(search_result) - # Check that the product is not out of stock - is_available = ( - listing.availability is None - or listing.availability.type is None - or listing.availability.type != "OutOfStock" + asins = pick_item_asins(found_items) + items = api.get_items( + build_item_ids(asins, api.marketplace), + include_unavailable=True, ) - return has_price and is_available + variations = None + variation_asin = pick_variation_asin(found_items) + if variation_asin: + with contextlib.suppress(ItemsNotFoundError): + variations = api.get_variations(variation_asin) - @classmethod - def _find_variation_asin(cls, items: list[Item]) -> str | None: - """Find ASIN to use for variations lookup.""" - item_with_variations = next( - (item for item in items if item.parent_asin), - None, + browse_node_ids = pick_browse_node_ids(found_items) + browse_nodes = ( + api.get_browse_nodes(browse_node_ids) if browse_node_ids else None ) - if item_with_variations: - return item_with_variations.parent_asin - - if cls.search_result.items: - return cls.search_result.items[0].asin + # Releasing the connections must not stop the client from working + api.close() + feeds = list_feeds(api) + feed_url = get_feed_url(api, feeds) + + # Dropping the token must make the next call authenticate again + api._clear_token() + reports = list_reports(api) + report_url = get_report_url(api, reports) + + return ApiSnapshot( + tag=credentials.tag, + search_result=search_result, + requested_ids=get_expected_item_ids(asins), + items=items, + variations=variations, + browse_node_ids=browse_node_ids, + browse_nodes=browse_nodes, + feeds=feeds, + feed_url=feed_url, + reports=reports, + report_url=report_url, + ) - return None - @classmethod - def _setup_variations_result(cls, items: list[Item]) -> None: - """Set up variations result if possible.""" - target_asin = cls._find_variation_asin(items) +@skipUnless(has_credentials(), SKIP_NO_CREDENTIALS) +class IntegrationTest(IntegrationAssertions): + """Run the shared assertions against the sync client.""" - if target_asin: - with contextlib.suppress(ItemsNotFoundError): - cls.variations_result = cls.api.get_variations(target_asin) + __test__ = True @classmethod - def _setup_browse_nodes_result(cls, items: list[Item]) -> None: - """Set up browse nodes result if possible.""" - item_with_browse_nodes = next( - ( - item - for item in items - if item.browse_node_info and item.browse_node_info.browse_nodes - ), - None, - ) - - if not item_with_browse_nodes: - cls.browse_nodes_result = [] - return - - browse_node_info = item_with_browse_nodes.browse_node_info - if not browse_node_info or not browse_node_info.browse_nodes: - cls.browse_nodes_result = [] - return + def setUpClass(cls) -> None: + """Collect the snapshot of the API once for the whole suite.""" + credentials = load_credentials() - browse_node_id = browse_node_info.browse_nodes[0].id - if browse_node_id: - cls.browse_nodes_result = cls.api.get_browse_nodes([browse_node_id]) - else: - cls.browse_nodes_result = [] + if credentials is None: + raise SkipTest(SKIP_NO_CREDENTIALS) - @classmethod - def setUpClass(cls) -> None: - """Set up API client and make shared API calls once for all tests.""" - ( - credential_id, - credential_secret, - api_version, - affiliate_tag, - marketplace, - country_code, - ) = get_api_credentials() - - cls.api = AmazonCreatorsApi( - credential_id=credential_id, - credential_secret=credential_secret, - version=api_version, - tag=affiliate_tag, - marketplace=marketplace, - country=country_code, # type: ignore[arg-type] - throttling=1, - ) - cls.affiliate_tag = affiliate_tag - - cls.search_result = cls.api.search_items(keywords="laptop") - items = cls.search_result.items or [] - - # Pick an item that has offers - cls.item_with_offers = cls._find_item_with_offers(items) - - # Get items by ASIN - if cls.item_with_offers.asin: - cls.get_items_result = cls.api.get_items([cls.item_with_offers.asin]) - else: - cls.get_items_result = [] - - # Set up variations and browse nodes - cls._setup_variations_result(items) - cls._setup_browse_nodes_result(items) - - def test_search_items_returns_expected_count(self) -> None: - """Test that search returns no more items than the default page size.""" - # API defaults to a page size of 10, but may return fewer items - items = self.search_result.items - if items: - self.assertLessEqual(len(items), 10) - - def test_search_items_includes_affiliate_tag(self) -> None: - """Test that search results include the affiliate tag in URLs.""" - if self.search_result.items: - searched_item = self.search_result.items[0] - if searched_item.detail_page_url: - self.assertIn(self.affiliate_tag, searched_item.detail_page_url) - - def test_search_items_returns_offers_v2(self) -> None: - """Test that search results include OffersV2 data.""" - items = self.search_result.items - self.assertIsNotNone(items) - if items: - self.assertGreater(len(items), 0) - - item = self.item_with_offers - self.assertIsNotNone(item.offers_v2) - - if item.offers_v2 and item.offers_v2.listings: - listing = item.offers_v2.listings[0] - self.assertIsNotNone(listing) - - def test_offers_v2_listing_has_price_info(self) -> None: - """Test that OffersV2 listings include price information.""" - item = self.item_with_offers - self.assertIsNotNone(item.offers_v2) - - if item.offers_v2 and item.offers_v2.listings: - listing = item.offers_v2.listings[0] - self.assertIsNotNone(listing.price) - - if listing.price and listing.price.money: - self.assertIsNotNone(listing.price.money.amount) - self.assertIsNotNone(listing.price.money.currency) - self.assertIsNotNone(listing.price.money.display_amount) - - def test_offers_v2_listing_has_merchant_info(self) -> None: - """Test that OffersV2 listings include merchant information.""" - item = self.item_with_offers - self.assertIsNotNone(item.offers_v2) - - if item.offers_v2 and item.offers_v2.listings: - listing = item.offers_v2.listings[0] - - if listing.merchant_info: - self.assertIsNotNone(listing.merchant_info.name) - - def test_offers_v2_listing_has_condition(self) -> None: - """Test that OffersV2 listings include condition information.""" - item = self.item_with_offers - self.assertIsNotNone(item.offers_v2) - - if item.offers_v2 and item.offers_v2.listings: - listing = item.offers_v2.listings[0] - - if listing.condition: - self.assertIsNotNone(listing.condition.value) - - def test_offers_v2_listing_has_availability(self) -> None: - """Test that OffersV2 listings include availability information.""" - item = self.item_with_offers - self.assertIsNotNone(item.offers_v2) - - if item.offers_v2 and item.offers_v2.listings: - listing = item.offers_v2.listings[0] - - if listing.availability: - self.assertIsNotNone(listing.availability.type) - - def test_get_items_returns_single_result(self) -> None: - """Test that get_items returns exactly one item when given one ASIN.""" - result = self.get_items_result - self.assertEqual(1, len(result)) - - def test_get_items_includes_affiliate_tag(self) -> None: - """Test that get_items results include the affiliate tag in URLs.""" - if self.get_items_result: - detail_url = self.get_items_result[0].detail_page_url - if detail_url: - self.assertIn(self.affiliate_tag, detail_url) - - def test_get_items_returns_offers_v2(self) -> None: - """Test that get_items returns OffersV2 data with price details.""" - if self.get_items_result: - item = self.get_items_result[0] - self.assertIsNotNone(item.offers_v2) - - if item.offers_v2 and item.offers_v2.listings: - listing = item.offers_v2.listings[0] - self.assertIsNotNone(listing) - - if listing.price and listing.price.money: - self.assertIsNotNone(listing.price.money.amount) - self.assertIsNotNone(listing.price.money.display_amount) - - def test_get_variations_returns_items(self) -> None: - """Test that get_variations returns a list of variation items.""" - if hasattr(self, "variations_result") and self.variations_result: - self.assertIsNotNone(self.variations_result) - items = self.variations_result.items - self.assertIsNotNone(items) - # We already checked self.variations_result is truthy, but items could - # be None/empty - if items: - self.assertGreater(len(items), 0) - else: - self.skipTest(self.NO_VARIATIONS_FOUND_MSG) - - def test_get_variations_returns_variation_summary(self) -> None: - """Test that get_variations returns variation summary.""" - if hasattr(self, "variations_result") and self.variations_result: - summary = self.variations_result.variation_summary - self.assertIsNotNone(summary) - if summary: - self.assertIsNotNone(summary.variation_count) - # Cast to int to satisfy mypy, assertIsNotNone guarantees it's not None - self.assertGreater(summary.variation_count, 0) # type: ignore[arg-type] - - else: - self.skipTest(self.NO_VARIATIONS_FOUND_MSG) - - def test_get_variations_items_include_affiliate_tag(self) -> None: - """Test that variation items include the affiliate tag in URLs.""" - if hasattr(self, "variations_result") and self.variations_result: - items = self.variations_result.items - if items: - item = items[0] - if item.detail_page_url: - self.assertIn(self.affiliate_tag, item.detail_page_url) - else: - self.skipTest(self.NO_VARIATIONS_FOUND_MSG) - - def test_get_browse_nodes_returns_results(self) -> None: - """Test that get_browse_nodes returns browse node information.""" - if not hasattr(self, "browse_nodes_result"): - self.fail("browse_nodes_result not set") - self.assertGreater(len(self.browse_nodes_result), 0) - - def test_get_browse_nodes_returns_node_info(self) -> None: - """Test that browse nodes contain expected information.""" - if self.browse_nodes_result: - node = self.browse_nodes_result[0] - self.assertIsNotNone(node.id) - self.assertIsNotNone(node.display_name) - - def test_offers_v2_listing_has_is_buy_box_winner(self) -> None: - """Test that OffersV2 listings include is_buy_box_winner attribute.""" - item = self.item_with_offers - self.assertIsNotNone(item.offers_v2) - - if item.offers_v2 and item.offers_v2.listings: - listing = item.offers_v2.listings[0] - self.assertIsInstance(listing.is_buy_box_winner, bool) - - def test_offers_v2_listing_has_type(self) -> None: - """Test that OffersV2 listings include offer type.""" - item = self.item_with_offers - self.assertIsNotNone(item.offers_v2) - - if item.offers_v2 and item.offers_v2.listings: - listing = item.offers_v2.listings[0] - if listing.type: - self.assertIsNotNone(listing.type) - - def test_offers_v2_price_has_savings_when_available(self) -> None: - """Test that OffersV2 price includes savings info when available.""" - item = self.item_with_offers - self.assertIsNotNone(item.offers_v2) - - if item.offers_v2 and item.offers_v2.listings: - listing = item.offers_v2.listings[0] - if listing.price and listing.price.savings: - savings = listing.price.savings - if savings.money: - self.assertIsNotNone(savings.money.amount) - if savings.percentage is not None: - self.assertIsInstance(savings.percentage, (int, float)) - - def test_search_items_returns_item_info(self) -> None: - """Test that search results include item info with title.""" - if self.search_result.items: - item = self.search_result.items[0] - self.assertIsNotNone(item.item_info) - if item.item_info: - self.assertIsNotNone(item.item_info.title) - if item.item_info.title: - self.assertIsNotNone(item.item_info.title.display_value) - if item.item_info.title: - title_display = item.item_info.title.display_value - self.assertIsNotNone(title_display) - self.assertIsInstance(title_display, str) - if title_display: - self.assertGreater(len(title_display), 0) - - def test_search_items_returns_valid_asin(self) -> None: - """Test that search results return valid ASIN format.""" - if self.search_result.items: - item = self.search_result.items[0] - self.assertIsNotNone(item.asin) - if item.asin: - self.assertEqual(len(item.asin), 10) - self.assertTrue(item.asin.isalnum()) - - def test_search_items_returns_images(self) -> None: - """Test that search results include product images.""" - if self.search_result.items: - item = self.search_result.items[0] - self.assertIsNotNone(item.images) - if item.images and item.images.primary: - self.assertIsNotNone(item.images.primary.large) - if item.images.primary.large: - large_url = item.images.primary.large.url - self.assertIsNotNone(large_url) - if large_url: - self.assertTrue(large_url.startswith("http")) - - def test_get_items_returns_item_info(self) -> None: - """Test that get_items returns item info with title.""" - if self.get_items_result: - item = self.get_items_result[0] - self.assertIsNotNone(item.item_info) - if item.item_info and item.item_info.title: - self.assertIsNotNone(item.item_info.title.display_value) - - def test_get_variations_returns_offers_v2(self) -> None: - """Test that get_variations returns OffersV2 data for variation items.""" - if ( - hasattr(self, "variations_result") - and self.variations_result - and self.variations_result.items - ): - item_with_offers = next( - (item for item in self.variations_result.items if item.offers_v2), - None, - ) - # Not all variations might have offers, but if we found one - if item_with_offers and item_with_offers.offers_v2: - self.assertIsNotNone(item_with_offers.offers_v2) - if item_with_offers.offers_v2.listings: - listing = item_with_offers.offers_v2.listings[0] - self.assertIsNotNone(listing) - else: - self.skipTest(self.NO_VARIATIONS_FOUND_MSG) + cls.snapshot = collect_snapshot(credentials) diff --git a/tests/integration_support.py b/tests/integration_support.py new file mode 100644 index 0000000..4001fea --- /dev/null +++ b/tests/integration_support.py @@ -0,0 +1,540 @@ +"""Shared pieces for the integration tests that hit the real Creators API. + +The account running these tests has a limited amount of requests per day, so +every call is made once, cached in an ApiSnapshot and then asserted from as +many angles as possible. Both the sync and the async suites reuse the +assertions defined here, which holds the two clients to the same contract. + +Request budget per client, all of them made while building the snapshot: + + 1. search_items also the discovery call for the ones below + 2. get_items mixes a URL, a duplicate and a missing ASIN + 3. get_variations only when the search found an item with variations + 4. get_browse_nodes only when the search found browse nodes + 5. list_feeds skipped for credentials without the feeds program + 6. get_feed only when the account has feeds + 7. list_reports skipped for credentials without reports + 8. get_report only when the account has reports + +Nothing else reaches the network: the tests only read the snapshot, and the +extra behaviours worth checking, such as releasing the connections or getting +a new token, are folded into the calls above instead of costing a request. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING, ClassVar, cast +from unittest import SkipTest, TestCase + +from dotenv import load_dotenv + +if TYPE_CHECKING: + from amazon_creatorsapi.core.marketplaces import CountryCode + from amazon_creatorsapi.core.results import ResultList + from creatorsapi_python_sdk.models.browse_node import BrowseNode + from creatorsapi_python_sdk.models.feed import Feed + from creatorsapi_python_sdk.models.item import Item + from creatorsapi_python_sdk.models.offer_listing_v2 import OfferListingV2 + from creatorsapi_python_sdk.models.report_metadata import ReportMetadata + from creatorsapi_python_sdk.models.search_result import SearchResult + from creatorsapi_python_sdk.models.variations_result import VariationsResult + +load_dotenv(Path(__file__).parents[1] / ".env") + +# Keyword broad enough to return offers, variations and browse nodes in any +# marketplace, as the country of the credentials is not known beforehand +SEARCH_KEYWORDS = "laptop" +SEARCH_ITEM_COUNT = 10 +MAX_BROWSE_NODES = 2 +MAX_ITEM_ASINS = 2 +ASIN_LENGTH = 10 + +# Well formed identifier that Amazon does not know, used to check the partial +# errors and the placeholders of a request for several items +MISSING_ASIN = "0000000000" + +SKIP_NO_CREDENTIALS = "Needs Amazon Creators API credentials" +SKIP_NO_SNAPSHOT = "The suite collected no snapshot of the API" +SKIP_NO_VARIATIONS = "The search returned no item with variations" +SKIP_NO_BROWSE_NODES = "The search returned no browse nodes" +SKIP_NO_FEEDS = "The credentials have no feeds available" +SKIP_NO_REPORTS = "The credentials have no reports available" + + +@dataclass(frozen=True) +class Credentials: + """Credentials and marketplace taken from the environment.""" + + credential_id: str + credential_secret: str + version: str + tag: str + marketplace: str | None + country: CountryCode | None + + +@dataclass(frozen=True) +class ApiSnapshot: + """Every result of the API calls made by a suite, cached for its tests. + + A field set to None means the call was not made, either because the + discovery call gave nothing to ask for or because the credentials cannot + reach that part of the API. The tests reading it skip instead of failing. + """ + + tag: str + search_result: SearchResult + requested_ids: list[str] + items: ResultList[Item] + variations: VariationsResult | None + browse_node_ids: list[str] + browse_nodes: ResultList[BrowseNode] | None + feeds: list[Feed] | None + feed_url: str | None + reports: list[ReportMetadata] | None + report_url: str | None + + +def load_credentials() -> Credentials | None: + """Read the credentials of the API from the environment. + + Returns: + The credentials, or None when any of them is missing. A marketplace + or a country is enough, as the client derives one from the other. + + """ + credential_id = os.environ.get("CREDENTIAL_ID") + credential_secret = os.environ.get("CREDENTIAL_SECRET") + version = os.environ.get("API_VERSION") + tag = os.environ.get("AFFILIATE_TAG") + marketplace = os.environ.get("MARKETPLACE") + country = os.environ.get("COUNTRY_CODE") + + if not (credential_id and credential_secret and version and tag): + return None + + if not marketplace and not country: + return None + + return Credentials( + credential_id=credential_id, + credential_secret=credential_secret, + version=version, + tag=tag, + marketplace=marketplace or None, + country=cast("CountryCode", country) if country else None, + ) + + +def has_credentials() -> bool: + """Tell whether the environment holds every credential the tests need.""" + return load_credentials() is not None + + +def get_found_items(search_result: SearchResult) -> list[Item]: + """Return the items of a search result, empty when it carried none.""" + return list(search_result.items or []) + + +def has_usable_offer(item: Item) -> bool: + """Tell whether an item is in stock and carries a price in its offer.""" + if item.offers_v2 is None or not item.offers_v2.listings: + return False + + listing = item.offers_v2.listings[0] + + has_price = ( + listing.price is not None + and listing.price.money is not None + and listing.price.money.amount is not None + ) + is_available = ( + listing.availability is None + or listing.availability.type is None + or listing.availability.type != "OutOfStock" + ) + + return has_price and is_available + + +def pick_item_asins(items: list[Item], limit: int = MAX_ITEM_ASINS) -> list[str]: + """Return the ASINs to ask get_items for, preferring items with offers. + + Args: + items: Items returned by the search call. + limit: Amount of ASINs to return at most. + + Returns: + Up to limit ASINs, the ones of items with a usable offer first, so + the offers of the response can be asserted. + + """ + with_offers = [item for item in items if has_usable_offer(item)] + without_offers = [item for item in items if not has_usable_offer(item)] + + return [item.asin for item in [*with_offers, *without_offers] if item.asin][:limit] + + +def pick_variation_asin(items: list[Item]) -> str | None: + """Return the parent ASIN of the first item that belongs to a family.""" + return next((item.parent_asin for item in items if item.parent_asin), None) + + +def pick_browse_node_ids(items: list[Item], limit: int = MAX_BROWSE_NODES) -> list[str]: + """Return distinct browse node identifiers found in the search results. + + Args: + items: Items returned by the search call. + limit: Amount of identifiers to return at most. + + Returns: + Up to limit browse node identifiers, without duplicates, so a single + call can check that every requested node comes back. + + """ + node_ids: list[str] = [] + + for item in items: + info = item.browse_node_info + for node in (info.browse_nodes or []) if info else []: + if node.id and node.id not in node_ids: + node_ids.append(node.id) + + return node_ids[:limit] + + +def build_item_ids(asins: list[str], marketplace: str) -> list[str]: + """Build the identifiers of the single get_items call of a suite. + + The list mixes an Amazon URL, the plain ASINs of the other items, a + duplicate of the first one and an identifier Amazon does not know, so one + request checks the parsing of URLs, the removal of duplicates, the order + of the response, its partial errors and the placeholders added for the + items that Amazon did not return. + + Args: + asins: ASINs discovered by the search call. + marketplace: Marketplace of the client, used to build the URL. + + Returns: + The identifiers to request, with the duplicate still in place. + + Raises: + RuntimeError: If the search returned no ASIN to request. + + """ + if not asins: + msg = "The search returned no item to request by ASIN" + raise RuntimeError(msg) + + return [ + f"https://{marketplace}/dp/{asins[0]}", + *asins[1:], + asins[0], + MISSING_ASIN, + ] + + +def get_expected_item_ids(asins: list[str]) -> list[str]: + """Return the ASINs that a get_items call should answer with, in order.""" + return [*asins, MISSING_ASIN] + + +class IntegrationAssertions(TestCase): + """Assertions run against the snapshot collected by each client. + + The suites of both clients inherit from this class, so the sync and the + async implementations are held to the same contract. It is not collected + on its own, as it has no snapshot to assert. + """ + + __test__ = False + + snapshot: ClassVar[ApiSnapshot] + + @classmethod + def setUpClass(cls) -> None: + """Skip the class when the suite collected no snapshot.""" + if not hasattr(cls, "snapshot"): + raise SkipTest(SKIP_NO_SNAPSHOT) + + def test_search_returns_items_within_the_requested_count(self) -> None: + """Test that the search honours the requested amount of items.""" + items = get_found_items(self.snapshot.search_result) + self.assertGreater(len(items), 0) + self.assertLessEqual(len(items), SEARCH_ITEM_COUNT) + + def test_search_returns_the_total_amount_of_results(self) -> None: + """Test that the search reports how many results Amazon has.""" + total = self.snapshot.search_result.total_result_count + + if total is None: + self.skipTest("The marketplace reported no total result count") + + self.assertGreater(total, 0) + + def test_search_returns_the_url_of_the_results(self) -> None: + """Test that the search returns the URL of its results page.""" + search_url = self.snapshot.search_result.search_url + + if search_url is None: + self.skipTest("The marketplace reported no search URL") + + self.assertTrue(search_url.startswith("http")) + + def test_search_items_have_a_valid_asin(self) -> None: + """Test that every item of the search carries a well formed ASIN.""" + for item in get_found_items(self.snapshot.search_result): + self.assertIsNotNone(item.asin) + self.assertEqual(ASIN_LENGTH, len(item.asin or "")) + self.assertTrue((item.asin or "").isalnum()) + + def test_search_items_include_the_affiliate_tag(self) -> None: + """Test that every detail page URL of the search carries the tag.""" + urls = [ + item.detail_page_url + for item in get_found_items(self.snapshot.search_result) + if item.detail_page_url + ] + + self.assertGreater(len(urls), 0) + for url in urls: + self.assertIn(self.snapshot.tag, url) + + def test_search_items_include_a_title(self) -> None: + """Test that the items of the search carry a non empty title.""" + titles = [ + item.item_info.title.display_value + for item in get_found_items(self.snapshot.search_result) + if item.item_info and item.item_info.title + ] + + self.assertGreater(len(titles), 0) + for title in titles: + self.assertIsNotNone(title) + self.assertGreater(len(title or ""), 0) + + def test_search_items_include_images(self) -> None: + """Test that the items of the search carry the URL of their image.""" + images = [ + item.images.primary.large + for item in get_found_items(self.snapshot.search_result) + if item.images and item.images.primary and item.images.primary.large + ] + + self.assertGreater(len(images), 0) + for image in images: + self.assertIsNotNone(image.url) + self.assertTrue((image.url or "").startswith("http")) + + def test_search_items_include_their_browse_nodes(self) -> None: + """Test that the items of the search carry identified browse nodes.""" + if not self.snapshot.browse_node_ids: + self.skipTest(SKIP_NO_BROWSE_NODES) + + for node_id in self.snapshot.browse_node_ids: + self.assertGreater(len(node_id), 0) + + def test_search_returns_an_item_with_a_usable_offer(self) -> None: + """Test that the search returns offers with a price and stock.""" + items = get_found_items(self.snapshot.search_result) + self.assertTrue(any(has_usable_offer(item) for item in items)) + + def test_offers_include_a_complete_price(self) -> None: + """Test that the listings of an offer carry amount and currency.""" + listing = self._get_listing_with_offer() + price = listing.price + + if price is None or price.money is None: + self.fail("The listing of the offer carried no price") + + self.assertIsNotNone(price.money.amount) + self.assertIsNotNone(price.money.currency) + self.assertIsNotNone(price.money.display_amount) + + def test_offers_describe_how_the_item_is_sold(self) -> None: + """Test that the listings of an offer describe condition and stock.""" + listing = self._get_listing_with_offer() + + self.assertIsNotNone(listing.condition) + self.assertIsNotNone(listing.availability) + self.assertIsInstance(listing.is_buy_box_winner, bool) + + if listing.condition: + self.assertIsNotNone(listing.condition.value) + if listing.availability: + self.assertIsNotNone(listing.availability.type) + if listing.merchant_info: + self.assertIsNotNone(listing.merchant_info.name) + + def test_offers_savings_are_consistent_when_present(self) -> None: + """Test that the savings of an offer come with amount and percentage.""" + listing = self._get_listing_with_offer() + savings = listing.price.savings if listing.price else None + + if savings is None: + self.skipTest("The offer carried no savings") + + if savings.money: + self.assertIsNotNone(savings.money.amount) + if savings.percentage is not None: + self.assertGreater(savings.percentage, 0) + + def test_get_items_returns_the_requested_ids_in_order(self) -> None: + """Test the parsing of URLs, the deduplication and the ordering. + + A single request asks for a URL, the ASIN it points to, a duplicate of + it and an identifier Amazon does not know, so its answer proves that + the client parses URLs, drops duplicates and keeps the asked order. + """ + self.assertEqual( + self.snapshot.requested_ids, + [item.asin for item in self.snapshot.items], + ) + + def test_get_items_reports_the_errors_of_the_missing_item(self) -> None: + """Test that the partial errors of Amazon reach the caller.""" + errors = self.snapshot.items.errors + + self.assertGreater(len(errors), 0) + for error in errors: + self.assertGreater(len(error.code), 0) + self.assertGreater(len(error.message), 0) + + def test_get_items_adds_a_placeholder_for_the_missing_item(self) -> None: + """Test that include_unavailable adds an item with only its ASIN.""" + placeholder = self.snapshot.items[-1] + + self.assertEqual(MISSING_ASIN, placeholder.asin) + self.assertIsNone(placeholder.item_info) + self.assertIsNone(placeholder.detail_page_url) + + def test_get_items_returns_complete_items(self) -> None: + """Test that the found items carry their info, URL and offers.""" + found = self.snapshot.items[:-1] + self.assertGreater(len(found), 0) + + for item in found: + self.assertIsNotNone(item.item_info) + self.assertIsNotNone(item.detail_page_url) + self.assertIn(self.snapshot.tag, item.detail_page_url or "") + + self.assertTrue(any(item.offers_v2 for item in found)) + + def test_get_variations_returns_items_of_the_same_family(self) -> None: + """Test that the variations of a product come with their attributes.""" + items = self._get_variations().items or [] + self.assertGreater(len(items), 0) + + for item in items: + self.assertIsNotNone(item.asin) + if item.detail_page_url: + self.assertIn(self.snapshot.tag, item.detail_page_url) + + self.assertTrue(any(item.variation_attributes for item in items)) + + def test_get_variations_returns_a_summary(self) -> None: + """Test that the variations come with a summary counting them.""" + summary = self._get_variations().variation_summary + + if summary is None: + self.fail("The variations came with no summary") + + if summary.variation_count is None: + self.skipTest("The marketplace reported no variation count") + + self.assertGreater(summary.variation_count, 0) + + def test_get_browse_nodes_returns_every_requested_node(self) -> None: + """Test that a request for several nodes answers with all of them.""" + nodes = self._get_browse_nodes() + + self.assertEqual( + set(self.snapshot.browse_node_ids), + {node.id for node in nodes}, + ) + + def test_get_browse_nodes_returns_named_nodes(self) -> None: + """Test that the browse nodes carry the names Amazon displays.""" + for node in self._get_browse_nodes(): + self.assertIsNotNone(node.display_name) + self.assertIsNotNone(node.context_free_name) + self.assertIsInstance(node.is_root, bool) + + def test_get_browse_nodes_returns_the_tree_of_a_node(self) -> None: + """Test that the browse nodes carry their ancestors or children.""" + nodes = self._get_browse_nodes() + self.assertTrue(any(node.ancestor or node.children for node in nodes)) + + def test_list_feeds_returns_described_feeds(self) -> None: + """Test that the feeds of the account come fully described.""" + for feed in self._get_feeds(): + self.assertGreater(len(feed.feed_name), 0) + self.assertGreater(len(feed.md5), 0) + self.assertGreater(len(feed.last_updated), 0) + self.assertGreater(feed.size, 0) + + def test_get_feed_returns_a_download_url(self) -> None: + """Test that a feed of the account can be turned into a URL.""" + self._get_feeds() + feed_url = self.snapshot.feed_url + + self.assertIsNotNone(feed_url) + self.assertTrue((feed_url or "").startswith("https://")) + + def test_list_reports_returns_described_reports(self) -> None: + """Test that the reports of the account come fully described.""" + for report in self._get_reports(): + self.assertGreater(len(report.filename), 0) + self.assertGreater(len(report.md5), 0) + self.assertGreater(len(report.last_modified), 0) + self.assertGreater(report.size, 0) + + def test_get_report_returns_a_download_url(self) -> None: + """Test that a report of the account can be turned into a URL.""" + self._get_reports() + report_url = self.snapshot.report_url + + self.assertIsNotNone(report_url) + self.assertTrue((report_url or "").startswith("https://")) + + def _get_listing_with_offer(self) -> OfferListingV2: + """Return the first listing of the search that has a usable offer.""" + items = get_found_items(self.snapshot.search_result) + item = next((item for item in items if has_usable_offer(item)), None) + + if item is None or item.offers_v2 is None or not item.offers_v2.listings: + self.skipTest("The search returned no item with a usable offer") + + return item.offers_v2.listings[0] + + def _get_variations(self) -> VariationsResult: + """Return the cached variations, skipping the test without them.""" + if self.snapshot.variations is None: + self.skipTest(SKIP_NO_VARIATIONS) + + return self.snapshot.variations + + def _get_browse_nodes(self) -> ResultList[BrowseNode]: + """Return the cached browse nodes, skipping the test without them.""" + if self.snapshot.browse_nodes is None: + self.skipTest(SKIP_NO_BROWSE_NODES) + + return self.snapshot.browse_nodes + + def _get_feeds(self) -> list[Feed]: + """Return the cached feeds, skipping the test without them.""" + if not self.snapshot.feeds: + self.skipTest(SKIP_NO_FEEDS) + + return self.snapshot.feeds + + def _get_reports(self) -> list[ReportMetadata]: + """Return the cached reports, skipping the test without them.""" + if not self.snapshot.reports: + self.skipTest(SKIP_NO_REPORTS) + + return self.snapshot.reports From 59c379a7a4b30ffc655a28462fefd395ba159504 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 13:58:49 +0000 Subject: [PATCH 2/3] docs: record the rework of the integration tests in the changelog Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_017ZLhDcp9SziFvYduax9evq --- CHANGELOG.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index c6c98e2..66d224c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - A `version` of a family that the library cannot authenticate is rejected even when `auth_endpoint` is given, instead of being sent with the Cognito flow and rejected by Amazon without an explanation - The error of an unsupported version tells that a newer version of a known family can be used by providing its `auth_endpoint` - The auth flow of a version and the `Authorization` header it expects are decided in a single place, and the copies bundled in the SDK are pinned to them by tests, so a bump of the SDK cannot leave both halves disagreeing +- The integration tests reach every operation of the API, feeds and reports included, and both clients share the same assertions, so a difference between them is a failure instead of a gap +- Every integration test reads the results of a single round of calls, which loads each request with as much as it can check, so the whole API is covered without spending more requests of the account +- The async integration tests make their calls when the suite runs instead of when the module is imported, so they are not sent when the tests are deselected and a failure is reported as such ## [7.4.0] - 2026-09-04 From e4cffdfbd1f7e854decfbf958149852a83dc026c Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 14:19:36 +0000 Subject: [PATCH 3/3] ci: keep the support of the integration tests out of the naming hook The name-tests-test hook asks every file under tests to be named after the pattern of a test module, and tests/integration_support.py is the support shared by both integration suites, holding no test of its own. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_017ZLhDcp9SziFvYduax9evq --- .pre-commit-config.yaml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 83d294e..47054bd 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -28,6 +28,8 @@ repos: - id: check-merge-conflict - id: debug-statements - id: name-tests-test + # Support shared by the integration tests, which holds no test of its own + exclude: ^tests/integration_support\.py$ - repo: https://github.com/lk16/detect-missing-init rev: v0.1.6