feat: Add watcher mode support for dbt test node states

[michal-mrazek]

Description

When using ExecutionMode.WATCHER, dbt test nodes now have their results aggregated per parent model and pushed as a single XCom value ("pass" or "fail"), enabling downstream consumer sensors to observe the test outcome for each model. This lays the groundwork for a future DbtTestConsumerWatcherSensor.

I also introduced some helpful function for handling node status. This can later be easily enhanced.

I removed using hardcoded string in TriggerEvent - this was prone for typos.

I think that documentation should be changed in following PR - where we change the behaviour of test tasks. For now, there is no change for end users.

Future work:

• Implement the TestOperator to read the xcom
• also implement for k8 execution mode

Related Issue(s)

Related to: #2311

Breaking Change?

No

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

This PR adds watcher mode support for dbt test node states by introducing helper functions to handle different node status values (models use "success"/"failed", tests use "pass"/"fail") and updating the codebase to use these helpers and constants instead of hardcoded strings.

Changes:

• Introduced helper functions (is_node_status_success, is_node_status_failed, is_node_status_terminal) to handle both model and test status values
• Replaced hardcoded status strings with EventStatus constants in trigger events
• Updated log messages to use generic "node" terminology instead of "model"

Reviewed changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
cosmos/operators/_watcher/state.py	Adds helper functions and constants for checking node status states
cosmos/operators/_watcher/triggerer.py	Updates to use new helper functions and EventStatus constants instead of hardcoded strings
cosmos/operators/_watcher/base.py	Updates store_dbt_resource_status_from_log to use is_node_status_terminal helper
cosmos/operators/_watcher/init.py	Exports new helper functions in all
tests/operators/_watcher/test_state.py	Adds comprehensive unit tests for the new node status helper functions
tests/operators/test_watcher.py	Adds tests verifying test node status ("pass"/"fail") is correctly stored in XCom
tests/operators/_watcher/test_triggerer.py	Updates test assertion to use "node" instead of "model" terminology

---

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

[netlify]

✅ Deploy Preview for astronomer-cosmos ready!

Name	Link
🔨 Latest commit	d00e36a
🔍 Latest deploy log	https://app.netlify.com/projects/astronomer-cosmos/deploys/6991c516128eb300087503dc
😎 Deploy Preview	https://deploy-preview-2318--astronomer-cosmos.netlify.app
📱 Preview on mobile	Toggle QR Code... Use your smartphone camera to open QR code link.

---

To edit notification comments on pull requests, go to your Netlify project configuration.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 97.97%. Comparing base (4d86173) to head (0826b5b).
⚠️ Report is 4 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2318      +/-   ##
==========================================
+ Coverage   97.96%   97.97%   +0.01%     
==========================================
  Files         102      103       +1     
  Lines        7013     7079      +66     
==========================================
+ Hits         6870     6936      +66     
  Misses        143      143              

☔ 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.

[tatiana]

@michal-mrazek this is looking great, thank you for the improvement! Really excited that this could also fix a significant part of #2291.

I believe we're very close to merge this change. I left some feedback inline - it would be great if we could disambiguate where we are referencing ariflow task state x dbt model state, since we don't have 1:1 parity.

Lastly, could you confirm if this works both:

• when using InvocationMode.DBT_RUNNER
• when using InvocationMode.SUBPROCESS (I believe you tested with this use case, referencing log lines)

[michal-mrazek]

@tatiana, during testing subprocess/dbt_runner, I realized that there is one caveat.

There can be multiple tests (so dbt nodes) for one model. Moreover, the test node name is more complex. So to make the AFTER_EACH work lately we have to map airflow task to multiple dbt test nodes:

1. Aggregate during the xcom push
2. Aggregate during the xcom pull

I tend to incline to option 2. But since this is architecture decision, I would be curious about your opinion :)

This is not strictly related to this PR, but we have to keep this in mind. But yeah, works with both subprocess/dbt_runner :)

[michal-mrazek]

@tatiana, once we merge this, I will start working on the test operator for after_each. I think it makes sense, we can reuse a lot of code, we will just have to pull for multiple xcoms :)

[Copilot]

Pull request overview

Copilot reviewed 7 out of 7 changed files in this pull request and generated no new comments.

---

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

[michal-mrazek]

@tatiana @pankajkoti Can you pls check if you have some time? :) I will then follow up with the next PR :)

[Copilot]

Pull request overview

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

---

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

[michal-mrazek]

Added the dbt test aggregation before pushing to xcom. This was a bit more complex than I expected :)

We need to pass the tests_per_model but also keep the state of existing tests (but be aware of multithreading, so lock was introduced). I will change the PR description.

Both subprocess and dbt_runner were tested - k8 should follow in another PR, I feel like this PR is already growing a lot.

[Copilot]

Pull request overview

Copilot reviewed 15 out of 15 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]

Thanks a lot for advancing the implementation further, @michal-mrazek!

I just tried out commit 0826b5b, and I noticed that we're still generating one XCom per test - could you adjust this, please?

I used the DAG that's in the repo example_watcher.py and airflow standalone to try it out.

[michal-mrazek]

@tatiana That is strange 🤔

git log --oneline -5
0826b5bf (HEAD -> feat/watcher-test-operator-support, origin/feat/watcher-test-operator-support) tests
5f016214 comments
9f5332f9 handle the dbt test aggregation
bda49418 rename all dbt node variables for cleaner code
da0ac024 feat: Add watcher mode support for dbt test node states

I just pulled the branch and run via docker compose you guys provide 🤔

[michal-mrazek]

@tatiana I did not manage to match your results with this branch. I specifically desinged tests for this case https://github.com/astronomer/astronomer-cosmos/pull/2318/changes#diff-b6f62de0effcda8f032c7203047c74394bfd2ee93ab65392462714a9910dbdfeR687 Can you please pull the branch and try it from scratch? 🙏

[tatiana]

Thanks a lot, @michal-mrazek , I'm not sure about what happened. Thanks for the amazing work here. I created a new Airflow deployment locally, and I was able to confirm the expected behaviour 🎉