test_dags

compare_submit_files

compare_submit_files(file1, file2)

Compare two submit files by loading them as Submit objects.

This provides a semantic comparison rather than exact string matching, making tests more robust to formatting changes.

Source code in ezdag/tests/test_dags.py

def compare_submit_files(file1: Path, file2: Path) -> bool:
    """Compare two submit files by loading them as Submit objects.

    This provides a semantic comparison rather than exact string matching,
    making tests more robust to formatting changes.
    """
    # Read both files
    submit1_text = file1.read_text()
    submit2_text = file2.read_text()

    # Parse into Submit objects
    submit1 = htcondor.Submit(submit1_text)
    submit2 = htcondor.Submit(submit2_text)

    # Get all keys from both submits
    keys1 = set(submit1.keys())
    keys2 = set(submit2.keys())

    # Check if keys match
    if keys1 != keys2:
        missing_in_1 = keys2 - keys1
        missing_in_2 = keys1 - keys2
        if missing_in_1:
            print(f"Keys in {file2.name} but not in {file1.name}: {missing_in_1}")
        if missing_in_2:
            print(f"Keys in {file1.name} but not in {file2.name}: {missing_in_2}")
        return False

    # Compare values for each key
    for key in keys1:
        val1 = submit1.get(key, "").strip()
        val2 = submit2.get(key, "").strip()
        if val1 != val2:
            print(f"Value mismatch for key '{key}':")
            print(f"  {file1.name}: {val1}")
            print(f"  {file2.name}: {val2}")
            return False

    return True

test_absolute_log_dir

test_absolute_log_dir(mock_which, tmp_path)

Layers with absolute log_dir create that directory directly.

Source code in ezdag/tests/test_dags.py

@mock.patch("shutil.which", side_effect=lambda x: f"/path/to/{x}")
def test_absolute_log_dir(mock_which, tmp_path):
    """Layers with absolute log_dir create that directory directly."""
    dag = DAG("abs_log")
    abs_log = str(tmp_path / "absolute_logs")

    layer = Layer("my_exec", log_dir=abs_log)
    layer += Node(arguments=Argument("x", 1))
    dag.attach(layer)

    dag.write(tmp_path)

    assert Path(abs_log).is_dir()

test_duplicate_layer_names

test_duplicate_layer_names(mock_which, tmp_path)

Attaching layers with the same name appends a counter.

Source code in ezdag/tests/test_dags.py

@mock.patch("shutil.which", side_effect=lambda x: f"/path/to/{x}")
def test_duplicate_layer_names(mock_which, tmp_path):
    """Attaching layers with the same name appends a counter."""
    dag = DAG("dup_dag")

    for i in range(2):
        layer = Layer("same_name")
        layer += Node(
            arguments=Argument("idx", i),
            outputs=Argument("out", f"out_{i}.txt"),
        )
        dag.attach(layer)

    dag.write(tmp_path)

    # second layer should be renamed to same_name2
    assert "same_name" in dag._layers
    assert "same_name2" in dag._layers

test_multi_parent_edges

test_multi_parent_edges(mock_which, tmp_path)

A child layer consuming outputs from two different parent layers.

Source code in ezdag/tests/test_dags.py

@mock.patch("shutil.which", side_effect=lambda x: f"/path/to/{x}")
def test_multi_parent_edges(mock_which, tmp_path):
    """A child layer consuming outputs from two different parent layers."""
    dag = DAG("multi_parent")

    # parent layer A
    layer_a = Layer("parent_a")
    layer_a += Node(outputs=Argument("out", "a_out.txt"))
    dag.attach(layer_a)

    # parent layer B
    layer_b = Layer("parent_b")
    layer_b += Node(outputs=Argument("out", "b_out.txt"))
    dag.attach(layer_b)

    # child depends on both
    child = Layer("child")
    child += Node(
        inputs=Argument("data", ["a_out.txt", "b_out.txt"]),
        outputs=Argument("out", "child_out.txt"),
    )
    dag.attach(child)

    dag.write(tmp_path)

    # child should have edges from both parents
    assert "parent_a" in dag._all_edges["child"]
    assert "parent_b" in dag._all_edges["child"]

test_visualize

test_visualize(mock_which, tmp_path)

Test DAG visualization produces an output file.

Source code in ezdag/tests/test_dags.py

@mock.patch("shutil.which", side_effect=lambda x: f"/path/to/{x}")
def test_visualize(mock_which, tmp_path):
    """Test DAG visualization produces an output file."""
    pytest.importorskip("graphviz")

    dag = DAG("viz_dag")

    process_layer = Layer("process")
    for i in range(2):
        process_layer += Node(
            inputs=Option("input", "data.txt"),
            outputs=Argument("out", f"out_{i}.txt"),
        )
    dag.attach(process_layer)

    combine_layer = Layer("combine")
    combine_layer += Node(
        inputs=Argument("data", ["out_0.txt", "out_1.txt"]),
    )
    dag.attach(combine_layer)

    dag.write(tmp_path)
    dag.visualize(tmp_path)

    assert (tmp_path / "viz_dag.svg").exists()