pytest 由于 ModuleNotFoundError 而失败

Question

我正在使用 pytest 对我的 Airflow DAG 执行完整性测试，这是我当前的文件夹结构：

|-- dags
|   |-- 01_lasic_retraining_overview.py
|   |-- 02_lasic_retraining_sagemaker_autopilot.py
|   |-- 03_lasic_retraining_h20_automl.py
|   |-- __init__.py
|   `-- common
|       |-- __init__.py
|       `-- helper.py
|-- docker-compose.yaml
|-- newrelic.ini
|-- plugins
|-- requirements.txt
|-- sample.env
|-- setup.sh
|-- test.sh
`-- tests
    |-- common
    |   `-- test_helper.py
    `-- dags
        |-- test_02_lasic_retraining_sagemaker_autopilot.py
        |-- test_03_lasic_retraining_h20_automl.py
        `-- test_dag_integrity.py

除了 01_lasic_retraining_overview.py（未测试）之外，我所有的 dags 中，我都从 dags/common/helper.py 向它们导入辅助函数，这是测试失败的原因：

import airflow
    from airflow import DAG
    from airflow.exceptions import AirflowFailException
    from airflow.operators.python import PythonOperator
    from airflow.providers.amazon.aws.hooks.s3 import S3Hook
>   from common.helper import _create_connection, _etl_lasic
E   ModuleNotFoundError: No module named 'common'

dags/03_lasic_retraining_h20_automl.py:6: ModuleNotFoundError
=================================== short test summary info ===================================
FAILED tests/dags/test_dag_integrity.py::test_dag_integrity[/Users/yravindranath/algo_lasic2_ct_pipeline/tests/dags/../../dags/02_lasic_retraining_sagemaker_autopilot.py]
FAILED tests/dags/test_dag_integrity.py::test_dag_integrity[/Users/yravindranath/algo_lasic2_ct_pipeline/tests/dags/../../dags/03_lasic_retraining_h20_automl.py]

现在这段代码在我的 docker 容器中运行没有问题。我尝试过但没有奏效的事情：

正在将 __init__py 添加到 tests 文件夹。
运行 python -m pytest tests/
正在删除 dags

__init__.py

设置PYTHONPATH=. pytest

完整性测试代码位于 `/tests/dags/test_dag_integrity.py`

import re
import glob
import importlib.util
import os

import pytest
from airflow.models import DAG

# go to the root dir and browse for any files that match the pattern
# this will find all the dag files
DAG_PATH = os.path.join(
    os.path.dirname(__file__),
    "..",
    "..",
    "dags/**/0*.py",
)

# holds a list of all the dag files
DAG_FILES = glob.glob(
    DAG_PATH,
    recursive=True,
)
# filter the files to exclude the 01 dag run as that is just a plan of the
# pipeline
DAG_FILES = [file for file in DAG_FILES if not re.search("/01", file)]


@pytest.mark.parametrize("dag_file", DAG_FILES)
def test_dag_integrity(dag_file):
    # Load file
    module_name, _ = os.path.splitext(dag_file)
    module_path = os.path.join(DAG_PATH, dag_file)
    mod_spec = importlib.util.spec_from_file_location(
        module_name,
        module_path,
    )
    module = importlib.util.module_from_spec(
        mod_spec,  # type: ignore
    )
    mod_spec.loader.exec_module(module)  # type: ignore
    # all objects of class DAG found in file
    dag_objects = [
        var
        for var in vars(module).values()
        if isinstance(
            var,
            DAG,
        )
    ]
    # check if DAG objects were found in the file
    assert dag_objects
    # check if there are no cycles in the dags
    for dag in dag_objects:
        dag.test_cycle()  # type: ignore

Answer 1

您需要检查一下您的 PYTHONPATH 是什么。您的 PYTHONPATH 中可能没有 dags。可能您的 PYTHONPATH 指向文件结构的根目录，因此导入“公共”文件夹的正确方法是

import dags.common

和你常用的测试代码一样

import tests.common

Python（甚至 python 3）没有很好的机制来导入相对于当前加载文件的内容。即使存在“相对”导入（前面有“.”）——它们也会令人困惑，并且工作方式与您想象的不同。避免使用它们。只需确保您的。

同时避免将 PYTHONPATH 设置为“.”。它使您的导入工作有所不同，具体取决于您当前的目录。最好的方法是设置一次并导出。

export PYTHONPATH="$(pwd)"

以上将 PYTHONPATH 设置为您当前所在的目录，并将其设置为绝对路径。

Answer 2

我也是运行 Docker 容器中的应用程序，@Jarek Potiuk 提供的答案在实际上运行 DAG 时不起作用，所以我正在使用一种超级 hack 方式，只包括在 docker 中工作的导入部分和在本地工作的部分。

try:
    # Works locally with tests
    from common.helper import _create_connection, _etl_lasic
except ImportError:
    # Works in docker container
    from dags.common.helper import _create_connection, _etl_lasic

Answer 3

在这里抛出一个疯狂的想法，尝试将 __init__.py 同时添加到 */dag 或 */common 和 */tests。

Answer 4

制作tests/conftest.py文件
在 conftest.py

确保您的模块 `common` 的路径名是正确的

import pytest
import sys

@pytest.fixture(scope='session)
def append_path():
    sys.path.insert(0 , 'absolute_path_to_common_module' )
    yield

现在将此夹具用作：

@pytest.mark.usefixtures("append_path") 
@pytest.mark.parametrize("dag_file", DAG_FILES) 
def test_dag_integrity(dag_file):
    .....

我们在做什么？

确保模块对python可见。

注意：您可以将自定义模块 common 重命名为不太常见但更独特的名称。没有双关语的意思。为了避免任何冲突。

pytest 由于 ModuleNotFoundError 而失败

pytest failing due to ModuleNotFoundError

python

pytest

airflow

完整性测试代码位于 /tests/dags/test_dag_integrity.py

确保您的模块 common 的路径名是正确的

完整性测试代码位于 `/tests/dags/test_dag_integrity.py`

确保您的模块 `common` 的路径名是正确的