Developer Console
Gracias por tu visita. Esta página solo está disponible en inglés.

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']