forgejo-runner - poller

  1// Copyright The Forgejo Authors.
  2// SPDX-License-Identifier: MIT
  3
  4package poll
  5
  6import (
  7	"context"
  8	"fmt"
  9	"testing"
 10	"time"
 11
 12	"connectrpc.com/connect"
 13
 14	"code.gitea.io/actions-proto-go/ping/v1/pingv1connect"
 15	runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
 16	"code.gitea.io/actions-proto-go/runner/v1/runnerv1connect"
 17	"gitea.com/gitea/act_runner/internal/pkg/config"
 18
 19	log "github.com/sirupsen/logrus"
 20	"github.com/stretchr/testify/assert"
 21)
 22
 23type mockPoller struct {
 24	poller
 25}
 26
 27func (o *mockPoller) Poll() {
 28	o.poller.Poll()
 29}
 30
 31type mockClient struct {
 32	pingv1connect.PingServiceClient
 33	runnerv1connect.RunnerServiceClient
 34
 35	sleep  time.Duration
 36	cancel bool
 37	err    error
 38	noTask bool
 39}
 40
 41func (o mockClient) Address() string {
 42	return ""
 43}
 44
 45func (o mockClient) Insecure() bool {
 46	return true
 47}
 48
 49func (o *mockClient) FetchTask(ctx context.Context, req *connect.Request[runnerv1.FetchTaskRequest]) (*connect.Response[runnerv1.FetchTaskResponse], error) {
 50	if o.sleep > 0 {
 51		select {
 52		case <-ctx.Done():
 53			log.Trace("fetch task done")
 54			return nil, context.DeadlineExceeded
 55		case <-time.After(o.sleep):
 56			log.Trace("slept")
 57			return nil, fmt.Errorf("unexpected")
 58		}
 59	}
 60	if o.cancel {
 61		return nil, context.Canceled
 62	}
 63	if o.err != nil {
 64		return nil, o.err
 65	}
 66	task := &runnerv1.Task{}
 67	if o.noTask {
 68		task = nil
 69		o.noTask = false
 70	}
 71
 72	return connect.NewResponse(&runnerv1.FetchTaskResponse{
 73		Task:         task,
 74		TasksVersion: int64(1),
 75	}), nil
 76}
 77
 78type mockRunner struct {
 79	cfg    *config.Runner
 80	log    chan string
 81	panics bool
 82	err    error
 83}
 84
 85func (o *mockRunner) Run(ctx context.Context, task *runnerv1.Task) error {
 86	o.log <- "runner starts"
 87	if o.panics {
 88		log.Trace("panics")
 89		o.log <- "runner panics"
 90		o.panics = false
 91		panic("whatever")
 92	}
 93	if o.err != nil {
 94		log.Trace("error")
 95		o.log <- "runner error"
 96		err := o.err
 97		o.err = nil
 98		return err
 99	}
100	for {
101		select {
102		case <-ctx.Done():
103			log.Trace("shutdown")
104			o.log <- "runner shutdown"
105			return nil
106		case <-time.After(o.cfg.Timeout):
107			log.Trace("after")
108			o.log <- "runner timeout"
109			return nil
110		}
111	}
112}
113
114func setTrace(t *testing.T) {
115	t.Helper()
116	log.SetReportCaller(true)
117	log.SetLevel(log.TraceLevel)
118}
119
120func TestPoller_New(t *testing.T) {
121	p := New(&config.Config{}, &mockClient{}, &mockRunner{})
122	assert.NotNil(t, p)
123}
124
125func TestPoller_Runner(t *testing.T) {
126	setTrace(t)
127	for _, testCase := range []struct {
128		name           string
129		timeout        time.Duration
130		noTask         bool
131		panics         bool
132		err            error
133		expected       string
134		contextTimeout time.Duration
135	}{
136		{
137			name:     "Simple",
138			timeout:  10 * time.Second,
139			expected: "runner shutdown",
140		},
141		{
142			name:     "Panics",
143			timeout:  10 * time.Second,
144			panics:   true,
145			expected: "runner panics",
146		},
147		{
148			name:     "Error",
149			timeout:  10 * time.Second,
150			err:      fmt.Errorf("ERROR"),
151			expected: "runner error",
152		},
153		{
154			name:     "PollTaskError",
155			timeout:  10 * time.Second,
156			noTask:   true,
157			expected: "runner shutdown",
158		},
159		{
160			name:           "ShutdownTimeout",
161			timeout:        1 * time.Second,
162			contextTimeout: 1 * time.Minute,
163			expected:       "runner timeout",
164		},
165	} {
166		t.Run(testCase.name, func(t *testing.T) {
167			runnerLog := make(chan string, 3)
168			configRunner := config.Runner{
169				FetchInterval: 1,
170				Capacity:      1,
171				Timeout:       testCase.timeout,
172			}
173			p := &mockPoller{}
174			p.init(
175				&config.Config{
176					Runner: configRunner,
177				},
178				&mockClient{
179					noTask: testCase.noTask,
180				},
181				&mockRunner{
182					cfg:    &configRunner,
183					log:    runnerLog,
184					panics: testCase.panics,
185					err:    testCase.err,
186				})
187			go p.Poll()
188			assert.Equal(t, "runner starts", <-runnerLog)
189			var ctx context.Context
190			var cancel context.CancelFunc
191			if testCase.contextTimeout > 0 {
192				ctx, cancel = context.WithTimeout(context.Background(), testCase.contextTimeout)
193				defer cancel()
194			} else {
195				ctx, cancel = context.WithCancel(context.Background())
196				cancel()
197			}
198			p.Shutdown(ctx)
199			<-p.done
200			assert.Equal(t, testCase.expected, <-runnerLog)
201		})
202	}
203}
204
205func TestPoller_Fetch(t *testing.T) {
206	setTrace(t)
207	for _, testCase := range []struct {
208		name    string
209		noTask  bool
210		sleep   time.Duration
211		err     error
212		cancel  bool
213		success bool
214	}{
215		{
216			name:    "Success",
217			success: true,
218		},
219		{
220			name:  "Timeout",
221			sleep: 100 * time.Millisecond,
222		},
223		{
224			name:   "Canceled",
225			cancel: true,
226		},
227		{
228			name:   "NoTask",
229			noTask: true,
230		},
231		{
232			name: "Error",
233			err:  fmt.Errorf("random error"),
234		},
235	} {
236		t.Run(testCase.name, func(t *testing.T) {
237			configRunner := config.Runner{
238				FetchTimeout: 1 * time.Millisecond,
239			}
240			p := &mockPoller{}
241			p.init(
242				&config.Config{
243					Runner: configRunner,
244				},
245				&mockClient{
246					sleep:  testCase.sleep,
247					cancel: testCase.cancel,
248					noTask: testCase.noTask,
249					err:    testCase.err,
250				},
251				&mockRunner{},
252			)
253			task, ok := p.fetchTask(context.Background())
254			if testCase.success {
255				assert.True(t, ok)
256				assert.NotNil(t, task)
257			} else {
258				assert.False(t, ok)
259				assert.Nil(t, task)
260			}
261		})
262	}
263}