Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions codebeaver.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
from: pytest
# This file was generated automatically by CodeBeaver based on your repository. Learn how to customize it here: https://docs.codebeaver.ai/open-source/codebeaver-yml/
78 changes: 78 additions & 0 deletions tests/test_opsworks.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

from troposphere import Template
from troposphere.opsworks import Stack
from troposphere.opsworks import App, Instance, Layer, BlockDeviceMapping, EbsBlockDevice, SslConfiguration


class TestOpsWorksStack(unittest.TestCase):
Expand Down Expand Up @@ -66,5 +67,82 @@ def test_custom_json(self):
stack.CustomJson = True


def test_app_minimal(self):
"""Test that a minimal valid App resource converts to JSON successfully."""
app = App(
"myapp",
Name="TestApp",
StackId="stack123",
Type="other",
)
t = Template()
t.add_resource(app)
json_output = t.to_json()
self.assertIn("TestApp", json_output)

def test_instance_minimal(self):
"""Test that a minimal valid Instance resource converts to JSON successfully."""
instance = Instance(
"myinstance",
InstanceType="t2.micro",
LayerIds=["layer1"],
StackId="stack123",
)
t = Template()
t.add_resource(instance)
json_output = t.to_json()
self.assertIn("t2.micro", json_output)

def test_layer_minimal(self):
"""Test that a minimal valid Layer resource converts to JSON successfully."""
layer = Layer(
"mylayer",
AutoAssignElasticIps=True,
AutoAssignPublicIps=True,
EnableAutoHealing=True,
Name="TestLayer",
Shortname="tlayer",
StackId="stack123",
Type="custom",
)
t = Template()
t.add_resource(layer)
json_output = t.to_json()
self.assertIn("TestLayer", json_output)

def test_block_device_mapping(self):
"""Test that BlockDeviceMapping validation does not raise for valid configuration."""
ebs = EbsBlockDevice(
DeleteOnTermination=True,
Iops=100,
VolumeSize=30,
VolumeType="gp2",
)
mapping = BlockDeviceMapping(
DeviceName="/dev/sdh",
Ebs=ebs,
)
# This should not raise any exception.
mapping.validate()

def test_ssl_configuration_in_app(self):
"""Test that an App with SslConfiguration converts to JSON successfully."""
ssl_config = SslConfiguration(
Certificate="cert-data",
PrivateKey="key-data",
Chain="chain-data",
)
app = App(
"myappssl",
Name="SecureApp",
StackId="stack123",
Type="php",
SslConfiguration=ssl_config,
)
t = Template()
t.add_resource(app)
json_output = t.to_json()
self.assertIn("SecureApp", json_output)
self.assertIn("cert-data", json_output)
if __name__ == "__main__":
unittest.main()
91 changes: 91 additions & 0 deletions tests/test_resiliencehub.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import unittest

import pytest
from troposphere.resiliencehub import FailurePolicy, ResiliencyPolicy


Expand Down Expand Up @@ -40,5 +41,95 @@ def test_invalid_policy_tier(self):
).to_dict()


def test_failurepolicy_invalid_rpo_type(self):
"""Test FailurePolicy raises error when RpoInSecs is not an integer."""
with self.assertRaises(ValueError):
# Passing a string instead of an integer should raise an error.
FailurePolicy(RpoInSecs="ten", RtoInSecs=10)

def test_failurepolicy_missing_rto(self):
"""Test FailurePolicy raises an error when RtoInSecs is missing."""
with self.assertRaises(ValueError):
FailurePolicy(RpoInSecs=10).to_dict()
def test_valid_policy_with_tags_and_constraint(self):
"""Test ResiliencyPolicy creates a valid dictionary when optional parameters are provided."""
policy_dict = ResiliencyPolicy(
"policy",
Policy={"Hardware": FailurePolicy(RpoInSecs=20, RtoInSecs=30)},
PolicyName="test_policy",
Tier="MissionCritical",
DataLocationConstraint="us-west-2",
Tags={"Env": "Dev"},
).to_dict()
# In troposphere, the to_dict() output contains "Type" and "Properties" keys.
props = policy_dict.get("Properties", {})
self.assertIn("DataLocationConstraint", props)
self.assertEqual(props["DataLocationConstraint"], "us-west-2")
self.assertIn("Tags", props)
self.assertEqual(props["Tags"], {"Env": "Dev"})
def test_empty_policy_dict(self):
"""Test ResiliencyPolicy converts an empty Policy dictionary correctly into the Properties structure."""
policy_dict = ResiliencyPolicy(
"policy",
Policy={},
PolicyName="foo",
Tier="MissionCritical",
).to_dict()
self.assertIn("Properties", policy_dict)
self.assertIn("Policy", policy_dict["Properties"])
self.assertEqual(policy_dict["Properties"]["Policy"], {})
"""Test ResiliencyPolicy raises an error when Policy is provided with a non-dictionary type."""
with self.assertRaises(ValueError):
ResiliencyPolicy(
"policy",
Policy=[{"Hardware": FailurePolicy(RpoInSecs=10, RtoInSecs=10)}],
PolicyName="foo",
Tier="MissionCritical",
).to_dict()

def test_failurepolicy_zero_values(self):
"""Test FailurePolicy accepts zero values and returns the correct dictionary representation."""
fp = FailurePolicy(RpoInSecs=0, RtoInSecs=0)
d = fp.to_dict()
self.assertEqual(d["RpoInSecs"], 0)
self.assertEqual(d["RtoInSecs"], 0)
def test_failurepolicy_negative_values(self):
"""Test FailurePolicy accepts negative integer values."""
fp = FailurePolicy(RpoInSecs=-10, RtoInSecs=-20)
d = fp.to_dict()
self.assertEqual(d["RpoInSecs"], -10)
self.assertEqual(d["RtoInSecs"], -20)

def test_invalid_tags_type(self):
"""Test ResiliencyPolicy raises an error when Tags is not a dictionary."""
with self.assertRaises((ValueError, TypeError)):
ResiliencyPolicy(
"policy",
Policy={"Hardware": FailurePolicy(RpoInSecs=10, RtoInSecs=10)},
PolicyName="foo",
Tier="MissionCritical",
Tags=["Env", "Dev"],
).to_dict()

def test_invalid_policyname_type(self):
"""Test ResiliencyPolicy raises an error when PolicyName is not a string."""
with self.assertRaises((ValueError, TypeError)):
ResiliencyPolicy(
"policy",
Policy={"Hardware": FailurePolicy(RpoInSecs=10, RtoInSecs=10)},
PolicyName=123,
Tier="MissionCritical",
).to_dict()

def test_invalid_datalocationconstraint_type(self):
"""Test ResiliencyPolicy raises an error when DataLocationConstraint is not a string."""
with self.assertRaises((ValueError, TypeError)):
ResiliencyPolicy(
"policy",
Policy={"Hardware": FailurePolicy(RpoInSecs=10, RtoInSecs=10)},
PolicyName="foo",
Tier="MissionCritical",
DataLocationConstraint=123,
).to_dict()
if __name__ == "__main__":
unittest.main()
172 changes: 172 additions & 0 deletions tests/test_resourcegroups.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,172 @@
import pytest
from troposphere.resourcegroups import Group, ResourceQuery, Query, TagFilter, ConfigurationItem, ConfigurationParameter
from troposphere import Tags

class TestResourceGroups:
"""Tests for the resource groups module of troposphere."""

def test_group_minimum(self):
"""Test creating a Group with only the required 'Name' property."""
group = Group("TestGroupTitle", Name="TestGroup")
# Accessing group properties; AWSObject is expected to store properties in 'properties'
props = getattr(group, "properties", {})
assert props.get("Name") == "TestGroup"
# Optional properties should not be set if not provided
for key in ["Configuration", "Description", "ResourceQuery", "Resources", "Tags"]:
assert key not in props

def test_group_full(self):
"""Test creating a Group with all properties including nested configuration."""
config_param = ConfigurationParameter(Name="Param1", Values=["value1", "value2"])
config_item = ConfigurationItem(Parameters=[config_param], Type="CustomType")
tag_filter = TagFilter(Key="Env", Values=["Prod", "Dev"])
query = Query(ResourceTypeFilters=["AWS::EC2::Instance"], StackIdentifier="stack-123", TagFilters=[tag_filter])
resource_query = ResourceQuery(Query=query, Type="TAG_FILTERS_1_0")
group = Group("FullGroupTitle",
Name="FullGroup",
Configuration=[config_item],
Description="A fully configured group",
ResourceQuery=resource_query,
Resources=["resource1", "resource2"],
Tags=Tags({"Key1": "Value1", "Key2": "Value2"})
)
props = getattr(group, "properties", {})
assert props.get("Name") == "FullGroup"
assert props.get("Description") == "A fully configured group"
assert isinstance(props.get("Configuration"), list)
# Check nested configuration parameter structure
config = props["Configuration"][0]
assert config.properties.get("Type") == "CustomType"
parameters = config.properties.get("Parameters", [])
assert isinstance(parameters, list)
param = parameters[0]
assert param.properties.get("Name") == "Param1"
assert param.properties.get("Values") == ["value1", "value2"]

def test_resource_query(self):
"""Test creating a ResourceQuery with a Query containing TagFilters."""
tag_filter1 = TagFilter(Key="Owner", Values=["Alice"])
tag_filter2 = TagFilter(Key="Status", Values=["Active", "Pending"])
query = Query(ResourceTypeFilters=["AWS::S3::Bucket"], TagFilters=[tag_filter1, tag_filter2])
resource_query = ResourceQuery(Query=query, Type="TAG_FILTERS_1_0")
rq_props = getattr(resource_query, "properties", {})
query_instance = rq_props.get("Query")
query_props = getattr(query_instance, "properties", {})
assert query_props.get("ResourceTypeFilters") == ["AWS::S3::Bucket"]
tag_filters = query_props.get("TagFilters", [])
assert len(tag_filters) == 2
# Validate TagFilter entries using properties dict
assert tag_filters[0].properties.get("Key") == "Owner"
assert tag_filters[0].properties.get("Values") == ["Alice"]
assert tag_filters[1].properties.get("Key") == "Status"
assert tag_filters[1].properties.get("Values") == ["Active", "Pending"]
# Also validate that Type is set correctly in ResourceQuery
assert rq_props.get("Type") == "TAG_FILTERS_1_0"

def test_missing_required_property(self):
"""Test that missing required property 'Name' in Group raises an error if validated."""
with pytest.raises(Exception):
# If the AWSObject validates required properties during instantiation,
# missing the required 'Name' should raise an exception.
Group()

def test_invalid_property_type(self):
"""Test providing an invalid type for a property raises an error."""
with pytest.raises(Exception):
# Passing an integer for 'Name' is invalid; expecting an exception during validation.
Group("InvalidTitle", Name=123)
def test_empty_optional_properties(self):
"""Test that providing empty lists/dict for optional properties sets the properties accordingly."""
group = Group("EmptyGroupTitle", Name="EmptyGroup", Configuration=[], Resources=[], Tags=Tags({}))
props = getattr(group, "properties", {})
# Check that these properties are set exactly as provided.
assert props.get("Configuration") == []
assert props.get("Resources") == []
tags_obj = props.get("Tags")
assert isinstance(tags_obj, Tags)
# Instead of converting to dict, check the underlying tags attribute
assert hasattr(tags_obj, "tags"), "Tags object should have a 'tags' attribute"
assert tags_obj.tags == [], "Expected tags attribute to be an empty list"

def test_invalid_configuration_parameter(self):
"""Test that providing an invalid type for ConfigurationParameter.Values raises an error."""
with pytest.raises(Exception):
# Values is expected to be a list of strings; providing an integer in the list should trigger a validation error.
ConfigurationParameter(Name="InvalidParam", Values=["valid", 100])

def test_query_with_empty_tag_filters(self):
"""Test that providing an empty list for TagFilters in Query sets the property correctly."""
query = Query(ResourceTypeFilters=["AWS::Lambda::Function"], TagFilters=[])
query_props = getattr(query, "properties", {})
assert query_props.get("TagFilters") == []
# Validate that when the query is used within ResourceQuery the empty TagFilters are maintained.
resource_query = ResourceQuery(Query=query, Type="TAG_FILTERS_1_0")
rq_props = getattr(resource_query, "properties", {})
query_obj = rq_props.get("Query")
query_obj_props = getattr(query_obj, "properties", {})
assert query_obj_props.get("TagFilters") == []
def test_configuration_item_empty_parameters(self):
"""Test creating a ConfigurationItem with no Parameters specified."""
# Create ConfigurationItem with only the required Type value
config_item = ConfigurationItem(Type="EmptyType")
props = getattr(config_item, "properties", {})
# Optional Parameters should not be set if not provided
assert "Parameters" not in props

def test_query_with_missing_optional(self):
"""Test that a Query defined only with ResourceTypeFilters omits the other optional properties."""
query = Query(ResourceTypeFilters=["AWS::Lambda::Function"])
props = getattr(query, "properties", {})
assert props.get("ResourceTypeFilters") == ["AWS::Lambda::Function"]
assert "StackIdentifier" not in props
assert "TagFilters" not in props

def test_tag_filter_optional(self):
"""Test creating a TagFilter without providing Key or Values results in an empty properties dict."""
tag_filter = TagFilter()
props = getattr(tag_filter, "properties", {})
assert "Key" not in props
assert "Values" not in props

def test_invalid_resource_query_type(self):
"""Test that providing an invalid type for ResourceQuery raises an error."""
with pytest.raises(Exception):
query = Query(ResourceTypeFilters=["AWS::DynamoDB::Table"])
# "INVALID_TYPE" should trigger a validation error via resourcequery_type
ResourceQuery(Query=query, Type="INVALID_TYPE")

def test_minimal_resource_query(self):
"""Test creating a minimal ResourceQuery with an empty Query property."""
resource_query = ResourceQuery(Query=Query(), Type="TAG_FILTERS_1_0")
props = getattr(resource_query, "properties", {})
query_obj = props.get("Query")
query_props = getattr(query_obj, "properties", {}) if query_obj else {}
# All optional properties in Query should not be set
assert "ResourceTypeFilters" not in query_props
assert "StackIdentifier" not in query_props
assert "TagFilters" not in query_props
# Verify that the ResourceQuery "Type" property is set correctly
assert props.get("Type") == "TAG_FILTERS_1_0"
def test_invalid_query_resource_type_filters(self):
"""Test that providing a non-string element in ResourceTypeFilters raises an error."""
with pytest.raises(Exception):
# ResourceTypeFilters should be a list of strings
Query(ResourceTypeFilters=[123])

def test_invalid_tagfilter_wrong_type(self):
"""Test that providing a non-string for TagFilter Key raises an error."""
with pytest.raises(Exception):
# TagFilter Key is expected to be a string
TagFilter(Key=123, Values=["valid"])

def test_group_invalid_resources_item(self):
"""Test that providing a non-string element in Resources list raises an error."""
with pytest.raises(Exception):
# Each resource in Resources should be a string
Group("InvalidResourcesTitle", Name="InvalidResources", Resources=[{"not": "a string"}])

def test_configuration_item_invalid_parameters_content(self):
"""Test that providing an invalid element in ConfigurationItem.Parameters (not a ConfigurationParameter) raises an error."""
with pytest.raises(Exception):
# Parameters should be a list of ConfigurationParameter objects
ConfigurationItem(Parameters=["invalid parameter"], Type="SomeType")