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) }