Add e2e test for XComObjectStorageBackend#62907

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jason810496 merged 8 commits into from
Mar 9, 2026
Merged

Add e2e test for XComObjectStorageBackend #62907

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
0635994
Add e2e test for XComObjectStorageBackend
jason810496 Mar 5, 2026
0600b02
Update GitHub Action and Breeze cmd
jason810496 Mar 5, 2026
1e9a06a
Add aws s3fs deps
jason810496 Mar 5, 2026
2bc77f0
Add debug log
jason810496 Mar 5, 2026
fcd0685
Add required AWS env to fix the e2e test failure
jason810496 Mar 6, 2026
e7865bf
Harden the client in e2e test
jason810496 Mar 6, 2026
b82137f
Fix nits
jason810496 Mar 6, 2026
0011121
Fix copilot suggestion
jason810496 Mar 6, 2026
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
11 changes: 11 additions & 0 deletions .github/workflows/additional-prod-image-tests.yml
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -226,6 +226,17 @@ jobs:
use-uv: ${{ inputs.use-uv }}
e2e_test_mode: "remote_log"

test-e2e-integration-tests-xcom-object-storage:
name: "XCom object storage backend tests with PROD image"
uses: ./.github/workflows/airflow-e2e-tests.yml
with:
workflow-name: "XCom object storage backend e2e test"
runners: ${{ inputs.runners }}
platform: ${{ inputs.platform }}
default-python-version: "${{ inputs.default-python-version }}"
use-uv: ${{ inputs.use-uv }}
e2e_test_mode: "xcom_object_storage"

test-ui-e2e-chromium:
name: "Chromium UI e2e tests with PROD image"
uses: ./.github/workflows/ui-e2e-tests.yml
Expand Down
4 changes: 2 additions & 2 deletions .github/workflows/airflow-e2e-tests.yml
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -49,7 +49,7 @@ on: # yamllint disable-line rule:truthy
type: string
required: true
e2e_test_mode:
description: "Test mode - basicor remote_log"
description: "Test mode - basic, remote_log, or xcom_object_storage"
type: string
default: "basic"

Expand DownExpand Up@@ -80,7 +80,7 @@ on: # yamllint disable-line rule:truthy
type: string
default: ""
e2e_test_mode:
description: "Test mode - quick or full"
description: "Test mode - basic, remote_log, or xcom_object_storage"
type: string
default: "basic"

Expand Down
1 change: 1 addition & 0 deletions airflow-e2e-tests/scripts/init-aws.sh
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -17,4 +17,5 @@
# under the License.

aws --endpoint-url=http://localstack:4566 s3 mb s3://test-airflow-logs
aws --endpoint-url=http://localstack:4566 s3 mb s3://test-xcom-objectstorage-backend
aws --endpoint-url=http://localstack:4566 s3 ls
32 changes: 31 additions & 1 deletion airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -36,6 +36,7 @@
LOCALSTACK_PATH,
LOGS_FOLDER,
TEST_REPORT_FILE,
XCOM_BUCKET,
)

from tests_common.test_utils.fernet import generate_fernet_key_string
Expand All@@ -48,13 +49,18 @@ class _E2ETestState:
airflow_logs_path: Path | None = None


def _setup_s3_integration(dot_env_file, tmp_dir):
def _copy_localstack_files(tmp_dir):
"""Copy localstack compose file and init script into the temp directory."""
copyfile(LOCALSTACK_PATH, tmp_dir / "localstack.yml")

copyfile(AWS_INIT_PATH, tmp_dir / "init-aws.sh")
current_permissions = os.stat(tmp_dir / "init-aws.sh").st_mode
os.chmod(tmp_dir / "init-aws.sh", current_permissions | 0o111)


def _setup_s3_integration(dot_env_file, tmp_dir):
_copy_localstack_files(tmp_dir)

dot_env_file.write_text(
f"AIRFLOW_UID={os.getuid()}\n"
"AWS_DEFAULT_REGION=us-east-1\n"
Expand All@@ -68,6 +74,27 @@ def _setup_s3_integration(dot_env_file, tmp_dir):
os.environ["ENV_FILE_PATH"] = str(dot_env_file)


def _setup_xcom_object_storage_integration(dot_env_file, tmp_dir):
_copy_localstack_files(tmp_dir)

dot_env_file.write_text(
f"AIRFLOW_UID={os.getuid()}\n"
# XComObjectStorageBackend requires AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY as env vars
# because `universal-path` uses boto3's native S3 client, which relies on environment variables
# for authentication rather than parsing credentials from the connection URI
"AWS_ACCESS_KEY_ID=test\n"
"AWS_SECRET_ACCESS_KEY=test\n"
"AWS_DEFAULT_REGION=us-east-1\n"
"AWS_ENDPOINT_URL_S3=http://localstack:4566\n"
"AIRFLOW_CONN_AWS_DEFAULT=aws://test:test@\n"
"AIRFLOW__CORE__XCOM_BACKEND=airflow.providers.common.io.xcom.backend.XComObjectStorageBackend\n"
f"AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_PATH=s3://aws_default@{XCOM_BUCKET}/xcom\n"
"AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_THRESHOLD=0\n"
"_PIP_ADDITIONAL_REQUIREMENTS=apache-airflow-providers-amazon[s3fs]\n"
)
os.environ["ENV_FILE_PATH"] = str(dot_env_file)


def spin_up_airflow_environment(tmp_path_factory: pytest.TempPathFactory):
tmp_dir = tmp_path_factory.mktemp("airflow-e2e-tests")

Expand DownExpand Up@@ -97,6 +124,9 @@ def spin_up_airflow_environment(tmp_path_factory: pytest.TempPathFactory):
if E2E_TEST_MODE == "remote_log":
compose_file_names.append("localstack.yml")
_setup_s3_integration(dot_env_file, tmp_dir)
elif E2E_TEST_MODE == "xcom_object_storage":
compose_file_names.append("localstack.yml")
_setup_xcom_object_storage_integration(dot_env_file, tmp_dir)

#
# Please Do not use this Fernet key in any deployments! Please generate your own key.
Expand Down
3 changes: 3 additions & 0 deletions airflow-e2e-tests/tests/airflow_e2e_tests/constants.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -42,3 +42,6 @@
LOCALSTACK_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "docker" / "localstack.yml"
E2E_TEST_MODE = os.environ.get("E2E_TEST_MODE", "basic")
AWS_INIT_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "scripts" / "init-aws.sh"

# s3 bucket name for XComObjectStorageBackend tests. This bucket will be created in the `init-aws.sh` script that is run as part of the LocalStack container initialization.
XCOM_BUCKET = "test-xcom-objectstorage-backend"
51 changes: 43 additions & 8 deletions airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -20,6 +20,7 @@
from datetime import datetime, timezone
from functools import cached_property

import boto3
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
Expand All@@ -31,19 +32,41 @@
)


def get_s3_client():
"""Return a boto3 S3 client configured to use the local LocalStack endpoint."""
return boto3.client(
"s3",
endpoint_url="http://localhost:4566",
aws_access_key_id="test",
aws_secret_access_key="test",
region_name="us-east-1",
)


def create_request_session_with_retries(status_forcelist: list[int]):
"""Create a requests Session with retry logic for handling transient errors."""
Retry.DEFAULT_BACKOFF_MAX = 32
retry_strategy = Retry(
total=10,
backoff_factor=1,
status_forcelist=status_forcelist,
)
session = requests.Session()
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session


class AirflowClient:
"""Client for interacting with the Airflow REST API."""

def __init__(self):
self.session = requests.Session()
self.session = create_request_session_with_retries(status_forcelist=[429])

@cached_property
def token(self):
Retry.DEFAULT_BACKOFF_MAX = 32
retry = Retry(total=10, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504])
session = requests.Session()
session.mount("http://", HTTPAdapter(max_retries=retry))
session.mount("https://", HTTPAdapter(max_retries=retry))
session = create_request_session_with_retries(status_forcelist=[429, 500, 502, 503, 504])

api_server_url = DOCKER_COMPOSE_HOST_PORT
if not api_server_url.startswith(("http://", "https://")):
Expand DownExpand Up@@ -121,11 +144,23 @@ def trigger_dag_and_wait(self, dag_id: str, json=None):
run_id=resp["dag_run_id"],
)

def get_task_logs(self, dag_id: str, run_id: str, task_id: str, try_number: int = 1):
def get_task_instances(self, dag_id: str, run_id: str):
"""Get task instances for a given DAG run."""
return self._make_request(
method="GET",
endpoint=f"dags/{dag_id}/dagRuns/{run_id}/taskInstances",
)

def get_task_logs(
self, dag_id: str, run_id: str, task_id: str, try_number: int = 1, map_index: int | None = None
):
"""Get task logs via API."""
endpoint = f"dags/{dag_id}/dagRuns/{run_id}/taskInstances/{task_id}/logs/{try_number}"
if map_index is not None:
endpoint += f"?map_index={map_index}"
return self._make_request(
method="GET",
endpoint=f"dags/{dag_id}/dagRuns/{run_id}/taskInstances/{task_id}/logs/{try_number}",
endpoint=endpoint,
)


Expand Down
12 changes: 3 additions & 9 deletions airflow-e2e-tests/tests/airflow_e2e_tests/remote_log_tests/test_remote_logging.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -19,10 +19,9 @@
import time
from datetime import datetime, timezone

import boto3
import pytest

from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient
from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient, get_s3_client


class TestRemoteLogging:
Expand DownExpand Up@@ -56,15 +55,10 @@ def test_remote_logging_s3(self):

# This bucket will be created part of the docker-compose setup in
bucket_name = "test-airflow-logs"
s3_client = boto3.client(
"s3",
endpoint_url="http://localhost:4566",
aws_access_key_id="test",
aws_secret_access_key="test",
region_name="us-east-1",
)
s3_client = get_s3_client()

# Wait for logs to be available in S3 before we call `get_task_logs`
contents = []
for _ in range(self.max_retries):
response = s3_client.list_objects_v2(Bucket=bucket_name)
contents = response.get("Contents", [])
Expand Down
16 changes: 16 additions & 0 deletions airflow-e2e-tests/tests/airflow_e2e_tests/xcom_object_storage_tests/__init__.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
# 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.
89 changes: 89 additions & 0 deletions ...sts/tests/airflow_e2e_tests/xcom_object_storage_tests/test_xcom_object_storage_backend.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
# 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 time
from datetime import datetime, timezone
from pprint import pprint
from uuid import uuid4

import pytest

from airflow_e2e_tests.constants import XCOM_BUCKET
from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient, get_s3_client


class TestXComObjectStorageBackend:
airflow_client = AirflowClient()
dag_id = "example_xcom_test"
retry_interval_in_seconds = 5
max_retries = 12

def test_dag_succeeds_and_xcom_values_stored_in_s3(self):
"""Test that a DAG using XComObjectStorageBackend completes successfully and persists XCom values to S3."""
self.airflow_client.un_pause_dag(self.dag_id)

trigger_resp = self.airflow_client.trigger_dag(
self.dag_id,
json={
"dag_run_id": f"test_xcom_object_storage_backend_{uuid4()}",
"logical_date": datetime.now(timezone.utc).isoformat(),
},
)
dag_run_id = trigger_resp["dag_run_id"]
state = self.airflow_client.wait_for_dag_run(
dag_id=self.dag_id,
run_id=dag_run_id,
)

# try to get all the logs to help debugging
if state != "success":
task_instances_resp = self.airflow_client.get_task_instances(self.dag_id, dag_run_id)
for task_instance in task_instances_resp["task_instances"]:
task_id = task_instance["task_id"]
try_number = task_instance["try_number"]
try:
print(f"\nLogs for task {task_id} (try {try_number}):")
task_logs_resp = self.airflow_client.get_task_logs(
dag_id=self.dag_id, task_id=task_id, run_id=dag_run_id, try_number=try_number
)
pprint(task_logs_resp)
except Exception as e:
print(f"Could not get logs for task {task_id} (try {try_number}): {e}")

assert state == "success", f"DAG {self.dag_id} did not complete successfully. Final state: {state}"

s3_client = get_s3_client()

contents = []
for _ in range(self.max_retries):
response = s3_client.list_objects_v2(Bucket=XCOM_BUCKET)
contents = response.get("Contents", [])
if contents:
break

print(f"No XCom objects found in S3 bucket {XCOM_BUCKET!r} yet. Retrying...")
time.sleep(self.retry_interval_in_seconds)

if not contents:
pytest.fail(
f"Expected XCom objects in S3 bucket {XCOM_BUCKET!r}, but bucket is empty.\n"
f"List Objects Response: {response}"
)

keys = [obj["Key"] for obj in contents]
print(f"Found {len(keys)} XCom object(s) in S3: {keys}")
Loading
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { // Add copy buttons to all 
 blocks
(function() {
function addCopyButtons() {
document.querySelectorAll('pre code').forEach(function(codeBlock) {
if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;
codeBlock.parentElement.setAttribute('data-copy-added', 'true');
var btn = document.createElement('button');
btn.textContent = 'Copy';
btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';
btn.onmouseover = function() { this.style.opacity = '1'; };
btn.onmouseout = function() { this.style.opacity = '0.7'; };
btn.onclick = function() {
navigator.clipboard.writeText(codeBlock.textContent).then(function() {
btn.textContent = 'Copied!';
setTimeout(function() { btn.textContent = 'Copy'; }, 1500);
});
};
codeBlock.parentElement.style.position = 'relative';
codeBlock.parentElement.appendChild(btn);
});
}
addCopyButtons();
// Re-run on dynamic content
var observer = new MutationObserver(addCopyButtons);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '

Skip to content
Add e2e test for XComObjectStorageBackend#62907

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

jason810496
merged 8 commits into

from

Mar 9, 2026
 Merged


Add e2e test for XComObjectStorageBackend
#62907
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


8 commits


Select commit
Hold shift + click to select a range
0635994

Add e2e test for XComObjectStorageBackend


jason810496 Mar 5, 2026
0600b02

Update GitHub Action and Breeze cmd


jason810496 Mar 5, 2026
1e9a06a

Add aws s3fs deps


jason810496 Mar 5, 2026
2bc77f0

Add debug log


jason810496 Mar 5, 2026
fcd0685

Add required AWS env to fix the e2e test failure


jason810496 Mar 6, 2026
e7865bf

Harden the client in e2e test


jason810496 Mar 6, 2026
b82137f

Fix nits


jason810496 Mar 6, 2026
0011121

Fix copilot suggestion


jason810496 Mar 6, 2026

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


11 changes: 11 additions & 0 deletions
.github/workflows/additional-prod-image-tests.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -226,6 +226,17 @@ jobs:
 use-uv: ${{ inputs.use-uv }}
 e2e_test_mode: "remote_log"

 test-e2e-integration-tests-xcom-object-storage:
 name: "XCom object storage backend tests with PROD image"
 uses: ./.github/workflows/airflow-e2e-tests.yml
 with:
 workflow-name: "XCom object storage backend e2e test"
 runners: ${{ inputs.runners }}
 platform: ${{ inputs.platform }}
 default-python-version: "${{ inputs.default-python-version }}"
 use-uv: ${{ inputs.use-uv }}
 e2e_test_mode: "xcom_object_storage"

 test-ui-e2e-chromium:
 name: "Chromium UI e2e tests with PROD image"
 uses: ./.github/workflows/ui-e2e-tests.yml
Expand Down

4 changes: 2 additions & 2 deletions
.github/workflows/airflow-e2e-tests.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -49,7 +49,7 @@ on: # yamllint disable-line rule:truthy
 type: string
 required: true
 e2e_test_mode:
 description: "Test mode - basicor remote_log"
 description: "Test mode - basic, remote_log, or xcom_object_storage"
 type: string
 default: "basic"

Expand DownExpand Up@@ -80,7 +80,7 @@ on: # yamllint disable-line rule:truthy
 type: string
 default: ""
 e2e_test_mode:
 description: "Test mode - quick or full"
 description: "Test mode - basic, remote_log, or xcom_object_storage"
 type: string
 default: "basic"

Expand Down

1 change: 1 addition & 0 deletions
airflow-e2e-tests/scripts/init-aws.sh

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,4 +17,5 @@
# under the License.

aws --endpoint-url=http://localstack:4566 s3 mb s3://test-airflow-logs
aws --endpoint-url=http://localstack:4566 s3 mb s3://test-xcom-objectstorage-backend
aws --endpoint-url=http://localstack:4566 s3 ls

32 changes: 31 additions & 1 deletion
airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -36,6 +36,7 @@
 LOCALSTACK_PATH,
 LOGS_FOLDER,
 TEST_REPORT_FILE,
 XCOM_BUCKET,
)

from tests_common.test_utils.fernet import generate_fernet_key_string
Expand All@@ -48,13 +49,18 @@ class _E2ETestState:
 airflow_logs_path: Path | None = None


def _setup_s3_integration(dot_env_file, tmp_dir):
def _copy_localstack_files(tmp_dir):
 """Copy localstack compose file and init script into the temp directory."""
 copyfile(LOCALSTACK_PATH, tmp_dir / "localstack.yml")

 copyfile(AWS_INIT_PATH, tmp_dir / "init-aws.sh")
 current_permissions = os.stat(tmp_dir / "init-aws.sh").st_mode
 os.chmod(tmp_dir / "init-aws.sh", current_permissions | 0o111)


def _setup_s3_integration(dot_env_file, tmp_dir):
 _copy_localstack_files(tmp_dir)

 dot_env_file.write_text(
 f"AIRFLOW_UID={os.getuid()}\n"
 "AWS_DEFAULT_REGION=us-east-1\n"
Expand All@@ -68,6 +74,27 @@ def _setup_s3_integration(dot_env_file, tmp_dir):
 os.environ["ENV_FILE_PATH"] = str(dot_env_file)


def _setup_xcom_object_storage_integration(dot_env_file, tmp_dir):
 _copy_localstack_files(tmp_dir)

 dot_env_file.write_text(
 f"AIRFLOW_UID={os.getuid()}\n"
 # XComObjectStorageBackend requires AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY as env vars
 # because `universal-path` uses boto3's native S3 client, which relies on environment variables
 # for authentication rather than parsing credentials from the connection URI
 "AWS_ACCESS_KEY_ID=test\n"
 "AWS_SECRET_ACCESS_KEY=test\n"
 "AWS_DEFAULT_REGION=us-east-1\n"
 "AWS_ENDPOINT_URL_S3=http://localstack:4566\n"
 "AIRFLOW_CONN_AWS_DEFAULT=aws://test:test@\n"
 "AIRFLOW__CORE__XCOM_BACKEND=airflow.providers.common.io.xcom.backend.XComObjectStorageBackend\n"
 f"AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_PATH=s3://aws_default@{XCOM_BUCKET}/xcom\n"
 "AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_THRESHOLD=0\n"
 "_PIP_ADDITIONAL_REQUIREMENTS=apache-airflow-providers-amazon[s3fs]\n"
 )
 os.environ["ENV_FILE_PATH"] = str(dot_env_file)


def spin_up_airflow_environment(tmp_path_factory: pytest.TempPathFactory):
 tmp_dir = tmp_path_factory.mktemp("airflow-e2e-tests")

Expand DownExpand Up@@ -97,6 +124,9 @@ def spin_up_airflow_environment(tmp_path_factory: pytest.TempPathFactory):
 if E2E_TEST_MODE == "remote_log":
 compose_file_names.append("localstack.yml")
 _setup_s3_integration(dot_env_file, tmp_dir)
 elif E2E_TEST_MODE == "xcom_object_storage":
 compose_file_names.append("localstack.yml")
 _setup_xcom_object_storage_integration(dot_env_file, tmp_dir)

 #
 # Please Do not use this Fernet key in any deployments! Please generate your own key.
Expand Down

3 changes: 3 additions & 0 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/constants.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -42,3 +42,6 @@
LOCALSTACK_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "docker" / "localstack.yml"
E2E_TEST_MODE = os.environ.get("E2E_TEST_MODE", "basic")
AWS_INIT_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "scripts" / "init-aws.sh"

# s3 bucket name for XComObjectStorageBackend tests. This bucket will be created in the `init-aws.sh` script that is run as part of the LocalStack container initialization.
XCOM_BUCKET = "test-xcom-objectstorage-backend"

51 changes: 43 additions & 8 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -20,6 +20,7 @@
from datetime import datetime, timezone
from functools import cached_property

import boto3
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
Expand All@@ -31,19 +32,41 @@
)


def get_s3_client():
 """Return a boto3 S3 client configured to use the local LocalStack endpoint."""
 return boto3.client(
 "s3",
 endpoint_url="http://localhost:4566",
 aws_access_key_id="test",
 aws_secret_access_key="test",
 region_name="us-east-1",
 )


def create_request_session_with_retries(status_forcelist: list[int]):
 """Create a requests Session with retry logic for handling transient errors."""
 Retry.DEFAULT_BACKOFF_MAX = 32
 retry_strategy = Retry(
 total=10,
 backoff_factor=1,
 status_forcelist=status_forcelist,
 )
 session = requests.Session()
 adapter = HTTPAdapter(max_retries=retry_strategy)
 session.mount("http://", adapter)
 session.mount("https://", adapter)
 return session


class AirflowClient:
 """Client for interacting with the Airflow REST API."""

 def __init__(self):
 self.session = requests.Session()
 self.session = create_request_session_with_retries(status_forcelist=[429])

 @cached_property
 def token(self):
 Retry.DEFAULT_BACKOFF_MAX = 32
 retry = Retry(total=10, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504])
 session = requests.Session()
 session.mount("http://", HTTPAdapter(max_retries=retry))
 session.mount("https://", HTTPAdapter(max_retries=retry))
 session = create_request_session_with_retries(status_forcelist=[429, 500, 502, 503, 504])

 api_server_url = DOCKER_COMPOSE_HOST_PORT
 if not api_server_url.startswith(("http://", "https://")):
Expand DownExpand Up@@ -121,11 +144,23 @@ def trigger_dag_and_wait(self, dag_id: str, json=None):
 run_id=resp["dag_run_id"],
 )

 def get_task_logs(self, dag_id: str, run_id: str, task_id: str, try_number: int = 1):
 def get_task_instances(self, dag_id: str, run_id: str):
 """Get task instances for a given DAG run."""
 return self._make_request(
 method="GET",
 endpoint=f"dags/{dag_id}/dagRuns/{run_id}/taskInstances",
 )

 def get_task_logs(
 self, dag_id: str, run_id: str, task_id: str, try_number: int = 1, map_index: int | None = None
 ):
 """Get task logs via API."""
 endpoint = f"dags/{dag_id}/dagRuns/{run_id}/taskInstances/{task_id}/logs/{try_number}"
 if map_index is not None:
 endpoint += f"?map_index={map_index}"
 return self._make_request(
 method="GET",
 endpoint=f"dags/{dag_id}/dagRuns/{run_id}/taskInstances/{task_id}/logs/{try_number}",
 endpoint=endpoint,
 )


Expand Down

12 changes: 3 additions & 9 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/remote_log_tests/test_remote_logging.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -19,10 +19,9 @@
import time
from datetime import datetime, timezone

import boto3
import pytest

from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient
from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient, get_s3_client


class TestRemoteLogging:
Expand DownExpand Up@@ -56,15 +55,10 @@ def test_remote_logging_s3(self):

 # This bucket will be created part of the docker-compose setup in
 bucket_name = "test-airflow-logs"
 s3_client = boto3.client(
 "s3",
 endpoint_url="http://localhost:4566",
 aws_access_key_id="test",
 aws_secret_access_key="test",
 region_name="us-east-1",
 )
 s3_client = get_s3_client()

 # Wait for logs to be available in S3 before we call `get_task_logs`
 contents = []
 for _ in range(self.max_retries):
 response = s3_client.list_objects_v2(Bucket=bucket_name)
 contents = response.get("Contents", [])
Expand Down

16 changes: 16 additions & 0 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/xcom_object_storage_tests/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
# 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.

89 changes: 89 additions & 0 deletions
...sts/tests/airflow_e2e_tests/xcom_object_storage_tests/test_xcom_object_storage_backend.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
# 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 time
from datetime import datetime, timezone
from pprint import pprint
from uuid import uuid4

import pytest

from airflow_e2e_tests.constants import XCOM_BUCKET
from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient, get_s3_client


class TestXComObjectStorageBackend:
 airflow_client = AirflowClient()
 dag_id = "example_xcom_test"
 retry_interval_in_seconds = 5
 max_retries = 12

 def test_dag_succeeds_and_xcom_values_stored_in_s3(self):
 """Test that a DAG using XComObjectStorageBackend completes successfully and persists XCom values to S3."""
 self.airflow_client.un_pause_dag(self.dag_id)

 trigger_resp = self.airflow_client.trigger_dag(
 self.dag_id,
 json={
 "dag_run_id": f"test_xcom_object_storage_backend_{uuid4()}",
 "logical_date": datetime.now(timezone.utc).isoformat(),
 },
 )
 dag_run_id = trigger_resp["dag_run_id"]
 state = self.airflow_client.wait_for_dag_run(
 dag_id=self.dag_id,
 run_id=dag_run_id,
 )

 # try to get all the logs to help debugging
 if state != "success":
 task_instances_resp = self.airflow_client.get_task_instances(self.dag_id, dag_run_id)
 for task_instance in task_instances_resp["task_instances"]:
 task_id = task_instance["task_id"]
 try_number = task_instance["try_number"]
 try:
 print(f"\nLogs for task {task_id} (try {try_number}):")
 task_logs_resp = self.airflow_client.get_task_logs(
 dag_id=self.dag_id, task_id=task_id, run_id=dag_run_id, try_number=try_number
 )
 pprint(task_logs_resp)
 except Exception as e:
 print(f"Could not get logs for task {task_id} (try {try_number}): {e}")

 assert state == "success", f"DAG {self.dag_id} did not complete successfully. Final state: {state}"

 s3_client = get_s3_client()

 contents = []
 for _ in range(self.max_retries):
 response = s3_client.list_objects_v2(Bucket=XCOM_BUCKET)
 contents = response.get("Contents", [])
 if contents:
 break

 print(f"No XCom objects found in S3 bucket {XCOM_BUCKET!r} yet. Retrying...")
 time.sleep(self.retry_interval_in_seconds)

 if not contents:
 pytest.fail(
 f"Expected XCom objects in S3 bucket {XCOM_BUCKET!r}, but bucket is empty.\n"
 f"List Objects Response: {response}"
 )

 keys = [obj["Key"] for obj in contents]
 print(f"Found {len(keys)} XCom object(s) in S3: {keys}")
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Force GitHub README to respect dark mode
(function() {
var style = document.createElement('style');
style.textContent = '
.markdown-body {
color-scheme: dark light;
}
.markdown-body pre { background: #161b22 !important; }
.markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }
.markdown-body table th, .markdown-body table td { border-color: #30363d !important; }
.markdown-body img { background: #0d1117; }
.markdown-body blockquote { border-left-color: #8b949e; }
.markdown-body hr { border-color: #30363d; }
';
document.head.appendChild(style);
})();
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Add e2e test for XComObjectStorageBackend#62907

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

jason810496
merged 8 commits into

from

Mar 9, 2026
 Merged


Add e2e test for XComObjectStorageBackend
#62907
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


8 commits


Select commit
Hold shift + click to select a range
0635994

Add e2e test for XComObjectStorageBackend


jason810496 Mar 5, 2026
0600b02

Update GitHub Action and Breeze cmd


jason810496 Mar 5, 2026
1e9a06a

Add aws s3fs deps


jason810496 Mar 5, 2026
2bc77f0

Add debug log


jason810496 Mar 5, 2026
fcd0685

Add required AWS env to fix the e2e test failure


jason810496 Mar 6, 2026
e7865bf

Harden the client in e2e test


jason810496 Mar 6, 2026
b82137f

Fix nits


jason810496 Mar 6, 2026
0011121

Fix copilot suggestion


jason810496 Mar 6, 2026

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


11 changes: 11 additions & 0 deletions
.github/workflows/additional-prod-image-tests.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -226,6 +226,17 @@ jobs:
 use-uv: ${{ inputs.use-uv }}
 e2e_test_mode: "remote_log"

 test-e2e-integration-tests-xcom-object-storage:
 name: "XCom object storage backend tests with PROD image"
 uses: ./.github/workflows/airflow-e2e-tests.yml
 with:
 workflow-name: "XCom object storage backend e2e test"
 runners: ${{ inputs.runners }}
 platform: ${{ inputs.platform }}
 default-python-version: "${{ inputs.default-python-version }}"
 use-uv: ${{ inputs.use-uv }}
 e2e_test_mode: "xcom_object_storage"

 test-ui-e2e-chromium:
 name: "Chromium UI e2e tests with PROD image"
 uses: ./.github/workflows/ui-e2e-tests.yml
Expand Down

4 changes: 2 additions & 2 deletions
.github/workflows/airflow-e2e-tests.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -49,7 +49,7 @@ on: # yamllint disable-line rule:truthy
 type: string
 required: true
 e2e_test_mode:
 description: "Test mode - basicor remote_log"
 description: "Test mode - basic, remote_log, or xcom_object_storage"
 type: string
 default: "basic"

Expand DownExpand Up@@ -80,7 +80,7 @@ on: # yamllint disable-line rule:truthy
 type: string
 default: ""
 e2e_test_mode:
 description: "Test mode - quick or full"
 description: "Test mode - basic, remote_log, or xcom_object_storage"
 type: string
 default: "basic"

Expand Down

1 change: 1 addition & 0 deletions
airflow-e2e-tests/scripts/init-aws.sh

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,4 +17,5 @@
# under the License.

aws --endpoint-url=http://localstack:4566 s3 mb s3://test-airflow-logs
aws --endpoint-url=http://localstack:4566 s3 mb s3://test-xcom-objectstorage-backend
aws --endpoint-url=http://localstack:4566 s3 ls

32 changes: 31 additions & 1 deletion
airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -36,6 +36,7 @@
 LOCALSTACK_PATH,
 LOGS_FOLDER,
 TEST_REPORT_FILE,
 XCOM_BUCKET,
)

from tests_common.test_utils.fernet import generate_fernet_key_string
Expand All@@ -48,13 +49,18 @@ class _E2ETestState:
 airflow_logs_path: Path | None = None


def _setup_s3_integration(dot_env_file, tmp_dir):
def _copy_localstack_files(tmp_dir):
 """Copy localstack compose file and init script into the temp directory."""
 copyfile(LOCALSTACK_PATH, tmp_dir / "localstack.yml")

 copyfile(AWS_INIT_PATH, tmp_dir / "init-aws.sh")
 current_permissions = os.stat(tmp_dir / "init-aws.sh").st_mode
 os.chmod(tmp_dir / "init-aws.sh", current_permissions | 0o111)


def _setup_s3_integration(dot_env_file, tmp_dir):
 _copy_localstack_files(tmp_dir)

 dot_env_file.write_text(
 f"AIRFLOW_UID={os.getuid()}\n"
 "AWS_DEFAULT_REGION=us-east-1\n"
Expand All@@ -68,6 +74,27 @@ def _setup_s3_integration(dot_env_file, tmp_dir):
 os.environ["ENV_FILE_PATH"] = str(dot_env_file)


def _setup_xcom_object_storage_integration(dot_env_file, tmp_dir):
 _copy_localstack_files(tmp_dir)

 dot_env_file.write_text(
 f"AIRFLOW_UID={os.getuid()}\n"
 # XComObjectStorageBackend requires AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY as env vars
 # because `universal-path` uses boto3's native S3 client, which relies on environment variables
 # for authentication rather than parsing credentials from the connection URI
 "AWS_ACCESS_KEY_ID=test\n"
 "AWS_SECRET_ACCESS_KEY=test\n"
 "AWS_DEFAULT_REGION=us-east-1\n"
 "AWS_ENDPOINT_URL_S3=http://localstack:4566\n"
 "AIRFLOW_CONN_AWS_DEFAULT=aws://test:test@\n"
 "AIRFLOW__CORE__XCOM_BACKEND=airflow.providers.common.io.xcom.backend.XComObjectStorageBackend\n"
 f"AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_PATH=s3://aws_default@{XCOM_BUCKET}/xcom\n"
 "AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_THRESHOLD=0\n"
 "_PIP_ADDITIONAL_REQUIREMENTS=apache-airflow-providers-amazon[s3fs]\n"
 )
 os.environ["ENV_FILE_PATH"] = str(dot_env_file)


def spin_up_airflow_environment(tmp_path_factory: pytest.TempPathFactory):
 tmp_dir = tmp_path_factory.mktemp("airflow-e2e-tests")

Expand DownExpand Up@@ -97,6 +124,9 @@ def spin_up_airflow_environment(tmp_path_factory: pytest.TempPathFactory):
 if E2E_TEST_MODE == "remote_log":
 compose_file_names.append("localstack.yml")
 _setup_s3_integration(dot_env_file, tmp_dir)
 elif E2E_TEST_MODE == "xcom_object_storage":
 compose_file_names.append("localstack.yml")
 _setup_xcom_object_storage_integration(dot_env_file, tmp_dir)

 #
 # Please Do not use this Fernet key in any deployments! Please generate your own key.
Expand Down

3 changes: 3 additions & 0 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/constants.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -42,3 +42,6 @@
LOCALSTACK_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "docker" / "localstack.yml"
E2E_TEST_MODE = os.environ.get("E2E_TEST_MODE", "basic")
AWS_INIT_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "scripts" / "init-aws.sh"

# s3 bucket name for XComObjectStorageBackend tests. This bucket will be created in the `init-aws.sh` script that is run as part of the LocalStack container initialization.
XCOM_BUCKET = "test-xcom-objectstorage-backend"

51 changes: 43 additions & 8 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -20,6 +20,7 @@
from datetime import datetime, timezone
from functools import cached_property

import boto3
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
Expand All@@ -31,19 +32,41 @@
)


def get_s3_client():
 """Return a boto3 S3 client configured to use the local LocalStack endpoint."""
 return boto3.client(
 "s3",
 endpoint_url="http://localhost:4566",
 aws_access_key_id="test",
 aws_secret_access_key="test",
 region_name="us-east-1",
 )


def create_request_session_with_retries(status_forcelist: list[int]):
 """Create a requests Session with retry logic for handling transient errors."""
 Retry.DEFAULT_BACKOFF_MAX = 32
 retry_strategy = Retry(
 total=10,
 backoff_factor=1,
 status_forcelist=status_forcelist,
 )
 session = requests.Session()
 adapter = HTTPAdapter(max_retries=retry_strategy)
 session.mount("http://", adapter)
 session.mount("https://", adapter)
 return session


class AirflowClient:
 """Client for interacting with the Airflow REST API."""

 def __init__(self):
 self.session = requests.Session()
 self.session = create_request_session_with_retries(status_forcelist=[429])

 @cached_property
 def token(self):
 Retry.DEFAULT_BACKOFF_MAX = 32
 retry = Retry(total=10, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504])
 session = requests.Session()
 session.mount("http://", HTTPAdapter(max_retries=retry))
 session.mount("https://", HTTPAdapter(max_retries=retry))
 session = create_request_session_with_retries(status_forcelist=[429, 500, 502, 503, 504])

 api_server_url = DOCKER_COMPOSE_HOST_PORT
 if not api_server_url.startswith(("http://", "https://")):
Expand DownExpand Up@@ -121,11 +144,23 @@ def trigger_dag_and_wait(self, dag_id: str, json=None):
 run_id=resp["dag_run_id"],
 )

 def get_task_logs(self, dag_id: str, run_id: str, task_id: str, try_number: int = 1):
 def get_task_instances(self, dag_id: str, run_id: str):
 """Get task instances for a given DAG run."""
 return self._make_request(
 method="GET",
 endpoint=f"dags/{dag_id}/dagRuns/{run_id}/taskInstances",
 )

 def get_task_logs(
 self, dag_id: str, run_id: str, task_id: str, try_number: int = 1, map_index: int | None = None
 ):
 """Get task logs via API."""
 endpoint = f"dags/{dag_id}/dagRuns/{run_id}/taskInstances/{task_id}/logs/{try_number}"
 if map_index is not None:
 endpoint += f"?map_index={map_index}"
 return self._make_request(
 method="GET",
 endpoint=f"dags/{dag_id}/dagRuns/{run_id}/taskInstances/{task_id}/logs/{try_number}",
 endpoint=endpoint,
 )


Expand Down

12 changes: 3 additions & 9 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/remote_log_tests/test_remote_logging.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -19,10 +19,9 @@
import time
from datetime import datetime, timezone

import boto3
import pytest

from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient
from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient, get_s3_client


class TestRemoteLogging:
Expand DownExpand Up@@ -56,15 +55,10 @@ def test_remote_logging_s3(self):

 # This bucket will be created part of the docker-compose setup in
 bucket_name = "test-airflow-logs"
 s3_client = boto3.client(
 "s3",
 endpoint_url="http://localhost:4566",
 aws_access_key_id="test",
 aws_secret_access_key="test",
 region_name="us-east-1",
 )
 s3_client = get_s3_client()

 # Wait for logs to be available in S3 before we call `get_task_logs`
 contents = []
 for _ in range(self.max_retries):
 response = s3_client.list_objects_v2(Bucket=bucket_name)
 contents = response.get("Contents", [])
Expand Down

16 changes: 16 additions & 0 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/xcom_object_storage_tests/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
# 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.

89 changes: 89 additions & 0 deletions
...sts/tests/airflow_e2e_tests/xcom_object_storage_tests/test_xcom_object_storage_backend.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
# 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 time
from datetime import datetime, timezone
from pprint import pprint
from uuid import uuid4

import pytest

from airflow_e2e_tests.constants import XCOM_BUCKET
from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient, get_s3_client


class TestXComObjectStorageBackend:
 airflow_client = AirflowClient()
 dag_id = "example_xcom_test"
 retry_interval_in_seconds = 5
 max_retries = 12

 def test_dag_succeeds_and_xcom_values_stored_in_s3(self):
 """Test that a DAG using XComObjectStorageBackend completes successfully and persists XCom values to S3."""
 self.airflow_client.un_pause_dag(self.dag_id)

 trigger_resp = self.airflow_client.trigger_dag(
 self.dag_id,
 json={
 "dag_run_id": f"test_xcom_object_storage_backend_{uuid4()}",
 "logical_date": datetime.now(timezone.utc).isoformat(),
 },
 )
 dag_run_id = trigger_resp["dag_run_id"]
 state = self.airflow_client.wait_for_dag_run(
 dag_id=self.dag_id,
 run_id=dag_run_id,
 )

 # try to get all the logs to help debugging
 if state != "success":
 task_instances_resp = self.airflow_client.get_task_instances(self.dag_id, dag_run_id)
 for task_instance in task_instances_resp["task_instances"]:
 task_id = task_instance["task_id"]
 try_number = task_instance["try_number"]
 try:
 print(f"\nLogs for task {task_id} (try {try_number}):")
 task_logs_resp = self.airflow_client.get_task_logs(
 dag_id=self.dag_id, task_id=task_id, run_id=dag_run_id, try_number=try_number
 )
 pprint(task_logs_resp)
 except Exception as e:
 print(f"Could not get logs for task {task_id} (try {try_number}): {e}")

 assert state == "success", f"DAG {self.dag_id} did not complete successfully. Final state: {state}"

 s3_client = get_s3_client()

 contents = []
 for _ in range(self.max_retries):
 response = s3_client.list_objects_v2(Bucket=XCOM_BUCKET)
 contents = response.get("Contents", [])
 if contents:
 break

 print(f"No XCom objects found in S3 bucket {XCOM_BUCKET!r} yet. Retrying...")
 time.sleep(self.retry_interval_in_seconds)

 if not contents:
 pytest.fail(
 f"Expected XCom objects in S3 bucket {XCOM_BUCKET!r}, but bucket is empty.\n"
 f"List Objects Response: {response}"
 )

 keys = [obj["Key"] for obj in contents]
 print(f"Found {len(keys)} XCom object(s) in S3: {keys}")
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Highlight search terms from Google/DuckDuckGo/Bing referrer
(function() {
var ref = document.referrer;
var terms = [];
if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {
var url = new URL(ref);
var q = url.searchParams.get('q') || url.searchParams.get('p');
if (q) {
terms = q.split(/\s+/).filter(function(t) { return t.length > 2; });
}
}
if (terms.length === 0) return;
var style = document.createElement('style');
style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';
document.head.appendChild(style);
function highlight(node) {
if (node.nodeType === 3) { // text node
var text = node.textContent;
var found = false;
terms.forEach(function(term) {
var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\]\\]/g, '\\') + ')', 'gi');
if (regex.test(text)) {
found = true;
var frag = document.createDocumentFragment();
var parts = text.split(regex);
parts.forEach(function(part, i) {
if (i % 2 === 0) {
frag.appendChild(document.createTextNode(part));
} else {
var span = document.createElement('span');
span.className = 'userscript-highlight';
span.textContent = part;
frag.appendChild(span);
}
});
node.parentNode.replaceChild(frag, node);
}
});
} else if (node.nodeType === 1 && node.childNodes) { // element
var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];
if (!skipTags.includes(node.tagName)) {
Array.from(node.childNodes).forEach(highlight);
}
}
}
highlight(document.body);
// Re-highlight on dynamic content
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1 || node.nodeType === 3) highlight(node);
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Add e2e test for XComObjectStorageBackend#62907

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

jason810496
merged 8 commits into

from

Mar 9, 2026
 Merged


Add e2e test for XComObjectStorageBackend
#62907
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


8 commits


Select commit
Hold shift + click to select a range
0635994

Add e2e test for XComObjectStorageBackend


jason810496 Mar 5, 2026
0600b02

Update GitHub Action and Breeze cmd


jason810496 Mar 5, 2026
1e9a06a

Add aws s3fs deps


jason810496 Mar 5, 2026
2bc77f0

Add debug log


jason810496 Mar 5, 2026
fcd0685

Add required AWS env to fix the e2e test failure


jason810496 Mar 6, 2026
e7865bf

Harden the client in e2e test


jason810496 Mar 6, 2026
b82137f

Fix nits


jason810496 Mar 6, 2026
0011121

Fix copilot suggestion


jason810496 Mar 6, 2026

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


11 changes: 11 additions & 0 deletions
.github/workflows/additional-prod-image-tests.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -226,6 +226,17 @@ jobs:
 use-uv: ${{ inputs.use-uv }}
 e2e_test_mode: "remote_log"

 test-e2e-integration-tests-xcom-object-storage:
 name: "XCom object storage backend tests with PROD image"
 uses: ./.github/workflows/airflow-e2e-tests.yml
 with:
 workflow-name: "XCom object storage backend e2e test"
 runners: ${{ inputs.runners }}
 platform: ${{ inputs.platform }}
 default-python-version: "${{ inputs.default-python-version }}"
 use-uv: ${{ inputs.use-uv }}
 e2e_test_mode: "xcom_object_storage"

 test-ui-e2e-chromium:
 name: "Chromium UI e2e tests with PROD image"
 uses: ./.github/workflows/ui-e2e-tests.yml
Expand Down

4 changes: 2 additions & 2 deletions
.github/workflows/airflow-e2e-tests.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -49,7 +49,7 @@ on: # yamllint disable-line rule:truthy
 type: string
 required: true
 e2e_test_mode:
 description: "Test mode - basicor remote_log"
 description: "Test mode - basic, remote_log, or xcom_object_storage"
 type: string
 default: "basic"

Expand DownExpand Up@@ -80,7 +80,7 @@ on: # yamllint disable-line rule:truthy
 type: string
 default: ""
 e2e_test_mode:
 description: "Test mode - quick or full"
 description: "Test mode - basic, remote_log, or xcom_object_storage"
 type: string
 default: "basic"

Expand Down

1 change: 1 addition & 0 deletions
airflow-e2e-tests/scripts/init-aws.sh

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,4 +17,5 @@
# under the License.

aws --endpoint-url=http://localstack:4566 s3 mb s3://test-airflow-logs
aws --endpoint-url=http://localstack:4566 s3 mb s3://test-xcom-objectstorage-backend
aws --endpoint-url=http://localstack:4566 s3 ls

32 changes: 31 additions & 1 deletion
airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -36,6 +36,7 @@
 LOCALSTACK_PATH,
 LOGS_FOLDER,
 TEST_REPORT_FILE,
 XCOM_BUCKET,
)

from tests_common.test_utils.fernet import generate_fernet_key_string
Expand All@@ -48,13 +49,18 @@ class _E2ETestState:
 airflow_logs_path: Path | None = None


def _setup_s3_integration(dot_env_file, tmp_dir):
def _copy_localstack_files(tmp_dir):
 """Copy localstack compose file and init script into the temp directory."""
 copyfile(LOCALSTACK_PATH, tmp_dir / "localstack.yml")

 copyfile(AWS_INIT_PATH, tmp_dir / "init-aws.sh")
 current_permissions = os.stat(tmp_dir / "init-aws.sh").st_mode
 os.chmod(tmp_dir / "init-aws.sh", current_permissions | 0o111)


def _setup_s3_integration(dot_env_file, tmp_dir):
 _copy_localstack_files(tmp_dir)

 dot_env_file.write_text(
 f"AIRFLOW_UID={os.getuid()}\n"
 "AWS_DEFAULT_REGION=us-east-1\n"
Expand All@@ -68,6 +74,27 @@ def _setup_s3_integration(dot_env_file, tmp_dir):
 os.environ["ENV_FILE_PATH"] = str(dot_env_file)


def _setup_xcom_object_storage_integration(dot_env_file, tmp_dir):
 _copy_localstack_files(tmp_dir)

 dot_env_file.write_text(
 f"AIRFLOW_UID={os.getuid()}\n"
 # XComObjectStorageBackend requires AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY as env vars
 # because `universal-path` uses boto3's native S3 client, which relies on environment variables
 # for authentication rather than parsing credentials from the connection URI
 "AWS_ACCESS_KEY_ID=test\n"
 "AWS_SECRET_ACCESS_KEY=test\n"
 "AWS_DEFAULT_REGION=us-east-1\n"
 "AWS_ENDPOINT_URL_S3=http://localstack:4566\n"
 "AIRFLOW_CONN_AWS_DEFAULT=aws://test:test@\n"
 "AIRFLOW__CORE__XCOM_BACKEND=airflow.providers.common.io.xcom.backend.XComObjectStorageBackend\n"
 f"AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_PATH=s3://aws_default@{XCOM_BUCKET}/xcom\n"
 "AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_THRESHOLD=0\n"
 "_PIP_ADDITIONAL_REQUIREMENTS=apache-airflow-providers-amazon[s3fs]\n"
 )
 os.environ["ENV_FILE_PATH"] = str(dot_env_file)


def spin_up_airflow_environment(tmp_path_factory: pytest.TempPathFactory):
 tmp_dir = tmp_path_factory.mktemp("airflow-e2e-tests")

Expand DownExpand Up@@ -97,6 +124,9 @@ def spin_up_airflow_environment(tmp_path_factory: pytest.TempPathFactory):
 if E2E_TEST_MODE == "remote_log":
 compose_file_names.append("localstack.yml")
 _setup_s3_integration(dot_env_file, tmp_dir)
 elif E2E_TEST_MODE == "xcom_object_storage":
 compose_file_names.append("localstack.yml")
 _setup_xcom_object_storage_integration(dot_env_file, tmp_dir)

 #
 # Please Do not use this Fernet key in any deployments! Please generate your own key.
Expand Down

3 changes: 3 additions & 0 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/constants.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -42,3 +42,6 @@
LOCALSTACK_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "docker" / "localstack.yml"
E2E_TEST_MODE = os.environ.get("E2E_TEST_MODE", "basic")
AWS_INIT_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "scripts" / "init-aws.sh"

# s3 bucket name for XComObjectStorageBackend tests. This bucket will be created in the `init-aws.sh` script that is run as part of the LocalStack container initialization.
XCOM_BUCKET = "test-xcom-objectstorage-backend"

51 changes: 43 additions & 8 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -20,6 +20,7 @@
from datetime import datetime, timezone
from functools import cached_property

import boto3
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
Expand All@@ -31,19 +32,41 @@
)


def get_s3_client():
 """Return a boto3 S3 client configured to use the local LocalStack endpoint."""
 return boto3.client(
 "s3",
 endpoint_url="http://localhost:4566",
 aws_access_key_id="test",
 aws_secret_access_key="test",
 region_name="us-east-1",
 )


def create_request_session_with_retries(status_forcelist: list[int]):
 """Create a requests Session with retry logic for handling transient errors."""
 Retry.DEFAULT_BACKOFF_MAX = 32
 retry_strategy = Retry(
 total=10,
 backoff_factor=1,
 status_forcelist=status_forcelist,
 )
 session = requests.Session()
 adapter = HTTPAdapter(max_retries=retry_strategy)
 session.mount("http://", adapter)
 session.mount("https://", adapter)
 return session


class AirflowClient:
 """Client for interacting with the Airflow REST API."""

 def __init__(self):
 self.session = requests.Session()
 self.session = create_request_session_with_retries(status_forcelist=[429])

 @cached_property
 def token(self):
 Retry.DEFAULT_BACKOFF_MAX = 32
 retry = Retry(total=10, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504])
 session = requests.Session()
 session.mount("http://", HTTPAdapter(max_retries=retry))
 session.mount("https://", HTTPAdapter(max_retries=retry))
 session = create_request_session_with_retries(status_forcelist=[429, 500, 502, 503, 504])

 api_server_url = DOCKER_COMPOSE_HOST_PORT
 if not api_server_url.startswith(("http://", "https://")):
Expand DownExpand Up@@ -121,11 +144,23 @@ def trigger_dag_and_wait(self, dag_id: str, json=None):
 run_id=resp["dag_run_id"],
 )

 def get_task_logs(self, dag_id: str, run_id: str, task_id: str, try_number: int = 1):
 def get_task_instances(self, dag_id: str, run_id: str):
 """Get task instances for a given DAG run."""
 return self._make_request(
 method="GET",
 endpoint=f"dags/{dag_id}/dagRuns/{run_id}/taskInstances",
 )

 def get_task_logs(
 self, dag_id: str, run_id: str, task_id: str, try_number: int = 1, map_index: int | None = None
 ):
 """Get task logs via API."""
 endpoint = f"dags/{dag_id}/dagRuns/{run_id}/taskInstances/{task_id}/logs/{try_number}"
 if map_index is not None:
 endpoint += f"?map_index={map_index}"
 return self._make_request(
 method="GET",
 endpoint=f"dags/{dag_id}/dagRuns/{run_id}/taskInstances/{task_id}/logs/{try_number}",
 endpoint=endpoint,
 )


Expand Down

12 changes: 3 additions & 9 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/remote_log_tests/test_remote_logging.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -19,10 +19,9 @@
import time
from datetime import datetime, timezone

import boto3
import pytest

from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient
from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient, get_s3_client


class TestRemoteLogging:
Expand DownExpand Up@@ -56,15 +55,10 @@ def test_remote_logging_s3(self):

 # This bucket will be created part of the docker-compose setup in
 bucket_name = "test-airflow-logs"
 s3_client = boto3.client(
 "s3",
 endpoint_url="http://localhost:4566",
 aws_access_key_id="test",
 aws_secret_access_key="test",
 region_name="us-east-1",
 )
 s3_client = get_s3_client()

 # Wait for logs to be available in S3 before we call `get_task_logs`
 contents = []
 for _ in range(self.max_retries):
 response = s3_client.list_objects_v2(Bucket=bucket_name)
 contents = response.get("Contents", [])
Expand Down

16 changes: 16 additions & 0 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/xcom_object_storage_tests/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
# 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.

89 changes: 89 additions & 0 deletions
...sts/tests/airflow_e2e_tests/xcom_object_storage_tests/test_xcom_object_storage_backend.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
# 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 time
from datetime import datetime, timezone
from pprint import pprint
from uuid import uuid4

import pytest

from airflow_e2e_tests.constants import XCOM_BUCKET
from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient, get_s3_client


class TestXComObjectStorageBackend:
 airflow_client = AirflowClient()
 dag_id = "example_xcom_test"
 retry_interval_in_seconds = 5
 max_retries = 12

 def test_dag_succeeds_and_xcom_values_stored_in_s3(self):
 """Test that a DAG using XComObjectStorageBackend completes successfully and persists XCom values to S3."""
 self.airflow_client.un_pause_dag(self.dag_id)

 trigger_resp = self.airflow_client.trigger_dag(
 self.dag_id,
 json={
 "dag_run_id": f"test_xcom_object_storage_backend_{uuid4()}",
 "logical_date": datetime.now(timezone.utc).isoformat(),
 },
 )
 dag_run_id = trigger_resp["dag_run_id"]
 state = self.airflow_client.wait_for_dag_run(
 dag_id=self.dag_id,
 run_id=dag_run_id,
 )

 # try to get all the logs to help debugging
 if state != "success":
 task_instances_resp = self.airflow_client.get_task_instances(self.dag_id, dag_run_id)
 for task_instance in task_instances_resp["task_instances"]:
 task_id = task_instance["task_id"]
 try_number = task_instance["try_number"]
 try:
 print(f"\nLogs for task {task_id} (try {try_number}):")
 task_logs_resp = self.airflow_client.get_task_logs(
 dag_id=self.dag_id, task_id=task_id, run_id=dag_run_id, try_number=try_number
 )
 pprint(task_logs_resp)
 except Exception as e:
 print(f"Could not get logs for task {task_id} (try {try_number}): {e}")

 assert state == "success", f"DAG {self.dag_id} did not complete successfully. Final state: {state}"

 s3_client = get_s3_client()

 contents = []
 for _ in range(self.max_retries):
 response = s3_client.list_objects_v2(Bucket=XCOM_BUCKET)
 contents = response.get("Contents", [])
 if contents:
 break

 print(f"No XCom objects found in S3 bucket {XCOM_BUCKET!r} yet. Retrying...")
 time.sleep(self.retry_interval_in_seconds)

 if not contents:
 pytest.fail(
 f"Expected XCom objects in S3 bucket {XCOM_BUCKET!r}, but bucket is empty.\n"
 f"List Objects Response: {response}"
 )

 keys = [obj["Key"] for obj in contents]
 print(f"Found {len(keys)} XCom object(s) in S3: {keys}")
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Strip utm_, fbclid, gclid, etc. from all links on page
(function() {
var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',
'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',
'ref', 'ref_src', 'source', 'medium', 'campaign'];
function cleanUrl(url) {
try {
var u = new URL(url, window.location.origin);
var changed = false;
trackingParams.forEach(function(p) {
if (u.searchParams.has(p)) {
u.searchParams.delete(p);
changed = true;
}
});
return changed ? u.toString() : url;
} catch (e) {
return url;
}
}
function cleanLinks() {
document.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
cleanLinks();
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1) {
if (node.tagName === 'A') cleanLinks();
node.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); }
})();
(function(){
try {
var __m = "youtube.com";
var __re = new RegExp('^' + "youtube\\.com" + '

Skip to content
Add e2e test for XComObjectStorageBackend#62907

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

jason810496
merged 8 commits into

from

Mar 9, 2026
 Merged


Add e2e test for XComObjectStorageBackend
#62907
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


8 commits


Select commit
Hold shift + click to select a range
0635994

Add e2e test for XComObjectStorageBackend


jason810496 Mar 5, 2026
0600b02

Update GitHub Action and Breeze cmd


jason810496 Mar 5, 2026
1e9a06a

Add aws s3fs deps


jason810496 Mar 5, 2026
2bc77f0

Add debug log


jason810496 Mar 5, 2026
fcd0685

Add required AWS env to fix the e2e test failure


jason810496 Mar 6, 2026
e7865bf

Harden the client in e2e test


jason810496 Mar 6, 2026
b82137f

Fix nits


jason810496 Mar 6, 2026
0011121

Fix copilot suggestion


jason810496 Mar 6, 2026

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


11 changes: 11 additions & 0 deletions
.github/workflows/additional-prod-image-tests.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -226,6 +226,17 @@ jobs:
 use-uv: ${{ inputs.use-uv }}
 e2e_test_mode: "remote_log"

 test-e2e-integration-tests-xcom-object-storage:
 name: "XCom object storage backend tests with PROD image"
 uses: ./.github/workflows/airflow-e2e-tests.yml
 with:
 workflow-name: "XCom object storage backend e2e test"
 runners: ${{ inputs.runners }}
 platform: ${{ inputs.platform }}
 default-python-version: "${{ inputs.default-python-version }}"
 use-uv: ${{ inputs.use-uv }}
 e2e_test_mode: "xcom_object_storage"

 test-ui-e2e-chromium:
 name: "Chromium UI e2e tests with PROD image"
 uses: ./.github/workflows/ui-e2e-tests.yml
Expand Down

4 changes: 2 additions & 2 deletions
.github/workflows/airflow-e2e-tests.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -49,7 +49,7 @@ on: # yamllint disable-line rule:truthy
 type: string
 required: true
 e2e_test_mode:
 description: "Test mode - basicor remote_log"
 description: "Test mode - basic, remote_log, or xcom_object_storage"
 type: string
 default: "basic"

Expand DownExpand Up@@ -80,7 +80,7 @@ on: # yamllint disable-line rule:truthy
 type: string
 default: ""
 e2e_test_mode:
 description: "Test mode - quick or full"
 description: "Test mode - basic, remote_log, or xcom_object_storage"
 type: string
 default: "basic"

Expand Down

1 change: 1 addition & 0 deletions
airflow-e2e-tests/scripts/init-aws.sh

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,4 +17,5 @@
# under the License.

aws --endpoint-url=http://localstack:4566 s3 mb s3://test-airflow-logs
aws --endpoint-url=http://localstack:4566 s3 mb s3://test-xcom-objectstorage-backend
aws --endpoint-url=http://localstack:4566 s3 ls

32 changes: 31 additions & 1 deletion
airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -36,6 +36,7 @@
 LOCALSTACK_PATH,
 LOGS_FOLDER,
 TEST_REPORT_FILE,
 XCOM_BUCKET,
)

from tests_common.test_utils.fernet import generate_fernet_key_string
Expand All@@ -48,13 +49,18 @@ class _E2ETestState:
 airflow_logs_path: Path | None = None


def _setup_s3_integration(dot_env_file, tmp_dir):
def _copy_localstack_files(tmp_dir):
 """Copy localstack compose file and init script into the temp directory."""
 copyfile(LOCALSTACK_PATH, tmp_dir / "localstack.yml")

 copyfile(AWS_INIT_PATH, tmp_dir / "init-aws.sh")
 current_permissions = os.stat(tmp_dir / "init-aws.sh").st_mode
 os.chmod(tmp_dir / "init-aws.sh", current_permissions | 0o111)


def _setup_s3_integration(dot_env_file, tmp_dir):
 _copy_localstack_files(tmp_dir)

 dot_env_file.write_text(
 f"AIRFLOW_UID={os.getuid()}\n"
 "AWS_DEFAULT_REGION=us-east-1\n"
Expand All@@ -68,6 +74,27 @@ def _setup_s3_integration(dot_env_file, tmp_dir):
 os.environ["ENV_FILE_PATH"] = str(dot_env_file)


def _setup_xcom_object_storage_integration(dot_env_file, tmp_dir):
 _copy_localstack_files(tmp_dir)

 dot_env_file.write_text(
 f"AIRFLOW_UID={os.getuid()}\n"
 # XComObjectStorageBackend requires AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY as env vars
 # because `universal-path` uses boto3's native S3 client, which relies on environment variables
 # for authentication rather than parsing credentials from the connection URI
 "AWS_ACCESS_KEY_ID=test\n"
 "AWS_SECRET_ACCESS_KEY=test\n"
 "AWS_DEFAULT_REGION=us-east-1\n"
 "AWS_ENDPOINT_URL_S3=http://localstack:4566\n"
 "AIRFLOW_CONN_AWS_DEFAULT=aws://test:test@\n"
 "AIRFLOW__CORE__XCOM_BACKEND=airflow.providers.common.io.xcom.backend.XComObjectStorageBackend\n"
 f"AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_PATH=s3://aws_default@{XCOM_BUCKET}/xcom\n"
 "AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_THRESHOLD=0\n"
 "_PIP_ADDITIONAL_REQUIREMENTS=apache-airflow-providers-amazon[s3fs]\n"
 )
 os.environ["ENV_FILE_PATH"] = str(dot_env_file)


def spin_up_airflow_environment(tmp_path_factory: pytest.TempPathFactory):
 tmp_dir = tmp_path_factory.mktemp("airflow-e2e-tests")

Expand DownExpand Up@@ -97,6 +124,9 @@ def spin_up_airflow_environment(tmp_path_factory: pytest.TempPathFactory):
 if E2E_TEST_MODE == "remote_log":
 compose_file_names.append("localstack.yml")
 _setup_s3_integration(dot_env_file, tmp_dir)
 elif E2E_TEST_MODE == "xcom_object_storage":
 compose_file_names.append("localstack.yml")
 _setup_xcom_object_storage_integration(dot_env_file, tmp_dir)

 #
 # Please Do not use this Fernet key in any deployments! Please generate your own key.
Expand Down

3 changes: 3 additions & 0 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/constants.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -42,3 +42,6 @@
LOCALSTACK_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "docker" / "localstack.yml"
E2E_TEST_MODE = os.environ.get("E2E_TEST_MODE", "basic")
AWS_INIT_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "scripts" / "init-aws.sh"

# s3 bucket name for XComObjectStorageBackend tests. This bucket will be created in the `init-aws.sh` script that is run as part of the LocalStack container initialization.
XCOM_BUCKET = "test-xcom-objectstorage-backend"

51 changes: 43 additions & 8 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -20,6 +20,7 @@
from datetime import datetime, timezone
from functools import cached_property

import boto3
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
Expand All@@ -31,19 +32,41 @@
)


def get_s3_client():
 """Return a boto3 S3 client configured to use the local LocalStack endpoint."""
 return boto3.client(
 "s3",
 endpoint_url="http://localhost:4566",
 aws_access_key_id="test",
 aws_secret_access_key="test",
 region_name="us-east-1",
 )


def create_request_session_with_retries(status_forcelist: list[int]):
 """Create a requests Session with retry logic for handling transient errors."""
 Retry.DEFAULT_BACKOFF_MAX = 32
 retry_strategy = Retry(
 total=10,
 backoff_factor=1,
 status_forcelist=status_forcelist,
 )
 session = requests.Session()
 adapter = HTTPAdapter(max_retries=retry_strategy)
 session.mount("http://", adapter)
 session.mount("https://", adapter)
 return session


class AirflowClient:
 """Client for interacting with the Airflow REST API."""

 def __init__(self):
 self.session = requests.Session()
 self.session = create_request_session_with_retries(status_forcelist=[429])

 @cached_property
 def token(self):
 Retry.DEFAULT_BACKOFF_MAX = 32
 retry = Retry(total=10, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504])
 session = requests.Session()
 session.mount("http://", HTTPAdapter(max_retries=retry))
 session.mount("https://", HTTPAdapter(max_retries=retry))
 session = create_request_session_with_retries(status_forcelist=[429, 500, 502, 503, 504])

 api_server_url = DOCKER_COMPOSE_HOST_PORT
 if not api_server_url.startswith(("http://", "https://")):
Expand DownExpand Up@@ -121,11 +144,23 @@ def trigger_dag_and_wait(self, dag_id: str, json=None):
 run_id=resp["dag_run_id"],
 )

 def get_task_logs(self, dag_id: str, run_id: str, task_id: str, try_number: int = 1):
 def get_task_instances(self, dag_id: str, run_id: str):
 """Get task instances for a given DAG run."""
 return self._make_request(
 method="GET",
 endpoint=f"dags/{dag_id}/dagRuns/{run_id}/taskInstances",
 )

 def get_task_logs(
 self, dag_id: str, run_id: str, task_id: str, try_number: int = 1, map_index: int | None = None
 ):
 """Get task logs via API."""
 endpoint = f"dags/{dag_id}/dagRuns/{run_id}/taskInstances/{task_id}/logs/{try_number}"
 if map_index is not None:
 endpoint += f"?map_index={map_index}"
 return self._make_request(
 method="GET",
 endpoint=f"dags/{dag_id}/dagRuns/{run_id}/taskInstances/{task_id}/logs/{try_number}",
 endpoint=endpoint,
 )


Expand Down

12 changes: 3 additions & 9 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/remote_log_tests/test_remote_logging.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -19,10 +19,9 @@
import time
from datetime import datetime, timezone

import boto3
import pytest

from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient
from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient, get_s3_client


class TestRemoteLogging:
Expand DownExpand Up@@ -56,15 +55,10 @@ def test_remote_logging_s3(self):

 # This bucket will be created part of the docker-compose setup in
 bucket_name = "test-airflow-logs"
 s3_client = boto3.client(
 "s3",
 endpoint_url="http://localhost:4566",
 aws_access_key_id="test",
 aws_secret_access_key="test",
 region_name="us-east-1",
 )
 s3_client = get_s3_client()

 # Wait for logs to be available in S3 before we call `get_task_logs`
 contents = []
 for _ in range(self.max_retries):
 response = s3_client.list_objects_v2(Bucket=bucket_name)
 contents = response.get("Contents", [])
Expand Down

16 changes: 16 additions & 0 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/xcom_object_storage_tests/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
# 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.

89 changes: 89 additions & 0 deletions
...sts/tests/airflow_e2e_tests/xcom_object_storage_tests/test_xcom_object_storage_backend.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
# 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 time
from datetime import datetime, timezone
from pprint import pprint
from uuid import uuid4

import pytest

from airflow_e2e_tests.constants import XCOM_BUCKET
from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient, get_s3_client


class TestXComObjectStorageBackend:
 airflow_client = AirflowClient()
 dag_id = "example_xcom_test"
 retry_interval_in_seconds = 5
 max_retries = 12

 def test_dag_succeeds_and_xcom_values_stored_in_s3(self):
 """Test that a DAG using XComObjectStorageBackend completes successfully and persists XCom values to S3."""
 self.airflow_client.un_pause_dag(self.dag_id)

 trigger_resp = self.airflow_client.trigger_dag(
 self.dag_id,
 json={
 "dag_run_id": f"test_xcom_object_storage_backend_{uuid4()}",
 "logical_date": datetime.now(timezone.utc).isoformat(),
 },
 )
 dag_run_id = trigger_resp["dag_run_id"]
 state = self.airflow_client.wait_for_dag_run(
 dag_id=self.dag_id,
 run_id=dag_run_id,
 )

 # try to get all the logs to help debugging
 if state != "success":
 task_instances_resp = self.airflow_client.get_task_instances(self.dag_id, dag_run_id)
 for task_instance in task_instances_resp["task_instances"]:
 task_id = task_instance["task_id"]
 try_number = task_instance["try_number"]
 try:
 print(f"\nLogs for task {task_id} (try {try_number}):")
 task_logs_resp = self.airflow_client.get_task_logs(
 dag_id=self.dag_id, task_id=task_id, run_id=dag_run_id, try_number=try_number
 )
 pprint(task_logs_resp)
 except Exception as e:
 print(f"Could not get logs for task {task_id} (try {try_number}): {e}")

 assert state == "success", f"DAG {self.dag_id} did not complete successfully. Final state: {state}"

 s3_client = get_s3_client()

 contents = []
 for _ in range(self.max_retries):
 response = s3_client.list_objects_v2(Bucket=XCOM_BUCKET)
 contents = response.get("Contents", [])
 if contents:
 break

 print(f"No XCom objects found in S3 bucket {XCOM_BUCKET!r} yet. Retrying...")
 time.sleep(self.retry_interval_in_seconds)

 if not contents:
 pytest.fail(
 f"Expected XCom objects in S3 bucket {XCOM_BUCKET!r}, but bucket is empty.\n"
 f"List Objects Response: {response}"
 )

 keys = [obj["Key"] for obj in contents]
 print(f"Found {len(keys)} XCom object(s) in S3: {keys}")
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Auto-enable theater mode on YouTube
(function() {
function tryTheater() {
var btn = document.querySelector('button[aria-label="Theater mode"], ytd-player #player button[title="Theater mode"]');
if (btn && !btn.classList.contains('activated')) {
btn.click();
}
}
// Try immediately
tryTheater();
// Try after navigation (SPA)
var lastUrl = location.href;
setInterval(function() {
if (location.href !== lastUrl) {
lastUrl = location.href;
setTimeout(tryTheater, 500);
}
}, 1000);
// Also try on player load
var observer = new MutationObserver(tryTheater);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Add e2e test for XComObjectStorageBackend#62907

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

jason810496
merged 8 commits into

from

Mar 9, 2026
 Merged


Add e2e test for XComObjectStorageBackend
#62907
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


8 commits


Select commit
Hold shift + click to select a range
0635994

Add e2e test for XComObjectStorageBackend


jason810496 Mar 5, 2026
0600b02

Update GitHub Action and Breeze cmd


jason810496 Mar 5, 2026
1e9a06a

Add aws s3fs deps


jason810496 Mar 5, 2026
2bc77f0

Add debug log


jason810496 Mar 5, 2026
fcd0685

Add required AWS env to fix the e2e test failure


jason810496 Mar 6, 2026
e7865bf

Harden the client in e2e test


jason810496 Mar 6, 2026
b82137f

Fix nits


jason810496 Mar 6, 2026
0011121

Fix copilot suggestion


jason810496 Mar 6, 2026

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


11 changes: 11 additions & 0 deletions
.github/workflows/additional-prod-image-tests.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -226,6 +226,17 @@ jobs:
 use-uv: ${{ inputs.use-uv }}
 e2e_test_mode: "remote_log"

 test-e2e-integration-tests-xcom-object-storage:
 name: "XCom object storage backend tests with PROD image"
 uses: ./.github/workflows/airflow-e2e-tests.yml
 with:
 workflow-name: "XCom object storage backend e2e test"
 runners: ${{ inputs.runners }}
 platform: ${{ inputs.platform }}
 default-python-version: "${{ inputs.default-python-version }}"
 use-uv: ${{ inputs.use-uv }}
 e2e_test_mode: "xcom_object_storage"

 test-ui-e2e-chromium:
 name: "Chromium UI e2e tests with PROD image"
 uses: ./.github/workflows/ui-e2e-tests.yml
Expand Down

4 changes: 2 additions & 2 deletions
.github/workflows/airflow-e2e-tests.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -49,7 +49,7 @@ on: # yamllint disable-line rule:truthy
 type: string
 required: true
 e2e_test_mode:
 description: "Test mode - basicor remote_log"
 description: "Test mode - basic, remote_log, or xcom_object_storage"
 type: string
 default: "basic"

Expand DownExpand Up@@ -80,7 +80,7 @@ on: # yamllint disable-line rule:truthy
 type: string
 default: ""
 e2e_test_mode:
 description: "Test mode - quick or full"
 description: "Test mode - basic, remote_log, or xcom_object_storage"
 type: string
 default: "basic"

Expand Down

1 change: 1 addition & 0 deletions
airflow-e2e-tests/scripts/init-aws.sh

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,4 +17,5 @@
# under the License.

aws --endpoint-url=http://localstack:4566 s3 mb s3://test-airflow-logs
aws --endpoint-url=http://localstack:4566 s3 mb s3://test-xcom-objectstorage-backend
aws --endpoint-url=http://localstack:4566 s3 ls

32 changes: 31 additions & 1 deletion
airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -36,6 +36,7 @@
 LOCALSTACK_PATH,
 LOGS_FOLDER,
 TEST_REPORT_FILE,
 XCOM_BUCKET,
)

from tests_common.test_utils.fernet import generate_fernet_key_string
Expand All@@ -48,13 +49,18 @@ class _E2ETestState:
 airflow_logs_path: Path | None = None


def _setup_s3_integration(dot_env_file, tmp_dir):
def _copy_localstack_files(tmp_dir):
 """Copy localstack compose file and init script into the temp directory."""
 copyfile(LOCALSTACK_PATH, tmp_dir / "localstack.yml")

 copyfile(AWS_INIT_PATH, tmp_dir / "init-aws.sh")
 current_permissions = os.stat(tmp_dir / "init-aws.sh").st_mode
 os.chmod(tmp_dir / "init-aws.sh", current_permissions | 0o111)


def _setup_s3_integration(dot_env_file, tmp_dir):
 _copy_localstack_files(tmp_dir)

 dot_env_file.write_text(
 f"AIRFLOW_UID={os.getuid()}\n"
 "AWS_DEFAULT_REGION=us-east-1\n"
Expand All@@ -68,6 +74,27 @@ def _setup_s3_integration(dot_env_file, tmp_dir):
 os.environ["ENV_FILE_PATH"] = str(dot_env_file)


def _setup_xcom_object_storage_integration(dot_env_file, tmp_dir):
 _copy_localstack_files(tmp_dir)

 dot_env_file.write_text(
 f"AIRFLOW_UID={os.getuid()}\n"
 # XComObjectStorageBackend requires AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY as env vars
 # because `universal-path` uses boto3's native S3 client, which relies on environment variables
 # for authentication rather than parsing credentials from the connection URI
 "AWS_ACCESS_KEY_ID=test\n"
 "AWS_SECRET_ACCESS_KEY=test\n"
 "AWS_DEFAULT_REGION=us-east-1\n"
 "AWS_ENDPOINT_URL_S3=http://localstack:4566\n"
 "AIRFLOW_CONN_AWS_DEFAULT=aws://test:test@\n"
 "AIRFLOW__CORE__XCOM_BACKEND=airflow.providers.common.io.xcom.backend.XComObjectStorageBackend\n"
 f"AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_PATH=s3://aws_default@{XCOM_BUCKET}/xcom\n"
 "AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_THRESHOLD=0\n"
 "_PIP_ADDITIONAL_REQUIREMENTS=apache-airflow-providers-amazon[s3fs]\n"
 )
 os.environ["ENV_FILE_PATH"] = str(dot_env_file)


def spin_up_airflow_environment(tmp_path_factory: pytest.TempPathFactory):
 tmp_dir = tmp_path_factory.mktemp("airflow-e2e-tests")

Expand DownExpand Up@@ -97,6 +124,9 @@ def spin_up_airflow_environment(tmp_path_factory: pytest.TempPathFactory):
 if E2E_TEST_MODE == "remote_log":
 compose_file_names.append("localstack.yml")
 _setup_s3_integration(dot_env_file, tmp_dir)
 elif E2E_TEST_MODE == "xcom_object_storage":
 compose_file_names.append("localstack.yml")
 _setup_xcom_object_storage_integration(dot_env_file, tmp_dir)

 #
 # Please Do not use this Fernet key in any deployments! Please generate your own key.
Expand Down

3 changes: 3 additions & 0 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/constants.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -42,3 +42,6 @@
LOCALSTACK_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "docker" / "localstack.yml"
E2E_TEST_MODE = os.environ.get("E2E_TEST_MODE", "basic")
AWS_INIT_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "scripts" / "init-aws.sh"

# s3 bucket name for XComObjectStorageBackend tests. This bucket will be created in the `init-aws.sh` script that is run as part of the LocalStack container initialization.
XCOM_BUCKET = "test-xcom-objectstorage-backend"

51 changes: 43 additions & 8 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -20,6 +20,7 @@
from datetime import datetime, timezone
from functools import cached_property

import boto3
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
Expand All@@ -31,19 +32,41 @@
)


def get_s3_client():
 """Return a boto3 S3 client configured to use the local LocalStack endpoint."""
 return boto3.client(
 "s3",
 endpoint_url="http://localhost:4566",
 aws_access_key_id="test",
 aws_secret_access_key="test",
 region_name="us-east-1",
 )


def create_request_session_with_retries(status_forcelist: list[int]):
 """Create a requests Session with retry logic for handling transient errors."""
 Retry.DEFAULT_BACKOFF_MAX = 32
 retry_strategy = Retry(
 total=10,
 backoff_factor=1,
 status_forcelist=status_forcelist,
 )
 session = requests.Session()
 adapter = HTTPAdapter(max_retries=retry_strategy)
 session.mount("http://", adapter)
 session.mount("https://", adapter)
 return session


class AirflowClient:
 """Client for interacting with the Airflow REST API."""

 def __init__(self):
 self.session = requests.Session()
 self.session = create_request_session_with_retries(status_forcelist=[429])

 @cached_property
 def token(self):
 Retry.DEFAULT_BACKOFF_MAX = 32
 retry = Retry(total=10, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504])
 session = requests.Session()
 session.mount("http://", HTTPAdapter(max_retries=retry))
 session.mount("https://", HTTPAdapter(max_retries=retry))
 session = create_request_session_with_retries(status_forcelist=[429, 500, 502, 503, 504])

 api_server_url = DOCKER_COMPOSE_HOST_PORT
 if not api_server_url.startswith(("http://", "https://")):
Expand DownExpand Up@@ -121,11 +144,23 @@ def trigger_dag_and_wait(self, dag_id: str, json=None):
 run_id=resp["dag_run_id"],
 )

 def get_task_logs(self, dag_id: str, run_id: str, task_id: str, try_number: int = 1):
 def get_task_instances(self, dag_id: str, run_id: str):
 """Get task instances for a given DAG run."""
 return self._make_request(
 method="GET",
 endpoint=f"dags/{dag_id}/dagRuns/{run_id}/taskInstances",
 )

 def get_task_logs(
 self, dag_id: str, run_id: str, task_id: str, try_number: int = 1, map_index: int | None = None
 ):
 """Get task logs via API."""
 endpoint = f"dags/{dag_id}/dagRuns/{run_id}/taskInstances/{task_id}/logs/{try_number}"
 if map_index is not None:
 endpoint += f"?map_index={map_index}"
 return self._make_request(
 method="GET",
 endpoint=f"dags/{dag_id}/dagRuns/{run_id}/taskInstances/{task_id}/logs/{try_number}",
 endpoint=endpoint,
 )


Expand Down

12 changes: 3 additions & 9 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/remote_log_tests/test_remote_logging.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -19,10 +19,9 @@
import time
from datetime import datetime, timezone

import boto3
import pytest

from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient
from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient, get_s3_client


class TestRemoteLogging:
Expand DownExpand Up@@ -56,15 +55,10 @@ def test_remote_logging_s3(self):

 # This bucket will be created part of the docker-compose setup in
 bucket_name = "test-airflow-logs"
 s3_client = boto3.client(
 "s3",
 endpoint_url="http://localhost:4566",
 aws_access_key_id="test",
 aws_secret_access_key="test",
 region_name="us-east-1",
 )
 s3_client = get_s3_client()

 # Wait for logs to be available in S3 before we call `get_task_logs`
 contents = []
 for _ in range(self.max_retries):
 response = s3_client.list_objects_v2(Bucket=bucket_name)
 contents = response.get("Contents", [])
Expand Down

16 changes: 16 additions & 0 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/xcom_object_storage_tests/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
# 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.

89 changes: 89 additions & 0 deletions
...sts/tests/airflow_e2e_tests/xcom_object_storage_tests/test_xcom_object_storage_backend.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
# 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 time
from datetime import datetime, timezone
from pprint import pprint
from uuid import uuid4

import pytest

from airflow_e2e_tests.constants import XCOM_BUCKET
from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient, get_s3_client


class TestXComObjectStorageBackend:
 airflow_client = AirflowClient()
 dag_id = "example_xcom_test"
 retry_interval_in_seconds = 5
 max_retries = 12

 def test_dag_succeeds_and_xcom_values_stored_in_s3(self):
 """Test that a DAG using XComObjectStorageBackend completes successfully and persists XCom values to S3."""
 self.airflow_client.un_pause_dag(self.dag_id)

 trigger_resp = self.airflow_client.trigger_dag(
 self.dag_id,
 json={
 "dag_run_id": f"test_xcom_object_storage_backend_{uuid4()}",
 "logical_date": datetime.now(timezone.utc).isoformat(),
 },
 )
 dag_run_id = trigger_resp["dag_run_id"]
 state = self.airflow_client.wait_for_dag_run(
 dag_id=self.dag_id,
 run_id=dag_run_id,
 )

 # try to get all the logs to help debugging
 if state != "success":
 task_instances_resp = self.airflow_client.get_task_instances(self.dag_id, dag_run_id)
 for task_instance in task_instances_resp["task_instances"]:
 task_id = task_instance["task_id"]
 try_number = task_instance["try_number"]
 try:
 print(f"\nLogs for task {task_id} (try {try_number}):")
 task_logs_resp = self.airflow_client.get_task_logs(
 dag_id=self.dag_id, task_id=task_id, run_id=dag_run_id, try_number=try_number
 )
 pprint(task_logs_resp)
 except Exception as e:
 print(f"Could not get logs for task {task_id} (try {try_number}): {e}")

 assert state == "success", f"DAG {self.dag_id} did not complete successfully. Final state: {state}"

 s3_client = get_s3_client()

 contents = []
 for _ in range(self.max_retries):
 response = s3_client.list_objects_v2(Bucket=XCOM_BUCKET)
 contents = response.get("Contents", [])
 if contents:
 break

 print(f"No XCom objects found in S3 bucket {XCOM_BUCKET!r} yet. Retrying...")
 time.sleep(self.retry_interval_in_seconds)

 if not contents:
 pytest.fail(
 f"Expected XCom objects in S3 bucket {XCOM_BUCKET!r}, but bucket is empty.\n"
 f"List Objects Response: {response}"
 )

 keys = [obj["Key"] for obj in contents]
 print(f"Found {len(keys)} XCom object(s) in S3: {keys}")
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Remove or un-stick sticky/fixed headers that block content
(function() {
function unstick() {
document.querySelectorAll('header, nav, [role="banner"], .header, .navbar, .sticky, .fixed-top, [style*="position: fixed"], [style*="position:sticky"]').forEach(function(el) {
if (el.style.position === 'fixed' || el.style.position === 'sticky' || getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {
el.style.position = 'static';
el.style.top = 'auto';
el.style.zIndex = 'auto';
}
});
}
unstick();
var observer = new MutationObserver(unstick);
observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });
})();
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Add e2e test for XComObjectStorageBackend#62907

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

jason810496
merged 8 commits into

from

Mar 9, 2026
 Merged


Add e2e test for XComObjectStorageBackend
#62907
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


8 commits


Select commit
Hold shift + click to select a range
0635994

Add e2e test for XComObjectStorageBackend


jason810496 Mar 5, 2026
0600b02

Update GitHub Action and Breeze cmd


jason810496 Mar 5, 2026
1e9a06a

Add aws s3fs deps


jason810496 Mar 5, 2026
2bc77f0

Add debug log


jason810496 Mar 5, 2026
fcd0685

Add required AWS env to fix the e2e test failure


jason810496 Mar 6, 2026
e7865bf

Harden the client in e2e test


jason810496 Mar 6, 2026
b82137f

Fix nits


jason810496 Mar 6, 2026
0011121

Fix copilot suggestion


jason810496 Mar 6, 2026

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


11 changes: 11 additions & 0 deletions
.github/workflows/additional-prod-image-tests.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -226,6 +226,17 @@ jobs:
 use-uv: ${{ inputs.use-uv }}
 e2e_test_mode: "remote_log"

 test-e2e-integration-tests-xcom-object-storage:
 name: "XCom object storage backend tests with PROD image"
 uses: ./.github/workflows/airflow-e2e-tests.yml
 with:
 workflow-name: "XCom object storage backend e2e test"
 runners: ${{ inputs.runners }}
 platform: ${{ inputs.platform }}
 default-python-version: "${{ inputs.default-python-version }}"
 use-uv: ${{ inputs.use-uv }}
 e2e_test_mode: "xcom_object_storage"

 test-ui-e2e-chromium:
 name: "Chromium UI e2e tests with PROD image"
 uses: ./.github/workflows/ui-e2e-tests.yml
Expand Down

4 changes: 2 additions & 2 deletions
.github/workflows/airflow-e2e-tests.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -49,7 +49,7 @@ on: # yamllint disable-line rule:truthy
 type: string
 required: true
 e2e_test_mode:
 description: "Test mode - basicor remote_log"
 description: "Test mode - basic, remote_log, or xcom_object_storage"
 type: string
 default: "basic"

Expand DownExpand Up@@ -80,7 +80,7 @@ on: # yamllint disable-line rule:truthy
 type: string
 default: ""
 e2e_test_mode:
 description: "Test mode - quick or full"
 description: "Test mode - basic, remote_log, or xcom_object_storage"
 type: string
 default: "basic"

Expand Down

1 change: 1 addition & 0 deletions
airflow-e2e-tests/scripts/init-aws.sh

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,4 +17,5 @@
# under the License.

aws --endpoint-url=http://localstack:4566 s3 mb s3://test-airflow-logs
aws --endpoint-url=http://localstack:4566 s3 mb s3://test-xcom-objectstorage-backend
aws --endpoint-url=http://localstack:4566 s3 ls

32 changes: 31 additions & 1 deletion
airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -36,6 +36,7 @@
 LOCALSTACK_PATH,
 LOGS_FOLDER,
 TEST_REPORT_FILE,
 XCOM_BUCKET,
)

from tests_common.test_utils.fernet import generate_fernet_key_string
Expand All@@ -48,13 +49,18 @@ class _E2ETestState:
 airflow_logs_path: Path | None = None


def _setup_s3_integration(dot_env_file, tmp_dir):
def _copy_localstack_files(tmp_dir):
 """Copy localstack compose file and init script into the temp directory."""
 copyfile(LOCALSTACK_PATH, tmp_dir / "localstack.yml")

 copyfile(AWS_INIT_PATH, tmp_dir / "init-aws.sh")
 current_permissions = os.stat(tmp_dir / "init-aws.sh").st_mode
 os.chmod(tmp_dir / "init-aws.sh", current_permissions | 0o111)


def _setup_s3_integration(dot_env_file, tmp_dir):
 _copy_localstack_files(tmp_dir)

 dot_env_file.write_text(
 f"AIRFLOW_UID={os.getuid()}\n"
 "AWS_DEFAULT_REGION=us-east-1\n"
Expand All@@ -68,6 +74,27 @@ def _setup_s3_integration(dot_env_file, tmp_dir):
 os.environ["ENV_FILE_PATH"] = str(dot_env_file)


def _setup_xcom_object_storage_integration(dot_env_file, tmp_dir):
 _copy_localstack_files(tmp_dir)

 dot_env_file.write_text(
 f"AIRFLOW_UID={os.getuid()}\n"
 # XComObjectStorageBackend requires AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY as env vars
 # because `universal-path` uses boto3's native S3 client, which relies on environment variables
 # for authentication rather than parsing credentials from the connection URI
 "AWS_ACCESS_KEY_ID=test\n"
 "AWS_SECRET_ACCESS_KEY=test\n"
 "AWS_DEFAULT_REGION=us-east-1\n"
 "AWS_ENDPOINT_URL_S3=http://localstack:4566\n"
 "AIRFLOW_CONN_AWS_DEFAULT=aws://test:test@\n"
 "AIRFLOW__CORE__XCOM_BACKEND=airflow.providers.common.io.xcom.backend.XComObjectStorageBackend\n"
 f"AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_PATH=s3://aws_default@{XCOM_BUCKET}/xcom\n"
 "AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_THRESHOLD=0\n"
 "_PIP_ADDITIONAL_REQUIREMENTS=apache-airflow-providers-amazon[s3fs]\n"
 )
 os.environ["ENV_FILE_PATH"] = str(dot_env_file)


def spin_up_airflow_environment(tmp_path_factory: pytest.TempPathFactory):
 tmp_dir = tmp_path_factory.mktemp("airflow-e2e-tests")

Expand DownExpand Up@@ -97,6 +124,9 @@ def spin_up_airflow_environment(tmp_path_factory: pytest.TempPathFactory):
 if E2E_TEST_MODE == "remote_log":
 compose_file_names.append("localstack.yml")
 _setup_s3_integration(dot_env_file, tmp_dir)
 elif E2E_TEST_MODE == "xcom_object_storage":
 compose_file_names.append("localstack.yml")
 _setup_xcom_object_storage_integration(dot_env_file, tmp_dir)

 #
 # Please Do not use this Fernet key in any deployments! Please generate your own key.
Expand Down

3 changes: 3 additions & 0 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/constants.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -42,3 +42,6 @@
LOCALSTACK_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "docker" / "localstack.yml"
E2E_TEST_MODE = os.environ.get("E2E_TEST_MODE", "basic")
AWS_INIT_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "scripts" / "init-aws.sh"

# s3 bucket name for XComObjectStorageBackend tests. This bucket will be created in the `init-aws.sh` script that is run as part of the LocalStack container initialization.
XCOM_BUCKET = "test-xcom-objectstorage-backend"

51 changes: 43 additions & 8 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -20,6 +20,7 @@
from datetime import datetime, timezone
from functools import cached_property

import boto3
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
Expand All@@ -31,19 +32,41 @@
)


def get_s3_client():
 """Return a boto3 S3 client configured to use the local LocalStack endpoint."""
 return boto3.client(
 "s3",
 endpoint_url="http://localhost:4566",
 aws_access_key_id="test",
 aws_secret_access_key="test",
 region_name="us-east-1",
 )


def create_request_session_with_retries(status_forcelist: list[int]):
 """Create a requests Session with retry logic for handling transient errors."""
 Retry.DEFAULT_BACKOFF_MAX = 32
 retry_strategy = Retry(
 total=10,
 backoff_factor=1,
 status_forcelist=status_forcelist,
 )
 session = requests.Session()
 adapter = HTTPAdapter(max_retries=retry_strategy)
 session.mount("http://", adapter)
 session.mount("https://", adapter)
 return session


class AirflowClient:
 """Client for interacting with the Airflow REST API."""

 def __init__(self):
 self.session = requests.Session()
 self.session = create_request_session_with_retries(status_forcelist=[429])

 @cached_property
 def token(self):
 Retry.DEFAULT_BACKOFF_MAX = 32
 retry = Retry(total=10, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504])
 session = requests.Session()
 session.mount("http://", HTTPAdapter(max_retries=retry))
 session.mount("https://", HTTPAdapter(max_retries=retry))
 session = create_request_session_with_retries(status_forcelist=[429, 500, 502, 503, 504])

 api_server_url = DOCKER_COMPOSE_HOST_PORT
 if not api_server_url.startswith(("http://", "https://")):
Expand DownExpand Up@@ -121,11 +144,23 @@ def trigger_dag_and_wait(self, dag_id: str, json=None):
 run_id=resp["dag_run_id"],
 )

 def get_task_logs(self, dag_id: str, run_id: str, task_id: str, try_number: int = 1):
 def get_task_instances(self, dag_id: str, run_id: str):
 """Get task instances for a given DAG run."""
 return self._make_request(
 method="GET",
 endpoint=f"dags/{dag_id}/dagRuns/{run_id}/taskInstances",
 )

 def get_task_logs(
 self, dag_id: str, run_id: str, task_id: str, try_number: int = 1, map_index: int | None = None
 ):
 """Get task logs via API."""
 endpoint = f"dags/{dag_id}/dagRuns/{run_id}/taskInstances/{task_id}/logs/{try_number}"
 if map_index is not None:
 endpoint += f"?map_index={map_index}"
 return self._make_request(
 method="GET",
 endpoint=f"dags/{dag_id}/dagRuns/{run_id}/taskInstances/{task_id}/logs/{try_number}",
 endpoint=endpoint,
 )


Expand Down

12 changes: 3 additions & 9 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/remote_log_tests/test_remote_logging.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -19,10 +19,9 @@
import time
from datetime import datetime, timezone

import boto3
import pytest

from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient
from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient, get_s3_client


class TestRemoteLogging:
Expand DownExpand Up@@ -56,15 +55,10 @@ def test_remote_logging_s3(self):

 # This bucket will be created part of the docker-compose setup in
 bucket_name = "test-airflow-logs"
 s3_client = boto3.client(
 "s3",
 endpoint_url="http://localhost:4566",
 aws_access_key_id="test",
 aws_secret_access_key="test",
 region_name="us-east-1",
 )
 s3_client = get_s3_client()

 # Wait for logs to be available in S3 before we call `get_task_logs`
 contents = []
 for _ in range(self.max_retries):
 response = s3_client.list_objects_v2(Bucket=bucket_name)
 contents = response.get("Contents", [])
Expand Down

16 changes: 16 additions & 0 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/xcom_object_storage_tests/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
# 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.

89 changes: 89 additions & 0 deletions
...sts/tests/airflow_e2e_tests/xcom_object_storage_tests/test_xcom_object_storage_backend.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
# 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 time
from datetime import datetime, timezone
from pprint import pprint
from uuid import uuid4

import pytest

from airflow_e2e_tests.constants import XCOM_BUCKET
from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient, get_s3_client


class TestXComObjectStorageBackend:
 airflow_client = AirflowClient()
 dag_id = "example_xcom_test"
 retry_interval_in_seconds = 5
 max_retries = 12

 def test_dag_succeeds_and_xcom_values_stored_in_s3(self):
 """Test that a DAG using XComObjectStorageBackend completes successfully and persists XCom values to S3."""
 self.airflow_client.un_pause_dag(self.dag_id)

 trigger_resp = self.airflow_client.trigger_dag(
 self.dag_id,
 json={
 "dag_run_id": f"test_xcom_object_storage_backend_{uuid4()}",
 "logical_date": datetime.now(timezone.utc).isoformat(),
 },
 )
 dag_run_id = trigger_resp["dag_run_id"]
 state = self.airflow_client.wait_for_dag_run(
 dag_id=self.dag_id,
 run_id=dag_run_id,
 )

 # try to get all the logs to help debugging
 if state != "success":
 task_instances_resp = self.airflow_client.get_task_instances(self.dag_id, dag_run_id)
 for task_instance in task_instances_resp["task_instances"]:
 task_id = task_instance["task_id"]
 try_number = task_instance["try_number"]
 try:
 print(f"\nLogs for task {task_id} (try {try_number}):")
 task_logs_resp = self.airflow_client.get_task_logs(
 dag_id=self.dag_id, task_id=task_id, run_id=dag_run_id, try_number=try_number
 )
 pprint(task_logs_resp)
 except Exception as e:
 print(f"Could not get logs for task {task_id} (try {try_number}): {e}")

 assert state == "success", f"DAG {self.dag_id} did not complete successfully. Final state: {state}"

 s3_client = get_s3_client()

 contents = []
 for _ in range(self.max_retries):
 response = s3_client.list_objects_v2(Bucket=XCOM_BUCKET)
 contents = response.get("Contents", [])
 if contents:
 break

 print(f"No XCom objects found in S3 bucket {XCOM_BUCKET!r} yet. Retrying...")
 time.sleep(self.retry_interval_in_seconds)

 if not contents:
 pytest.fail(
 f"Expected XCom objects in S3 bucket {XCOM_BUCKET!r}, but bucket is empty.\n"
 f"List Objects Response: {response}"
 )

 keys = [obj["Key"] for obj in contents]
 print(f"Found {len(keys)} XCom object(s) in S3: {keys}")
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Universal Dark Mode - works on any site
(function() {
var enabled = true;
function applyDarkMode() {
if (!enabled) return;
// Create style element if it doesn't exist
var style = document.getElementById('universal-dark-mode-style');
if (!style) {
style = document.createElement('style');
style.id = 'universal-dark-mode-style';
document.head.appendChild(style);
}
// Dark mode CSS - inverts colors but preserves images/video
style.textContent = '
/* Invert everything except media */
html {
filter: invert(1) hue-rotate(180deg) !important;
background: #1a1a2e !important;
}
/* Restore images, videos, iframes, canvas */
img, video, iframe, canvas, svg, picture, [style*="background-image"] {
filter: invert(1) hue-rotate(180deg) !important;
}
/* Preserve specific elements that should not be inverted */
.no-dark-mode, .no-dark-mode *,
[data-theme="light"], [data-theme="light"],
.ace_editor, .ace_editor *,
.CodeMirror, .CodeMirror *,
.monaco-editor, .monaco-editor *,
.markdown-body pre, .markdown-body pre *,
.highlight, .highlight *,
pre code, pre code * {
filter: none !important;
}
/* Fix common UI elements */
.modal, .popup, .dropdown-menu, .tooltip, .popover {
filter: invert(1) hue-rotate(180deg) !important;
background: #2d2d44 !important;
border-color: #444 !important;
}
/* Scrollbars */
::-webkit-scrollbar { background: #1a1a2e !important; }
::-webkit-scrollbar-thumb { background: #444 !important; }
::-webkit-scrollbar-thumb:hover { background: #555 !important; }
/* Selection */
::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
';
}
function removeDarkMode() {
var style = document.getElementById('universal-dark-mode-style');
if (style) style.remove();
}
// Toggle with Alt+Shift+D
document.addEventListener('keydown', function(e) {
if (e.altKey && e.shiftKey && e.key === 'D') {
e.preventDefault();
enabled = !enabled;
if (enabled) {
applyDarkMode();
console.log('[Universal Dark Mode] Enabled');
} else {
removeDarkMode();
console.log('[Universal Dark Mode] Disabled');
}
}
});
// Apply on load
applyDarkMode();
// Re-apply on dynamic content
var observer = new MutationObserver(function(mutations) {
if (enabled && !document.getElementById('universal-dark-mode-style')) {
applyDarkMode();
}
});
observer.observe(document.head, { childList: true });
console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');
})();
}
} catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); }
})();
})();

Skip to content
Add e2e test for XComObjectStorageBackend#62907

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

jason810496
merged 8 commits into

from

Mar 9, 2026
 Merged


Add e2e test for XComObjectStorageBackend
#62907
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


8 commits


Select commit
Hold shift + click to select a range
0635994

Add e2e test for XComObjectStorageBackend


jason810496 Mar 5, 2026
0600b02

Update GitHub Action and Breeze cmd


jason810496 Mar 5, 2026
1e9a06a

Add aws s3fs deps


jason810496 Mar 5, 2026
2bc77f0

Add debug log


jason810496 Mar 5, 2026
fcd0685

Add required AWS env to fix the e2e test failure


jason810496 Mar 6, 2026
e7865bf

Harden the client in e2e test


jason810496 Mar 6, 2026
b82137f

Fix nits


jason810496 Mar 6, 2026
0011121

Fix copilot suggestion


jason810496 Mar 6, 2026

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


11 changes: 11 additions & 0 deletions
.github/workflows/additional-prod-image-tests.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -226,6 +226,17 @@ jobs:
 use-uv: ${{ inputs.use-uv }}
 e2e_test_mode: "remote_log"

 test-e2e-integration-tests-xcom-object-storage:
 name: "XCom object storage backend tests with PROD image"
 uses: ./.github/workflows/airflow-e2e-tests.yml
 with:
 workflow-name: "XCom object storage backend e2e test"
 runners: ${{ inputs.runners }}
 platform: ${{ inputs.platform }}
 default-python-version: "${{ inputs.default-python-version }}"
 use-uv: ${{ inputs.use-uv }}
 e2e_test_mode: "xcom_object_storage"

 test-ui-e2e-chromium:
 name: "Chromium UI e2e tests with PROD image"
 uses: ./.github/workflows/ui-e2e-tests.yml
Expand Down

4 changes: 2 additions & 2 deletions
.github/workflows/airflow-e2e-tests.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -49,7 +49,7 @@ on: # yamllint disable-line rule:truthy
 type: string
 required: true
 e2e_test_mode:
 description: "Test mode - basicor remote_log"
 description: "Test mode - basic, remote_log, or xcom_object_storage"
 type: string
 default: "basic"

Expand DownExpand Up@@ -80,7 +80,7 @@ on: # yamllint disable-line rule:truthy
 type: string
 default: ""
 e2e_test_mode:
 description: "Test mode - quick or full"
 description: "Test mode - basic, remote_log, or xcom_object_storage"
 type: string
 default: "basic"

Expand Down

1 change: 1 addition & 0 deletions
airflow-e2e-tests/scripts/init-aws.sh

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,4 +17,5 @@
# under the License.

aws --endpoint-url=http://localstack:4566 s3 mb s3://test-airflow-logs
aws --endpoint-url=http://localstack:4566 s3 mb s3://test-xcom-objectstorage-backend
aws --endpoint-url=http://localstack:4566 s3 ls

32 changes: 31 additions & 1 deletion
airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -36,6 +36,7 @@
 LOCALSTACK_PATH,
 LOGS_FOLDER,
 TEST_REPORT_FILE,
 XCOM_BUCKET,
)

from tests_common.test_utils.fernet import generate_fernet_key_string
Expand All@@ -48,13 +49,18 @@ class _E2ETestState:
 airflow_logs_path: Path | None = None


def _setup_s3_integration(dot_env_file, tmp_dir):
def _copy_localstack_files(tmp_dir):
 """Copy localstack compose file and init script into the temp directory."""
 copyfile(LOCALSTACK_PATH, tmp_dir / "localstack.yml")

 copyfile(AWS_INIT_PATH, tmp_dir / "init-aws.sh")
 current_permissions = os.stat(tmp_dir / "init-aws.sh").st_mode
 os.chmod(tmp_dir / "init-aws.sh", current_permissions | 0o111)


def _setup_s3_integration(dot_env_file, tmp_dir):
 _copy_localstack_files(tmp_dir)

 dot_env_file.write_text(
 f"AIRFLOW_UID={os.getuid()}\n"
 "AWS_DEFAULT_REGION=us-east-1\n"
Expand All@@ -68,6 +74,27 @@ def _setup_s3_integration(dot_env_file, tmp_dir):
 os.environ["ENV_FILE_PATH"] = str(dot_env_file)


def _setup_xcom_object_storage_integration(dot_env_file, tmp_dir):
 _copy_localstack_files(tmp_dir)

 dot_env_file.write_text(
 f"AIRFLOW_UID={os.getuid()}\n"
 # XComObjectStorageBackend requires AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY as env vars
 # because `universal-path` uses boto3's native S3 client, which relies on environment variables
 # for authentication rather than parsing credentials from the connection URI
 "AWS_ACCESS_KEY_ID=test\n"
 "AWS_SECRET_ACCESS_KEY=test\n"
 "AWS_DEFAULT_REGION=us-east-1\n"
 "AWS_ENDPOINT_URL_S3=http://localstack:4566\n"
 "AIRFLOW_CONN_AWS_DEFAULT=aws://test:test@\n"
 "AIRFLOW__CORE__XCOM_BACKEND=airflow.providers.common.io.xcom.backend.XComObjectStorageBackend\n"
 f"AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_PATH=s3://aws_default@{XCOM_BUCKET}/xcom\n"
 "AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_THRESHOLD=0\n"
 "_PIP_ADDITIONAL_REQUIREMENTS=apache-airflow-providers-amazon[s3fs]\n"
 )
 os.environ["ENV_FILE_PATH"] = str(dot_env_file)


def spin_up_airflow_environment(tmp_path_factory: pytest.TempPathFactory):
 tmp_dir = tmp_path_factory.mktemp("airflow-e2e-tests")

Expand DownExpand Up@@ -97,6 +124,9 @@ def spin_up_airflow_environment(tmp_path_factory: pytest.TempPathFactory):
 if E2E_TEST_MODE == "remote_log":
 compose_file_names.append("localstack.yml")
 _setup_s3_integration(dot_env_file, tmp_dir)
 elif E2E_TEST_MODE == "xcom_object_storage":
 compose_file_names.append("localstack.yml")
 _setup_xcom_object_storage_integration(dot_env_file, tmp_dir)

 #
 # Please Do not use this Fernet key in any deployments! Please generate your own key.
Expand Down

3 changes: 3 additions & 0 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/constants.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -42,3 +42,6 @@
LOCALSTACK_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "docker" / "localstack.yml"
E2E_TEST_MODE = os.environ.get("E2E_TEST_MODE", "basic")
AWS_INIT_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "scripts" / "init-aws.sh"

# s3 bucket name for XComObjectStorageBackend tests. This bucket will be created in the `init-aws.sh` script that is run as part of the LocalStack container initialization.
XCOM_BUCKET = "test-xcom-objectstorage-backend"

51 changes: 43 additions & 8 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -20,6 +20,7 @@
from datetime import datetime, timezone
from functools import cached_property

import boto3
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
Expand All@@ -31,19 +32,41 @@
)


def get_s3_client():
 """Return a boto3 S3 client configured to use the local LocalStack endpoint."""
 return boto3.client(
 "s3",
 endpoint_url="http://localhost:4566",
 aws_access_key_id="test",
 aws_secret_access_key="test",
 region_name="us-east-1",
 )


def create_request_session_with_retries(status_forcelist: list[int]):
 """Create a requests Session with retry logic for handling transient errors."""
 Retry.DEFAULT_BACKOFF_MAX = 32
 retry_strategy = Retry(
 total=10,
 backoff_factor=1,
 status_forcelist=status_forcelist,
 )
 session = requests.Session()
 adapter = HTTPAdapter(max_retries=retry_strategy)
 session.mount("http://", adapter)
 session.mount("https://", adapter)
 return session


class AirflowClient:
 """Client for interacting with the Airflow REST API."""

 def __init__(self):
 self.session = requests.Session()
 self.session = create_request_session_with_retries(status_forcelist=[429])

 @cached_property
 def token(self):
 Retry.DEFAULT_BACKOFF_MAX = 32
 retry = Retry(total=10, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504])
 session = requests.Session()
 session.mount("http://", HTTPAdapter(max_retries=retry))
 session.mount("https://", HTTPAdapter(max_retries=retry))
 session = create_request_session_with_retries(status_forcelist=[429, 500, 502, 503, 504])

 api_server_url = DOCKER_COMPOSE_HOST_PORT
 if not api_server_url.startswith(("http://", "https://")):
Expand DownExpand Up@@ -121,11 +144,23 @@ def trigger_dag_and_wait(self, dag_id: str, json=None):
 run_id=resp["dag_run_id"],
 )

 def get_task_logs(self, dag_id: str, run_id: str, task_id: str, try_number: int = 1):
 def get_task_instances(self, dag_id: str, run_id: str):
 """Get task instances for a given DAG run."""
 return self._make_request(
 method="GET",
 endpoint=f"dags/{dag_id}/dagRuns/{run_id}/taskInstances",
 )

 def get_task_logs(
 self, dag_id: str, run_id: str, task_id: str, try_number: int = 1, map_index: int | None = None
 ):
 """Get task logs via API."""
 endpoint = f"dags/{dag_id}/dagRuns/{run_id}/taskInstances/{task_id}/logs/{try_number}"
 if map_index is not None:
 endpoint += f"?map_index={map_index}"
 return self._make_request(
 method="GET",
 endpoint=f"dags/{dag_id}/dagRuns/{run_id}/taskInstances/{task_id}/logs/{try_number}",
 endpoint=endpoint,
 )


Expand Down

12 changes: 3 additions & 9 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/remote_log_tests/test_remote_logging.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -19,10 +19,9 @@
import time
from datetime import datetime, timezone

import boto3
import pytest

from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient
from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient, get_s3_client


class TestRemoteLogging:
Expand DownExpand Up@@ -56,15 +55,10 @@ def test_remote_logging_s3(self):

 # This bucket will be created part of the docker-compose setup in
 bucket_name = "test-airflow-logs"
 s3_client = boto3.client(
 "s3",
 endpoint_url="http://localhost:4566",
 aws_access_key_id="test",
 aws_secret_access_key="test",
 region_name="us-east-1",
 )
 s3_client = get_s3_client()

 # Wait for logs to be available in S3 before we call `get_task_logs`
 contents = []
 for _ in range(self.max_retries):
 response = s3_client.list_objects_v2(Bucket=bucket_name)
 contents = response.get("Contents", [])
Expand Down

16 changes: 16 additions & 0 deletions
airflow-e2e-tests/tests/airflow_e2e_tests/xcom_object_storage_tests/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
# 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.

89 changes: 89 additions & 0 deletions
...sts/tests/airflow_e2e_tests/xcom_object_storage_tests/test_xcom_object_storage_backend.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
# 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 time
from datetime import datetime, timezone
from pprint import pprint
from uuid import uuid4

import pytest

from airflow_e2e_tests.constants import XCOM_BUCKET
from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient, get_s3_client


class TestXComObjectStorageBackend:
 airflow_client = AirflowClient()
 dag_id = "example_xcom_test"
 retry_interval_in_seconds = 5
 max_retries = 12

 def test_dag_succeeds_and_xcom_values_stored_in_s3(self):
 """Test that a DAG using XComObjectStorageBackend completes successfully and persists XCom values to S3."""
 self.airflow_client.un_pause_dag(self.dag_id)

 trigger_resp = self.airflow_client.trigger_dag(
 self.dag_id,
 json={
 "dag_run_id": f"test_xcom_object_storage_backend_{uuid4()}",
 "logical_date": datetime.now(timezone.utc).isoformat(),
 },
 )
 dag_run_id = trigger_resp["dag_run_id"]
 state = self.airflow_client.wait_for_dag_run(
 dag_id=self.dag_id,
 run_id=dag_run_id,
 )

 # try to get all the logs to help debugging
 if state != "success":
 task_instances_resp = self.airflow_client.get_task_instances(self.dag_id, dag_run_id)
 for task_instance in task_instances_resp["task_instances"]:
 task_id = task_instance["task_id"]
 try_number = task_instance["try_number"]
 try:
 print(f"\nLogs for task {task_id} (try {try_number}):")
 task_logs_resp = self.airflow_client.get_task_logs(
 dag_id=self.dag_id, task_id=task_id, run_id=dag_run_id, try_number=try_number
 )
 pprint(task_logs_resp)
 except Exception as e:
 print(f"Could not get logs for task {task_id} (try {try_number}): {e}")

 assert state == "success", f"DAG {self.dag_id} did not complete successfully. Final state: {state}"

 s3_client = get_s3_client()

 contents = []
 for _ in range(self.max_retries):
 response = s3_client.list_objects_v2(Bucket=XCOM_BUCKET)
 contents = response.get("Contents", [])
 if contents:
 break

 print(f"No XCom objects found in S3 bucket {XCOM_BUCKET!r} yet. Retrying...")
 time.sleep(self.retry_interval_in_seconds)

 if not contents:
 pytest.fail(
 f"Expected XCom objects in S3 bucket {XCOM_BUCKET!r}, but bucket is empty.\n"
 f"List Objects Response: {response}"
 )

 keys = [obj["Key"] for obj in contents]
 print(f"Found {len(keys)} XCom object(s) in S3: {keys}")
Loading
Loading