Skip to content
117 changes: 74 additions & 43 deletions tests/integration/init/schemas/test_init_with_schemas_command.py
Original file line number Diff line number Diff line change
@@ -1,19 +1,29 @@
import os
import tempfile
import pytest
from functools import lru_cache
from pathlib import Path
from unittest import skipIf

from boto3.session import Session
from click.testing import CliRunner

from samcli.commands.init import cli as init_cmd
from samcli.commands.init.init_templates import InitTemplates
from samcli.commands.init.interactive_init_flow import get_sorted_runtimes
from tests.integration.init.schemas.schemas_test_data_setup import SchemaTestDataSetup
from tests.testing_utils import RUNNING_ON_CI, RUNNING_TEST_FOR_MASTER_ON_CI, RUN_BY_CANARY

# Schemas tests require credentials. This is to skip running the test where credentials are not available.
SKIP_SCHEMA_TESTS = RUNNING_ON_CI and RUNNING_TEST_FOR_MASTER_ON_CI and not RUN_BY_CANARY

EVENT_BRIDGE_USE_CASE = "Infrastructure event management"

# Prompt labels, not runtime ids -- the go entry displays as "go (provided.al2)".
JAVA_RUNTIME_FOR_INIT = "java17.al2023"
PYTHON_RUNTIME_FOR_INIT = "python3.9"
GO_RUNTIME_FOR_INIT = "go (provided.al2)"


def _get_registry_position(registry_name):
"""Query EventBridge Schema registries and return the 1-based menu position for the given registry name.
Expand All @@ -34,38 +44,61 @@ def _get_registry_position(registry_name):
raise ValueError(f"Registry '{registry_name}' not found. Available: {registries}")


@lru_cache(maxsize=1)
def _get_manifest():
"""Fetch once per process: each call re-fetches MANIFEST_URL, and two calls can differ."""
return InitTemplates().get_preprocessed_manifest(None, None, None, None)


def _get_runtime_position(runtime_name):
"""Resolve a runtime's 1-based position; the order shifts as runtimes are added/removed."""
runtime_options = _get_manifest()[EVENT_BRIDGE_USE_CASE]
runtimes = get_sorted_runtimes(list(runtime_options.keys()))
for i, name in enumerate(runtimes, 1):
if name == runtime_name:
return i
raise ValueError(f"Runtime '{runtime_name}' not found. Available: {runtimes}")


def _get_use_case_position(use_case_name):
"""Resolve a use case's 1-based position; the manifest grows over time."""
use_cases = list(_get_manifest().keys())
for i, name in enumerate(use_cases, 1):
if name == use_case_name:
return i
raise ValueError(f"Use case '{use_case_name}' not found. Available: {use_cases}")


@skipIf(SKIP_SCHEMA_TESTS, "Skip schema test")
@pytest.mark.xdist_group(name="sam_init")
class TestBasicInitWithEventBridgeCommand(SchemaTestDataSetup):
@pytest.mark.timeout(300)
def test_init_interactive_with_event_bridge_app_aws_registry(self):
# WHEN the user follows interactive init prompts
# 1: AWS Quick Start Templates
# 8: Infrastructure event management - Use case
# 4: Java Runtime
# {use_case_pos}: Infrastructure event management - Use case (dynamic position)
# {java_runtime_pos}: JAVA_RUNTIME_FOR_INIT (dynamic position)
# 2: Maven
# 2: select event-bridge app from scratch
# N: disable adding xray tracing
# N: disable cloudwatch insights
# N: disable structured logging
# eb-app-maven: response to name
# Y: Use default aws configuration
# 1: select schema from cli_paginator
# {aws_registry_pos}: select aws.events as registries (dynamic position)
# 9: select schema AWSAPICallViaCloudTrail
aws_registry_pos = _get_registry_position("aws.events")
user_input = f"""
1
8
4
{_get_use_case_position(EVENT_BRIDGE_USE_CASE)}
{_get_runtime_position(JAVA_RUNTIME_FOR_INIT)}
2
2
N
N
N
eb-app-maven
Y
1
{aws_registry_pos}
9
"""
Expand All @@ -85,8 +118,8 @@ def test_init_interactive_with_event_bridge_app_aws_registry(self):
def test_init_interactive_with_event_bridge_app_partner_registry(self):
# WHEN the user follows interactive init prompts
# 1: AWS Quick Start Templates
# 8: Infrastructure event management - Use case
# 4: Java Runtime
# {use_case_pos}: Infrastructure event management - Use case (dynamic position)
# {java_runtime_pos}: JAVA_RUNTIME_FOR_INIT (dynamic position)
# 2: Maven
# 2: select event-bridge app from scratch
# N: disable adding xray tracing
Expand All @@ -99,8 +132,8 @@ def test_init_interactive_with_event_bridge_app_partner_registry(self):
partner_registry_pos = _get_registry_position("partner-registry")
user_input = f"""
1
8
4
{_get_use_case_position(EVENT_BRIDGE_USE_CASE)}
{_get_runtime_position(JAVA_RUNTIME_FOR_INIT)}
2
2
N
Expand Down Expand Up @@ -139,8 +172,8 @@ def test_init_interactive_with_event_bridge_app_partner_registry(self):
def test_init_interactive_with_event_bridge_app_pagination(self):
# WHEN the user follows interactive init prompts
# 1: AWS Quick Start Templates
# 8: Infrastructure event management - Use case
# 4: Java Runtime
# {use_case_pos}: Infrastructure event management - Use case (dynamic position)
# {java_runtime_pos}: JAVA_RUNTIME_FOR_INIT (dynamic position)
# 2: Maven
# 2: select event-bridge app from scratch
# N: disable adding xray tracing
Expand All @@ -155,8 +188,8 @@ def test_init_interactive_with_event_bridge_app_pagination(self):
pagination_registry_pos = _get_registry_position("test-pagination")
user_input = f"""
1
8
4
{_get_use_case_position(EVENT_BRIDGE_USE_CASE)}
{_get_runtime_position(JAVA_RUNTIME_FOR_INIT)}
2
2
N
Expand Down Expand Up @@ -186,8 +219,8 @@ def test_init_interactive_with_event_bridge_app_pagination(self):
def test_init_interactive_with_event_bridge_app_customer_registry(self):
# WHEN the user follows interactive init prompts
# 1: AWS Quick Start Templates
# 8: Infrastructure event management - Use case
# 4: Java Runtime
# {use_case_pos}: Infrastructure event management - Use case (dynamic position)
# {java_runtime_pos}: JAVA_RUNTIME_FOR_INIT (dynamic position)
# 2: Maven
# 2: select event-bridge app from scratch
# N: disable adding xray tracing
Expand All @@ -200,8 +233,8 @@ def test_init_interactive_with_event_bridge_app_customer_registry(self):
other_schema_pos = _get_registry_position("other-schema")
user_input = f"""
1
8
4
{_get_use_case_position(EVENT_BRIDGE_USE_CASE)}
{_get_runtime_position(JAVA_RUNTIME_FOR_INIT)}
2
2
N
Expand Down Expand Up @@ -240,29 +273,27 @@ def test_init_interactive_with_event_bridge_app_customer_registry(self):
def test_init_interactive_with_event_bridge_app_aws_schemas_python(self):
# WHEN the user follows interactive init prompts
# 1: AWS Quick Start Templates
# 8: Infrastructure event management - Use case
# 8: Python 3.9
# {use_case_pos}: Infrastructure event management - Use case (dynamic position)
# {python_runtime_pos}: PYTHON_RUNTIME_FOR_INIT (dynamic position)
# 2: select event-bridge app from scratch
# N: disable adding xray tracing
# N: disable cloudwatch insights
# N: disable structured logging
# eb-app-python39: response to name
# Y: Use default aws configuration
# 1: select schema from cli_paginator
# {aws_registry_pos}: select aws.events as registries (dynamic position)
# 1: select aws schema
aws_registry_pos = _get_registry_position("aws.events")
user_input = f"""
1
8
8
{_get_use_case_position(EVENT_BRIDGE_USE_CASE)}
{_get_runtime_position(PYTHON_RUNTIME_FOR_INIT)}
Comment thread
roger-zhangg marked this conversation as resolved.
2
N
N
N
eb-app-python39
Y
1
{aws_registry_pos}
1
"""
Expand All @@ -280,28 +311,28 @@ def test_init_interactive_with_event_bridge_app_aws_schemas_python(self):
def test_init_interactive_with_event_bridge_app_aws_schemas_go(self):
# WHEN the user follows interactive init prompts
# 1: AWS Quick Start Templates
# 8: Infrastructure event management - Use case
# 1: Go 1.x
# {use_case_pos}: Infrastructure event management - Use case (dynamic position)
# {go_runtime_pos}: GO_RUNTIME_FOR_INIT (dynamic position)
# 2: select event-bridge app from scratch
# N: disable adding xray tracing
# N: disable cloudwatch insights
# N: disable structured logging
# eb-app-go: response to name
# Y: Use default aws configuration
# 4: select aws.events as registries
# {aws_registry_pos}: select aws.events as registries (dynamic position)
# 1: select aws schema

user_input = """
1
8
aws_registry_pos = _get_registry_position("aws.events")
user_input = f"""
1
{_get_use_case_position(EVENT_BRIDGE_USE_CASE)}
{_get_runtime_position(GO_RUNTIME_FOR_INIT)}
2
N
N
N
eb-app-go
Y
1
{aws_registry_pos}
1
"""
with tempfile.TemporaryDirectory() as temp:
Expand All @@ -319,8 +350,8 @@ def test_init_interactive_with_event_bridge_app_non_default_profile_selection(se
self._init_custom_config("mynewprofile", "us-west-2")
# WHEN the user follows interactive init prompts
# 1: AWS Quick Start Templates
# 8: Infrastructure event management - Use case
# 8: Python 3.9
# {use_case_pos}: Infrastructure event management - Use case (dynamic position)
# {python_runtime_pos}: PYTHON_RUNTIME_FOR_INIT (dynamic position)
# 2: select event-bridge app from scratch
# N: disable adding xray tracing
# N: disable cloudwatch insights
Expand All @@ -331,17 +362,17 @@ def test_init_interactive_with_event_bridge_app_non_default_profile_selection(se
# schemas aws region us-east-1
# 1: select aws.events as registries
# 1: select aws schema
# Registry hardcoded: non-default profile/region resolves a different account.

user_input = """
user_input = f"""
1
8
8
{_get_use_case_position(EVENT_BRIDGE_USE_CASE)}
{_get_runtime_position(PYTHON_RUNTIME_FOR_INIT)}
2
N
N
N
eb-app-python39
3
N
2
us-east-1
Expand All @@ -363,8 +394,8 @@ def test_init_interactive_with_event_bridge_app_non_supported_schemas_region(sel
self._init_custom_config("default", "cn-north-1")
# WHEN the user follows interactive init prompts
# 1: AWS Quick Start Templates
# 8: Infrastructure event management - Use case
# 7: Python 3.9
# {use_case_pos}: Infrastructure event management - Use case (dynamic position)
# {python_runtime_pos}: PYTHON_RUNTIME_FOR_INIT (dynamic position)
# 2: select event-bridge app from scratch
# N: disable adding xray tracing
# N: disable cloudwatch insights
Expand All @@ -374,10 +405,10 @@ def test_init_interactive_with_event_bridge_app_non_supported_schemas_region(sel
# 1: select aws.events as registries
# 1: select aws schema

user_input = """
user_input = f"""
1
8
8
{_get_use_case_position(EVENT_BRIDGE_USE_CASE)}
{_get_runtime_position(PYTHON_RUNTIME_FOR_INIT)}
2
N
N
Expand Down
Loading