Files
distributed-execution/tests/test_tightly_coupled.py
ILay 9b1b7056b0 [SIMPL-30451] Close the distributed execution reference service
Check the chart's actual DAGSTER_PG_PASSWORD injection in the tightly coupled preflight instead of DAGSTER_POSTGRES_*, which the Simpl chart never sets; the old check warned about a misconfiguration on a correctly configured run pod. Evidence reports the variable name only, never its value.

Record the platform k8s_job_executor run as cluster-verified in the guide and readiness checklist, add the Dagster UI screenshots covering job list, graph, code-configured tags and a successful run.

Changelog: fixed
2026-09-02 18:47:27 +02:00

98 lines
3.3 KiB
Python

"""Verifies that execution-target linkage is what the guide claims it is."""
import pytest
from distributed_execution.preflight import check_env_vars, describe_pod_identity
from distributed_execution.repository import defs
from distributed_execution.tightly_coupled.jobs import (
SHARED_IO_BASE_DIR,
STEP_K8S_CONFIG,
tightly_coupled_in_process_job,
tightly_coupled_k8s_job,
tightly_coupled_local_job,
)
EXPECTED_JOBS = {
"tightly_coupled_in_process_job",
"tightly_coupled_local_job",
"tightly_coupled_k8s_job",
}
def test_repository_registers_tightly_coupled_jobs():
names = {job.name for job in defs.jobs}
assert EXPECTED_JOBS <= names
@pytest.mark.parametrize(
("job", "expected_executor"),
[
(tightly_coupled_in_process_job, "in_process"),
(tightly_coupled_local_job, "multiprocess"),
(tightly_coupled_k8s_job, "k8s_job"),
],
)
def test_each_job_declares_its_execution_target(job, expected_executor):
assert job.tags["execution_target"] == "tightly_coupled"
assert job.tags["executor"] == expected_executor
def test_k8s_step_pods_mount_shared_dagster_storage():
container_config = STEP_K8S_CONFIG["container_config"]
pod_spec_config = STEP_K8S_CONFIG["pod_spec_config"]
assert {"name": "dagster-shared-storage", "mount_path": "/dagster/shared"} in container_config["volume_mounts"]
assert {
"name": "dagster-shared-storage",
"persistent_volume_claim": {"claim_name": "dagster-shared-pvc"},
} in pod_spec_config["volumes"]
assert SHARED_IO_BASE_DIR.startswith("/dagster/shared/")
def test_k8s_job_pins_io_manager_to_shared_storage():
io_manager = tightly_coupled_k8s_job.resource_defs["io_manager"]
resolved = io_manager.apply_config_mapping({}).value
assert resolved["config"]["base_dir"] == SHARED_IO_BASE_DIR
def test_in_process_job_runs_end_to_end():
result = tightly_coupled_in_process_job.execute_in_process()
assert result.success
summary = result.output_for_node("summarise_results")
assert summary["units"] == 4
assert summary["total"] == 14 # 0 + 1 + 4 + 9
# In-process execution keeps every step in the run worker's own process.
assert len(summary["contributing_workers"]) == 1
def test_work_units_fan_out_into_one_step_each():
result = tightly_coupled_in_process_job.execute_in_process()
mapped = result.output_for_node("process_work_unit")
# Without this the executor choice would be meaningless: one step cannot span pods.
assert set(mapped) == {"unit_0", "unit_1", "unit_2", "unit_3"}
def test_env_var_check_reports_missing_names():
result = check_env_vars(("DEFINITELY_NOT_SET_12345",))
assert result["passed"] is False
assert result["missing"] == ["DEFINITELY_NOT_SET_12345"]
def test_env_var_check_reports_present_names(monkeypatch):
monkeypatch.setenv("DAGSTER_PG_PASSWORD", "not-exposed-in-evidence")
result = check_env_vars(("DAGSTER_PG_PASSWORD",))
assert result["passed"] is True
assert result["present"] == ["DAGSTER_PG_PASSWORD"]
assert "not-exposed-in-evidence" not in repr(result)
def test_pod_identity_falls_back_outside_kubernetes(monkeypatch):
monkeypatch.delenv("DAGSTER_K8S_PIPELINE_RUN_NAMESPACE", raising=False)
assert describe_pod_identity()["namespace"] == "<not-in-kubernetes>"