test

package
v2.12.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 6, 2026 License: GPL-3.0 Imports: 58 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func AssertInlineRescheduledRunParams 

func AssertInlineRescheduledRunParams(t *testing.T, server Server, dagName, dagRunID string)

func CreateInlineDAGRunForReschedule 

func CreateInlineDAGRunForReschedule(t *testing.T, server Server, dagName string, enqueue bool) (string, string)

func CreateStaleLegacyProcFile 

func CreateStaleLegacyProcFile(
	t *testing.T,
	procDir string,
	groupName string,
	dagRun exec.DAGRunRef,
	startedAt time.Time,
	age time.Duration,
) string

CreateStaleLegacyProcFile writes a stale legacy .proc heartbeat file for the given dag-run.

func CreateStaleLegacyProcFileWithAttempt 

func CreateStaleLegacyProcFileWithAttempt(
	t *testing.T,
	procDir string,
	groupName string,
	dagRun exec.DAGRunRef,
	attemptID string,
	startedAt time.Time,
	age time.Duration,
) string

CreateStaleLegacyProcFileWithAttempt writes a stale legacy .proc heartbeat file for the given dag-run and attempt.

func EnvOutput 

func EnvOutput(names ...string) string

func EnvOutputWithSeparator 

func EnvOutputWithSeparator(separator string, names ...string) string

func ExpandedOutput 

func ExpandedOutput(ref string) string

ExpandedOutput emits a Dagu-resolved ${...} value while keeping shell quoting valid on each platform. The input should be a Dagu reference, not an arbitrary literal string.

func ExpectedInlineTempPath 

func ExpectedInlineTempPath(name, dagRunID string) string

func ForOS 

func ForOS(posix, windows string) string

func JoinLines 

func JoinLines(lines ...string) string

func JoinShellCommands 

func JoinShellCommands(commands ...string) string

func LabeledExpandedOutput 

func LabeledExpandedOutput(prefix, ref string) string

func Output 

func Output(value string) string

func OutputEscaped 

func OutputEscaped(value string) string

func PosixQuote 

func PosixQuote(value string) string

func PowerShellQuote 

func PowerShellQuote(value string) string

func ProcessQueuedInlineRun 

func ProcessQueuedInlineRun(t *testing.T, server Server, queueName string)

func ReadRunStatus 

func ReadRunStatus(ctx context.Context, t *testing.T, store exec.DAGRunStore, dagRun exec.DAGRunRef) *exec.DAGRunStatus

ReadRunStatus loads the persisted status for the given dag-run reference.

func ReadTestdata 

func ReadTestdata(t *testing.T, filename string) []byte

ReadTestdata reads the content of a testdata file.

func RequireProcHeartbeatAdvance 

func RequireProcHeartbeatAdvance(
	t *testing.T,
	ctx context.Context,
	procStore ProcHeartbeatObserver,
	groupName string,
	dagRun exec.DAGRunRef,
	timeout time.Duration,
)

RequireProcHeartbeatAdvance verifies dagRun's proc heartbeat updates within the timeout.

func ReserveServerListener 

func ReserveServerListener(t *testing.T) (net.Listener, string)

ReserveServerListener binds a loopback listener and closes it during test cleanup.

func RunBuiltCLI 

func RunBuiltCLI(t *testing.T, th Helper, extraEnv []string, args ...string) string

func RunBuiltCLICommand 

func RunBuiltCLICommand(tb testing.TB, th Helper, extraEnv []string, args ...string) ([]byte, error)

func ShellPath 

func ShellPath(path string) string

func ShellQuote 

func ShellQuote(value string) string

func Sleep 

func Sleep(d time.Duration) string

func StatusOutputValue 

func StatusOutputValue(t *testing.T, status *coreexec.DAGRunStatus, key string) string

func Stderr 

func Stderr(value string) string

func SubprocessRunTimeout 

func SubprocessRunTimeout(base time.Duration) time.Duration

SubprocessRunTimeout returns a timeout appropriate for waiting on the result of a subprocess-executed DAG run. On Windows, process creation is significantly slower so a larger multiplier is used.

func TestdataPath 

func TestdataPath(t *testing.T, filename string) string

TestdataPath returns the path to a testdata file.

func WaitForAttemptSnapshot 

func WaitForAttemptSnapshot(t *testing.T, server Server, dagName, dagRunID string) exec.DAGRunAttempt

func WaitForAttemptSnapshotWithDAG 

func WaitForAttemptSnapshotWithDAG(t *testing.T, server Server, dagName, dagRunID string) (exec.DAGRunAttempt, *core.DAG)

func WaitForProcHeartbeat 

func WaitForProcHeartbeat(
	t *testing.T,
	ctx context.Context,
	procStore ProcHeartbeatObserver,
	groupName string,
	dagRun exec.DAGRunRef,
	timeout time.Duration,
) exec.ProcHeartbeat

WaitForProcHeartbeat returns the latest heartbeat observation for dagRun once it exists.

func WithConfigFlag 

func WithConfigFlag(args []string, cfg *config.Config) []string

WithConfigFlag appends --config <file> unless already present.

Types

type APIClient 

type APIClient struct {
	// contains filtered or unexported fields
}

APIClient handles HTTP requests to the test server

func (*APIClient) Delete 

func (c *APIClient) Delete(path string) *Request

Delete prepares a DELETE request

func (*APIClient) Get 

func (c *APIClient) Get(path string) *Request

Get prepares a GET request

func (*APIClient) Patch 

func (c *APIClient) Patch(path string, body any) *Request

Patch prepares a PATCH request with the given body

func (*APIClient) Post 

func (c *APIClient) Post(path string, body any) *Request

Post prepares a POST request with the given body

func (*APIClient) Put 

func (c *APIClient) Put(path string, body any) *Request

Put prepares a PUT request with the given body

type Agent 

type Agent struct {
	*Helper
	*core.DAG
	*agent.Agent
	// contains filtered or unexported fields
}

func (*Agent) Abort 

func (a *Agent) Abort()

func (*Agent) RunCancel 

func (a *Agent) RunCancel(t *testing.T)

func (*Agent) RunCheckErr 

func (a *Agent) RunCheckErr(t *testing.T, expectedErr string)

func (*Agent) RunError 

func (a *Agent) RunError(t *testing.T)

func (*Agent) RunSuccess 

func (a *Agent) RunSuccess(t *testing.T)

type AgentOption 

type AgentOption func(*Agent)

func WithAgentOptions 

func WithAgentOptions(options agent.Options) AgentOption

func WithDAGRunID 

func WithDAGRunID(dagRunID string) AgentOption

type CmdTest 

type CmdTest struct {
	Name        string   // Name of the test.
	Args        []string // Arguments to pass to the command.
	ExpectedOut []string // Expected output to be present in the standard output / error.
}

CmdTest is a helper struct to test commands.

type Command 

type Command struct {
	Helper
}

Command is a helper struct to test commands.

func SetupCommand 

func SetupCommand(t *testing.T, opts ...HelperOption) Command

func (Command) CreateDAGFile 

func (c Command) CreateDAGFile(t *testing.T, name string, content string) string

CreateDAGFile creates a DAG file in the DAGsDir for command tests

func (Command) ExecuteCommand 

func (th Command) ExecuteCommand(cmd *cobra.Command, testCase CmdTest) error

ExecuteCommand runs a command and validates the expected output without touching testing.T. It is safe to use from background goroutines in tests.

func (Command) RunCommand 

func (th Command) RunCommand(t *testing.T, cmd *cobra.Command, testCase CmdTest)

func (Command) RunCommandWithError 

func (th Command) RunCommandWithError(t *testing.T, cmd *cobra.Command, testCase CmdTest) error

RunCommandWithError runs a command and returns the error (if any) without failing the test.

type Contains 

type Contains string

type Coordinator 

type Coordinator struct {
	Helper
	// contains filtered or unexported fields
}

Coordinator represents a test gRPC coordinator instance

func SetupCoordinator 

func SetupCoordinator(t *testing.T, opts ...HelperOption) *Coordinator

SetupCoordinator creates and starts a test coordinator instance

func (*Coordinator) Address 

func (c *Coordinator) Address() string

Address returns the address the coordinator is listening on

func (*Coordinator) DispatchTask 

func (c *Coordinator) DispatchTask(t *testing.T, task *coordinatorv1.Task) error

DispatchTask dispatches a task to a waiting worker

func (*Coordinator) GetCoordinatorClient 

func (c *Coordinator) GetCoordinatorClient(t *testing.T) coordinator.Client

GetCoordinatorClient returns a coordinator client for this coordinator

func (*Coordinator) Handler 

func (c *Coordinator) Handler() *coordinator.Handler

Handler returns the coordinator handler for direct testing

func (*Coordinator) LogDir 

func (c *Coordinator) LogDir() string

LogDir returns the log directory path for verifying log persistence

func (*Coordinator) Port 

func (c *Coordinator) Port() int

Port returns the port the coordinator is listening on

func (*Coordinator) Stop 

func (c *Coordinator) Stop() error

Stop gracefully shuts down the coordinator

type DAG 

type DAG struct {
	*Helper
	*core.DAG
}

func (*DAG) Agent 

func (d *DAG) Agent(opts ...AgentOption) *Agent

func (*DAG) AssertCurrentStatus 

func (d *DAG) AssertCurrentStatus(t *testing.T, expected core.Status)

func (*DAG) AssertDAGRunCount 

func (d *DAG) AssertDAGRunCount(t *testing.T, expected int)

func (*DAG) AssertLatestStatus 

func (d *DAG) AssertLatestStatus(t *testing.T, expected core.Status)

func (*DAG) AssertOutputs 

func (d *DAG) AssertOutputs(t *testing.T, outputs map[string]any)

AssertOutputs checks the given outputs against the actual outputs of the DAG Note that this function does not respect dependencies between nodes making the outputs with the same key indeterministic

func (*DAG) ReadOutputs 

func (d *DAG) ReadOutputs(t *testing.T) map[string]string

ReadOutputs reads the collected outputs from the outputs.json file.

type Helper 

type Helper struct {
	Context                   context.Context
	Cancel                    context.CancelFunc
	Config                    *config.Config
	ChildEnv                  []string
	LoggingOutput             *SyncBuffer
	DAGStore                  exec1.DAGStore
	DAGRunStore               exec1.DAGRunStore
	DAGRunMgr                 runtimepkg.Manager
	ProcStore                 exec1.ProcStore
	QueueStore                exec1.QueueStore
	StateStore                dagstate.Store
	ServiceRegistry           exec1.ServiceRegistry
	DispatchTaskStore         exec1.DispatchTaskStore
	WorkerHeartbeatStore      exec1.WorkerHeartbeatStore
	DAGRunLeaseStore          exec1.DAGRunLeaseStore
	ActiveDistributedRunStore exec1.ActiveDistributedRunStore
	SubCmdBuilder             *launcher.SubCmdBuilder
	ServerOptions             []frontend.ServerOption
	StaleHeartbeatThreshold   time.Duration
	StaleLeaseThreshold       time.Duration
	// contains filtered or unexported fields
}

Helper provides test utilities and configuration

func Setup 

func Setup(t *testing.T, opts ...HelperOption) Helper

Setup creates and returns a Helper preconfigured for tests.

Setup prepares an isolated test environment: it creates a temporary DAGU_HOME, writes a minimal config file, initializes stores and a runtime manager, sets key environment variables (e.g. DEBUG, CI, TZ, DAGU_EXECUTABLE, DAGU_CONFIG, SHELL), installs a cancellable context, and registers cleanup to restore the working directory and remove the temp directory. Use the returned Helper to interact with the test runtime and stores.

func (Helper) Cleanup 

func (h Helper) Cleanup()

Cleanup removes temporary test directories

func (Helper) CreateDAGFile 

func (h Helper) CreateDAGFile(t *testing.T, dir string, name string, yamlContent []byte) string

CreateDAGFile creates a DAG file in a given directory for tests that need separate DAG files

func (Helper) DAG 

func (h Helper) DAG(t *testing.T, yamlContent string) DAG

DAG creates a test DAG from YAML content

func (Helper) DAGExpectError 

func (h Helper) DAGExpectError(t *testing.T, name string, expectedErr string)

func (Helper) TempFile 

func (h Helper) TempFile(t *testing.T, name string, data []byte) string

TempFile creates a temp file with specified name and content.

type HelperOption 

type HelperOption func(*Options)

HelperOption defines functional options for Helper

func WithArtifactPersistence 

func WithArtifactPersistence() HelperOption

WithArtifactPersistence enables artifact persistence to filesystem on the coordinator handler. Use this for testing remote artifact uploads from workers.

func WithBuiltExecutable 

func WithBuiltExecutable() HelperOption

WithBuiltExecutable makes Setup build the current ./cmd binary once and use it as cfg.Paths.Executable for subprocess-based tests.

func WithCaptureLoggingOutput 

func WithCaptureLoggingOutput() HelperOption

WithCaptureLoggingOutput creates a logging capture option

func WithConfigMutator 

func WithConfigMutator(mutator func(*config.Config)) HelperOption

WithConfigMutator applies mutations to the loaded configuration after defaults are set.

func WithCoordinatorConfig 

func WithCoordinatorConfig(host string, port int) HelperOption

WithCoordinatorConfig creates a coordinator configuration option

func WithCoordinatorEnabled 

func WithCoordinatorEnabled() HelperOption

WithCoordinatorEnabled re-enables the coordinator in test configuration. By default, tests disable the coordinator since no coordinator is running.

func WithDAGsDir 

func WithDAGsDir(dir string) HelperOption

func WithLogPersistence 

func WithLogPersistence() HelperOption

WithLogPersistence enables log persistence to filesystem on the coordinator handler. Use this for testing remote log streaming from workers.

func WithSchedulerTestDAGs 

func WithSchedulerTestDAGs(dagsDir string) HelperOption

WithSchedulerTestDAGs creates a scheduler option for setting up test DAGs directory

func WithServerConfig 

func WithServerConfig(cfg *config.Server) HelperOption

func WithServerOptions 

func WithServerOptions(serverOpts ...frontend.ServerOption) HelperOption

WithServerOptions appends frontend.ServerOption values to be passed when creating the test server.

func WithStaleThresholds 

func WithStaleThresholds(heartbeat, lease time.Duration) HelperOption

WithStaleThresholds overrides the shared heartbeat and lease staleness thresholds used by distributed test helpers. Useful for tests that need faster zombie detection or dispatch reservation expiry.

func WithStatusPersistence 

func WithStatusPersistence() HelperOption

WithStatusPersistence enables status persistence via DAGRunStore on the coordinator handler. Use this for testing remote status pushing from workers.

type NotEmpty 

type NotEmpty struct{}

type Options 

type Options struct {
	CaptureLoggingOutput bool // CaptureLoggingOutput enables capturing of logging output
	DAGsDir              string
	ServerConfig         *config.Server
	ConfigMutators       []func(*config.Config)
	CoordinatorHost      string
	CoordinatorPort      int
	ServerOptions        []frontend.ServerOption
	UseBuiltExecutable   bool // UseBuiltExecutable builds the current ./cmd binary for subprocess-based tests
	// Coordinator handler options for worker tests
	WithStatusPersistence   bool          // Enable status persistence via DAGRunStore
	WithLogPersistence      bool          // Enable log persistence to filesystem
	WithArtifactPersistence bool          // Enable artifact persistence to filesystem
	StaleHeartbeatThreshold time.Duration // Override for handler's stale heartbeat threshold
	StaleLeaseThreshold     time.Duration // Override for handler's stale lease threshold
}

type ProcHeartbeatObserver 

type ProcHeartbeatObserver interface {
	LatestHeartbeat(ctx context.Context, groupName string, dagRun exec.DAGRunRef) (*exec.ProcHeartbeat, error)
}

ProcHeartbeatObserver is the proc-store surface needed by heartbeat liveness tests.

type Request 

type Request struct {
	// contains filtered or unexported fields
}

Request represents an HTTP request being prepared

func (*Request) ExpectStatus 

func (r *Request) ExpectStatus(code int) *Request

ExpectStatus sets the expected HTTP status code

func (*Request) Send 

func (r *Request) Send(t *testing.T) *Response

Send executes the request and returns the response

func (*Request) WithBasicAuth 

func (r *Request) WithBasicAuth(username, password string) *Request

WithBasicAuth adds Basic authentication to the request

func (*Request) WithBearerToken 

func (r *Request) WithBearerToken(token string) *Request

WithBearerToken adds a Bearer token to the Authorization header

func (*Request) WithHeader 

func (r *Request) WithHeader(key, value string) *Request

WithHeader adds a header to the request

type Response 

type Response struct {
	Body     string
	Response *resty.Response
}

Response represents an HTTP response

func (*Response) Unmarshal 

func (r *Response) Unmarshal(t *testing.T, v any)

Unmarshal parses the response body into the provided value

type Scheduler 

type Scheduler struct {
	Helper
	EntryReader    scheduler.EntryReader
	QueueStore     exec.QueueStore
	CoordinatorCli exec.Dispatcher
}

Scheduler represents a test scheduler instance

func SetupScheduler 

func SetupScheduler(t *testing.T, opts ...HelperOption) *Scheduler

SetupScheduler creates a test scheduler instance with all dependencies

func (*Scheduler) NewSchedulerInstance 

func (s *Scheduler) NewSchedulerInstance(t *testing.T) (*scheduler.Scheduler, error)

NewSchedulerInstance creates a new scheduler instance for testing

func (*Scheduler) Start 

func (s *Scheduler) Start(t *testing.T, ctx context.Context) (*scheduler.Scheduler, chan error)

Start starts the scheduler instance

func (*Scheduler) StartAsync 

func (s *Scheduler) StartAsync(t *testing.T) (*scheduler.Scheduler, chan error)

StartAsync starts the scheduler instance asynchronously

type Server 

type Server struct {
	Helper
}

Server represents a test HTTP server instance

func SetupServer 

func SetupServer(t *testing.T, opts ...HelperOption) Server

SetupServer creates and starts a test server instance

func (*Server) Client 

func (srv *Server) Client() *APIClient

Client returns an HTTP client for the server

type SyncBuffer 

type SyncBuffer struct {
	// contains filtered or unexported fields
}

SyncBuffer provides thread-safe buffer operations

func (*SyncBuffer) Reset 

func (b *SyncBuffer) Reset()

func (*SyncBuffer) String 

func (b *SyncBuffer) String() string

func (*SyncBuffer) Write 

func (b *SyncBuffer) Write(p []byte) (n int, err error)

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.