test: Add dagbag tests

.github/workflows/check.yml

@@ -44,3 +44,8 @@ jobs:
           SKIP: sqlfluff-fix,sqlfluff-lint
         with:
           extra_args: --color=always --show-diff-on-failure
+
+      - name: pytest
+        run: |
+          poetry install;
+          make test;

.pre-commit-config.yaml

@@ -54,7 +54,7 @@ repos:
     hooks:
       - id: ruff
         name: 'python ruff'
-        files: ^(dags|dbts|notebook)/
+        files: ^(dags|dbts|dags-test)/
         args: [ --fix, --exit-non-zero-on-fix ]
         types_or: [ python, pyi, jupyter ]
 
@@ -64,7 +64,7 @@ repos:
       - id: mypy
         name: 'python type check'
         verbose: true
-        files: ^(dags|dbts)/
+        files: ^(dags|dbts|dags-test)/
         types: [ python ]
         args: [ "--config-file", "./mypy.ini", ]
         additional_dependencies:

Makefile

@@ -34,6 +34,15 @@ compose.trino:
 compose.dbt:
 	COMPOSE_PROFILES=trino,airflow docker-compose up
 
+.PHONY:test
+test:
+	@ echo ""
+	@ echo ""
+	@ echo "[$(TAG)] ($(shell date '+%H:%M:%S')) - Executing pytest"
+	@ AIRFLOW_HOME=$(shell pwd) poetry run pytest dags-test/
+	@ echo ""
+	@ echo ""
+
 .PHONY:check
 check:
 	@ echo ""

dags-test/test_dag_validation.py

@@ -0,0 +1,23 @@
+import os
+from pathlib import Path
+
+from airflow.models import DagBag
+
+CONST_DAG_DIRECTORY = "dags"
+
+CONST_AIRFLOW_HOME = os.environ.get("AIRFLOW_HOME", Path(__file__).parent.parent)
+
+
+def test_no_import_errors() -> None:
+    dag_folder = f"{CONST_AIRFLOW_HOME}/{CONST_DAG_DIRECTORY}"
+    dag_bag = DagBag(dag_folder=dag_folder, include_examples=False)
+    assert len(dag_bag.import_errors) == 0, "No Import Failures"
+
+
+def test_retries_present() -> None:
+    dag_folder = f"{CONST_AIRFLOW_HOME}/{CONST_DAG_DIRECTORY}"
+    dag_bag = DagBag(dag_folder=dag_folder, include_examples=False)
+    for dag in dag_bag.dags:
+        retries = dag_bag.dags[dag].default_args.get("retries", [])
+        error_msg = f"Retries are set for DAG {dag}"
+        assert retries >= 1, error_msg

dags/config/dag.py

@@ -1,12 +1,12 @@
-from typing import Any, Optional, Dict
+from typing import Any
 
 CONST_DEFAULT_ARGS = {
     "owner": "airflow",
     "retries": 1,
 }
 
 
-def build_default_args(custom_args: Optional[Dict[Any, Any]] = None) -> Dict[Any, Any]:
+def build_default_args(custom_args: dict[Any, Any] | None = None) -> dict[Any, Any]:
     actual_args = CONST_DEFAULT_ARGS.copy()
 
     if custom_args:

dags/dag_dbt.py

@@ -8,50 +8,50 @@
 
 import pendulum
 from airflow.decorators import dag
-
 from airflow_dbt.operators.dbt_operator import (
-    DbtRunOperator,
-    DbtDepsOperator,
+  DbtDepsOperator,
+  DbtRunOperator,
 )
+from config.dag import build_default_args
 
-import config.dag
-
-dag_args = config.dag.build_default_args()
+dag_args = build_default_args()
 
 
 @dag(
     dag_id=Path(__file__).stem,
     tags=["dbt"],
-    description=__doc__[0 : __doc__.find(".")],
+    description=__doc__[0: __doc__.find(".")],
     doc_md=__doc__,
     default_args=dag_args,
     start_date=pendulum.datetime(2023, 9, 1, tz="Asia/Seoul"),
-    schedule='5 0 * * *',
+    schedule="5 0 * * *",
     catchup=False,
     dagrun_timeout=dt.timedelta(minutes=60),
 )
 def generate_dag() -> None:
+  dbt_home = "/opt/airflow/dbts"
+  dbt_target = "dev"
 
-    dbt_home = "/opt/airflow/dbts"
-    dbt_target = "dev"
+  task_dbt_deps = DbtDepsOperator(
+      task_id="task_dbt_deps",
+      dir=dbt_home,
+      profiles_dir=dbt_home,
+      target=dbt_target,
+      full_refresh=False,
+  )
 
-    task_dbt_deps = DbtDepsOperator(
-        task_id="task_dbt_deps",
-        dir=dbt_home,
-        profiles_dir=dbt_home,
-        target=dbt_target,
-        full_refresh=False,
-    )
+  task_dbt_run = DbtRunOperator(
+      task_id="task_dbt_run",
+      dir=dbt_home,
+      profiles_dir=dbt_home,
+      target=dbt_target,
+      full_refresh=False,
+  )
 
-    task_dbt_run = DbtRunOperator(
-        task_id="task_dbt_run",
-        dir=dbt_home,
-        profiles_dir=dbt_home,
-        target=dbt_target,
-        full_refresh=False,
-    )
-
-    task_dbt_deps >> task_dbt_run
+  task_dbt_deps >> task_dbt_run
 
 
 generate_dag()
+
+if __name__ == "__main__":
+  dag.test()

mypy.ini

@@ -1,7 +1,12 @@
 [mypy]
 
-mypy_path=./dags:./dbts
+python_version=3.11
+mypy_path=./dags:./dbts:./dags-test
 
 strict=True
 ignore_missing_imports=True
 disallow_untyped_decorators=False
+show_error_codes = True
+show_error_context = True
+show_column_numbers = True
+warn_no_return = false

pyproject.toml

@@ -21,7 +21,57 @@ sqlfluff = "^2.3.1"
 sqlfluff-templater-dbt = "^2.3.1"
 ruff = "^0.0.287"
 black = "^23.7.0"
+pytest = "^7.4.1"
 
 [build-system]
 requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"
+
+
+[tool.ruff]
+line-length = 100  # defaults to 88 like black
+target-version = "py311"  # the python version to target, useful when considering code upgrades, defaults to "py310"
+fix = true
+show-source = true
+
+format = "grouped"
+select = [
+    # pycodestyle
+    "E",
+    "W",
+    # pyflakes
+    "F",
+    # pylint
+    "PLC",
+    "PLE",
+    "PLW",
+    # isort
+    "I",
+    # flake8-builtins
+    "A",
+    # flake8-bugbear
+    "B",
+    # flake8-commas
+    "COM",
+    # flake8-comprehensions
+    "C4",
+    # flake8-datetimez
+    "DTZ",
+    # flake8-implicit-str-concat
+    "ISC",
+    # flake8-pie
+    "PIE",
+    # flake8-pytest-style
+    "PT",
+    # flake8-quotes
+    "Q",
+    # flake8-tidy-imports
+    "TID",
+    # explicitly select rules under the nursery-flag introduced in ruff 0.0.269
+    # remove once enabled via the "E" selector
+    "E111", "E112", "E113", "E114", "E115", "E116", "E117",
+    "E201", "E202", "E203", "E211", "E221", "E222", "E223",
+    "E224", "E225", "E226", "E227", "E228", "E231", "E251",
+    "E252", "E261", "E262", "E265", "E266", "E271", "E272",
+    "E273", "E274", "E275",
+]

pytest.ini

@@ -0,0 +1,7 @@
+[pytest]
+filterwarnings =
+  ignore::DeprecationWarning
+
+log_cli=true
+log_level=INFO
+log_cli_level=INFO