act_runner/internal/app/cmd/daemon_test.go

// Copyright 2026 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT

package cmd

import (
	"context"
	"crypto/tls"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"gitea.com/gitea/runner/act/common/logging"
	"gitea.com/gitea/runner/internal/pkg/config"

	"connectrpc.com/connect/v2"
	"connectrpc.com/connect/v2/connecthttp"
	runnerv1 "gitea.dev/actionslib/runner/v1"
	"gitea.dev/actionslib/runner/v1/runnerv1connect"
	"github.com/stretchr/testify/require"
	"google.golang.org/protobuf/types/known/structpb"
)

func TestResolveLabels(t *testing.T) {
	var (
		cfgLabels = []string{"cfg:host"}
		regLabels = []string{"reg:host"}
	)

	tests := []struct {
		name string
		arg  string
		cfg  []string
		reg  []string
		want []string
	}{
		{"flag wins", "flag:host,other", cfgLabels, regLabels, []string{"flag:host", "other"}},
		{"config wins over registration", "", cfgLabels, regLabels, cfgLabels},
		{"registration is the fallback", "", nil, regLabels, regLabels},
		{"blank flag is ignored", " , ", cfgLabels, regLabels, cfgLabels},
		{"nothing configured", "", nil, nil, nil},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			require.Equal(t, tt.want, resolveLabels(tt.arg, tt.cfg, tt.reg))
		})
	}
}

func TestGetDockerSocketPathUsesConfigAndEnvironment(t *testing.T) {
	got, err := getDockerSocketPath("tcp://docker.example:2376")
	require.NoError(t, err)
	require.Equal(t, "tcp://docker.example:2376", got)

	t.Setenv("DOCKER_HOST", "unix:///tmp/docker.sock")
	got, err = getDockerSocketPath("-")
	require.NoError(t, err)
	require.Equal(t, "unix:///tmp/docker.sock", got)
}

func TestInitLoggingSetsLevelAndCaller(t *testing.T) {
	oldLevel, oldOut := logging.GetLevel(), logging.Output()
	t.Cleanup(func() {
		logging.SetLevel(oldLevel)
		logging.SetOutput(oldOut)
		logging.SetReportCaller(false)
		logging.SetTextFormat(logging.TextOptions{})
	})

	cfg := &config.Config{}
	cfg.Log.Level = "debug"
	initLogging(cfg)

	require.Equal(t, logging.LevelDebug, logging.GetLevel())
	var buf strings.Builder
	logging.SetOutput(&buf)
	logging.Debugf("line")
	require.Contains(t, buf.String(), "[daemon_test.go:")
}

var errUnregisteredRunner = connect.NewError(connect.CodeUnknown, "rpc error: code = Unauthenticated desc = unregistered runner")

type daemonTestService struct {
	runnerv1connect.UnimplementedRunnerServiceHandler
	declareErr, fetchErr error
	declareCalls         atomic.Int32
	beforeTask           bool
	cancel               context.CancelFunc
}

func (s *daemonTestService) Declare(context.Context, *runnerv1.DeclareRequest) (*runnerv1.DeclareResponse, error) {
	if s.declareCalls.Add(1) == 1 {
		return nil, connect.NewError(connect.CodeUnavailable, "bad gateway")
	}
	if s.declareErr != nil {
		if s.beforeTask {
			s.cancel()
		}
		return nil, s.declareErr
	}
	return &runnerv1.DeclareResponse{Runner: &runnerv1.Runner{Name: "test"}}, nil
}

func (s *daemonTestService) FetchTask(context.Context, *runnerv1.FetchTaskRequest) (*runnerv1.FetchTaskResponse, error) {
	if s.fetchErr != nil {
		return nil, s.fetchErr
	}
	if s.beforeTask {
		s.cancel()
		return nil, context.Canceled
	}
	return &runnerv1.FetchTaskResponse{Task: &runnerv1.Task{Id: 1, Context: &structpb.Struct{}, WorkflowPayload: []byte("invalid: [")}}, nil
}

func (*daemonTestService) UpdateTask(_ context.Context, req *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) {
	return &runnerv1.UpdateTaskResponse{State: req.State}, nil
}

func (*daemonTestService) UpdateLog(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) {
	return &runnerv1.UpdateLogResponse{AckIndex: req.Index + int64(len(req.Rows))}, nil
}

func TestDaemonRetriesDeclareOutagesAndRemovesOnlyConsumedEphemeralRegistration(t *testing.T) {
	defer func(delay time.Duration) { declareRetryDelay = delay }(declareRetryDelay)
	declareRetryDelay = 0
	for _, tc := range []struct {
		name                  string
		ephemeral, beforeTask bool
		declareErr, fetchErr  error
		removed               bool
	}{
		{"ephemeral failed task", true, false, nil, nil, true},
		{"ephemeral interrupted before task", true, true, nil, nil, false},
		{"persistent once", false, false, nil, nil, false},
		{"ephemeral rejected on fetch", true, false, nil, errUnregisteredRunner, true},
		{"ephemeral rejected on fetch with 401", true, false, nil, connect.NewError(connect.CodeUnauthenticated, "unregistered runner"), true},
		{"ephemeral proxy 401 on fetch", true, false, nil, connect.NewError(connect.CodeUnauthenticated, "HTTP status 401 Unauthorized"), false},
		{"ephemeral rejected on declare", true, false, errUnregisteredRunner, nil, true},
		{"persistent rejected on declare", false, false, errUnregisteredRunner, nil, false},
		{"ephemeral stopped during declare outage", true, true, connect.NewError(connect.CodeUnavailable, "bad gateway"), nil, false},
	} {
		t.Run(tc.name, func(t *testing.T) {
			ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
			defer cancel()
			service := &daemonTestService{declareErr: tc.declareErr, fetchErr: tc.fetchErr, beforeTask: tc.beforeTask, cancel: cancel}
			rpcServer := connect.NewServer()
			runnerv1connect.RegisterRunnerServiceHandler(rpcServer, service)
			mux := http.NewServeMux()
			connecthttp.Mount(mux, rpcServer, connecthttp.WithReadMaxBytes(0))
			server := httptest.NewServer(http.StripPrefix("/api/actions", mux))
			defer server.Close()
			dir := t.TempDir()
			regFile := filepath.Join(dir, "registration.json")
			configFile := filepath.Join(dir, "config.yaml")
			require.NoError(t, os.WriteFile(configFile, []byte("runner:\n  file: "+regFile+"\n  idle_cleanup_interval: 0s\ncache:\n  enabled: false\n"), 0o600))
			require.NoError(t, config.SaveRegistration(regFile, &config.Registration{Address: server.URL, UUID: "test", Name: "test", Token: "test", Labels: []string{"host:host"}, Ephemeral: tc.ephemeral}))
			err := runDaemon(ctx, &daemonArgs{Once: true}, &configFile)(nil, nil)
			if tc.declareErr == nil && tc.fetchErr == nil || tc.beforeTask {
				require.NoError(t, err)
			} else {
				require.Error(t, err)
			}
			require.Equal(t, int32(2), service.declareCalls.Load())
			if tc.removed {
				require.NoFileExists(t, regFile)
			} else {
				require.FileExists(t, regFile)
			}
		})
	}
}

func TestShouldRetryDeclare(t *testing.T) {
	for _, tc := range []struct {
		err   error
		retry bool
	}{
		{connect.NewError(connect.CodeUnavailable, "bad gateway"), true},
		{connect.NewError(connect.CodeInternal, "update runner").WithRemote(), true},
		{connect.NewError(connect.CodeUnavailable, "certificate verification").WithCause(&tls.CertificateVerificationError{}), false},
		{connect.NewError(connect.CodeUnavailable, "scheme mismatch").WithCause(http.ErrSchemeMismatch), false},
		{connect.NewError(connect.CodeUnavailable, "TLS alert").WithCause(tls.AlertError(42)), false},
		{connect.NewError(connect.CodePermissionDenied, "forbidden"), false},
		{connect.NewError(connect.CodeUnknown, "unregistered runner").WithRemote(), false},
	} {
		require.Equal(t, tc.retry, shouldRetryDeclare(tc.err), tc.err.Error())
	}
}