This is an automated email from the ASF dual-hosted git repository.
vincbeck pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new d9d0d2cc07a Add tests for the AWS Batch executor boto schemas (#72609)
d9d0d2cc07a is described below
commit d9d0d2cc07a85052b23349038e38194806d31570
Author: JiHo Kim <[email protected]>
AuthorDate: Wed Sep 9 23:41:45 2026 +0900
Add tests for the AWS Batch executor boto schemas (#72609)
The three marshmallow schemas in boto_schema.py sit on the
OVERLOOKED_TESTS allowlist and, unlike most entries there, are not
covered indirectly either - nothing under providers/*/tests references
them. They parse live AWS Batch API responses, so the behaviour worth
pinning down is the data_key renaming from the API's camelCase, the
unknown=EXCLUDE policy that tolerates fields AWS adds over time, and
the post_load hook that returns a BatchJob rather than a plain dict.
Removes the module's entry from OVERLOOKED_TESTS in the same commit, as
that list fails once a listed test file exists.
related: #35442
---
.../tests/unit/always/test_project_structure.py | 1 -
.../amazon/aws/executors/batch/test_boto_schema.py | 124 +++++++++++++++++++++
2 files changed, 124 insertions(+), 1 deletion(-)
diff --git a/airflow-core/tests/unit/always/test_project_structure.py
b/airflow-core/tests/unit/always/test_project_structure.py
index 28eac70d585..2efbc1d7470 100644
--- a/airflow-core/tests/unit/always/test_project_structure.py
+++ b/airflow-core/tests/unit/always/test_project_structure.py
@@ -64,7 +64,6 @@ class TestProjectStructure:
"providers/amazon/tests/unit/amazon/aws/auth_manager/datamodels/test_login.py",
"providers/amazon/tests/unit/amazon/aws/auth_manager/security_manager/test_aws_security_manager_override.py",
"providers/amazon/tests/unit/amazon/aws/executors/batch/test_batch_executor_config.py",
-
"providers/amazon/tests/unit/amazon/aws/executors/batch/test_boto_schema.py",
"providers/amazon/tests/unit/amazon/aws/executors/ecs/test_ecs_executor_config.py",
"providers/amazon/tests/unit/amazon/aws/executors/aws_lambda/test_utils.py",
"providers/amazon/tests/unit/amazon/aws/executors/aws_lambda/docker/test_app.py",
diff --git
a/providers/amazon/tests/unit/amazon/aws/executors/batch/test_boto_schema.py
b/providers/amazon/tests/unit/amazon/aws/executors/batch/test_boto_schema.py
new file mode 100644
index 00000000000..823fb5f88dc
--- /dev/null
+++ b/providers/amazon/tests/unit/amazon/aws/executors/batch/test_boto_schema.py
@@ -0,0 +1,124 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import pytest
+from marshmallow import ValidationError
+
+from airflow.providers.amazon.aws.executors.batch.boto_schema import (
+ BatchDescribeJobsResponseSchema,
+ BatchJobDetailSchema,
+ BatchSubmitJobResponseSchema,
+)
+from airflow.providers.amazon.aws.executors.batch.utils import BatchJob
+
+
+class TestBatchSubmitJobResponseSchema:
+ def test_load_maps_camel_case_job_id_to_snake_case(self):
+ loaded = BatchSubmitJobResponseSchema().load({"jobId": "job-1"})
+
+ assert loaded == {"job_id": "job-1"}
+
+ def test_load_ignores_fields_the_submit_job_api_adds(self):
+ loaded = BatchSubmitJobResponseSchema().load(
+ {"jobId": "job-1", "jobName": "airflow-task", "jobArn":
"arn:aws:batch:job-1"}
+ )
+
+ assert loaded == {"job_id": "job-1"}
+
+ def test_load_rejects_response_without_job_id(self):
+ with pytest.raises(ValidationError) as excinfo:
+ BatchSubmitJobResponseSchema().load({"jobName": "airflow-task"})
+
+ assert "jobId" in excinfo.value.messages
+
+ def test_load_rejects_snake_case_job_id(self):
+ with pytest.raises(ValidationError) as excinfo:
+ BatchSubmitJobResponseSchema().load({"job_id": "job-1"})
+
+ assert "jobId" in excinfo.value.messages
+
+
+class TestBatchJobDetailSchema:
+ def test_load_returns_batch_job_instead_of_dict(self):
+ job = BatchJobDetailSchema().load({"jobId": "job-1", "status":
"RUNNING", "statusReason": "started"})
+
+ assert isinstance(job, BatchJob)
+ assert job.job_id == "job-1"
+ assert job.status == "RUNNING"
+ assert job.status_reason == "started"
+
+ def test_load_without_status_reason_leaves_it_unset(self):
+ job = BatchJobDetailSchema().load({"jobId": "job-1", "status":
"SUCCEEDED"})
+
+ assert job.status_reason is None
+
+ def test_load_ignores_fields_the_describe_jobs_api_adds(self):
+ job = BatchJobDetailSchema().load(
+ {"jobId": "job-1", "status": "FAILED", "jobQueue": "queue-1",
"startedAt": 1700000000}
+ )
+
+ assert isinstance(job, BatchJob)
+ assert job.job_id == "job-1"
+
+ @pytest.mark.parametrize(
+ ("payload", "missing_key"),
+ [
+ pytest.param({"status": "RUNNING"}, "jobId", id="without-job-id"),
+ pytest.param({"jobId": "job-1"}, "status", id="without-status"),
+ ],
+ )
+ def test_load_rejects_detail_missing_a_required_field(self, payload,
missing_key):
+ with pytest.raises(ValidationError) as excinfo:
+ BatchJobDetailSchema().load(payload)
+
+ assert missing_key in excinfo.value.messages
+
+
+class TestBatchDescribeJobsResponseSchema:
+ def test_load_returns_a_batch_job_per_entry(self):
+ loaded = BatchDescribeJobsResponseSchema().load(
+ {
+ "jobs": [
+ {"jobId": "job-1", "status": "RUNNING"},
+ {"jobId": "job-2", "status": "FAILED", "statusReason":
"exit code 1"},
+ ]
+ }
+ )
+
+ assert [(job.job_id, job.status, job.status_reason) for job in
loaded["jobs"]] == [
+ ("job-1", "RUNNING", None),
+ ("job-2", "FAILED", "exit code 1"),
+ ]
+ assert all(isinstance(job, BatchJob) for job in loaded["jobs"])
+
+ def test_load_accepts_an_empty_job_list(self):
+ loaded = BatchDescribeJobsResponseSchema().load({"jobs": []})
+
+ assert loaded == {"jobs": []}
+
+ def test_load_rejects_response_without_jobs(self):
+ with pytest.raises(ValidationError) as excinfo:
+ BatchDescribeJobsResponseSchema().load({})
+
+ assert "jobs" in excinfo.value.messages
+
+ def test_load_propagates_validation_error_from_a_nested_job(self):
+ with pytest.raises(ValidationError) as excinfo:
+ BatchDescribeJobsResponseSchema().load({"jobs": [{"jobId":
"job-1"}]})
+
+ assert "status" in excinfo.value.messages["jobs"][0]
