dagu/engine_test.go at main · dagucloud/dagu · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
// Copyright (C) 2026 Yota Hamada
// SPDX-License-Identifier: GPL-3.0-or-later

package dagu_test

import (
	"context"
	"fmt"
	"os"
	"path/filepath"
	"runtime"
	"strings"
	"testing"
	"time"

	"github.com/dagucloud/dagu"
	"github.com/stretchr/testify/require"
)

func engineTestTimeout(timeout time.Duration) time.Duration {
	if runtime.GOOS == "windows" {
		return timeout * 3
	}
	return timeout
}

func TestEngineRunFile(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), engineTestTimeout(20*time.Second))
	defer cancel()

	home := t.TempDir()
	dagFile := filepath.Join(home, "embedded-file.yaml")
	testBinary, err := os.Executable()
	require.NoError(t, err, "Executable()")
	writeDAG(t, dagFile, fmt.Sprintf(`
name: embedded-file
steps:
  - name: check-param
    exec:
      command: %q
      args:
        - -test.run=TestEngineRunFileParamHelper
    env:
      DAGU_ENGINE_PARAM_HELPER: "1"
`, testBinary))

	engine, err := dagu.New(ctx, dagu.Options{HomeDir: home})
	require.NoError(t, err, "New()")
	defer func() {
		require.NoError(t, engine.Close(context.Background()))
	}()

	run, err := engine.RunFile(ctx, dagFile, dagu.WithParams(map[string]string{"FOO": "bar"}))
	require.NoError(t, err, "RunFile()")
	status, err := run.Wait(ctx)
	require.NoError(t, err, "Wait()")
	require.Equal(t, "succeeded", status.Status)
	require.Equal(t, "embedded-file", status.Name)
	require.NotEmpty(t, status.RunID)
}

func TestEngineRunFileParamHelper(t *testing.T) {
	if os.Getenv("DAGU_ENGINE_PARAM_HELPER") != "1" {
		return
	}
	require.Equal(t, "bar", os.Getenv("FOO"))
}

func TestEngineRunYAML(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), engineTestTimeout(20*time.Second))
	defer cancel()

	originalWorkingDir, err := os.Getwd()
	require.NoError(t, err, "Getwd()")

	engine, err := dagu.New(ctx, dagu.Options{HomeDir: t.TempDir()})
	require.NoError(t, err, "New()")
	defer func() {
		require.NoError(t, engine.Close(context.Background()))
	}()

	run, err := engine.RunYAML(ctx, []byte(`
name: embedded-yaml
steps:
  - name: hello
    command: echo hello
`))
	require.NoError(t, err, "RunYAML()")
	status, err := run.Wait(ctx)
	require.NoError(t, err, "Wait()")
	require.Equal(t, "succeeded", status.Status)
	require.Equal(t, "embedded-yaml", status.Name)
	currentWorkingDir, err := os.Getwd()
	require.NoError(t, err, "Getwd() after run")
	require.Equal(t, originalWorkingDir, currentWorkingDir)
}

func TestEngineDistributedRequiresCoordinator(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), engineTestTimeout(20*time.Second))
	defer cancel()

	engine, err := dagu.New(ctx, dagu.Options{HomeDir: t.TempDir()})
	require.NoError(t, err, "New()")
	defer func() {
		require.NoError(t, engine.Close(context.Background()))
	}()

	_, err = engine.RunYAML(ctx, []byte(`
name: embedded-distributed
steps:
  - name: hello
    command: echo hello
`), dagu.WithMode(dagu.ExecutionModeDistributed))
	require.Error(t, err)
	require.Contains(t, err.Error(), "coordinator")
}

func writeDAG(t *testing.T, path, content string) {
	t.Helper()
	if err := os.WriteFile(path, []byte(strings.TrimSpace(content)+"\n"), 0o600); err != nil {
		t.Fatalf("write DAG: %v", err)
	}
}