Add missing info population in the activity info. (#936) (#937)

Author: venkat1109

internal/internal_task_pollers.go

@@ -464,7 +464,12 @@ func (lath *localActivityTaskHandler) executeLocalActivityTask(task *localActivi
 		rootCtx = context.Background()
 	}
 
+	workflowTypeLocal := task.params.WorkflowInfo.WorkflowType
+
 	ctx := context.WithValue(rootCtx, activityEnvContextKey, &activityEnvironment{
+		workflowType:      &workflowTypeLocal,
+		workflowDomain:    task.params.WorkflowInfo.Domain,
+		taskList:          task.params.WorkflowInfo.TaskListName,
 		activityType:      ActivityType{Name: activityType},
 		activityID:        fmt.Sprintf("%v", task.activityID),
 		workflowExecution: task.params.WorkflowInfo.WorkflowExecution,
@@ -505,6 +510,7 @@ func (lath *localActivityTaskHandler) executeLocalActivityTask(task *localActivi
 		// this is attempt and expire time is before SCHEDULE_TO_CLOSE timeout
 		deadline = task.expireTime
 	}
+
 	ctx, cancel := context.WithDeadline(ctx, deadline)
 	task.Lock()
 	if task.canceled {

internal/internal_workflow_testsuite.go

@@ -537,7 +537,7 @@ func (env *testWorkflowEnvironmentImpl) executeActivity(
 func (env *testWorkflowEnvironmentImpl) executeLocalActivity(
 	activityFn interface{},
 	args ...interface{},
-) (Value, error) {
+) (val Value, result *localActivityResult, err error) {
 	params := executeLocalActivityParams{
 		localActivityOptions: localActivityOptions{
 			ScheduleToCloseTimeoutSeconds: common.Int32Ceil(env.testTimeout.Seconds()),
@@ -559,11 +559,11 @@ func (env *testWorkflowEnvironmentImpl) executeLocalActivity(
 		tracer:       opentracing.NoopTracer{},
 	}
 
-	result := taskHandler.executeLocalActivityTask(task)
+	result = taskHandler.executeLocalActivityTask(task)
 	if result.err != nil {
-		return nil, result.err
+		return nil, nil, result.err
 	}
-	return newEncodedValue(result.result, env.GetDataConverter()), nil
+	return newEncodedValue(result.result, env.GetDataConverter()), result, nil
 }
 
 func (env *testWorkflowEnvironmentImpl) startDecisionTask() {

internal/internal_workflow_testsuite_test.go

@@ -1616,8 +1616,11 @@ func (s *WorkflowTestSuiteUnitTest) Test_LocalActivity() {
 	}
 
 	env := s.NewTestActivityEnvironment()
-	result, err := env.ExecuteLocalActivity(localActivityFn, "local_activity")
+	result, localActivity, err := env.ExecuteLocalActivity(localActivityFn, "local_activity")
 	s.NoError(err)
+	s.Equal(WorkflowType{Name: workflowTypeNotSpecified}, localActivity.task.params.WorkflowInfo.WorkflowType)
+	s.Equal(defaultTestDomain, localActivity.task.params.WorkflowInfo.Domain)
+	s.Equal(defaultTestTaskList, localActivity.task.params.WorkflowInfo.TaskListName)
 	var laResult string
 	err = result.Get(&laResult)
 	s.NoError(err)

internal/workflow_testsuite.go

@@ -159,7 +159,7 @@ func (t *TestActivityEnvironment) ExecuteActivity(activityFn interface{}, args .
 
 // ExecuteLocalActivity executes a local activity. The tested activity will be executed synchronously in the calling goroutinue.
 // Caller should use Value.Get() to extract strong typed result value.
-func (t *TestActivityEnvironment) ExecuteLocalActivity(activityFn interface{}, args ...interface{}) (Value, error) {
+func (t *TestActivityEnvironment) ExecuteLocalActivity(activityFn interface{}, args ...interface{}) (val Value, result *localActivityResult, err error) {
 	return t.impl.executeLocalActivity(activityFn, args...)
 }
 

test/workflow_test.go

@@ -427,6 +427,12 @@ func (w *Workflows) ConsistentQueryWorkflow(ctx workflow.Context, delay time.Dur
 	laCtx := workflow.WithLocalActivityOptions(ctx, workflow.LocalActivityOptions{
 		ScheduleToCloseTimeout: 5 * time.Second,
 	})
+
+	workflowInfo := internal.GetWorkflowInfo(laCtx)
+	if &workflowInfo.WorkflowType == nil {
+		return errors.New("failed to get work flow type")
+	}
+
 	workflow.ExecuteLocalActivity(laCtx, LocalSleep, delay).Get(laCtx, nil)
 	queryResult = signalData
 	return nil