344 lines
13 KiB
Go
344 lines
13 KiB
Go
package jobs
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"io"
|
|
"log/slog"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"github.com/shcizo/package-updater/internal/master/events"
|
|
"github.com/shcizo/package-updater/internal/master/model"
|
|
"github.com/shcizo/package-updater/internal/protocol"
|
|
)
|
|
|
|
type fakeStore struct {
|
|
services map[int64]model.Service
|
|
jobs map[int64]model.Job
|
|
created []model.NewJob
|
|
active *model.Job // returned with ErrJobActive when set
|
|
|
|
revokedAgents map[int64]bool // CreateJob returns ErrAgentOffline for these
|
|
}
|
|
|
|
func (f *fakeStore) GetService(_ context.Context, id int64) (model.Service, error) {
|
|
svc, ok := f.services[id]
|
|
if !ok {
|
|
return model.Service{}, model.ErrNotFound
|
|
}
|
|
return svc, nil
|
|
}
|
|
|
|
func (f *fakeStore) CreateJob(_ context.Context, nj model.NewJob) (model.Job, error) {
|
|
if f.active != nil {
|
|
return *f.active, model.ErrJobActive
|
|
}
|
|
if nj.AgentID != nil && f.revokedAgents[*nj.AgentID] {
|
|
return model.Job{}, model.ErrAgentOffline
|
|
}
|
|
f.created = append(f.created, nj)
|
|
j := model.Job{ID: int64(len(f.created)), ServiceID: nj.ServiceID, AgentID: nj.AgentID, FromRef: nj.FromRef,
|
|
ToRef: nj.ToRef, Trigger: nj.Trigger, RequestedBy: nj.RequestedBy, Status: model.JobQueued, CreatedAt: nj.CreatedAt}
|
|
f.jobs[j.ID] = j
|
|
return j, nil
|
|
}
|
|
|
|
func (f *fakeStore) FailJob(_ context.Context, id int64, msg string, at time.Time) (bool, error) {
|
|
j := f.jobs[id]
|
|
j.Status, j.Error, j.FinishedAt = model.JobFailed, msg, &at
|
|
f.jobs[id] = j
|
|
return true, nil
|
|
}
|
|
|
|
func (f *fakeStore) GetJob(_ context.Context, id int64) (model.Job, error) { return f.jobs[id], nil }
|
|
|
|
type startCall struct {
|
|
route model.Route
|
|
id int64
|
|
start protocol.JobStart
|
|
}
|
|
|
|
type fakeHub struct {
|
|
route model.Route
|
|
routeErr error
|
|
startErr error
|
|
starts []startCall
|
|
}
|
|
|
|
func (h *fakeHub) Route(model.Service) (model.Route, error) { return h.route, h.routeErr }
|
|
|
|
func (h *fakeHub) StartJob(r model.Route, id int64, s protocol.JobStart) error {
|
|
h.starts = append(h.starts, startCall{r, id, s})
|
|
return h.startErr
|
|
}
|
|
|
|
type fakePub struct{ topics []events.Topic }
|
|
|
|
func (p *fakePub) Publish(t events.Topic) { p.topics = append(p.topics, t) }
|
|
|
|
const selfImage = "gitea.shcizo.se/shcizo/package-updater:latest"
|
|
|
|
var (
|
|
composeSvc = model.Service{ID: 1, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web",
|
|
AgentID: ptr(int64(7)), ActualRef: "gitea.shcizo.se/shcizo/slusk:1.4.2@sha256:d1"}
|
|
bareSvc = model.Service{ID: 3, Kind: protocol.KindCompose, ComposeProject: "db", ComposeService: "pg",
|
|
AgentID: ptr(int64(7)), ActualRef: "postgres"}
|
|
swarmSvc = model.Service{ID: 2, Kind: protocol.KindSwarm, SwarmClusterID: "c1", SwarmService: "gitea_gitea",
|
|
ActualRef: "gitea/gitea:1.27.3@sha256:87a6"}
|
|
)
|
|
|
|
func ptr[T any](v T) *T { return &v }
|
|
|
|
func newService(st *fakeStore, h *fakeHub) (*Service, *fakePub) {
|
|
pub := &fakePub{}
|
|
return New(st, h, pub, selfImage, slog.New(slog.NewTextHandler(io.Discard, nil))), pub
|
|
}
|
|
|
|
func TestUpdatable(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
svc model.Service
|
|
reason string
|
|
}{
|
|
{"removed", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", RemovedAt: ptr(time.Now())}, model.ReasonRemoved},
|
|
{"flagged self", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Self: true}, model.ReasonSelf},
|
|
{"same repo as the agent image", model.Service{Kind: protocol.KindCompose,
|
|
ActualRef: "gitea.shcizo.se/shcizo/package-updater:1.2.0@sha256:abc"}, model.ReasonSelf},
|
|
{"built from source", model.Service{Kind: protocol.KindCompose, ActualRef: "slusk:1", Built: true}, model.ReasonBuilt},
|
|
{"container", model.Service{Kind: protocol.KindContainer, ActualRef: "redis:7"}, model.ReasonContainer},
|
|
{"digest without tag", model.Service{Kind: protocol.KindCompose, ActualRef: "redis@sha256:abc"}, model.ReasonNoTag},
|
|
{"bare image means latest", model.Service{Kind: protocol.KindCompose, ActualRef: "postgres"}, ""},
|
|
{"bare image on a registry with port", model.Service{Kind: protocol.KindCompose, ActualRef: "localhost:5000/foo"}, ""},
|
|
{"tag", model.Service{Kind: protocol.KindCompose, ActualRef: "postgres:16"}, ""},
|
|
{"tag and digest", model.Service{Kind: protocol.KindCompose, ActualRef: "postgres:17.10@sha256:abc"}, ""},
|
|
{"registry with port and tag", model.Service{Kind: protocol.KindSwarm, ActualRef: "localhost:5000/foo:v1"}, ""},
|
|
}
|
|
s, _ := newService(&fakeStore{}, &fakeHub{})
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
err := s.Updatable(tt.svc)
|
|
if tt.reason == "" {
|
|
assert.NoError(t, err)
|
|
return
|
|
}
|
|
var nu *model.NotUpdatableError
|
|
require.True(t, errors.As(err, &nu))
|
|
assert.Equal(t, tt.reason, nu.Reason)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestTrigger(t *testing.T) {
|
|
ok := model.Route{AgentID: 7, ConnID: 70}
|
|
activeJob := model.Job{ID: 99, Status: model.JobRunning}
|
|
|
|
tests := []struct {
|
|
name string
|
|
serviceID int64
|
|
hub *fakeHub
|
|
active *model.Job
|
|
wantErr error
|
|
wantStatus model.JobStatus
|
|
wantError string
|
|
wantStart *protocol.JobStart
|
|
wantCreate bool
|
|
}{
|
|
{name: "compose happy path", serviceID: 1, hub: &fakeHub{route: ok}, wantStatus: model.JobQueued, wantCreate: true,
|
|
wantStart: &protocol.JobStart{Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", Ref: "gitea.shcizo.se/shcizo/slusk:1.4.2"}},
|
|
{name: "swarm", serviceID: 2, hub: &fakeHub{route: ok}, wantStatus: model.JobQueued, wantCreate: true,
|
|
wantStart: &protocol.JobStart{Kind: protocol.KindSwarm, SwarmClusterID: "c1", SwarmService: "gitea_gitea", Ref: "gitea/gitea:1.27.3"}},
|
|
{name: "bare image deploys latest", serviceID: 3, hub: &fakeHub{route: ok}, wantStatus: model.JobQueued, wantCreate: true,
|
|
wantStart: &protocol.JobStart{Kind: protocol.KindCompose, ComposeProject: "db", ComposeService: "pg", Ref: "postgres:latest"}},
|
|
{name: "active job", serviceID: 1, hub: &fakeHub{route: ok}, active: &activeJob, wantErr: model.ErrJobActive, wantStatus: model.JobRunning},
|
|
{name: "agent offline", serviceID: 1, hub: &fakeHub{routeErr: model.ErrAgentOffline}, wantStatus: model.JobFailed,
|
|
wantError: "agent offline", wantCreate: true},
|
|
{name: "agent too old", serviceID: 1, hub: &fakeHub{routeErr: model.ErrAgentTooOld}, wantStatus: model.JobFailed,
|
|
wantError: "agent too old for updates", wantCreate: true},
|
|
{name: "start fails", serviceID: 1, hub: &fakeHub{route: ok, startErr: model.ErrAgentOffline}, wantStatus: model.JobFailed,
|
|
wantError: "agent offline", wantCreate: true,
|
|
wantStart: &protocol.JobStart{Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", Ref: "gitea.shcizo.se/shcizo/slusk:1.4.2"}},
|
|
{name: "unknown service", serviceID: 5, hub: &fakeHub{route: ok}, wantErr: model.ErrNotFound},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
st := &fakeStore{
|
|
services: map[int64]model.Service{1: composeSvc, 2: swarmSvc, 3: bareSvc},
|
|
jobs: map[int64]model.Job{99: activeJob},
|
|
active: tt.active,
|
|
}
|
|
s, pub := newService(st, tt.hub)
|
|
|
|
job, err := s.Trigger(context.Background(), tt.serviceID, "sub-1")
|
|
|
|
if tt.wantErr != nil {
|
|
assert.ErrorIs(t, err, tt.wantErr)
|
|
} else {
|
|
require.NoError(t, err)
|
|
}
|
|
if tt.wantErr == nil || tt.active != nil {
|
|
assert.Equal(t, tt.wantStatus, job.Status)
|
|
assert.Equal(t, tt.wantError, job.Error)
|
|
}
|
|
if tt.wantStart == nil {
|
|
assert.Empty(t, tt.hub.starts)
|
|
} else {
|
|
require.Len(t, tt.hub.starts, 1)
|
|
assert.Equal(t, *tt.wantStart, tt.hub.starts[0].start)
|
|
assert.Equal(t, ok, tt.hub.starts[0].route)
|
|
assert.Equal(t, job.ID, tt.hub.starts[0].id)
|
|
}
|
|
if tt.wantCreate {
|
|
require.Len(t, st.created, 1)
|
|
assert.Equal(t, "sub-1", st.created[0].RequestedBy)
|
|
assert.Equal(t, model.TriggerManual, st.created[0].Trigger)
|
|
assert.Equal(t, st.services[tt.serviceID].ActualRef, st.created[0].FromRef)
|
|
assert.Contains(t, pub.topics, events.Jobs)
|
|
assert.Contains(t, pub.topics, events.JobTopic(job.ID))
|
|
} else {
|
|
assert.Empty(t, st.created)
|
|
assert.Empty(t, pub.topics)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// The routed agent can be revoked between Route and CreateJob; the store then
|
|
// refuses the job. Trigger records a failed job that was never sent.
|
|
func TestTrigger_AgentRevokedAfterRoute(t *testing.T) {
|
|
st := &fakeStore{
|
|
services: map[int64]model.Service{1: composeSvc},
|
|
jobs: map[int64]model.Job{},
|
|
revokedAgents: map[int64]bool{7: true},
|
|
}
|
|
h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
|
|
s, _ := newService(st, h)
|
|
|
|
job, err := s.Trigger(context.Background(), 1, "sub-1")
|
|
|
|
require.NoError(t, err)
|
|
assert.Equal(t, model.JobFailed, job.Status)
|
|
assert.Equal(t, "agent offline", job.Error)
|
|
assert.Nil(t, job.AgentID)
|
|
assert.Empty(t, h.starts)
|
|
}
|
|
|
|
func TestTrigger_NotUpdatableCreatesNoJob(t *testing.T) {
|
|
st := &fakeStore{
|
|
services: map[int64]model.Service{1: {ID: 1, Kind: protocol.KindContainer, ActualRef: "redis:7"}},
|
|
jobs: map[int64]model.Job{},
|
|
}
|
|
h := &fakeHub{}
|
|
s, pub := newService(st, h)
|
|
|
|
_, err := s.Trigger(context.Background(), 1, "sub-1")
|
|
|
|
var nu *model.NotUpdatableError
|
|
require.True(t, errors.As(err, &nu))
|
|
assert.Empty(t, st.created)
|
|
assert.Empty(t, h.starts)
|
|
assert.Empty(t, pub.topics)
|
|
}
|
|
|
|
func TestStart_WebhookTriggerWithExplicitRef(t *testing.T) {
|
|
st := &fakeStore{services: map[int64]model.Service{1: composeSvc}, jobs: map[int64]model.Job{}}
|
|
h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
|
|
s, _ := newService(st, h)
|
|
|
|
job, err := s.Start(context.Background(), composeSvc, StartRequest{
|
|
Trigger: model.TriggerWebhook, Ref: "gitea.shcizo.se/shcizo/slusk:2.0.0"})
|
|
|
|
require.NoError(t, err)
|
|
assert.Equal(t, model.JobQueued, job.Status)
|
|
require.Len(t, st.created, 1)
|
|
assert.Equal(t, model.TriggerWebhook, st.created[0].Trigger)
|
|
assert.Empty(t, st.created[0].RequestedBy)
|
|
assert.Equal(t, "gitea.shcizo.se/shcizo/slusk:2.0.0", st.created[0].ToRef)
|
|
require.Len(t, h.starts, 1)
|
|
assert.Equal(t, "gitea.shcizo.se/shcizo/slusk:2.0.0", h.starts[0].start.Ref)
|
|
}
|
|
|
|
func TestStart_ExplicitRefBypassesOnlyTheNoTagRefusal(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
svc model.Service
|
|
reason string // "" means the job starts
|
|
}{
|
|
{"digest only", model.Service{Kind: protocol.KindCompose, ActualRef: "redis@sha256:abc"}, ""},
|
|
{"removed", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", RemovedAt: ptr(time.Now())}, model.ReasonRemoved},
|
|
{"self", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Self: true}, model.ReasonSelf},
|
|
{"built", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Built: true}, model.ReasonBuilt},
|
|
{"container", model.Service{Kind: protocol.KindContainer, ActualRef: "redis:7"}, model.ReasonContainer},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
tt.svc.ID = 1
|
|
st := &fakeStore{services: map[int64]model.Service{1: tt.svc}, jobs: map[int64]model.Job{}}
|
|
h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
|
|
s, _ := newService(st, h)
|
|
|
|
_, err := s.Start(context.Background(), tt.svc, StartRequest{Trigger: model.TriggerWebhook, Ref: "x/y:2"})
|
|
|
|
if tt.reason == "" {
|
|
require.NoError(t, err)
|
|
assert.Len(t, st.created, 1)
|
|
return
|
|
}
|
|
var nu *model.NotUpdatableError
|
|
require.True(t, errors.As(err, &nu))
|
|
assert.Equal(t, tt.reason, nu.Reason)
|
|
assert.Empty(t, st.created)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestStart_WithoutRefStillRefusesNoTag(t *testing.T) {
|
|
svc := model.Service{ID: 1, Kind: protocol.KindCompose, ActualRef: "redis@sha256:abc"}
|
|
s, _ := newService(&fakeStore{jobs: map[int64]model.Job{}}, &fakeHub{})
|
|
|
|
_, err := s.Start(context.Background(), svc, StartRequest{Trigger: model.TriggerWebhook})
|
|
|
|
var nu *model.NotUpdatableError
|
|
require.True(t, errors.As(err, &nu))
|
|
assert.Equal(t, model.ReasonNoTag, nu.Reason)
|
|
}
|
|
|
|
func TestAutoAllowed(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
svc model.Service
|
|
reason string
|
|
}{
|
|
{"removed", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", RemovedAt: ptr(time.Now())}, model.ReasonRemoved},
|
|
{"self", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Self: true}, model.ReasonSelf},
|
|
{"agent image repo", model.Service{Kind: protocol.KindCompose, ActualRef: selfImage}, model.ReasonSelf},
|
|
{"built", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Built: true}, model.ReasonBuilt},
|
|
{"container", model.Service{Kind: protocol.KindContainer, ActualRef: "redis:7"}, model.ReasonContainer},
|
|
{"digest without tag is allowed", model.Service{Kind: protocol.KindCompose, ActualRef: "redis@sha256:abc"}, ""},
|
|
{"tag", model.Service{Kind: protocol.KindCompose, ActualRef: "postgres:16"}, ""},
|
|
}
|
|
s, _ := newService(&fakeStore{}, &fakeHub{})
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
err := s.AutoAllowed(tt.svc)
|
|
if tt.reason == "" {
|
|
assert.NoError(t, err)
|
|
return
|
|
}
|
|
var nu *model.NotUpdatableError
|
|
require.True(t, errors.As(err, &nu))
|
|
assert.Equal(t, tt.reason, nu.Reason)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestFollowedRef(t *testing.T) {
|
|
assert.Equal(t, "postgres:16", FollowedRef("postgres:16@sha256:abc"))
|
|
assert.Equal(t, "postgres:latest", FollowedRef("postgres"))
|
|
assert.Equal(t, "localhost:5000/foo:latest", FollowedRef("localhost:5000/foo"))
|
|
assert.Empty(t, FollowedRef("redis@sha256:abc"))
|
|
}
|