You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ta...@apache.org on 2024/03/22 23:47:23 UTC
(airflow) 01/01: Use `python-on-whales` in docker tests
This is an automated email from the ASF dual-hosted git repository.
taragolis pushed a commit to branch modify-docker-tests
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit d9549bbadeee2193037e4a2ae88bf255a315bb09
Author: Andrey Anshin <An...@taragol.is>
AuthorDate: Sat Mar 23 03:40:48 2024 +0400
Use `python-on-whales` in docker tests
---
docker_tests/constants.py | 5 +
.../{docker_tests_utils.py => docker_utils.py} | 102 ++++++++++-------
docker_tests/requirements.txt | 4 +-
docker_tests/test_ci_image.py | 28 ++---
docker_tests/test_docker_compose_quick_start.py | 122 +++++++--------------
.../test_examples_of_prod_image_building.py | 19 ++--
docker_tests/test_prod_image.py | 88 +++++----------
7 files changed, 160 insertions(+), 208 deletions(-)
diff --git a/docker_tests/constants.py b/docker_tests/constants.py
index db79d1d862..83c77e2fb4 100644
--- a/docker_tests/constants.py
+++ b/docker_tests/constants.py
@@ -16,6 +16,11 @@
# under the License.
from __future__ import annotations
+import os
from pathlib import Path
SOURCE_ROOT = Path(__file__).resolve().parents[1]
+
+DEFAULT_PYTHON_MAJOR_MINOR_VERSION = "3.8"
+DEFAULT_DOCKER_IMAGE = f"ghcr.io/apache/airflow/main/prod/python{DEFAULT_PYTHON_MAJOR_MINOR_VERSION}:latest"
+DOCKER_IMAGE = os.environ.get("DOCKER_IMAGE") or DEFAULT_DOCKER_IMAGE
diff --git a/docker_tests/docker_tests_utils.py b/docker_tests/docker_utils.py
similarity index 51%
rename from docker_tests/docker_tests_utils.py
rename to docker_tests/docker_utils.py
index 7eea98e9bd..2eb67c9c88 100644
--- a/docker_tests/docker_tests_utils.py
+++ b/docker_tests/docker_utils.py
@@ -16,50 +16,49 @@
# under the License.
from __future__ import annotations
-import os
-
-from docker_tests.command_utils import run_command
+from time import monotonic, sleep
+
+from python_on_whales import docker
+from python_on_whales.exceptions import NoSuchContainer
+
+from docker_tests.constants import DEFAULT_DOCKER_IMAGE
+
+
+def run_cmd_in_docker(
+ cmd: list[str] | None = None,
+ docker_image: str = DEFAULT_DOCKER_IMAGE,
+ entrypoint: str | None = None,
+ envs: dict[str, str] | None = None,
+ remove: bool = True,
+ **kwargs,
+):
+ kwargs.pop("image", None)
+ cmd = cmd or []
+ envs = envs or {}
+ return docker.run(
+ image=docker_image,
+ entrypoint=entrypoint,
+ command=cmd,
+ remove=remove,
+ envs={"COLUMNS": "180", **envs},
+ **kwargs,
+ )
-DEFAULT_PYTHON_MAJOR_MINOR_VERSION = "3.8"
-docker_image = os.environ.get(
- "DOCKER_IMAGE", f"ghcr.io/apache/airflow/main/prod/python{DEFAULT_PYTHON_MAJOR_MINOR_VERSION}:latest"
-)
+def run_bash_in_docker(bash_script: str, **kwargs):
+ kwargs.pop("entrypoint", None)
+ return run_cmd_in_docker(cmd=["-c", bash_script], entrypoint="/bin/bash", **kwargs)
-print("Using docker image: ", docker_image)
+def run_python_in_docker(python_script, **kwargs):
+ kwargs.pop("entrypoint", None)
+ envs = {"PYTHONDONTWRITEBYTECODE": "true", **kwargs.pop("envs", {})}
+ return run_cmd_in_docker(cmd=["python", "-c", python_script], envs=envs, **kwargs)
-def run_bash_in_docker(bash_script, **kwargs):
- docker_command = [
- "docker",
- "run",
- "--rm",
- "-e",
- "COLUMNS=180",
- "--entrypoint",
- "/bin/bash",
- docker_image,
- "-c",
- bash_script,
- ]
- return run_command(docker_command, **kwargs)
-
-def run_python_in_docker(python_script, **kwargs):
- docker_command = [
- "docker",
- "run",
- "--rm",
- "-e",
- "COLUMNS=180",
- "-e",
- "PYTHONDONTWRITEBYTECODE=true",
- docker_image,
- "python",
- "-c",
- python_script,
- ]
- return run_command(docker_command, **kwargs)
+def run_airflow_cmd_in_docker(cmd: list[str] | None = None, **kwargs):
+ kwargs.pop("entrypoint", None)
+ return run_cmd_in_docker(cmd=["airflow", *(cmd or [])], **kwargs)
def display_dependency_conflict_message():
@@ -103,3 +102,30 @@ Production image:
***** End of the instructions ****
"""
)
+
+
+def wait_for_container(container_id: str, timeout: int = 300):
+ print(f"Waiting for container: [{container_id}] for {timeout} more seconds.")
+ start_time = monotonic()
+ while True:
+ if timeout != 0 and monotonic() - start_time > timeout:
+ err_msg = f"Timeout. The operation takes longer than the maximum waiting time ({timeout}s)"
+ raise TimeoutError(err_msg)
+
+ try:
+ container = docker.container.inspect("container_id")
+ except NoSuchContainer:
+ sleep(5)
+ continue
+ container.export()
+ if (state := container.state).status in ("running", "restarting"):
+ if state.health_status is None or state.health_status.status == "healthy":
+ break
+ elif state.status == "exited" and state.exit_status == 0:
+ break
+
+ msg = f"Container {container.name}[{id}] has state:\n{state}"
+ if timeout != 0:
+ msg += f"\nWaiting for {int(timeout - (monotonic() - start_time))} more seconds"
+ print(msg)
+ sleep(1)
diff --git a/docker_tests/requirements.txt b/docker_tests/requirements.txt
index 60b4005325..61a70cded4 100644
--- a/docker_tests/requirements.txt
+++ b/docker_tests/requirements.txt
@@ -2,4 +2,6 @@
# Internal meta-task for track https://github.com/apache/airflow/issues/37156
pytest>=7.4.4,<8.0
pytest-xdist
-requests
+# Requests 3 if it will be released, will be heavily breaking.
+requests>=2.27.0,<3
+python-on-whales>=0.70.0
diff --git a/docker_tests/test_ci_image.py b/docker_tests/test_ci_image.py
index 7ec65c425c..a5b1c51214 100644
--- a/docker_tests/test_ci_image.py
+++ b/docker_tests/test_ci_image.py
@@ -16,34 +16,22 @@
# under the License.
from __future__ import annotations
-import subprocess
+from python_on_whales import DockerException
-from docker_tests.command_utils import run_command
-from docker_tests.docker_tests_utils import display_dependency_conflict_message, docker_image
+from docker_tests.docker_utils import display_dependency_conflict_message, run_bash_in_docker
def test_pip_dependencies_conflict():
try:
- run_command(["docker", "run", "--rm", "--entrypoint", "/bin/bash", docker_image, "-c", "pip check"])
- except subprocess.CalledProcessError as ex:
+ run_bash_in_docker("pip check")
+ except DockerException:
display_dependency_conflict_message()
- raise ex
+ raise
def test_providers_present():
try:
- run_command(
- [
- "docker",
- "run",
- "--rm",
- "--entrypoint",
- "/bin/bash",
- docker_image,
- "-c",
- "airflow providers list",
- ],
- )
- except subprocess.CalledProcessError as ex:
+ run_bash_in_docker("airflow providers list")
+ except DockerException:
display_dependency_conflict_message()
- raise ex
+ raise
diff --git a/docker_tests/test_docker_compose_quick_start.py b/docker_tests/test_docker_compose_quick_start.py
index 7c70c39f05..dbe745524f 100644
--- a/docker_tests/test_docker_compose_quick_start.py
+++ b/docker_tests/test_docker_compose_quick_start.py
@@ -19,18 +19,18 @@ from __future__ import annotations
import json
import os
import shlex
-import subprocess
-import sys
from pprint import pprint
from shutil import copyfile
-from time import monotonic, sleep
+from time import sleep
+import pytest
import requests
+from python_on_whales import DockerClient, docker
+from python_on_whales.exceptions import DockerException
# isort:off (needed to workaround isort bug)
from docker_tests.command_utils import run_command
-from docker_tests.constants import SOURCE_ROOT
-from docker_tests.docker_tests_utils import docker_image
+from docker_tests.constants import DOCKER_IMAGE, SOURCE_ROOT
# isort:on (needed to workaround isort bug)
@@ -52,56 +52,11 @@ def api_request(method: str, path: str, base_url: str = "http://localhost:8080/a
return response.json()
-def wait_for_container(container_id: str, timeout: int = 300):
- container_name = (
- subprocess.check_output(["docker", "inspect", container_id, "--format", "{{ .Name }}"])
- .decode()
- .strip()
- )
- print(f"Waiting for container: {container_name} [{container_id}] for {timeout} more seconds.")
- waiting_done = False
- start_time = monotonic()
- while not waiting_done:
- container_state = (
- subprocess.check_output(["docker", "inspect", container_id, "--format", "{{ .State.Status }}"])
- .decode()
- .strip()
- )
- if container_state in ("running", "restarting"):
- health_status = (
- subprocess.check_output(
- [
- "docker",
- "inspect",
- container_id,
- "--format",
- "{{ if .State.Health }}{{ .State.Health.Status }}{{ else }}no-check{{ end }}",
- ]
- )
- .decode()
- .strip()
- )
- current_time = monotonic()
- print(
- f"{container_name}: container_state={container_state}, health_status={health_status}. "
- f"Waiting for {int(timeout - (current_time - start_time))} more seconds"
- )
-
- if health_status == "healthy" or health_status == "no-check":
- waiting_done = True
- else:
- print(f"{container_name}: container_state={container_state}")
- waiting_done = True
- if timeout != 0 and monotonic() - start_time > timeout:
- raise Exception(f"Timeout. The operation takes longer than the maximum waiting time ({timeout}s)")
- sleep(1)
-
-
def wait_for_terminal_dag_state(dag_id, dag_run_id):
print(f" Simplified representation of DAG {dag_id} ".center(72, "="))
pprint(api_request("GET", f"dags/{DAG_ID}/details"))
- # Wait 80 seconds
+ # Wait 400 seconds
for _ in range(400):
dag_state = api_request("GET", f"dags/{dag_id}/dagRuns/{dag_run_id}").get("state")
print(f"Waiting for DAG Run: dag_state={dag_state}")
@@ -113,8 +68,7 @@ def wait_for_terminal_dag_state(dag_id, dag_run_id):
def test_trigger_dag_and_wait_for_result(tmp_path_factory, monkeypatch):
"""Simple test which reproduce setup docker-compose environment and trigger example dag."""
tmp_dir = tmp_path_factory.mktemp("airflow-quick-start")
- monkeypatch.chdir(tmp_dir)
- monkeypatch.setenv("AIRFLOW_IMAGE_NAME", docker_image)
+ monkeypatch.setenv("AIRFLOW_IMAGE_NAME", DOCKER_IMAGE)
compose_file_path = (
SOURCE_ROOT / "docs" / "apache-airflow" / "howto" / "docker-compose" / "docker-compose.yaml"
@@ -130,19 +84,24 @@ def test_trigger_dag_and_wait_for_result(tmp_path_factory, monkeypatch):
print(" .env file content ".center(72, "="))
print(dot_env_file.read_text())
- # check if docker-compose is available
- compose_command = ["docker", "compose"]
- success = run_command([*compose_command, "version"], check=False)
- if not success:
- print("ERROR: `docker compose` not available. Make sure compose plugin is installed")
- sys.exit(1)
- compose_command.extend(["--project-name", "quick-start"])
- run_command([*compose_command, "config"])
- run_command([*compose_command, "down", "--volumes", "--remove-orphans"])
- run_command([*compose_command, "up", "-d", "--wait"])
- api_request("PATCH", path=f"dags/{DAG_ID}", json={"is_paused": False})
- api_request("POST", path=f"dags/{DAG_ID}/dagRuns", json={"dag_run_id": DAG_RUN_ID})
+ compose_version = None
try:
+ compose_version = docker.compose.version()
+ except DockerException:
+ pytest.fail("`docker compose` not available. Make sure compose plugin is installed")
+ try:
+ docker_version = docker.version()
+ except NotImplementedError:
+ docker_version = run_command(["docker", "version"], return_output=True)
+
+ compose = DockerClient(compose_project_name="quick-start", compose_project_directory=tmp_dir).compose
+ compose.down(remove_orphans=True, volumes=True, quiet=True)
+ try:
+ compose.up(detach=True, wait=True, color=not os.environ.get("NO_COLOR"))
+
+ api_request("PATCH", path=f"dags/{DAG_ID}", json={"is_paused": False})
+ api_request("POST", path=f"dags/{DAG_ID}/dagRuns", json={"dag_run_id": DAG_RUN_ID})
+
wait_for_terminal_dag_state(dag_id=DAG_ID, dag_run_id=DAG_RUN_ID)
dag_state = api_request("GET", f"dags/{DAG_ID}/dagRuns/{DAG_RUN_ID}").get("state")
assert dag_state == "success"
@@ -153,26 +112,29 @@ def test_trigger_dag_and_wait_for_result(tmp_path_factory, monkeypatch):
pprint(api_request("GET", f"dags/{DAG_ID}/dagRuns"))
print(f"HTTP: GET dags/{DAG_ID}/dagRuns/{DAG_RUN_ID}/taskInstances")
pprint(api_request("GET", f"dags/{DAG_ID}/dagRuns/{DAG_RUN_ID}/taskInstances"))
- print(f"Current working directory: {os.getcwd()}")
- run_command(["docker", "version"])
- run_command([*compose_command, "version"])
- run_command(["docker", "ps"])
- run_command([*compose_command, "logs"])
- ps_output = run_command([*compose_command, "ps", "--format", "json"], return_output=True)
- container_names = [container["Name"] for container in json.loads(ps_output)]
- for container in container_names:
- print(f"Health check for {container}")
- result = run_command(
- ["docker", "inspect", "--format", "{{json .State}}", container], return_output=True
- )
- pprint(json.loads(result))
+ print(" Docker Version ".center(72, "="))
+ print(docker_version)
+ print(" Docker Compose Version ".center(72, "="))
+ print(compose_version)
+ print(" Compose Config ".center(72, "="))
+ print(json.dumps(compose.config(return_json=True), indent=4))
+
+ for service in compose.ps(all=True):
+ print(f" Service: {service.name} ".center(72, "-"))
+ print(" Service State ".center(72, "."))
+ pprint(service.state)
+ print(" Service Config ".center(72, "."))
+ pprint(service.config)
+ print(" Service Logs ".center(72, "."))
+ print(service.logs())
raise
finally:
if not os.environ.get("SKIP_DOCKER_COMPOSE_DELETION"):
- run_command([*compose_command, "down", "--volumes"])
+ compose.down(remove_orphans=True, volumes=True, quiet=True)
print("Docker compose instance deleted")
else:
print("Skipping docker-compose deletion")
print()
print("You can run inspect your docker-compose by running commands starting with:")
- print(" ".join([shlex.quote(arg) for arg in compose_command]))
+ quoted_command = map(shlex.quote, map(str, compose.docker_compose_cmd))
+ print(" ".join(quoted_command))
diff --git a/docker_tests/test_examples_of_prod_image_building.py b/docker_tests/test_examples_of_prod_image_building.py
index 991f9c8e2b..6931a78b79 100644
--- a/docker_tests/test_examples_of_prod_image_building.py
+++ b/docker_tests/test_examples_of_prod_image_building.py
@@ -24,6 +24,7 @@ from pathlib import Path
import pytest
import requests
+from python_on_whales import docker
# isort:off (needed to workaround isort bug)
from docker_tests.command_utils import run_command
@@ -52,17 +53,21 @@ def test_shell_script_example(script_file):
@pytest.mark.parametrize("dockerfile", glob.glob(f"{DOCKER_EXAMPLES_DIR}/**/Dockerfile", recursive=True))
-def test_dockerfile_example(dockerfile):
+def test_dockerfile_example(dockerfile, tmp_path):
rel_dockerfile_path = Path(dockerfile).relative_to(DOCKER_EXAMPLES_DIR)
image_name = str(rel_dockerfile_path).lower().replace("/", "-")
content = Path(dockerfile).read_text()
test_image = os.environ.get("TEST_IMAGE", get_latest_airflow_image())
- new_content = re.sub(r"FROM apache/airflow:.*", rf"FROM {test_image}", content)
+
+ test_image_file = tmp_path / image_name
+ test_image_file.write_text(re.sub(r"FROM apache/airflow:.*", rf"FROM {test_image}", content))
try:
- run_command(
- ["docker", "build", ".", "--tag", image_name, "-f", "-"],
- cwd=str(Path(dockerfile).parent),
- input=new_content.encode(),
+ image = docker.build(
+ context_path=Path(dockerfile).parent,
+ tags=image_name,
+ file=test_image_file,
+ load=True, # Load image to docker daemon
)
+ assert image
finally:
- run_command(["docker", "rmi", "--force", image_name])
+ docker.image.remove(image_name, force=True)
diff --git a/docker_tests/test_prod_image.py b/docker_tests/test_prod_image.py
index ac56cc6eff..d046c3d59a 100644
--- a/docker_tests/test_prod_image.py
+++ b/docker_tests/test_prod_image.py
@@ -18,18 +18,18 @@ from __future__ import annotations
import json
import os
-import subprocess
from importlib.util import find_spec
from pathlib import Path
import pytest
+from python_on_whales import DockerException
-from docker_tests.command_utils import run_command
from docker_tests.constants import SOURCE_ROOT
-from docker_tests.docker_tests_utils import (
+from docker_tests.docker_utils import (
display_dependency_conflict_message,
- docker_image,
+ run_airflow_cmd_in_docker,
run_bash_in_docker,
+ run_cmd_in_docker,
run_python_in_docker,
)
@@ -51,38 +51,29 @@ REGULAR_IMAGE_PROVIDERS = [
class TestCommands:
def test_without_command(self):
"""Checking the image without a command. It should return non-zero exit code."""
- with pytest.raises(subprocess.CalledProcessError) as ctx:
- run_command(["docker", "run", "--rm", "-e", "COLUMNS=180", docker_image])
- assert 2 == ctx.value.returncode
+ with pytest.raises(DockerException) as ctx:
+ run_cmd_in_docker()
+ assert 2 == ctx.value.return_code
def test_airflow_command(self):
- """Checking 'airflow' command It should return non-zero exit code."""
- with pytest.raises(subprocess.CalledProcessError) as ctx:
- run_command(["docker", "run", "--rm", "-e", "COLUMNS=180", docker_image, "airflow"])
- assert 2 == ctx.value.returncode
+ """Checking 'airflow' command. It should return non-zero exit code."""
+ with pytest.raises(DockerException) as ctx:
+ run_airflow_cmd_in_docker()
+ assert 2 == ctx.value.return_code
def test_airflow_version(self):
- """Checking 'airflow version' command It should return zero exit code."""
- output = run_command(
- ["docker", "run", "--rm", "-e", "COLUMNS=180", docker_image, "airflow", "version"],
- return_output=True,
- )
+ """Checking 'airflow version' command. It should return zero exit code."""
+ output = run_airflow_cmd_in_docker(["version"])
assert "2." in output
def test_python_version(self):
- """Checking 'python --version' command It should return zero exit code."""
- output = run_command(
- ["docker", "run", "--rm", "-e", "COLUMNS=180", docker_image, "python", "--version"],
- return_output=True,
- )
+ """Checking 'python --version' command. It should return zero exit code."""
+ output = run_cmd_in_docker(cmd=["python", "--version"])
assert "Python 3." in output
def test_bash_version(self):
"""Checking 'bash --version' command It should return zero exit code."""
- output = run_command(
- ["docker", "run", "--rm", "-e", "COLUMNS=180", docker_image, "bash", "--version"],
- return_output=True,
- )
+ output = run_cmd_in_docker(cmd=["bash", "--version"])
assert "GNU bash," in output
@@ -95,9 +86,7 @@ class TestPythonPackages:
packages_to_install = set(REGULAR_IMAGE_PROVIDERS)
package_file = PROD_IMAGE_PROVIDERS_FILE_PATH
assert len(packages_to_install) != 0
- output = run_bash_in_docker(
- "airflow providers list --output json", stderr=subprocess.DEVNULL, return_output=True
- )
+ output = run_bash_in_docker("airflow providers list --output json")
providers = json.loads(output)
packages_installed = set(d["package_name"] for d in providers)
assert len(packages_installed) != 0
@@ -109,9 +98,9 @@ class TestPythonPackages:
def test_pip_dependencies_conflict(self):
try:
run_bash_in_docker("pip check")
- except subprocess.CalledProcessError as ex:
+ except DockerException:
display_dependency_conflict_message()
- raise ex
+ raise
PACKAGE_IMPORTS = {
"amazon": ["boto3", "botocore", "watchtower"],
@@ -185,41 +174,16 @@ class TestPythonPackages:
class TestExecuteAsRoot:
def test_execute_airflow_as_root(self):
- run_command(
- [
- "docker",
- "run",
- "--rm",
- "--user",
- "0",
- "-e",
- "PYTHONDONTWRITEBYTECODE=true",
- docker_image,
- "airflow",
- "info",
- ]
- )
+ run_cmd_in_docker(cmd=["airflow", "info"], user=0)
def test_run_custom_python_packages_as_root(self, tmp_path):
(tmp_path / "__init__.py").write_text("")
(tmp_path / "awesome.py").write_text('print("Awesome")')
- run_command(
- [
- "docker",
- "run",
- "--rm",
- "-e",
- f"PYTHONPATH={tmp_path}",
- "-e",
- "PYTHONDONTWRITEBYTECODE=true",
- "-v",
- f"{tmp_path}:{tmp_path}",
- "--user",
- "0",
- docker_image,
- "python",
- "-c",
- "import awesome",
- ]
+ output = run_cmd_in_docker(
+ envs={"PYTHONPATH": "/custom/mount"},
+ volumes=[(tmp_path.as_posix(), "/custom/mount")],
+ user=0,
+ cmd=["python", "-c", "import awesome"],
)
+ assert output.strip() == "Awesome"