Disallow termination of protected EMR job flows (#4015)

Error message verified against real AWS backend.

moto/emr/exceptions.py

@@ -12,3 +12,10 @@ class InvalidRequestException(JsonRESTError):
         super(InvalidRequestException, self).__init__(
             "InvalidRequestException", message, **kwargs
         )
+
+
+class ValidationException(JsonRESTError):
+    def __init__(self, message, **kwargs):
+        super(ValidationException, self).__init__(
+            "ValidationException", message, **kwargs
+        )

moto/emr/models.py

@@ -8,7 +8,7 @@ import pytz
 from boto3 import Session
 from dateutil.parser import parse as dtparse
 from moto.core import ACCOUNT_ID, BaseBackend, BaseModel
-from moto.emr.exceptions import EmrError, InvalidRequestException
+from moto.emr.exceptions import EmrError, InvalidRequestException, ValidationException
 from .utils import (
     random_instance_group_id,
     random_cluster_id,
@@ -624,12 +624,20 @@ class ElasticMapReduceBackend(BaseBackend):
             cluster.set_termination_protection(value)
 
     def terminate_job_flows(self, job_flow_ids):
-        clusters = []
+        clusters_terminated = []
+        clusters_protected = []
         for job_flow_id in job_flow_ids:
             cluster = self.clusters[job_flow_id]
+            if cluster.termination_protected:
+                clusters_protected.append(cluster)
+                continue
             cluster.terminate()
-            clusters.append(cluster)
-        return clusters
+            clusters_terminated.append(cluster)
+        if clusters_protected:
+            raise ValidationException(
+                "Could not shut down one or more job flows since they are termination protected."
+            )
+        return clusters_terminated
 
     def put_auto_scaling_policy(self, instance_group_id, auto_scaling_policy):
         instance_groups = self.get_instance_groups(

moto/emr/responses.py

@@ -505,7 +505,7 @@ class ElasticMapReduceResponse(BaseResponse):
 
     @generate_boto3_response("SetTerminationProtection")
     def set_termination_protection(self):
-        termination_protection = self._get_param("TerminationProtected")
+        termination_protection = self._get_bool_param("TerminationProtected")
         job_ids = self._get_multi_param("JobFlowIds.member")
         self.backend.set_termination_protection(job_ids, termination_protection)
         template = self.response_template(SET_TERMINATION_PROTECTION_TEMPLATE)

tests/test_emr/test_emr_boto3.py

@@ -708,6 +708,25 @@ def test_set_termination_protection():
         resp["Cluster"]["TerminationProtected"].should.equal(expected)
 
 
+@mock_emr
+def test_terminate_protected_job_flow_raises_error():
+    client = boto3.client("emr", region_name="us-east-1")
+    resp = client.run_job_flow(**run_job_flow_args)
+    cluster_id = resp["JobFlowId"]
+    client.set_termination_protection(
+        JobFlowIds=[cluster_id], TerminationProtected=True
+    )
+    with pytest.raises(ClientError) as ex:
+        client.terminate_job_flows(
+            JobFlowIds=[cluster_id,]
+        )
+    error = ex.value.response["Error"]
+    error["Code"].should.equal("ValidationException")
+    error["Message"].should.equal(
+        "Could not shut down one or more job flows since they are termination protected."
+    )
+
+
 @mock_emr
 def test_set_visible_to_all_users():
     client = boto3.client("emr", region_name="us-east-1")