feat: implement DbtTestWatcherOperator

[michal-mrazek]

Description

Implementation of the DbtTestWatcherOperator (formerly only EmptyOperator) with a DbtConsumerWatcherSensor subclass that watches aggregated dbt test results. All logic is handled in the BaseConsumerSensor that just checks for is_test_sensor property.

I also added WatcherEventReason enum for type-safe trigger event payloads and changed model->node.

Related Issue(s)

closes #2441

Breaking Change?

Checklist

• I have made corresponding changes to the documentation (if required)
• I have added tests that prove my fix is effective or that my feature works

[Copilot]

Pull request overview

Implements DbtTestWatcherOperator as a real watcher-mode sensor that monitors aggregated dbt test outcomes via XCom, and introduces a typed WatcherEventReason for trigger → sensor payloads while standardizing terminology from “model” to “node”.

Changes:

• Implement DbtTestWatcherOperator as a DbtConsumerWatcherSensor subclass that watches the aggregated <model_uid>_tests_status XCom key.
• Add WatcherEventReason enum and update WatcherTrigger/BaseConsumerSensor to use typed reason codes and support test sensors via is_test_sensor.
• Update and extend tests to cover aggregated test status polling and the new reason codes.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
cosmos/operators/watcher.py	Replaces the placeholder DbtTestWatcherOperator with a real watcher-mode sensor implementation.
cosmos/operators/_watcher/triggerer.py	Adds WatcherEventReason and extends WatcherTrigger to handle aggregated test status via is_test_sensor.
cosmos/operators/_watcher/base.py	Adds test-sensor plumbing (is_test_sensor, _resource_label, _get_node_status) and updates event handling/logging.
tests/operators/_watcher/test_triggerer.py	Updates triggerer tests to assert enum-based reason payloads.
tests/operators/test_watcher.py	Updates existing watcher tests for enum reasons and adds new tests for DbtTestWatcherOperator.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

You can also share your feedback on Copilot code review. Take the survey.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 98.02%. Comparing base (5a09ad5) to head (614ada3).
⚠️ Report is 8 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2447      +/-   ##
==========================================
- Coverage   98.04%   98.02%   -0.03%     
==========================================
  Files         103      103              
  Lines        7229     7254      +25     
==========================================
+ Hits         7088     7111      +23     
- Misses        141      143       +2     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[michal-mrazek]

@tatiana I will rebase this ticket since we now have logging in sensors (which we are super grateful for!) and elt you know it is ready fore review :)

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 3 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

You can also share your feedback on Copilot code review. Take the survey.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 3 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

You can also share your feedback on Copilot code review. Take the survey.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 3 comments.

Comments suppressed due to low confidence (3)

cosmos/operators/_watcher/base.py:332

• For is_test_sensor retries, _fallback_to_non_watcher_run currently returns True, which will mark a retry attempt as successful even if the aggregated test status is still fail (or never produced). This can mask failing tests whenever Airflow retries are configured. Consider changing retry handling for test sensors to re-check the aggregated _tests_status XCom (and fail again if it's fail), or explicitly raise an exception indicating retries are unsupported instead of returning success.

    def _filter_flags(flags: list[str]) -> list[str]:
        """Filters out dbt flags that are incompatible with retry (e.g., --select, --exclude)."""
        filtered = []
        skip_next = False
        for token in flags:
            if skip_next:
                if token.startswith("--"):

cosmos/operators/_watcher/base.py:438

• The log message for NODE_NOT_RUN is framed as "was skipped by the dbt command" and mentions ephemeral/empty model SQL. When is_test_sensor=True, this reason more likely means no aggregated test status was produced (e.g., no tests selected), so the message becomes misleading and grammatically awkward ("Tests for model ... was"). Consider branching the message for test sensors (e.g., "No test results were reported...") or using a resource-specific verb phrasing.

    def execute(self, context: Context, **kwargs: Any) -> None:
        if not self.deferrable:
            super().execute(context)
        elif not self.poke(context):
            self.defer(

cosmos/operators/_watcher/base.py:464

• The failure messages format dbt {self._resource_label.lower()} ... failed / reporting results for {self._resource_label.lower()} ... leads to awkward output like "dbt tests for model '...' failed". Consider using a dedicated noun/label for error strings (e.g., "model" vs "tests") or a full sentence that avoids concatenating dbt with the label, so messages read naturally for both model and test sensors.

        if status == "success" and reason == WatcherEventReason.NODE_NOT_RUN:
            logger.info(
                "%s '%s' was skipped by the dbt command. This may happen if it is an ephemeral model or if the model sql file is empty.",
                self._resource_label,
                self.model_unique_id,
            )

        # Extract and store compiled_sql from the event if available

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 4 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[tatiana]

@michal-mrazek Thank you very much for continuing this work stream. I would like us to merge these changes before we release Cosmos 1.14.0. I just gave some feedback in-line - please, let me know your thoughts.

Also, please, could you mention how you tested these changes?

• InvocationMode.SUBPROCESS x InvocationMode.DBT_RUNNER
• Local x K8s (based on the files changed, I assume only local)

For us to have feature parity in K8s & Local, should we update the ticket #2427 or create a new one?

[michal-mrazek]

@michal-mrazek Thank you very much for continuing this work stream. I would like us to merge these changes before we release Cosmos 1.14.0. I just gave some feedback in-line - please, let me know your thoughts.

Also, please, could you mention how you tested these changes?

• InvocationMode.SUBPROCESS x InvocationMode.DBT_RUNNER
• Local x K8s (based on the files changed, I assume only local)

For us to have feature parity in K8s & Local, should we update the ticket #2427 or create a new one?

@tatiana I tested on both DBT_RUNNER and SUBPROCESS. Not K8 though, so I think the #2427 is still actual and we should work on it :)

Btw do you have any timeline for 1.14.0 release?

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 4 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[tatiana]

@michal-mrazek thanks a lot, I believe we're very close to merging this - please, could you fix the unit tests?

[michal-mrazek]

Adjusted tests, hopefully it will pass now :) Lets trigger it, I can fix in a minute if needed :)

[tatiana]

Thank you very much, @michal-mrazek , it's an excellent addition to Cosmos watcher to run test after each properly 🎉