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 diff --git a/CHANGELOG.md b/CHANGELOG.md index f2d999d..88933e7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - 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 usage guide covers everything the README does, so the documentation does not have to be read in both places, and the README links to it +- 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 ### Fixed 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