diff --git a/src/devbox/cli.py b/src/devbox/cli.py index e54fe27..bd3b57b 100644 --- a/src/devbox/cli.py +++ b/src/devbox/cli.py @@ -7,6 +7,7 @@ import click from typing import Optional +from .remote_client import normalize_param_prefix from .commands.status import run_status_command from .devbox_manager import DevBoxManager from .console_output import ConsoleOutput @@ -15,6 +16,16 @@ PARAM_PREFIX_ENV_VAR = "DEVBOX_PARAM_PREFIX" +def _validate_param_prefix( + _ctx: click.Context, _param: click.Parameter, value: str +) -> str: + """Normalize and validate shared ``--param-prefix`` values.""" + try: + return normalize_param_prefix(value) + except ValueError as exc: + raise click.BadParameter(str(exc)) from exc + + def param_prefix_option(func): """Attach the shared ``--param-prefix`` Click option. @@ -32,6 +43,7 @@ def param_prefix_option(func): "--param-prefix", default=DEFAULT_PARAM_PREFIX, envvar=PARAM_PREFIX_ENV_VAR, + callback=_validate_param_prefix, show_default=True, show_envvar=True, help="SSM parameter prefix", @@ -188,11 +200,20 @@ def launch( @cli.command() -@click.argument("project") -@click.option("--base-ami", required=True, help="Base AMI ID for the project") +@click.argument('project') +@click.option('--base-ami', required=True, help='Base AMI ID for the project') +@click.option('--instance-type', help='Default EC2 instance type for future launches') +@click.option('--key-pair', help='Default SSH key pair name for future launches') @param_prefix_option @click.pass_context -def new(ctx, project: str, base_ami: str, param_prefix: str): +def new( + ctx, + project: str, + base_ami: str, + instance_type: Optional[str], + key_pair: Optional[str], + param_prefix: str, +): """Create a new DevBox project without launching an instance. PROJECT is the name of the project to create. @@ -203,7 +224,11 @@ def new(ctx, project: str, base_ami: str, param_prefix: str): try: new_project_programmatic( - project=project, base_ami=base_ami, param_prefix=param_prefix + project=project, + base_ami=base_ami, + instance_type=instance_type, + key_pair=key_pair, + param_prefix=param_prefix ) except Exception as e: console.print_error(f"Failed to create project: {str(e)}") diff --git a/src/devbox/devbox_manager.py b/src/devbox/devbox_manager.py index c09273d..42774a2 100644 --- a/src/devbox/devbox_manager.py +++ b/src/devbox/devbox_manager.py @@ -296,12 +296,9 @@ def delete_ami_and_snapshots(self, ami_id: str) -> Dict[str, Any]: raise ValueError("No AMI ID provided.") try: - response = self.ec2.describe_images(ImageIds=[ami_id]) - images = response.get("Images", []) - if not images: + image = utils.get_image(ami_id, ec2_client=self.ec2) + if image is None: raise utils.ResourceNotFoundError(f"AMI {ami_id} not found.") - - image = images[0] snapshot_ids = [] for mapping in image.get("BlockDeviceMappings", []): ebs = mapping.get("Ebs", {}) diff --git a/src/devbox/launch.py b/src/devbox/launch.py index ff0f78c..fe13c9d 100644 --- a/src/devbox/launch.py +++ b/src/devbox/launch.py @@ -167,7 +167,7 @@ def get_volume_info( ValueError: If the image is not found or invalid """ try: - resp = ec2.describe_images(ImageIds=[image_id]) + image = utils.get_image(image_id, ec2_client=ec2) except ClientError as e: raise AWSClientError( f"Error fetching image details for {image_id}", @@ -175,10 +175,8 @@ def get_volume_info( original_exception=e, ) - if not resp.get("Images"): + if image is None: raise ValueError(f"AMI {image_id} not found") - - image = resp["Images"][0] volumes = image.get("BlockDeviceMappings", []).copy() # Find the largest volume @@ -860,9 +858,8 @@ def display_instance_info(ec2: Any, instance_id: str, project: str, table: Any) ami_id = resp["Item"].get("AMI", "") if ami_id: try: - ami_resp = ec2.describe_images(ImageIds=[ami_id]) - if ami_resp.get("Images"): - ami = ami_resp["Images"][0] + ami = utils.get_image(ami_id, ec2_client=ec2) + if ami is not None: ami_name = ami.get("Name", "") ami_description = ami.get("Description", "") determined_username = determine_ssh_username( @@ -880,6 +877,8 @@ def display_instance_info(ec2: Any, instance_id: str, project: str, table: Any) username = ( "" # placeholder for unknown ) + else: + username = "" except Exception: username = "" else: @@ -1054,9 +1053,8 @@ def launch_programmatic( existing_username = resp["Item"].get("Username", "") if not existing_username: # Determine username from AMI - ami_resp = aws["ec2"].describe_images(ImageIds=[image_id]) - if ami_resp.get("Images"): - ami = ami_resp["Images"][0] + ami = utils.get_image(image_id, ec2_client=aws["ec2"]) + if ami is not None: ami_name = ami.get("Name", "") ami_description = ami.get("Description", "") determined_username = determine_ssh_username( @@ -1069,6 +1067,10 @@ def launch_programmatic( UpdateExpression="SET Username = :u", ExpressionAttributeValues={":u": determined_username}, ) + else: + print( + f"Warning: Could not determine SSH username: AMI {image_id} not found" + ) except Exception as e: print(f"Warning: Could not determine SSH username: {e}") diff --git a/src/devbox/lifecycle/snapshots.py b/src/devbox/lifecycle/snapshots.py index b32657f..6b3b6a5 100644 --- a/src/devbox/lifecycle/snapshots.py +++ b/src/devbox/lifecycle/snapshots.py @@ -9,7 +9,7 @@ from boto3.dynamodb.conditions import Attr, Key from botocore.exceptions import ClientError -from devbox.utils import get_project_tag +from devbox.utils import get_image, get_project_tag logger = logging.getLogger(__name__) @@ -177,23 +177,12 @@ def cleanup_ami_and_snapshots( logger.info("waiting for ami to vanish ami_id=%s", ami_id) for _ in range(config.cleanup_max_attempts): time.sleep(config.cleanup_wait_seconds) - try: - resp = ec2_client.describe_images(ImageIds=[ami_id]) - except ClientError as exc: - code = exc.response.get("Error", {}).get("Code", "") - if code.startswith("InvalidAMIID"): - logger.info( - "ami no longer exists ami_id=%s error_code=%s", ami_id, code - ) - break - raise - images = resp.get("Images", []) - - if not images: + image = get_image(ami_id, ec2_client=ec2_client) + if image is None: logger.info("ami no longer exists ami_id=%s", ami_id) break logger.info( - "ami still present ami_id=%s image_count=%s", ami_id, len(images) + "ami still present ami_id=%s image_count=%s", ami_id, 1 ) else: raise RuntimeError(f"Timed out waiting for AMI '{ami_id}' to deregister") @@ -306,19 +295,12 @@ def make_mapping(item: Dict[str, Any]) -> Dict[str, Any]: old_ami = main_item.get("AMI") if old_ami: - desc = [] - old_ami_missing = False - try: - desc = ec2_client.describe_images(ImageIds=[old_ami])["Images"] - except ClientError as exc: - code = exc.response.get("Error", {}).get("Code", "") - if code.startswith("InvalidAMIID"): - logger.warning("old ami not found ami_id=%s", old_ami) - old_ami_missing = True - else: - raise - if desc: - image = desc[0] + image = get_image(old_ami, ec2_client=ec2_client) + old_ami_missing = image is None + if old_ami_missing: + logger.warning("old ami not found ami_id=%s", old_ami) + + if image is not None: if not virtualization_type: virtualization_type = image.get("VirtualizationType") if not architecture: @@ -338,8 +320,6 @@ def make_mapping(item: Dict[str, Any]) -> Dict[str, Any]: old_ami, project, ) - elif not old_ami_missing: - logger.warning("old ami not found ami_id=%s", old_ami) register_image_args: Dict[str, Any] = { "Name": f"{project}-ami", diff --git a/src/devbox/new.py b/src/devbox/new.py new file mode 100644 index 0000000..ad2a547 --- /dev/null +++ b/src/devbox/new.py @@ -0,0 +1,334 @@ +#!/usr/bin/env python3 +"""Create new DevBox projects without launching instances. + +This module creates project metadata in DynamoDB without starting EC2 +instances. It supports both the main CLI integration and a standalone +``python -m devbox.new`` entrypoint used for debugging. +""" +from __future__ import annotations + +import sys +from typing import Dict, Any, Optional, TYPE_CHECKING + +from botocore.exceptions import ClientError + +from .remote_client import normalize_param_prefix +from . import utils +from .utils import ResourceNotFoundError, AWSClientError + +if TYPE_CHECKING: + from mypy_boto3_ec2.client import EC2Client + from mypy_boto3_dynamodb.service_resource import Table as DynamoDBTable + from mypy_boto3_ssm.client import SSMClient + +# Type aliases for runtime use +SSMClient = Any +DynamoDBTable = Any +EC2Client = Any + + +def initialize_aws_clients() -> Dict[str, Any]: + """Initialize AWS clients needed for project creation. + + Returns + ------- + dict[str, Any] + Mapping of AWS client and resource handles used during project + creation. + + Raises + ------ + AWSClientError + Raised when any required client or resource cannot be initialized. + """ + try: + return { + "ssm": utils.get_ssm_client(), + "ec2": utils.get_ec2_client(), + "ddb": utils.get_dynamodb_resource() + } + except Exception as e: + raise AWSClientError(f"Failed to initialize AWS clients: {str(e)}") from e + + +def validate_ami_exists(ec2_client: Any, ami_id: str) -> Dict[str, Any]: + """Validate that the specified AMI exists and return its metadata. + + Parameters + ---------- + ec2_client : Any + Boto3 EC2 client instance. + ami_id : str + AMI identifier to validate. Expected to start with ``"ami-"``. + + Returns + ------- + dict[str, Any] + AMI metadata from ``describe_images`` for the requested image. + + Raises + ------ + ResourceNotFoundError + Raised when the AMI does not exist or is not accessible. + AWSClientError + Raised for other AWS API failures, such as permission errors. + """ + try: + image = utils.get_image(ami_id, ec2_client=ec2_client) + if image is None: + raise ResourceNotFoundError(f"AMI {ami_id} not found") + return image + except ClientError as e: + raise AWSClientError(f"Error validating AMI {ami_id}: {str(e)}") from e + + +def check_project_exists(table: Any, project_name: str) -> Optional[Dict[str, Any]]: + """Check whether a project already exists in DynamoDB. + + Parameters + ---------- + table : Any + DynamoDB table resource object. + project_name : str + Project name to query. + + Returns + ------- + dict[str, Any] or None + Existing project item when present; otherwise ``None``. + + Raises + ------ + AWSClientError + Raised for DynamoDB access errors other than a missing resource. + """ + try: + response = table.get_item(Key={"project": project_name}) + return response.get("Item") + except ClientError as e: + error_code = e.response.get("Error", {}).get("Code") + if error_code == "ResourceNotFoundException": + return None + raise AWSClientError(f"Error checking project existence: {str(e)}") from e + + +def create_project_entry( + table: Any, + project_name: str, + ami_info: Dict[str, Any], + instance_type: Optional[str] = None, + key_pair: Optional[str] = None, +) -> None: + """Create a new project entry in DynamoDB. + + Parameters + ---------- + table : Any + DynamoDB table resource object. + project_name : str + Name of the new project. This becomes the table partition key. + ami_info : dict[str, Any] + AMI metadata returned from ``describe_images``. + instance_type : str, optional + Default EC2 instance type for future launches. + key_pair : str, optional + Default SSH key pair for future launches. + + Raises + ------ + ValueError + Raised when a project with the same name already exists. + AWSClientError + Raised when the ``put_item`` request fails. + """ + try: + item = { + "project": project_name, + "Status": "READY", + "AMI": ami_info["ImageId"], + "VirtualizationType": ami_info.get("VirtualizationType", "hvm"), + "Architecture": ami_info.get("Architecture", "x86_64"), + "RootDeviceName": ami_info.get("RootDeviceName", "/dev/sda1"), + "LastUpdated": str(utils.get_utc_now()), + "State": "ready" + } + + # Add optional AMI metadata + if "Name" in ami_info: + item["AMIName"] = ami_info["Name"] + if "Description" in ami_info: + item["AMIDescription"] = ami_info["Description"] + if "CreationDate" in ami_info: + item["AMICreationDate"] = ami_info["CreationDate"] + if instance_type: + item["LastInstanceType"] = instance_type + if key_pair: + item["LastKeyPair"] = key_pair + + table.put_item( + Item=item, + ConditionExpression="attribute_not_exists(#project)", + ExpressionAttributeNames={"#project": "project"}, + ) + except ClientError as e: + error_code = e.response.get("Error", {}).get("Code") + if error_code == "ConditionalCheckFailedException": + raise ValueError(f"Project '{project_name}' already exists") from e + raise AWSClientError(f"Failed to create project entry: {str(e)}") from e + + +def new_project_programmatic( + project: str, + base_ami: str, + instance_type: Optional[str] = None, + key_pair: Optional[str] = None, + param_prefix: str = "/devbox" +) -> None: + """Create a new DevBox project without launching an instance. + + Parameters + ---------- + project : str + Project name. Must satisfy the command's naming constraints. + base_ami : str + Base AMI identifier for the project. + instance_type : str, optional + Default EC2 instance type for future launches. + key_pair : str, optional + Default SSH key pair name for future launches. + param_prefix : str, default="/devbox" + Prefix for AWS Systems Manager Parameter Store keys. + + Raises + ------ + ValueError + Raised when the project name or AMI identifier is invalid, or when + the project already exists. + ResourceNotFoundError + Raised when required AWS resources cannot be found. + AWSClientError + Raised for AWS API failures. + """ + # Validate project name + if not project: + raise ValueError("Project name cannot be empty") + + if not project.replace('-', '').isalnum(): + raise ValueError("Project name must be alphanumeric with optional hyphens") + + if len(project) < 1 or len(project) > 50: + raise ValueError("Project name must be between 1 and 50 characters") + + if project.startswith('-') or project.endswith('-'): + raise ValueError("Project name cannot start or end with hyphens") + + if '--' in project: + raise ValueError("Project name cannot contain consecutive hyphens") + + # Validate base AMI format + if not base_ami: + raise ValueError("Base AMI cannot be empty") + + if not base_ami.startswith('ami-'): + raise ValueError("Base AMI must be a valid AMI ID starting with 'ami-'") + + if len(base_ami) < 12: # ami- + at least 8 characters + raise ValueError("Base AMI ID appears to be too short to be valid") + + param_prefix = normalize_param_prefix(param_prefix) + + print(f"Creating new project: {project}") + print(f"Base AMI: {base_ami}") + if instance_type: + print(f"Default instance type: {instance_type}") + if key_pair: + print(f"Default key pair: {key_pair}") + print(f"Parameter prefix: {param_prefix}") + + # Initialize AWS clients + print("Initializing AWS clients...") + aws = initialize_aws_clients() + + # Validate AMI exists and get its details + print("Validating AMI...") + ami_info = validate_ami_exists(aws["ec2"], base_ami) + print(f"AMI validated: {ami_info.get('Name', 'Unknown Name')} ({ami_info.get('Architecture', 'unknown arch')})") + + # Get DynamoDB table + print("Getting DynamoDB table...") + try: + table_name = utils.get_ssm_parameter( + f"{param_prefix}/snapshotTable", ssm_client=aws["ssm"] + ) + table = utils.get_dynamodb_table(table_name, dynamodb_resource=aws["ddb"]) + except Exception as e: + raise AWSClientError(f"Failed to get DynamoDB table: {str(e)}") from e + + # Check if project already exists + print("Checking if project already exists...") + existing_project = check_project_exists(table, project) + if existing_project: + raise ValueError(f"Project '{project}' already exists with status: {existing_project.get('Status', 'unknown')}") + + # Create project entry + print("Creating project entry...") + create_project_entry( + table=table, + project_name=project, + ami_info=ami_info, + instance_type=instance_type, + key_pair=key_pair, + ) + + print(f"✅ Project '{project}' created successfully!") + print(" Status: READY") + print(f" Base AMI: {base_ami}") + print(" You can now launch instances for this project using the 'launch' command.") + + +def main(): + """Run the standalone ``python -m devbox.new`` entrypoint.""" + import argparse + + try: + parser = argparse.ArgumentParser(description="Create a new devbox project") + parser.add_argument("project", help="Project name") + parser.add_argument("--base-ami", required=True, help="Base AMI ID") + parser.add_argument( + "--instance-type", + help="Default EC2 instance type for future launches", + ) + parser.add_argument( + "--key-pair", + help="Default SSH key pair name for future launches", + ) + parser.add_argument("--param-prefix", default="/devbox", help="SSM parameter prefix") + + args = parser.parse_args() + + new_project_programmatic( + project=args.project, + base_ami=args.base_ami, + instance_type=args.instance_type, + key_pair=args.key_pair, + param_prefix=args.param_prefix + ) + + except KeyboardInterrupt: + print("\nOperation cancelled by user") + sys.exit(1) + except (ValueError, ResourceNotFoundError) as e: + print(f"Error: {str(e)}", file=sys.stderr) + sys.exit(2) + except AWSClientError as e: + print(f"AWS Error: {str(e)}", file=sys.stderr) + if hasattr(e, 'error_code'): + print(f"Error Code: {e.error_code}", file=sys.stderr) + sys.exit(3) + except Exception as e: + print(f"Unexpected error: {str(e)}", file=sys.stderr) + sys.exit(4) + + +if __name__ == "__main__": + main() diff --git a/src/devbox/remote_client.py b/src/devbox/remote_client.py index e5cfd3f..df6df39 100644 --- a/src/devbox/remote_client.py +++ b/src/devbox/remote_client.py @@ -42,10 +42,18 @@ def normalize_param_prefix(param_prefix: str) -> str: str Normalized prefix in ``/name`` form. Empty input falls back to ``/devbox``. + + Raises + ------ + ValueError + Raised when the normalized prefix would contain empty path + segments, such as consecutive slashes. """ stripped = param_prefix.strip("/") if not stripped: return "/devbox" + if "//" in stripped: + raise ValueError("Parameter prefix cannot contain consecutive slashes") return f"/{stripped}" diff --git a/src/devbox/utils.py b/src/devbox/utils.py index cf37790..459af6c 100644 --- a/src/devbox/utils.py +++ b/src/devbox/utils.py @@ -36,25 +36,33 @@ def get_dynamodb_resource() -> ServiceResource: return boto3.resource('dynamodb') -def get_dynamodb_table(table_name: str) -> Any: +def get_dynamodb_table( + table_name: str, dynamodb_resource: Optional[ServiceResource] = None +) -> Any: """Get a DynamoDB table resource. Args: table_name: Name of the DynamoDB table + dynamodb_resource: Optional pre-configured DynamoDB resource Returns: A DynamoDB Table resource """ - dynamodb = get_dynamodb_resource() + dynamodb = dynamodb_resource or get_dynamodb_resource() return dynamodb.Table(table_name) -def get_ssm_parameter(parameter_name: str, required: bool = True) -> str: +def get_ssm_parameter( + parameter_name: str, + required: bool = True, + ssm_client: Optional[BaseClient] = None, +) -> str: """Get a parameter from SSM Parameter Store. Args: parameter_name: Name of the parameter to fetch required: If True, raises an exception if parameter is not found + ssm_client: Optional pre-configured SSM client Returns: The parameter value as a string @@ -63,7 +71,7 @@ def get_ssm_parameter(parameter_name: str, required: bool = True) -> str: ValueError: If parameter is not found and required is True """ try: - ssm = get_ssm_client() + ssm = ssm_client or get_ssm_client() response = ssm.get_parameter(Name=parameter_name, WithDecryption=True) return response['Parameter']['Value'] except (ClientError, KeyError) as e: @@ -72,6 +80,38 @@ def get_ssm_parameter(parameter_name: str, required: bool = True) -> str: return "" +def get_image( + image_id: str, ec2_client: Optional[BaseClient] = None +) -> Optional[Dict[str, Any]]: + """Look up a single EC2 image and normalize missing-image cases. + + Args: + image_id: AMI ID to fetch + ec2_client: Optional pre-configured EC2 client + + Returns: + The first matching image dictionary, or None if the image does not exist + + Raises: + ClientError: If AWS returns an error other than an invalid or missing AMI + """ + ec2 = ec2_client or get_ec2_client() + + try: + response = ec2.describe_images(ImageIds=[image_id]) + except ClientError as e: + error_code = e.response.get("Error", {}).get("Code", "") + if error_code.startswith("InvalidAMIID"): + return None + raise + + images = response.get("Images", []) + if not images: + return None + + return images[0] + + def get_project_tag(tags: List[Dict[str, str]]) -> str: """Extract the Project tag value from a list of tags. diff --git a/tests/test_cli.py b/tests/test_cli.py index 29da50f..0611acd 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -4,7 +4,7 @@ from unittest.mock import MagicMock, patch, call from click.testing import CliRunner -from devbox.cli import cli, status, terminate, launch, delete_project, main +from devbox.cli import cli, status, terminate, launch, delete_project, new as new_command, main from devbox.utils import AWSClientError @@ -15,6 +15,7 @@ def test_cli_help(): assert result.exit_code == 0 assert "DevBox - AWS EC2 Development Environment Manager" in result.output assert "launch" in result.output + assert "new" in result.output assert "status" in result.output assert "terminate" in result.output @@ -782,6 +783,165 @@ def test_delete_project_with_param_prefix_option( mock_manager.delete_project_entry.assert_called_once_with("demo") +class TestNewCommand: + def setup_method(self): + self.runner = CliRunner() + + def test_new_help(self): + result = self.runner.invoke(new_command, ["--help"]) + + assert result.exit_code == 0 + assert "Create a new DevBox project without launching an instance" in result.output + assert "PROJECT" in result.output + assert "--base-ami" in result.output + assert "--instance-type" in result.output + assert "--key-pair" in result.output + assert "--param-prefix" in result.output + + @patch("devbox.new.new_project_programmatic") + @patch("devbox.cli.ConsoleOutput") + def test_new_success(self, mock_console_class, mock_new): + mock_console = MagicMock() + mock_console_class.return_value = mock_console + + result = self.runner.invoke( + cli, ["new", "test-project", "--base-ami", "ami-12345678"] + ) + + assert result.exit_code == 0 + mock_new.assert_called_once_with( + project="test-project", + base_ami="ami-12345678", + instance_type=None, + key_pair=None, + param_prefix="/devbox", + ) + + @patch("devbox.new.new_project_programmatic") + @patch("devbox.cli.ConsoleOutput") + def test_new_with_custom_param_prefix(self, mock_console_class, mock_new): + mock_console = MagicMock() + mock_console_class.return_value = mock_console + + result = self.runner.invoke( + cli, + [ + "new", + "my-project", + "--base-ami", + "ami-0abcdef1234567890", + "--param-prefix", + "/my-devbox", + ], + ) + + assert result.exit_code == 0 + mock_new.assert_called_once_with( + project="my-project", + base_ami="ami-0abcdef1234567890", + instance_type=None, + key_pair=None, + param_prefix="/my-devbox", + ) + + @patch("devbox.new.new_project_programmatic") + @patch("devbox.cli.ConsoleOutput") + def test_new_with_instance_type_and_key_pair(self, mock_console_class, mock_new): + mock_console = MagicMock() + mock_console_class.return_value = mock_console + + result = self.runner.invoke( + cli, + [ + "new", + "my-project", + "--base-ami", + "ami-0abcdef1234567890", + "--instance-type", + "m5.large", + "--key-pair", + "my-key", + ], + ) + + assert result.exit_code == 0 + mock_new.assert_called_once_with( + project="my-project", + base_ami="ami-0abcdef1234567890", + instance_type="m5.large", + key_pair="my-key", + param_prefix="/devbox", + ) + + @patch("devbox.new.new_project_programmatic") + @patch("devbox.cli.ConsoleOutput") + def test_new_exception(self, mock_console_class, mock_new): + mock_console = MagicMock() + mock_console_class.return_value = mock_console + mock_new.side_effect = Exception("Project creation failed") + + result = self.runner.invoke( + cli, ["new", "test-project", "--base-ami", "ami-12345678"] + ) + + assert result.exit_code == 1 + mock_console.print_error.assert_called_once() + error_call = mock_console.print_error.call_args[0][0] + assert "Failed to create project" in error_call + + def test_new_missing_base_ami(self): + result = self.runner.invoke(cli, ["new", "test-project"]) + + assert result.exit_code == 2 + assert "Missing option '--base-ami'" in result.output + + def test_new_missing_project(self): + result = self.runner.invoke(cli, ["new", "--base-ami", "ami-12345678"]) + + assert result.exit_code == 2 + assert "Missing argument 'PROJECT'" in result.output + + @patch("devbox.new.new_project_programmatic") + @patch("devbox.cli.ConsoleOutput") + def test_new_normalizes_slashless_param_prefix( + self, mock_console_class, mock_new + ): + mock_console_class.return_value = MagicMock() + + result = self.runner.invoke( + cli, + ["new", "test-project", "--base-ami", "ami-12345678", "--param-prefix", "devbox"], + ) + + assert result.exit_code == 0 + mock_new.assert_called_once_with( + project="test-project", + base_ami="ami-12345678", + instance_type=None, + key_pair=None, + param_prefix="/devbox", + ) + + @patch("devbox.new.new_project_programmatic") + def test_new_rejects_param_prefix_with_consecutive_slashes(self, mock_new): + result = self.runner.invoke( + cli, + [ + "new", + "test-project", + "--base-ami", + "ami-12345678", + "--param-prefix", + "/devbox//nested", + ], + ) + + assert result.exit_code == 2 + assert "Invalid value for '--param-prefix'" in result.output + assert "cannot contain consecutive slashes" in result.output + mock_new.assert_not_called() + + @patch("devbox.cli.cli") def test_main_calls_cli(mock_cli): """Test main function calls CLI.""" @@ -1099,3 +1259,24 @@ def test_launch_uses_param_prefix_from_env(self, mock_console_class, mock_launch assign_dns=True, dns_subdomain=None, ) + + @patch("devbox.new.new_project_programmatic") + @patch("devbox.cli.ConsoleOutput") + def test_new_uses_param_prefix_from_env(self, mock_console_class, mock_new): + mock_console = MagicMock() + mock_console_class.return_value = mock_console + + result = self.runner.invoke( + cli, + ["new", "test-project", "--base-ami", "ami-12345678"], + env={"DEVBOX_PARAM_PREFIX": "/env/devbox"}, + ) + + assert result.exit_code == 0 + mock_new.assert_called_once_with( + project="test-project", + base_ami="ami-12345678", + instance_type=None, + key_pair=None, + param_prefix="/env/devbox", + ) diff --git a/tests/test_launch.py b/tests/test_launch.py index e09fba8..c800439 100644 --- a/tests/test_launch.py +++ b/tests/test_launch.py @@ -1155,8 +1155,112 @@ def test_display_instance_info_uses_existing_username(): mock_table.update_item.assert_not_called() -def test_launch_programmatic_determines_username(): +@patch("devbox.launch.utils.get_image", return_value=None) +def test_display_instance_info_uses_placeholder_when_ami_metadata_missing( + mock_get_image, capsys +): + """Test that display_instance_info falls back to a placeholder username.""" + mock_ec2_client = MagicMock() + mock_table = MagicMock() + + mock_ec2_client.describe_instances.return_value = { + "Reservations": [ + { + "Instances": [ + { + "InstanceId": "i-12345", + "State": {"Name": "running"}, + "PublicIpAddress": "1.2.3.4", + "InstanceType": "t3.medium", + } + ] + } + ] + } + mock_table.get_item.return_value = {"Item": {"Username": "", "AMI": "ami-missing"}} + + display_instance_info(mock_ec2_client, "i-12345", "test-project", mock_table) + + output = capsys.readouterr().out + assert "@1.2.3.4" in output + mock_table.update_item.assert_not_called() + mock_get_image.assert_called_once_with("ami-missing", ec2_client=mock_ec2_client) + + +@patch("devbox.launch.initialize_aws_clients") +@patch("devbox.launch.get_launch_config") +@patch("devbox.launch.validate_project_status") +@patch("devbox.launch.determine_ami") +@patch("devbox.launch.get_volume_info") +@patch("devbox.launch.get_launch_template_info") +@patch("devbox.launch.launch_instance_in_azs") +@patch("devbox.launch.update_instance_status") +@patch("devbox.launch.display_instance_info") +def test_launch_programmatic_determines_username( + mock_display, + mock_update, + mock_launch_azs, + mock_get_lt_info, + mock_get_vol_info, + mock_determine_ami, + mock_validate, + mock_get_config, + mock_init_aws, +): """Test that launch_programmatic determines and stores SSH username when not set.""" + mock_aws = {"ec2": MagicMock(), "ec2_resource": MagicMock()} + mock_init_aws.return_value = mock_aws + + mock_table = MagicMock() + mock_config = { + "lt_ids": ["lt-12345"], + "table": mock_table, + "item": { + "Status": "READY", + "LastInstanceType": "t3.medium", + "LastKeyPair": "test-key", + }, + } + mock_get_config.return_value = mock_config + mock_validate.return_value = "READY" + mock_determine_ami.return_value = "ami-12345" + mock_get_vol_info.return_value = ([], 0) + mock_get_lt_info.return_value = {"lt-12345": {"name": "us-east-1a"}} + + mock_instance = MagicMock() + mock_instance.meta.data = {"State": {"Name": "running"}} + mock_launch_azs.return_value = ( + mock_instance, + "i-12345", + {"State": {"Name": "running"}}, + ) + + # Mock DynamoDB response with no username + mock_table.get_item.return_value = {"Item": {"Username": ""}} + + # Mock AMI response for username determination + mock_aws["ec2"].describe_images.return_value = { + "Images": [{"Name": "amzn2-ami-hvm", "Description": "Amazon Linux 2"}] + } + + launch_programmatic("test-project", assign_dns=False) + + # Verify that username was determined and stored + mock_table.update_item.assert_called() + update_calls = mock_table.update_item.call_args_list + username_update = None + for call in update_calls: + if "Username" in call[1].get("UpdateExpression", ""): + username_update = call + break + + assert username_update is not None + assert "ec2-user" in str(username_update[1]["ExpressionAttributeValues"]) + + +@patch("devbox.launch.utils.get_image", return_value=None) +def test_launch_programmatic_warns_when_ami_metadata_missing(mock_get_image, capsys): + """Test that launch_programmatic warns when username metadata is unavailable.""" with patch("devbox.launch.initialize_aws_clients") as mock_init_aws, patch( "devbox.launch.get_launch_config" ) as mock_get_config, patch( @@ -1187,7 +1291,7 @@ def test_launch_programmatic_determines_username(): } mock_get_config.return_value = mock_config mock_validate.return_value = "READY" - mock_determine_ami.return_value = "ami-12345" + mock_determine_ami.return_value = "ami-missing" mock_get_vol_info.return_value = ([], 0) mock_get_lt_info.return_value = {"lt-12345": {"name": "us-east-1a"}} @@ -1199,27 +1303,14 @@ def test_launch_programmatic_determines_username(): {"State": {"Name": "running"}}, ) - # Mock DynamoDB response with no username mock_table.get_item.return_value = {"Item": {"Username": ""}} - # Mock AMI response for username determination - mock_aws["ec2"].describe_images.return_value = { - "Images": [{"Name": "amzn2-ami-hvm", "Description": "Amazon Linux 2"}] - } - launch_programmatic("test-project", assign_dns=False) - # Verify that username was determined and stored - mock_table.update_item.assert_called() - update_calls = mock_table.update_item.call_args_list - username_update = None - for call in update_calls: - if "Username" in call[1].get("UpdateExpression", ""): - username_update = call - break - - assert username_update is not None - assert "ec2-user" in str(username_update[1]["ExpressionAttributeValues"]) + output = capsys.readouterr().out + assert "Warning: Could not determine SSH username: AMI ami-missing not found" in output + mock_table.update_item.assert_not_called() + mock_get_image.assert_called_once_with("ami-missing", ec2_client=mock_aws["ec2"]) # Test functions for main diff --git a/tests/test_new.py b/tests/test_new.py new file mode 100644 index 0000000..7000d6a --- /dev/null +++ b/tests/test_new.py @@ -0,0 +1,503 @@ +"""Unit tests for devbox.new module.""" + +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch + +import pytest +from botocore.exceptions import ClientError + +from devbox.new import ( + initialize_aws_clients, + validate_ami_exists, + check_project_exists, + create_project_entry, + new_project_programmatic, + main, +) +from devbox.utils import AWSClientError, ResourceNotFoundError + + +def _client_error(code: str, message: str, operation: str) -> ClientError: + return ClientError( + error_response={"Error": {"Code": code, "Message": message}}, + operation_name=operation, + ) + + +@patch("devbox.new.utils.get_dynamodb_resource") +@patch("devbox.new.utils.get_ec2_client") +@patch("devbox.new.utils.get_ssm_client") +def test_initialize_aws_clients_success(mock_ssm, mock_ec2, mock_ddb): + mock_ssm.return_value = MagicMock() + mock_ec2.return_value = MagicMock() + mock_ddb.return_value = MagicMock() + + aws = initialize_aws_clients() + + assert aws["ssm"] is mock_ssm.return_value + assert aws["ec2"] is mock_ec2.return_value + assert aws["ddb"] is mock_ddb.return_value + + +@patch("devbox.new.utils.get_ssm_client") +def test_initialize_aws_clients_error(mock_ssm): + mock_ssm.side_effect = Exception("boom") + + with pytest.raises(AWSClientError): + initialize_aws_clients() + + +def test_validate_ami_exists_success(): + mock_ec2 = MagicMock() + mock_ec2.describe_images.return_value = { + "Images": [{"ImageId": "ami-12345678", "Architecture": "x86_64"}] + } + + ami = validate_ami_exists(mock_ec2, "ami-12345678") + + assert ami["ImageId"] == "ami-12345678" + + +def test_validate_ami_exists_not_found_empty_images(): + mock_ec2 = MagicMock() + mock_ec2.describe_images.return_value = {"Images": []} + + with pytest.raises(ResourceNotFoundError): + validate_ami_exists(mock_ec2, "ami-12345678") + + +def test_validate_ami_exists_not_found_client_error(): + mock_ec2 = MagicMock() + mock_ec2.describe_images.side_effect = _client_error( + "InvalidAMIID.NotFound", "AMI not found", "DescribeImages" + ) + + with pytest.raises(ResourceNotFoundError): + validate_ami_exists(mock_ec2, "ami-12345678") + + +def test_validate_ami_exists_other_client_error(): + mock_ec2 = MagicMock() + mock_ec2.describe_images.side_effect = _client_error( + "UnauthorizedOperation", "Access denied", "DescribeImages" + ) + + with pytest.raises(AWSClientError): + validate_ami_exists(mock_ec2, "ami-12345678") + + +def test_check_project_exists_returns_item(): + mock_table = MagicMock() + mock_table.get_item.return_value = {"Item": {"project": "test", "Status": "READY"}} + + item = check_project_exists(mock_table, "test") + + assert item == {"project": "test", "Status": "READY"} + + +def test_check_project_exists_returns_none_when_missing(): + mock_table = MagicMock() + mock_table.get_item.return_value = {} + + item = check_project_exists(mock_table, "test") + + assert item is None + + +def test_check_project_exists_handles_resource_not_found(): + mock_table = MagicMock() + mock_table.get_item.side_effect = _client_error( + "ResourceNotFoundException", "No table", "GetItem" + ) + + item = check_project_exists(mock_table, "test") + + assert item is None + + +def test_check_project_exists_other_client_error(): + mock_table = MagicMock() + mock_table.get_item.side_effect = _client_error( + "AccessDeniedException", "Denied", "GetItem" + ) + + with pytest.raises(AWSClientError): + check_project_exists(mock_table, "test") + + +@patch("devbox.new.utils.get_utc_now") +def test_create_project_entry_success(mock_now): + fixed_time = datetime(2025, 1, 1, tzinfo=timezone.utc) + mock_now.return_value = fixed_time + mock_table = MagicMock() + ami_info = { + "ImageId": "ami-12345678", + "VirtualizationType": "hvm", + "Architecture": "x86_64", + "RootDeviceName": "/dev/xvda", + "Name": "Test AMI", + "Description": "Test Description", + "CreationDate": "2025-01-01T00:00:00.000Z", + } + + create_project_entry( + mock_table, + "my-project", + ami_info, + instance_type="m5.large", + key_pair="my-key", + ) + + put_item_call_args = mock_table.put_item.call_args[1] + put_item_args = put_item_call_args["Item"] + assert put_item_args["project"] == "my-project" + assert put_item_args["Status"] == "READY" + assert put_item_args["AMI"] == "ami-12345678" + assert put_item_args["LastUpdated"] == str(fixed_time) + assert put_item_args["AMIName"] == "Test AMI" + assert put_item_args["AMIDescription"] == "Test Description" + assert put_item_args["AMICreationDate"] == "2025-01-01T00:00:00.000Z" + assert put_item_args["LastInstanceType"] == "m5.large" + assert put_item_args["LastKeyPair"] == "my-key" + assert put_item_call_args["ConditionExpression"] == "attribute_not_exists(#project)" + assert put_item_call_args["ExpressionAttributeNames"] == {"#project": "project"} + + +def test_create_project_entry_raises_existing_project_error_on_conditional_failure(): + mock_table = MagicMock() + mock_table.put_item.side_effect = _client_error( + "ConditionalCheckFailedException", + "The conditional request failed", + "PutItem", + ) + + with pytest.raises(ValueError, match="Project 'my-project' already exists"): + create_project_entry(mock_table, "my-project", {"ImageId": "ami-12345678"}) + + +def test_create_project_entry_client_error(): + mock_table = MagicMock() + mock_table.put_item.side_effect = _client_error("AccessDeniedException", "Denied", "PutItem") + + with pytest.raises(AWSClientError): + create_project_entry(mock_table, "my-project", {"ImageId": "ami-12345678"}) + + +def test_create_project_entry_without_launch_defaults(): + mock_table = MagicMock() + ami_info = {"ImageId": "ami-12345678"} + + create_project_entry(mock_table, "my-project", ami_info) + + put_item_call_args = mock_table.put_item.call_args[1] + put_item_args = put_item_call_args["Item"] + assert "LastInstanceType" not in put_item_args + assert "LastKeyPair" not in put_item_args + assert put_item_call_args["ConditionExpression"] == "attribute_not_exists(#project)" + assert put_item_call_args["ExpressionAttributeNames"] == {"#project": "project"} + + +@patch("devbox.new.create_project_entry") +@patch("devbox.new.check_project_exists") +@patch("devbox.new.validate_ami_exists") +@patch("devbox.new.initialize_aws_clients") +@patch("devbox.new.utils.get_dynamodb_table") +@patch("devbox.new.utils.get_ssm_parameter") +def test_new_project_programmatic_success( + mock_get_ssm_parameter, + mock_get_dynamodb_table, + mock_init_aws, + mock_validate_ami, + mock_check_exists, + mock_create_entry, +): + aws = {"ec2": MagicMock(), "ssm": MagicMock(), "ddb": MagicMock()} + mock_init_aws.return_value = aws + mock_get_ssm_parameter.return_value = "snapshot-table" + mock_validate_ami.return_value = {"ImageId": "ami-12345678", "Architecture": "x86_64"} + mock_get_dynamodb_table.return_value = MagicMock() + mock_check_exists.return_value = None + + new_project_programmatic( + project="my-project", + base_ami="ami-12345678", + instance_type="m5.large", + key_pair="my-key", + param_prefix="/devbox", + ) + + mock_init_aws.assert_called_once() + mock_validate_ami.assert_called_once_with(aws["ec2"], "ami-12345678") + mock_get_ssm_parameter.assert_called_once_with( + "/devbox/snapshotTable", ssm_client=aws["ssm"] + ) + mock_get_dynamodb_table.assert_called_once_with( + "snapshot-table", dynamodb_resource=aws["ddb"] + ) + mock_check_exists.assert_called_once() + mock_create_entry.assert_called_once_with( + table=mock_get_dynamodb_table.return_value, + project_name="my-project", + ami_info=mock_validate_ami.return_value, + instance_type="m5.large", + key_pair="my-key", + ) + + +def test_new_project_programmatic_invalid_project_name(): + with pytest.raises(ValueError, match="Project name cannot be empty"): + new_project_programmatic(project="", base_ami="ami-12345678") + + +def test_new_project_programmatic_rejects_underscore_in_project_name(): + with pytest.raises( + ValueError, match="Project name must be alphanumeric with optional hyphens" + ): + new_project_programmatic(project="my_project", base_ami="ami-12345678") + + +def test_new_project_programmatic_invalid_ami(): + with pytest.raises(ValueError, match="Base AMI must be a valid AMI ID"): + new_project_programmatic(project="my-project", base_ami="not-an-ami") + + +@patch("devbox.new.create_project_entry") +@patch("devbox.new.check_project_exists") +@patch("devbox.new.validate_ami_exists") +@patch("devbox.new.initialize_aws_clients") +@patch("devbox.new.utils.get_dynamodb_table") +@patch("devbox.new.utils.get_ssm_parameter") +def test_new_project_programmatic_normalizes_slashless_param_prefix( + mock_get_ssm_parameter, + mock_get_dynamodb_table, + mock_init_aws, + mock_validate_ami, + mock_check_exists, + mock_create_entry, +): + aws = {"ec2": MagicMock(), "ssm": MagicMock(), "ddb": MagicMock()} + mock_init_aws.return_value = aws + mock_validate_ami.return_value = {"ImageId": "ami-12345678"} + mock_get_ssm_parameter.return_value = "snapshot-table" + mock_get_dynamodb_table.return_value = MagicMock() + mock_check_exists.return_value = None + + new_project_programmatic( + project="my-project", base_ami="ami-12345678", param_prefix="devbox" + ) + + mock_get_ssm_parameter.assert_called_once_with( + "/devbox/snapshotTable", ssm_client=aws["ssm"] + ) + mock_get_dynamodb_table.assert_called_once_with( + "snapshot-table", dynamodb_resource=aws["ddb"] + ) + mock_create_entry.assert_called_once() + + +@patch("devbox.new.create_project_entry") +@patch("devbox.new.check_project_exists") +@patch("devbox.new.validate_ami_exists") +@patch("devbox.new.initialize_aws_clients") +@patch("devbox.new.utils.get_dynamodb_table") +@patch("devbox.new.utils.get_ssm_parameter") +def test_new_project_programmatic_strips_trailing_slash_from_param_prefix( + mock_get_ssm_parameter, + mock_get_dynamodb_table, + mock_init_aws, + mock_validate_ami, + mock_check_exists, + mock_create_entry, +): + aws = {"ec2": MagicMock(), "ssm": MagicMock(), "ddb": MagicMock()} + mock_init_aws.return_value = aws + mock_validate_ami.return_value = {"ImageId": "ami-12345678"} + mock_get_ssm_parameter.return_value = "snapshot-table" + mock_get_dynamodb_table.return_value = MagicMock() + mock_check_exists.return_value = None + + new_project_programmatic( + project="my-project", base_ami="ami-12345678", param_prefix="/devbox/" + ) + + mock_get_ssm_parameter.assert_called_once_with( + "/devbox/snapshotTable", ssm_client=aws["ssm"] + ) + mock_get_dynamodb_table.assert_called_once_with( + "snapshot-table", dynamodb_resource=aws["ddb"] + ) + mock_create_entry.assert_called_once() + + +def test_new_project_programmatic_rejects_consecutive_slashes_in_param_prefix(): + with pytest.raises( + ValueError, match="Parameter prefix cannot contain consecutive slashes" + ): + new_project_programmatic( + project="my-project", + base_ami="ami-12345678", + param_prefix="/devbox//nested", + ) + + +@patch("devbox.new.create_project_entry") +@patch("devbox.new.check_project_exists") +@patch("devbox.new.validate_ami_exists") +@patch("devbox.new.initialize_aws_clients") +@patch("devbox.new.utils.get_dynamodb_table") +@patch("devbox.new.utils.get_ssm_parameter") +def test_new_project_programmatic_existing_project( + mock_get_ssm_parameter, + mock_get_dynamodb_table, + mock_init_aws, + mock_validate_ami, + mock_check_exists, + mock_create_entry, +): + mock_init_aws.return_value = { + "ec2": MagicMock(), + "ssm": MagicMock(), + "ddb": MagicMock(), + } + mock_get_ssm_parameter.return_value = "snapshot-table" + mock_validate_ami.return_value = {"ImageId": "ami-12345678"} + mock_get_dynamodb_table.return_value = MagicMock() + mock_check_exists.return_value = {"project": "my-project", "Status": "READY"} + + with pytest.raises(ValueError, match="already exists"): + new_project_programmatic(project="my-project", base_ami="ami-12345678") + + mock_create_entry.assert_not_called() + + +@patch("devbox.new.create_project_entry") +@patch("devbox.new.check_project_exists") +@patch("devbox.new.validate_ami_exists") +@patch("devbox.new.initialize_aws_clients") +@patch("devbox.new.utils.get_dynamodb_table") +@patch("devbox.new.utils.get_ssm_parameter") +def test_new_project_programmatic_surfaces_conditional_write_race_as_existing_project( + mock_get_ssm_parameter, + mock_get_dynamodb_table, + mock_init_aws, + mock_validate_ami, + mock_check_exists, + mock_create_entry, +): + mock_init_aws.return_value = { + "ec2": MagicMock(), + "ssm": MagicMock(), + "ddb": MagicMock(), + } + mock_get_ssm_parameter.return_value = "snapshot-table" + mock_validate_ami.return_value = {"ImageId": "ami-12345678"} + mock_get_dynamodb_table.return_value = MagicMock() + mock_check_exists.return_value = None + mock_create_entry.side_effect = ValueError("Project 'my-project' already exists") + + with pytest.raises(ValueError, match="Project 'my-project' already exists"): + new_project_programmatic(project="my-project", base_ami="ami-12345678") + + +@patch("argparse.ArgumentParser.parse_args") +@patch("devbox.new.new_project_programmatic") +def test_main_success(mock_new_project, mock_parse_args): + mock_args = MagicMock() + mock_args.project = "my-project" + mock_args.base_ami = "ami-12345678" + mock_args.instance_type = "m5.large" + mock_args.key_pair = "my-key" + mock_args.param_prefix = "/devbox" + mock_parse_args.return_value = mock_args + + main() + + mock_new_project.assert_called_once_with( + project="my-project", + base_ami="ami-12345678", + instance_type="m5.large", + key_pair="my-key", + param_prefix="/devbox", + ) + + +@patch("argparse.ArgumentParser.parse_args") +def test_main_keyboard_interrupt(mock_parse_args): + mock_parse_args.side_effect = KeyboardInterrupt() + + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 1 + + +@patch("argparse.ArgumentParser.parse_args") +@patch("devbox.new.new_project_programmatic") +def test_main_value_error(mock_new_project, mock_parse_args): + mock_args = MagicMock() + mock_args.project = "my-project" + mock_args.base_ami = "ami-12345678" + mock_args.instance_type = None + mock_args.key_pair = None + mock_args.param_prefix = "/devbox" + mock_parse_args.return_value = mock_args + mock_new_project.side_effect = ValueError("Invalid") + + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 2 + + +@patch("argparse.ArgumentParser.parse_args") +@patch("devbox.new.new_project_programmatic") +def test_main_resource_not_found_error(mock_new_project, mock_parse_args): + mock_args = MagicMock() + mock_args.project = "my-project" + mock_args.base_ami = "ami-12345678" + mock_args.instance_type = None + mock_args.key_pair = None + mock_args.param_prefix = "/devbox" + mock_parse_args.return_value = mock_args + mock_new_project.side_effect = ResourceNotFoundError("AMI not found") + + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 2 + + +@patch("argparse.ArgumentParser.parse_args") +@patch("devbox.new.new_project_programmatic") +def test_main_aws_client_error(mock_new_project, mock_parse_args): + mock_args = MagicMock() + mock_args.project = "my-project" + mock_args.base_ami = "ami-12345678" + mock_args.instance_type = None + mock_args.key_pair = None + mock_args.param_prefix = "/devbox" + mock_parse_args.return_value = mock_args + mock_new_project.side_effect = AWSClientError("AWS failed") + + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 3 + + +@patch("argparse.ArgumentParser.parse_args") +@patch("devbox.new.new_project_programmatic") +def test_main_unexpected_exception(mock_new_project, mock_parse_args): + mock_args = MagicMock() + mock_args.project = "my-project" + mock_args.base_ami = "ami-12345678" + mock_args.instance_type = None + mock_args.key_pair = None + mock_args.param_prefix = "/devbox" + mock_parse_args.return_value = mock_args + mock_new_project.side_effect = Exception("boom") + + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 4 diff --git a/tests/test_remote_client.py b/tests/test_remote_client.py index 45374b2..f2ca9a6 100644 --- a/tests/test_remote_client.py +++ b/tests/test_remote_client.py @@ -15,12 +15,26 @@ get_cli_function_url, get_function_url_region, invoke_action, + normalize_param_prefix, ) FUNCTION_URL = "https://abc123.lambda-url.us-east-1.on.aws/" +def test_normalize_param_prefix_adds_leading_slash(): + assert normalize_param_prefix("devbox") == "/devbox" + + +def test_normalize_param_prefix_strips_trailing_slash(): + assert normalize_param_prefix("/devbox/") == "/devbox" + + +def test_normalize_param_prefix_rejects_consecutive_slashes(): + with pytest.raises(ValueError, match="Parameter prefix cannot contain consecutive slashes"): + normalize_param_prefix("/devbox//nested") + + def test_get_cli_function_url_uses_expected_parameter_name(): with patch( "devbox.remote_client.utils.get_ssm_parameter", diff --git a/tests/test_utils.py b/tests/test_utils.py index e2a0689..0337564 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -2,6 +2,7 @@ import json from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock # Removed unused imports - using moto fixtures instead import pytest @@ -13,6 +14,7 @@ get_ec2_resource, get_dynamodb_resource, get_dynamodb_table, + get_image, get_ssm_parameter, get_project_tag, format_timedelta, @@ -63,6 +65,17 @@ def test_get_dynamodb_table(mock_dynamodb): assert table.name == table_name +def test_get_dynamodb_table_with_injected_resource(): + mock_resource = MagicMock() + mock_table = MagicMock() + mock_resource.Table.return_value = mock_table + + table = get_dynamodb_table("test-table", dynamodb_resource=mock_resource) + + assert table is mock_table + mock_resource.Table.assert_called_once_with("test-table") + + def test_get_ssm_parameter_success(mock_ssm): param_name = "/test/param" param_value = "test-value" @@ -72,6 +85,62 @@ def test_get_ssm_parameter_success(mock_ssm): result = get_ssm_parameter(param_name) assert result == param_value + +def test_get_ssm_parameter_with_injected_client(): + mock_ssm = MagicMock() + mock_ssm.get_parameter.return_value = {"Parameter": {"Value": "test-value"}} + + result = get_ssm_parameter("/test/param", ssm_client=mock_ssm) + + assert result == "test-value" + mock_ssm.get_parameter.assert_called_once_with( + Name="/test/param", WithDecryption=True + ) + + +def test_get_image_with_injected_client(): + mock_ec2 = MagicMock() + mock_ec2.describe_images.return_value = { + "Images": [{"ImageId": "ami-12345678", "Name": "test-image"}] + } + + image = get_image("ami-12345678", ec2_client=mock_ec2) + + assert image == {"ImageId": "ami-12345678", "Name": "test-image"} + mock_ec2.describe_images.assert_called_once_with(ImageIds=["ami-12345678"]) + + +def test_get_image_returns_none_for_empty_result(): + mock_ec2 = MagicMock() + mock_ec2.describe_images.return_value = {"Images": []} + + image = get_image("ami-12345678", ec2_client=mock_ec2) + + assert image is None + + +def test_get_image_returns_none_for_invalid_ami(): + mock_ec2 = MagicMock() + mock_ec2.describe_images.side_effect = ClientError( + {"Error": {"Code": "InvalidAMIID.NotFound", "Message": "Not found"}}, + "DescribeImages", + ) + + image = get_image("ami-12345678", ec2_client=mock_ec2) + + assert image is None + + +def test_get_image_reraises_other_client_error(): + mock_ec2 = MagicMock() + mock_ec2.describe_images.side_effect = ClientError( + {"Error": {"Code": "AccessDenied", "Message": "Denied"}}, + "DescribeImages", + ) + + with pytest.raises(ClientError): + get_image("ami-12345678", ec2_client=mock_ec2) + def test_get_ssm_parameter_not_found_required(mock_ssm): param_name = "/nonexistent/param"