Developer Console
感谢您的访问。此页面目前仅提供英语版本。我们正在开发中文版本。谢谢您的理解。

Catalog API Invoke API

Summary

The invoke catalog API is used to upload items to the Amazon Just WalkOut API.

Python Code

            import requests
            from requests_aws4auth import AWS4Auth
            import json
            import boto3
            import os

            def update_catalog():
            try:
                # Load catalog JSON from a file or environment variable
                with open('catalog.json', 'r') as f:
                request_parameters = json.load(f)
                # Get role ARN from environment variable
                role_arn = os.environ['CATALOG_API_ROLE_ARN']
                # Get temp credentials using STS
                client_sts = boto3.client('sts')
                response = client_sts.assume_role(
                    RoleArn=role_arn,
                    RoleSessionName='CatalogAPI'
                )
                credentials = response['Credentials']
                # Auth the outgoing API call
                auth = AWS4Auth(
                    credentials['AccessKeyId'],
                    credentials['SecretAccessKey'],
                    os.environ['AWS_REGION'],
                    'execute-api',
                    session_token=credentials['SessionToken']
                )
                 # Use requests library to post to the API end point
                 api_url = os.environ['CATALOG_API_ENDPOINT']
                 response = requests.post(api_url, auth=auth, json=request_parameters)
                 # Check response code and act accordingly
                if response.status_code == 201:
                    response_text = response.json()
                    print(f"Ingestion ID: {response_text['ingestionId']}")
                elif response.status_code == 403:
                    print("Error authenticating to the API")
                elif response.status_code == 400:
                    print("Error: Bad request")
                else:
                    print(f"Unexpected status code: {response.status_code}")
                    print(f"Response status code: {response.status_code}")
                    print(f"Response body: {response.text}")
                except requests.exceptions.RequestException as e:
                    print(f"Request failed: {e}")
                except json.JSONDecodeError:
                    print("Failed to parse response JSON")
                except Exception as e:
                    print(f"An unexpected error occurred: {e}")

Unit Tests

import pytest
import json
import requests
from unittest.mock import patch, MagicMock
from botocore.exceptions import ClientError

from src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function import (
    lambda_handler,
    sign_request,
    validate_payload,
    get_table
)

@pytest.fixture
def valid_event():
    return {
        "catalogItems": [
            {
                "item_sku": "Test-012000161155",
                "external_product_id": "012000161155",
                "external_product_id_type": "UPC",
                "item_name": "Life Water",
                "store_id": "SAMPLE_STORE",
                "standard_price": "1.00",
                "brand_name": "Life Water",
                "product_tax_code": "A_GEN_TAX",
                "product_category": "Drink",
                "product_subcategory": "Water"
            }
        ]
    }

@pytest.fixture
def mock_table():
    return MagicMock()

@pytest.fixture
def mock_dynamodb(mock_table):
    with patch('boto3.resource') as mock_dynamo:
        mock_dynamo.return_value.Table.return_value = mock_table
        yield mock_dynamo

class TestLambdaFunction:
    @patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.requests.post')
    @patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.sign_request')
    @patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.get_table')
    def test_successful_execution(self, mock_get_table, mock_sign, mock_post, valid_event):
        mock_table = MagicMock()
        mock_get_table.return_value = mock_table

        mock_response = MagicMock()
        mock_response.json.return_value = {"status": "success"}
        mock_response.raise_for_status.return_value = None
        mock_post.return_value = mock_response

        mock_sign.return_value = {"Authorization": "test-auth"}

        response = lambda_handler(valid_event, None)

        assert response['statusCode'] == 200
        assert 'Successfully processed catalog items' in response['body']
        mock_table.put_item.assert_called_once()
        mock_post.assert_called_once()

    @patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.requests.post')
    @patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.sign_request')
    def test_api_error(self, mock_sign, mock_post, valid_event):
        mock_error = requests.exceptions.RequestException("API Error")
        mock_post.side_effect = mock_error
        mock_sign.return_value = {"Authorization": "test-auth"}

        response = lambda_handler(valid_event, None)

        assert response['statusCode'] == 500
        assert 'Error processing catalog items: API Error' in response['body']

    def test_missing_api_url(self, monkeypatch, valid_event):
        monkeypatch.delenv('API_URL', raising=False)

        response = lambda_handler(valid_event, None)

        assert response['statusCode'] == 500
        assert 'API_URL environment variable is not set' in response['body']

    @patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.get_table')
    def test_unexpected_error(self, mock_get_table, valid_event):
        mock_get_table.side_effect = ClientError(
            error_response={'Error': {'Code': 'InternalServerError'}},
            operation_name='GetTable'
        )

        response = lambda_handler(valid_event, None)

        assert response['statusCode'] == 500
        assert 'Unexpected error processing catalog items' in response['body']
        mock_get_table.assert_called_once()

    def test_validate_payload_valid(self, valid_event):
        assert validate_payload(valid_event) is True

    @pytest.mark.parametrize("invalid_payload", [
        None,
        {},
        {"catalogItems": []},
        {"catalogItems": [{}]},
        {"catalogItems": [{"item_sku": "test"}]},
    ])
    def test_validate_payload_invalid(self, invalid_payload):
        assert validate_payload(invalid_payload) is False

    @patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.boto3.Session')
    def test_sign_request(self, mock_session):
        mock_credentials = MagicMock(
            access_key='test-key',
            secret_key='test-secret',
            token='test-token'
        )
        mock_session.return_value.get_credentials.return_value = mock_credentials

        url = 'https://api-test.amazonaws.com/dev/catalog'
        method = 'POST'
        body = {"test": "data"}

        headers = sign_request(url, method, body)

        assert isinstance(headers, dict)
        assert 'Content-Type' in headers
        assert headers['Content-Type'] == 'application/json'
        mock_session.return_value.get_credentials.assert_called_once()

    def test_get_table(self, mock_dynamodb):
        table = get_table()

        assert table is not None
        mock_dynamodb.assert_called_once_with('dynamodb')
        mock_dynamodb.return_value.Table.assert_called_once_with('CatalogAPICalls')

    @patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.requests.post')
    @patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.sign_request')
    @patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.get_table')
    def test_lambda_handler_invalid_payload(self, mock_get_table, mock_sign, mock_post):
        response = lambda_handler({"invalid": "payload"}, None)
        assert response['statusCode'] == 400
        assert 'Invalid payload structure' in response['body']

    @patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.requests.post')
    @patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.sign_request')
    @patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.get_table')
    def test_lambda_handler_empty_event(self, mock_get_table, mock_sign, mock_post):
        response = lambda_handler({}, None)
        assert response['statusCode'] == 400
        assert 'Invalid payload structure' in response['body']