From 036a93821d19633f10dc212b5fda65fc01cfe422 Mon Sep 17 00:00:00 2001 From: Justin Gasper Date: Wed, 4 Nov 2026 15:15:28 +1100 Subject: [PATCH] Accept optional ECS circuit breaker settings --- tcdeploy.py | 6 +++++- tests/test_security.py | 34 ++++++++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 1 deletion(-) diff --git a/tcdeploy.py b/tcdeploy.py index 15c2325..f1da01b 100644 --- a/tcdeploy.py +++ b/tcdeploy.py @@ -284,7 +284,11 @@ def read_service(aws, m): service = response['services'][0] require(service.get('status') == 'ACTIVE' and service.get('desiredCount', 0) > 0, 'Service must be active with positive desired capacity; bootstrap separately') require(service.get('deploymentController', {}).get('type', 'ECS') == 'ECS', 'Only ECS rolling deployments are supported') - require(service.get('deploymentConfiguration', {}).get('deploymentCircuitBreaker') == {'enable': True, 'rollback': True}, 'Enable ECS deployment circuit breaker with rollback before migration') + breaker = service.get('deploymentConfiguration', {}).get('deploymentCircuitBreaker') + # ECS also returns optional reset/threshold settings. Require the two safety + # flags without rejecting or changing those service-owned settings. + require(isinstance(breaker, dict) and breaker.get('enable') is True and breaker.get('rollback') is True, + 'Enable ECS deployment circuit breaker with rollback before migration') require(service.get('serviceArn') == arn(m, 'ecs', f"service/{m['cluster']}/{m['service']}"), 'Service ARN differs from manifest') return service diff --git a/tests/test_security.py b/tests/test_security.py index d3bcf20..44f3db0 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -391,6 +391,40 @@ def test_required_circuit_breaker(self): with self.assertRaises(d.DeployError): self.deploy() + def test_circuit_breaker_accepts_optional_ecs_settings(self): + breaker = self.aws.service['deploymentConfiguration']['deploymentCircuitBreaker'] + breaker.update(resetOnHealthyTask=True, + thresholdConfiguration={'type': 'BOUNDED_PERCENT', 'value': 50}) + original = deepcopy(breaker) + checked = d.read_service(self.aws, self.m) + self.assertEqual(checked['deploymentConfiguration']['deploymentCircuitBreaker'], original) + self.deploy() + updates = [params for _, op, params in self.aws.calls if op == 'update-service'] + self.assertEqual(len(updates), 1) + self.assertNotIn('deploymentConfiguration', updates[0]) + + def test_circuit_breaker_requires_literal_true_flags(self): + for flag in ('enable', 'rollback'): + for value in (False, None, 0, 1, 'true'): + with self.subTest(flag=flag, value=value): + breaker = {'enable': True, 'rollback': True, 'resetOnHealthyTask': True, + 'thresholdConfiguration': {'type': 'BOUNDED_PERCENT', 'value': 50}} + breaker[flag] = value + self.aws.service['deploymentConfiguration']['deploymentCircuitBreaker'] = breaker + with self.assertRaises(d.DeployError): + self.deploy() + self.assertEqual(self.publisher.calls, []) + self.assertIsNone(self.aws.registered) + + def test_circuit_breaker_rejects_missing_flags_and_malformed_objects(self): + for breaker in (None, [], 'enabled', {}, {'enable': True}, {'rollback': True}): + with self.subTest(breaker=breaker): + self.aws.service['deploymentConfiguration']['deploymentCircuitBreaker'] = breaker + with self.assertRaises(d.DeployError): + self.deploy() + self.assertEqual(self.publisher.calls, []) + self.assertIsNone(self.aws.registered) + def test_ssm_failure_cannot_be_silently_ignored_because_no_read_exists(self): def fail_secret_calls(api, operation, params): self.assertNotIn(api, ('ssm', 'secretsmanager', 'kms'))