(airflow) branch main updated: Adds a unit test to verify that if TaskInstances().Run() fails, (#51709)

[amoghdesai]

This is an automated email from the ASF dual-hosted git repository.

amoghdesai pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new d7ca9305dac Adds a unit test to verify that if TaskInstances().Run() 
fails, (#51709)
d7ca9305dac is described below

commit d7ca9305daced4b55e76c4dbd3007ebf4fa9a8f2
Author: Jenson Reggentin <108676720+jensonjohnat...@users.noreply.github.com>
AuthorDate: Mon Jun 16 07:21:19 2025 +0200

    Adds a unit test to verify that if TaskInstances().Run() fails, (#51709)
---
 go-sdk/worker/runner_test.go | 30 +++++++++++++++++++++++++++---
 1 file changed, 27 insertions(+), 3 deletions(-)

diff --git a/go-sdk/worker/runner_test.go b/go-sdk/worker/runner_test.go
index 61ec1452dfd..957a79a563f 100644
--- a/go-sdk/worker/runner_test.go
+++ b/go-sdk/worker/runner_test.go
@@ -152,10 +152,34 @@ func (s *WorkerSuite) TestTaskNotRegisteredErrors() {
 // TestStartContextErrorTaskDoesntStart checks that if the /run endpoint 
returns an error that task doesn't
 // start, but that it is logged
 func (s *WorkerSuite) TestStartContextErrorTaskDoesntStart() {
-       s.T().Skip("TODO: Not implemented yet")
+       s.T().Parallel()
+       id := uuid.New().String()
+       testWorkload := newTestWorkLoad(id, id[:8])
+
+       // Flag to see if the Task gets called
+       wasCalled := false
+
+       // Register a task that should NOT be called if everything works
+       s.worker.RegisterTaskWithName(testWorkload.TI.DagId, 
testWorkload.TI.TaskId, func() error {
+               wasCalled = true
+               return nil
+       })
+
+       // Setup the mock
+       s.ti.EXPECT().
+               Run(mock.Anything, uuid.MustParse(id), mock.Anything).
+               Return(nil, fmt.Errorf("simulated start context error"))
+
+       s.client.EXPECT().TaskInstances().Return(s.ti)
+
+       err := s.worker.ExecuteTaskWorkload(context.Background(), testWorkload)
+
+       s.Error(err)
+       s.Contains(err.Error(), "simulated start context error")
+       s.False(wasCalled, "Task function should not be executed when start 
context fails")
 }
 
-// TestTaskPanicReportsFailedState tests that when the task/user code panics 
that we catch it and report thr
+// TestTaskPanicReportsFailedState tests that when the task/user code panics 
that we catch it and report the
 // error upstream
 func (s *WorkerSuite) TestTaskPanicReportsFailedState() {
        s.T().Skip("TODO: Not implemented yet")
@@ -197,7 +221,7 @@ func (s *WorkerSuite) TestTaskHeartbeatsWhileRunning() {
                True(callCount < 11 && callCount > 9, fmt.Sprintf("Call count 
of %d was not within the margin of error of 10+/-1", callCount))
 }
 
-func (s *WorkerSuite) TestTaskHeatbeatErrorStopsTaskAndLogs() {
+func (s *WorkerSuite) TestTaskHeartbeatErrorStopsTaskAndLogs() {
        s.T().Skip("TODO: Not implemented yet")
 }