Airflow `dag.test()` raises `AirflowException: Cannot create DagRun for DAG because the dag is not serialized`

[tatiana]

Apache Airflow version

3.1.0

If "Other Airflow 2/3 version" selected, which one?

No response

What happened?

Airflow dag.test() stopped working in Airflow 3.1.

What you think should happen instead?

The command dag.test() should work as in Airlfow 3.0.

How to reproduce

Create this DAG file as example_bug.py:

from airflow import DAG
from airflow.decorators import dag, task

@dag(
    dag_id="simple_dag",
)
def simple_dag():
    """A simple DAG with two Python tasks."""

    @task
    def start():
        print("Starting the DAG...")
        return "Hello from start!"

    @task
    def end(message: str):
        print(f"Received message: {message}")
        print("Ending the DAG!")

    msg = start()
    end(msg)


dag = simple_dag()

if __name__ == "__main__":
    dag.test()

And run in a Python virtual environment with:

 python example_bug.py

Observe stacktrace similar to:

Traceback (most recent call last):
  File "/Users/tatiana.alchueyr/Code/astronomer-cosmos/example_bug.py", line 27, in <module>
    dag.test()
  File "/Users/tatiana.alchueyr/Library/Application Support/hatch/env/virtual/astronomer-cosmos/4lAAXoPP/tests.py3.10-3.1-1.10/lib/python3.10/site-packages/airflow/sdk/definitions/dag.py", line 1202, in test
    dr: DagRun = get_or_create_dagrun(
  File "/Users/tatiana.alchueyr/Library/Application Support/hatch/env/virtual/astronomer-cosmos/4lAAXoPP/tests.py3.10-3.1-1.10/lib/python3.10/site-packages/airflow/models/dagrun.py", line 2170, in get_or_create_dagrun
    dr = dag.create_dagrun(
  File "/Users/tatiana.alchueyr/Library/Application Support/hatch/env/virtual/astronomer-cosmos/4lAAXoPP/tests.py3.10-3.1-1.10/lib/python3.10/site-packages/airflow/utils/session.py", line 98, in wrapper
    return func(*args, **kwargs)
  File "/Users/tatiana.alchueyr/Library/Application Support/hatch/env/virtual/astronomer-cosmos/4lAAXoPP/tests.py3.10-3.1-1.10/lib/python3.10/site-packages/airflow/serialization/serialized_objects.py", line 3257, in create_dagrun
    orm_dagrun = _create_orm_dagrun(
  File "/Users/tatiana.alchueyr/Library/Application Support/hatch/env/virtual/astronomer-cosmos/4lAAXoPP/tests.py3.10-3.1-1.10/lib/python3.10/site-packages/airflow/utils/session.py", line 98, in wrapper
    return func(*args, **kwargs)
  File "/Users/tatiana.alchueyr/Library/Application Support/hatch/env/virtual/astronomer-cosmos/4lAAXoPP/tests.py3.10-3.1-1.10/lib/python3.10/site-packages/airflow/serialization/serialized_objects.py", line 2314, in _create_orm_dagrun
    raise AirflowException(f"Cannot create DagRun for DAG {dag.dag_id} because the dag is not serialized")
airflow.exceptions.AirflowException: Cannot create DagRun for DAG simple_dag because the dag is not serialized

Operating System

MacOS

Versions of Apache Airflow Providers

No response

Deployment

Other

Deployment details

This issue happens in the local environment, not in a deployed Airflow environment.

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[ndellosa95]

I'm still seeing this issue with Airflow 3.1.1 for tests where we construct a dag in the test and then run it using dag.test. I suppose this is not supported behavior for dag.test specifically anymore but I'm at my wits end trying to figure out how to replicate the old functionality. These dags are parametrized so there's no easy way to just copy them into a separate dag file.

[thehalleffect]

This is still a problem in 3.1.2 under python 3.10.18.

@ndellosa95 a suggestion: perhaps pre-generating the dags into discrete dagfiles in a temp dir, pointing a DagBag at that temp dir, and then invoking the parametrized tests against each generated test dag would work? Disclaimer: YMMV - just an idea, and I have not tried this myself.

[danielvdende]

I just tried this with the 3.1.3rc1 version, but I still have the same problem...

[thehalleffect]

For whoever is trying to work around this still, I've untangled whats going on a bit. Whenever you import ANYTHING that transitively imports the config module, it will eagerly load defaults in a way that kind of hoses dagbag loading in a pretty opaque fashion. i've come up with this as a (kinda crappy) workaround in one of the conftest.pys in our project:

@pytest.fixture(scope="session")
def monkeysession(request):
    """
    session-scoped monkeypatching
    """
    with pytest.MonkeyPatch.context() as mp:
        yield mp


...


@pytest.fixture()
def dagbag_dir(monkeysession):
    """
    very, ***very*** annoyingly, this does nothing, because the config value is eagerly
    loaded when importing airflow components *before* this has a chance to be monkeypatched.
    leaving this in, because they'll probably fix it at some point, and this is a cleaner
    way to point the dag processor at the correct dag dir.
    """
    monkeysession.setenv(
        "AIRFLOW__CORE__DAGS_FOLDER",
        str(path_for(resource="dags")),  # or wherever your dags live
    )


@pytest.fixture(scope="session", autouse=True)
def _force_correct_dag_bundle_loading(monkeysession):
    """
    force the airflow config crap to load things properly.
    any airflow component that touches `airflow.settings` for any reason will make that module
    eagerly load a bunch of constants. this means that if you import something like airflow.sdk.DAG
    in tests.integration.conftest and monkeypatch the `AIRFLOW__CORE__DAGS_FOLDER` env var,
    *it doesn't actually affect how the dags are loaded for testing*.
    
    this, combined with the `force_dag_serialization` util function (or the similar
    logic in DAG.test), will work around that.
    """
    monkeysession.setenv(
        "AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST",
        json.dumps(
            [
                dict(
                    name="dags-folder",
                    classpath="airflow.dag_processing.bundles.local.LocalDagBundle",
                    kwargs=dict(path=str(path_for(resource="dags"))),  # or wherever your dags live
                )
            ]
        ),
    )

(path_for is just a little utility shim for grabbing the absolute path of some project resource relative to the project root)

DAG.test should now work.

if you also need to run specific task instances manually in your tests without dealing with the boilerplate (and execution time) of invoking test on the WHOLE dag (which is a thing we do with our project), you can also add this hackiness (ripped from DAG.test and simplified/massaged a little):

from datetime import datetime

from airflow import settings
from airflow.sdk import DAG, timezone
from airflow.serialization.serialized_objects import SerializedDAG
from airflow.utils.types import ArgNotSet, NOTSET


def force_dag_serialization(
    *,
    dag: DAG,
    logical_date: datetime | None | ArgNotSet = NOTSET,
    session: settings.Session,
) -> SerializedDAG:
    """
    ripped from airflow.sdk.definitions.DAG.test and hacked for our use.
    """
    dag.validate()

    logical_date = (
        logical_date if not isinstance(logical_date, ArgNotSet) else timezone.utcnow()
    )

    SerializedDAG.clear_dags(
        dags=[dag],
        start_date=logical_date,
        end_date=logical_date,
        dag_run_state=False,
    )

    from airflow.models.dag_version import DagVersion

    version = DagVersion.get_version(dag.dag_id)
    if not version:
        from airflow.dag_processing.bundles.manager import DagBundlesManager
        from airflow.models.dagbag import DagBag, sync_bag_to_db
        from airflow.sdk.definitions._internal.dag_parsing_context import (
            _airflow_parsing_context_manager,
        )

        manager = DagBundlesManager()
        manager.sync_bundles_to_db(session=session)
        session.commit()

        for bundle in manager.get_all_dag_bundles():
            if bundle.name != "dags-folder":
                # skip crap we don't care about (example dags)
                continue
            if not bundle.is_initialized:
                bundle.initialize()
            with _airflow_parsing_context_manager(dag_id=dag.dag_id):
                dagbag = DagBag(
                    dag_folder=bundle.path.resolve(),
                    bundle_path=bundle.path.resolve(),
                    include_examples=False,
                )
                sync_bag_to_db(dagbag, bundle.name, bundle.version)
            version = DagVersion.get_version(dag.dag_id)
            if version:
                break
    serialized_dag = SerializedDAG.deserialize_dag(SerializedDAG.serialize_dag(dag))

    serialized_dag.on_success_callback = dag.on_success_callback  # type: ignore[attr-defined, union-attr]
    serialized_dag.on_failure_callback = dag.on_failure_callback  # type: ignore[attr-defined, union-attr]

    return serialized_dag

all that said: this was WAY too much digging to get functionality back that worked alright before. DAG.test definitely needs some regression testing (it's just tests all the way down, i guess? 🤪) wrapped around it, and the unreasonably eager conf loading needs to be looked at by someone more familiar with the intimate details of how Airflow bootstraps itself when modules are loaded.

@ephraimbuddy fyi