import unittest
from troposphere import Parameter, Ref
from troposphere.autoscaling import AutoScalingGroup
from troposphere.policies import (
AutoScalingCreationPolicy,
AutoScalingRollingUpdate,
CreationPolicy,
ResourceSignal,
UpdatePolicy,
)
[docs]class TestCreationPolicy(unittest.TestCase):
[docs] def test_invalid_pausetime(self):
with self.assertRaises(ValueError):
CreationPolicy(ResourceSignal=ResourceSignal(Count=2, Timeout="90"))
[docs] def test_works(self):
policy = CreationPolicy(ResourceSignal=ResourceSignal(Count=2, Timeout="PT10M"))
self.assertEqual(policy.ResourceSignal.Count, 2)
self.assertEqual(policy.ResourceSignal.Timeout, "PT10M")
[docs] def test_json(self):
policy = CreationPolicy(ResourceSignal=ResourceSignal(Count=2, Timeout="PT10M"))
p = policy.to_dict()
self.assertEqual(p["ResourceSignal"]["Count"], 2)
self.assertEqual(p["ResourceSignal"]["Timeout"], "PT10M")
[docs] def test_auto_scaling_creation_policy(self):
policy = CreationPolicy(
AutoScalingCreationPolicy=AutoScalingCreationPolicy(
MinSuccessfulInstancesPercent=50,
),
ResourceSignal=ResourceSignal(Count=2, Timeout="PT10M"),
)
self.assertEqual(
policy.AutoScalingCreationPolicy.MinSuccessfulInstancesPercent, 50
)
[docs] def test_auto_scaling_creation_policy_json(self):
policy = CreationPolicy(
AutoScalingCreationPolicy=AutoScalingCreationPolicy(
MinSuccessfulInstancesPercent=50,
),
ResourceSignal=ResourceSignal(Count=2, Timeout="PT10M"),
)
p = policy.to_dict()
self.assertEqual(
p["AutoScalingCreationPolicy"]["MinSuccessfulInstancesPercent"], 50
)
[docs]class TestUpdatePolicy(unittest.TestCase):
[docs] def test_invalid_pausetime(self):
with self.assertRaises(ValueError):
UpdatePolicy(
AutoScalingRollingUpdate=AutoScalingRollingUpdate(PauseTime="90")
)
[docs] def test_works(self):
p = UpdatePolicy(
AutoScalingRollingUpdate=AutoScalingRollingUpdate(
MaxBatchSize=2,
MinInstancesInService=1,
PauseTime="PT90S",
WaitOnResourceSignals=True,
)
)
self.assertEqual(p.AutoScalingRollingUpdate.MaxBatchSize, 2)
self.assertEqual(p.AutoScalingRollingUpdate.MinInstancesInService, 1)
self.assertEqual(p.AutoScalingRollingUpdate.PauseTime, "PT90S")
self.assertTrue(p.AutoScalingRollingUpdate.WaitOnResourceSignals)
[docs] def test_mininstances(self):
group = AutoScalingGroup(
"mygroup",
LaunchConfigurationName="I'm a test",
MaxSize=1,
MinSize=1,
UpdatePolicy=UpdatePolicy(
AutoScalingRollingUpdate=AutoScalingRollingUpdate(
PauseTime="PT1M5S", MinInstancesInService="1", MaxBatchSize="1"
)
),
)
with self.assertRaises(ValueError):
group.validate()
[docs] def test_mininstances_maxsize_is_ref(self):
paramMaxSize = Parameter("ParamMaxSize", Type="String")
group = AutoScalingGroup(
"mygroup",
AvailabilityZones=["eu-west-1a", "eu-west-1b"],
LaunchConfigurationName="I'm a test",
MaxSize=Ref(paramMaxSize),
MinSize="2",
UpdatePolicy=UpdatePolicy(
AutoScalingRollingUpdate=AutoScalingRollingUpdate(
PauseTime="PT1M5S", MinInstancesInService="2", MaxBatchSize="1"
)
),
)
self.assertIsNone(group.validate())
[docs] def test_mininstances_mininstancesinservice_is_ref(self):
paramMinInstancesInService = Parameter(
"ParamMinInstancesInService", Type="String"
)
group = AutoScalingGroup(
"mygroup",
AvailabilityZones=["eu-west-1a", "eu-west-1b"],
LaunchConfigurationName="I'm a test",
MaxSize="4",
MinSize="2",
UpdatePolicy=UpdatePolicy(
AutoScalingRollingUpdate=AutoScalingRollingUpdate(
PauseTime="PT1M5S",
MinInstancesInService=Ref(paramMinInstancesInService),
MaxBatchSize="2",
)
),
)
self.assertIsNone(group.validate())
[docs] def test_json(self):
p = UpdatePolicy(
AutoScalingRollingUpdate=AutoScalingRollingUpdate(
MaxBatchSize=2,
MinInstancesInService=1,
PauseTime="PT90S",
WaitOnResourceSignals=True,
)
)
p = p.to_dict()
self.assertEqual(p["AutoScalingRollingUpdate"]["MaxBatchSize"], 2)
self.assertEqual(p["AutoScalingRollingUpdate"]["MinInstancesInService"], 1)
self.assertEqual(p["AutoScalingRollingUpdate"]["PauseTime"], "PT90S")
self.assertTrue(p["AutoScalingRollingUpdate"]["WaitOnResourceSignals"])