executor: Enable command substitution and environment variables for `…

…ssh` and `subworkflow` executor config (#748)

cmd/common_test.go

@@ -48,7 +48,7 @@ func (td *testDAG) AssertCurrentStatus(t *testing.T, expected scheduler.Status)
 	dag, err := digraph.Load(td.Context, td.Config.Paths.BaseConfig, td.Path, "")
 	require.NoError(t, err)
 
-	cli := td.Client()
+	cli := td.Client
 	require.Eventually(t, func() bool {
 		status, err := cli.GetCurrentStatus(td.Context, dag)
 		require.NoError(t, err)
@@ -59,7 +59,7 @@ func (td *testDAG) AssertCurrentStatus(t *testing.T, expected scheduler.Status)
 func (th *testDAG) AssertLastStatus(t *testing.T, expected scheduler.Status) {
 	t.Helper()
 
-	hs := th.DataStore().HistoryStore()
+	hs := th.DataStores.HistoryStore()
 	require.Eventually(t, func() bool {
 		status := hs.ReadStatusRecent(th.Context, th.Path, 1)
 		if len(status) < 1 {

cmd/retry_test.go

@@ -23,7 +23,7 @@ func TestRetryCommand(t *testing.T) {
 		th.RunCommand(t, startCmd(), cmdTest{args: args})
 
 		// Find the request ID.
-		cli := th.Client()
+		cli := th.Client
 		ctx := context.Background()
 		status, err := cli.GetStatus(ctx, dagFile.Path)
 		require.NoError(t, err)

cmd/start_test.go

@@ -8,7 +8,10 @@ import (
 )
 
 func TestStartCommand(t *testing.T) {
+	t.Parallel()
+
 	th := testSetup(t)
+
 	tests := []cmdTest{
 		{
 			name:        "StartDAG",

cmd/status_test.go

@@ -24,7 +24,7 @@ func TestStatusCommand(t *testing.T) {
 			close(done)
 		}()
 
-		hs := th.DataStore().HistoryStore()
+		hs := th.DataStores.HistoryStore()
 		require.Eventually(t, func() bool {
 			status := hs.ReadStatusRecent(th.Context, dagFile.Path, 1)
 			if len(status) < 1 {

internal/agent/agent.go

@@ -116,6 +116,7 @@ func (a *Agent) Run(ctx context.Context) error {
 
 	// Check if the DAG is already running.
 	if err := a.checkIsAlreadyRunning(ctx); err != nil {
+		a.scheduler.Cancel(ctx, a.graph)
 		return err
 	}
 
@@ -203,7 +204,7 @@ func (a *Agent) Run(ctx context.Context) error {
 
 	// Send the execution report if necessary.
 	a.lastErr = lastErr
-	if err := a.reporter.send(a.dag, finishedStatus, lastErr); err != nil {
+	if err := a.reporter.send(ctx, a.dag, finishedStatus, lastErr); err != nil {
 		logger.Error(ctx, "Mail notification failed", "err", err)
 	}
 
@@ -314,14 +315,13 @@ func (a *Agent) setup(ctx context.Context) error {
 	defer a.lock.Unlock()
 
 	a.scheduler = a.newScheduler()
-	a.reporter = newReporter(
-		mailer.New(mailer.Config{
-			Host:     a.dag.SMTP.Host,
-			Port:     a.dag.SMTP.Port,
-			Username: a.dag.SMTP.Username,
-			Password: a.dag.SMTP.Password,
-		}),
-	)
+	mailer := mailer.New(mailer.Config{
+		Host:     a.dag.SMTP.Host,
+		Port:     a.dag.SMTP.Port,
+		Username: a.dag.SMTP.Username,
+		Password: a.dag.SMTP.Password,
+	})
+	a.reporter = newReporter(mailer)
 
 	return a.setupGraph(ctx)
 }
@@ -482,7 +482,7 @@ func (a *Agent) checkPreconditions(ctx context.Context) error {
 	// If one of the conditions does not met, cancel the execution.
 	if err := digraph.EvalConditions(a.dag.Preconditions); err != nil {
 		logger.Error(ctx, "Preconditions are not met", "err", err)
-		a.scheduler.Cancel(a.graph)
+		a.scheduler.Cancel(ctx, a.graph)
 		return err
 	}
 	return nil