report the same test status as dbt

.github/workflows/test-warehouse.yml

@@ -126,7 +126,7 @@ jobs:
 
       - name: Test
         working-directory: "${{ env.TESTS_DIR }}/tests"
-        run: py.test -n8 -vvv --target "${{ inputs.warehouse-type }}" --junit-xml=test-results.xml --html=detailed_report_${{ inputs.warehouse-type }}_dbt_${{ inputs.dbt-version }}.html --self-contained-html
+        run: py.test -n8 -vvv --target "${{ inputs.warehouse-type }}" --junit-xml=test-results.xml --html=detailed_report_${{ inputs.warehouse-type }}_dbt_${{ inputs.dbt-version }}.html --self-contained-html --clear-on-end
 
       - name: Upload test results
         if: always()

integration_tests/tests/conftest.py

@@ -2,18 +2,22 @@
 from pathlib import Path
 from tempfile import mkdtemp
 
-import env
 import pytest
 from dbt.version import __version__ as dbt_version
 from dbt_project import DbtProject
+from env import Environment
+from logger import get_logger
 from packaging import version
 
 DBT_PROJECT_PATH = Path(__file__).parent.parent / "dbt_project"
 
+logger = get_logger(__name__)
+
 
 def pytest_addoption(parser):
     parser.addoption("--target", action="store", default="postgres")
     parser.addoption("--skip-init", action="store_true", default=False)
+    parser.addoption("--clear-on-end", action="store_true", default=False)
 
 
 @pytest.fixture(scope="session")
@@ -32,9 +36,22 @@ def project_dir_copy():
 
 
 @pytest.fixture(scope="session", autouse=True)
-def init_tests_env(target, skip_init, project_dir_copy: str):
+def init_tests_env(
+    target: str, skip_init: bool, clear_on_end: bool, project_dir_copy: str
+):
+    env = Environment(target, project_dir_copy)
     if not skip_init:
-        env.init(target, project_dir_copy)
+        logger.info("Initializing test environment")
+        env.clear()
+        env.init()
+        logger.info("Initialization complete")
+
+    yield
+
+    if clear_on_end:
+        logger.info("Clearing tests environment")
+        env.clear()
+        logger.info("Clearing complete")
 
 
 @pytest.fixture(autouse=True)
@@ -78,10 +95,15 @@ def target(request) -> str:
 
 
 @pytest.fixture(scope="session")
-def skip_init(request) -> str:
+def skip_init(request) -> bool:
     return request.config.getoption("--skip-init")
 
 
+@pytest.fixture(scope="session")
+def clear_on_end(request) -> bool:
+    return request.config.getoption("--clear-on-end")
+
+
 @pytest.fixture
 def test_id(request) -> str:
     if request.cls:

integration_tests/tests/env.py

@@ -1,12 +1,6 @@
 import dbt_project
 
 
-def init(target: str, project_dir: str):
-    tests_env = Environment(target, project_dir)
-    tests_env.clear()
-    tests_env.init()
-
-
 class Environment:
     def __init__(self, target: str, project_dir: str):
         self.dbt_runner = dbt_project.get_dbt_runner(target, project_dir)

integration_tests/tests/test_failed_row_count.py

@@ -54,3 +54,19 @@ def test_custom_failed_row_count(test_id: str, dbt_project: DbtProject):
     )
     assert test_result["status"] == "fail"
     assert test_result["failed_row_count"] == overwrite_failed_row_count
+
+
+def test_warn_if_0(test_id: str, dbt_project: DbtProject):
+    # Edge case that we want to verify
+
+    null_count = 50
+    data = [{COLUMN_NAME: "pasten"} for _ in range(null_count)]
+    test_result = dbt_project.test(
+        test_id,
+        "not_null",
+        dict(column_name=COLUMN_NAME, warn_if="=0"),
+        data=data,
+        test_vars={"enable_elementary_test_materialization": True},
+    )
+    assert test_result["status"] == "warn"
+    assert test_result["failed_row_count"] == 0

macros/edr/tests/on_run_end/handle_tests_results.sql

@@ -41,7 +41,13 @@
 
     {% for elementary_test_results_row in elementary_test_results_rows %}
       {% set failures = elementary_test_results_row.get("failures", result.failures) %}
-      {% set status = "pass" if failures == 0 else result.status %}
+
+      {# For Elementary anomaly tests, we actually save more than one result per test, in that case the dbt status will be "fail"
+         even if one such result failed and the rest succeeded. To handle this, we make sure to mark the status as "pass" for these 
+         results if the number of failed rows is 0.
+         We don't want to do this for every test though - because otherwise it can break configurations like warn_if=0 #}
+      {% set status = "pass" if failures == 0 and elementary_test_results_row.get("test_type") == "anomaly_detection" else result.status %}
+
       {% do elementary_test_results_row.update({'status': status, 'failures': failures, 'invocation_id': invocation_id, 
                                                 'failed_row_count': elementary_test_failed_row_count}) %}
       {% do elementary_test_results_row.setdefault('test_results_description', result.message) %}