repositories / bugabinga.net
bugabinga.net
personal infrastructure for bugabinga!
owned by admin
services/toad/internal/api/api_test.go
Rawpackage api
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"sync"
"testing"
"time"
"bugabinga.net/toad/internal/credential"
"bugabinga.net/toad/internal/enroll"
"bugabinga.net/toad/internal/rollout"
"bugabinga.net/toad/internal/store"
)
const (
digestA = "sha256:1111111111111111111111111111111111111111111111111111111111111111"
digestB = "sha256:2222222222222222222222222222222222222222222222222222222222222222"
)
func TestDeploymentPromotesAfterStabilityWindow(t *testing.T) {
manager, host, _ := newHarness(t)
operation := submit(t, manager, digestA, 0)
manager.Drain()
operation = reload(t, manager, operation)
if operation.Phase != store.PhasePromoted || operation.Generation != 1 {
t.Fatalf("operation = %+v", operation)
}
state, err := manager.Store.ReadState("app")
if err != nil || state.Generation != 1 || state.Current == nil || state.Current.Digest != digestA || state.Rollback != nil {
t.Fatalf("state = %+v err = %v", state, err)
}
if !host.app().running || host.app().digest != digestA {
t.Fatalf("host = %+v", host.app())
}
}
func TestStaleGenerationAndIdempotentReplay(t *testing.T) {
manager, _, _ := newHarness(t)
first := submit(t, manager, digestA, 0)
manager.Drain()
if _, err := manager.Submit(service(t, manager), DeployRequest{Digest: digestB, ExpectedGeneration: 0, IdempotencyKey: "stale"}); !errors.Is(err, ErrConflict) {
t.Fatalf("stale generation accepted: %v", err)
}
replay, err := manager.Submit(service(t, manager), DeployRequest{Digest: digestA, ExpectedGeneration: 0, IdempotencyKey: "deploy-app-1"})
if err != nil || replay.ID != first.ID {
t.Fatalf("replay = %+v err = %v", replay, err)
}
manager.Drain()
state, _ := manager.Store.ReadState("app")
if state.Generation != 1 {
t.Fatalf("replay started a second rollout: generation = %d", state.Generation)
}
}
func TestConcurrentRolloutIsRejected(t *testing.T) {
manager, host, _ := newHarness(t)
gate := make(chan struct{})
host.pull = func(string) (string, error) {
<-gate
return "", errors.New("aborted")
}
if _, err := manager.Submit(service(t, manager), DeployRequest{Digest: digestA, ExpectedGeneration: 0, IdempotencyKey: "first"}); err != nil {
t.Fatal(err)
}
_, err := manager.Submit(service(t, manager), DeployRequest{Digest: digestB, ExpectedGeneration: 0, IdempotencyKey: "second"})
if !errors.Is(err, ErrConflict) {
t.Fatalf("second rollout accepted: %v", err)
}
close(gate)
manager.Drain()
}
func TestUnreadyCandidateWithoutPreviousRevisionFails(t *testing.T) {
manager, host, _ := newHarness(t)
host.probe = func(string) error { return errors.New("connection refused") }
operation := submit(t, manager, digestA, 0)
manager.Drain()
operation = reload(t, manager, operation)
if operation.Phase != store.PhaseFailed || !strings.Contains(operation.Reason, "deadline") {
t.Fatalf("operation = %+v", operation)
}
if host.app().running {
t.Fatal("failed candidate left running")
}
state, _ := manager.Store.ReadState("app")
if state.Generation != 0 || state.Current != nil {
t.Fatalf("state = %+v", state)
}
}
func TestRestartBudgetRollsBackToPreviousRevision(t *testing.T) {
manager, host, _ := newHarness(t)
submit(t, manager, digestA, 0)
manager.Drain()
host.onPoll = func(h *fakeHost, _ string) {
if h.app().digest == digestB {
h.app().restarts++
}
}
operation := submit(t, manager, digestB, 1)
manager.Drain()
operation = reload(t, manager, operation)
if operation.Phase != store.PhaseRolledBack || !strings.Contains(operation.Reason, "restart budget") {
t.Fatalf("operation = %+v", operation)
}
state, _ := manager.Store.ReadState("app")
if state.Current == nil || state.Current.Digest != digestA {
t.Fatalf("rollback changed current revision: %+v", state)
}
if state.Generation != 2 {
t.Fatalf("generation did not advance after rollback: %d", state.Generation)
}
if !host.app().running || host.app().digest != digestA {
t.Fatalf("previous revision not restored: %+v", host.app())
}
}
func TestReadinessFromWrongImageDoesNotPromote(t *testing.T) {
manager, host, clock := newHarness(t)
host.reportDigest = digestB
operation := submit(t, manager, digestA, 0)
manager.Drain()
operation = reload(t, manager, operation)
if operation.Phase != store.PhaseFailed {
t.Fatalf("unattributable readiness promoted: %+v", operation)
}
if clock.Now().Before(operation.Deadline) {
t.Fatal("failure reported before the rollout deadline")
}
}
func TestUnverifiedTerminationAbortsWithoutStartingCandidate(t *testing.T) {
manager, host, _ := newHarness(t)
submit(t, manager, digestA, 0)
manager.Drain()
host.stopRefused = true
operation := submit(t, manager, digestB, 1)
manager.Drain()
operation = reload(t, manager, operation)
if operation.Phase != store.PhaseFailed || !strings.Contains(operation.Reason, "still running") {
t.Fatalf("operation = %+v", operation)
}
if host.app().digest != digestA || !strings.Contains(host.app().installed, digestA) {
t.Fatalf("candidate started or unit not restored: %+v", host.app())
}
state, _ := manager.Store.ReadState("app")
if state.Current == nil || state.Current.Digest != digestA {
t.Fatalf("state = %+v", state)
}
}
func TestFailedRollbackIsTerminal(t *testing.T) {
manager, host, clock := newHarness(t)
submit(t, manager, digestA, 0)
manager.Drain()
breakAt := clock.Now().Add(time.Second)
host.probe = func(string) error {
if clock.Now().After(breakAt) {
return errors.New("connection refused")
}
return nil
}
operation := submit(t, manager, digestB, 1)
manager.Drain()
operation = reload(t, manager, operation)
if operation.Phase != store.PhaseFailed || !strings.Contains(operation.Reason, "rollback failed") {
t.Fatalf("operation = %+v", operation)
}
if operation.EndedAt == nil {
t.Fatal("failed rollback is not terminal")
}
}
func TestOperatorAbortRollsBackDuringProbation(t *testing.T) {
manager, host, _ := newHarness(t)
submit(t, manager, digestA, 0)
manager.Drain()
var once sync.Once
host.onPoll = func(h *fakeHost, _ string) {
if h.app().digest != digestB {
return
}
once.Do(func() {
if _, err := manager.Abort("app"); err != nil {
t.Error(err)
}
})
}
operation := submit(t, manager, digestB, 1)
manager.Drain()
operation = reload(t, manager, operation)
if operation.Phase != store.PhaseRolledBack || !strings.Contains(operation.Reason, "interrupted") {
t.Fatalf("operation = %+v", operation)
}
if !host.app().running || host.app().digest != digestA {
t.Fatalf("previous revision not restored after abort: %+v", host.app())
}
}
func TestRollbackWithoutActiveRolloutIsRejected(t *testing.T) {
manager, _, _ := newHarness(t)
if _, err := manager.Abort("app"); !errors.Is(err, ErrConflict) {
t.Fatalf("abort without rollout = %v", err)
}
}
func TestRecoveryResumesInterruptedOperation(t *testing.T) {
manager, host, clock := newHarness(t)
interrupted := interruptedOperation(t, manager, digestA, clock.Now().Add(time.Minute), nil)
if err := manager.Recover(); err != nil {
t.Fatal(err)
}
manager.Drain()
operation := reload(t, manager, interrupted)
if operation.Phase != store.PhasePromoted {
t.Fatalf("operation = %+v", operation)
}
if !host.app().running || host.app().digest != digestA {
t.Fatalf("host = %+v", host.app())
}
}
func TestRecoveryTerminalizesCommittedPromotionWithoutReplayingHost(t *testing.T) {
manager, host, clock := newHarness(t)
previous := store.Revision{Digest: digestA, Reference: service(t, manager).Reference(digestA), Unit: service(t, manager).RenderUnit(service(t, manager).Reference(digestA))}
candidate := store.Revision{Digest: digestB, Reference: service(t, manager).Reference(digestB), Unit: service(t, manager).RenderUnit(service(t, manager).Reference(digestB))}
operation := store.Operation{
ID: store.NewOperationID(clock.Now()),
Service: "app",
Phase: store.PhasePromoting,
ExpectedGeneration: 1,
Candidate: candidate,
Rollback: &previous,
StartedAt: clock.Now(),
Deadline: clock.Now().Add(time.Minute),
}
if err := manager.Store.WriteOperation(operation); err != nil {
t.Fatal(err)
}
// This is the durable state left by a crash after promotion committed but before terminalization.
if err := manager.Store.WriteState("app", store.State{Generation: 2, OperationID: operation.ID, Current: &candidate, Rollback: &previous}); err != nil {
t.Fatal(err)
}
if err := manager.Recover(); err != nil {
t.Fatal(err)
}
manager.Drain()
recovered := reload(t, manager, operation)
if recovered.Phase != store.PhasePromoted || recovered.Generation != 2 || recovered.EndedAt == nil {
t.Fatalf("operation = %+v", recovered)
}
state, err := manager.Store.ReadState("app")
if err != nil || state.Generation != 2 || state.OperationID != operation.ID || !sameRevision(state.Current, &candidate) || !sameRevision(state.Rollback, &previous) {
t.Fatalf("state = %+v err = %v", state, err)
}
if host.stops != 0 || host.starts != 0 {
t.Fatalf("recovery replayed host mutation: stops=%d starts=%d", host.stops, host.starts)
}
}
func TestRecoveryRejectsUnrelatedLaterPromotion(t *testing.T) {
manager, host, clock := newHarness(t)
candidate := store.Revision{Digest: digestA, Reference: service(t, manager).Reference(digestA), Unit: service(t, manager).RenderUnit(service(t, manager).Reference(digestA))}
operation := store.Operation{
ID: store.NewOperationID(clock.Now()),
Service: "app",
Phase: store.PhasePromoting,
ExpectedGeneration: 0,
Candidate: candidate,
StartedAt: clock.Now(),
Deadline: clock.Now().Add(time.Minute),
}
if err := manager.Store.WriteOperation(operation); err != nil {
t.Fatal(err)
}
later := store.Revision{Digest: digestB, Reference: service(t, manager).Reference(digestB), Unit: service(t, manager).RenderUnit(service(t, manager).Reference(digestB))}
if err := manager.Store.WriteState("app", store.State{Generation: 2, OperationID: "later-operation", Current: &later, Rollback: &candidate}); err != nil {
t.Fatal(err)
}
if err := manager.Recover(); err != nil {
t.Fatal(err)
}
manager.Drain()
recovered := reload(t, manager, operation)
if recovered.Phase != store.PhaseFailed || !strings.Contains(recovered.Reason, "does not match") {
t.Fatalf("operation = %+v", recovered)
}
state, _ := manager.Store.ReadState("app")
if state.Generation != 2 || !sameRevision(state.Current, &later) {
t.Fatalf("state was changed: %+v", state)
}
if host.stops != 0 || host.starts != 0 {
t.Fatalf("recovery replayed host mutation: stops=%d starts=%d", host.stops, host.starts)
}
}
func TestRecoveryDoesNotExtendAnExpiredDeadline(t *testing.T) {
manager, host, clock := newHarness(t)
submit(t, manager, digestA, 0)
manager.Drain()
previous := currentRevision(t, manager)
interrupted := interruptedOperation(t, manager, digestB, clock.Now().Add(-time.Second), previous)
if err := manager.Recover(); err != nil {
t.Fatal(err)
}
manager.Drain()
operation := reload(t, manager, interrupted)
if operation.Phase != store.PhaseRolledBack || !strings.Contains(operation.Reason, "deadline") {
t.Fatalf("operation = %+v", operation)
}
if host.app().digest != digestA {
t.Fatalf("expired candidate kept running: %+v", host.app())
}
}
func TestRecoveryFailsOlderInterruptedOperations(t *testing.T) {
manager, _, clock := newHarness(t)
older := interruptedOperation(t, manager, digestA, clock.Now().Add(time.Minute), nil)
_ = clock.Sleep(context.Background(), time.Second)
newer := interruptedOperation(t, manager, digestB, clock.Now().Add(time.Minute), nil)
if err := manager.Recover(); err != nil {
t.Fatal(err)
}
manager.Drain()
if reloaded := reload(t, manager, older); reloaded.Phase != store.PhaseFailed || reloaded.EndedAt == nil {
t.Fatalf("older operation = %+v", reloaded)
}
if reloaded := reload(t, manager, newer); reloaded.Phase != store.PhasePromoted {
t.Fatalf("newer operation = %+v", reloaded)
}
}
func TestEnrolledServicesDeployIndependently(t *testing.T) {
manager, host, _ := newHarness(t)
// The second service crashes itself out of probation while the first one stays healthy.
host.onPoll = func(h *fakeHost, unit string) {
if strings.HasPrefix(unit, "other") {
h.get(unit).restarts++
}
}
promoted := submitFor(t, manager, "app", digestA, 0)
failing := submitFor(t, manager, "other", digestB, 0)
manager.Drain()
if reloaded := reload(t, manager, promoted); reloaded.Phase != store.PhasePromoted {
t.Fatalf("app operation = %+v", reloaded)
}
if reloaded := reload(t, manager, failing); reloaded.Phase != store.PhaseFailed || !strings.Contains(reloaded.Reason, "restart budget") {
t.Fatalf("other operation = %+v", reloaded)
}
appState, _ := manager.Store.ReadState("app")
otherState, _ := manager.Store.ReadState("other")
if appState.Generation != 1 || appState.Current == nil || appState.Current.Digest != digestA {
t.Fatalf("app state = %+v", appState)
}
if otherState.Generation != 0 || otherState.Current != nil {
t.Fatalf("other state = %+v", otherState)
}
if !host.app().running || host.get("other").running {
t.Fatalf("workloads = %+v %+v", host.app(), host.get("other"))
}
if !strings.Contains(host.app().installed, "registry.invalid/app@") || !strings.Contains(host.get("other").installed, "registry.invalid/other@") {
t.Fatal("services shared a unit or repository")
}
}
func TestServiceTokenCannotReachAnotherService(t *testing.T) {
manager, _, _ := newHarness(t)
server := httptest.NewServer(manager.Handler())
defer server.Close()
appToken := issueToken(t, manager, "app")
otherToken := issueToken(t, manager, "other")
if response := get(t, server.URL+"/v1/services/other", appToken); response.StatusCode != http.StatusUnauthorized {
t.Fatalf("app token read other service: %d", response.StatusCode)
}
if response := get(t, server.URL+"/v1/services/other", otherToken); response.StatusCode != http.StatusOK {
t.Fatalf("own token rejected: %d", response.StatusCode)
}
}
func TestTokenAdministrationIsLocalOnlyAndRevocationIsImmediate(t *testing.T) {
manager, _, _ := newHarness(t)
remote := httptest.NewServer(manager.Handler())
defer remote.Close()
local := httptest.NewServer(manager.LocalHandler())
defer local.Close()
if response := call(t, remote.URL+"/v1/services/app/tokens", issueToken(t, manager, "app"), `{"label":"sneaky"}`); response.StatusCode != http.StatusNotFound {
t.Fatalf("token administration reachable over TCP: %d", response.StatusCode)
}
response := call(t, local.URL+"/v1/services/app/tokens", "", `{"label":"ci"}`)
if response.StatusCode != http.StatusCreated {
t.Fatalf("local issue = %d", response.StatusCode)
}
var issued struct {
ID string `json:"id"`
Token string `json:"token"`
}
if err := json.NewDecoder(response.Body).Decode(&issued); err != nil {
t.Fatal(err)
}
if issued.Token == "" || !strings.HasPrefix(issued.Token, issued.ID+".") {
t.Fatalf("issued = %+v", issued)
}
if response := get(t, remote.URL+"/v1/services/app", issued.Token); response.StatusCode != http.StatusOK {
t.Fatalf("issued token rejected: %d", response.StatusCode)
}
remove(t, local.URL+"/v1/services/app/tokens/"+issued.ID)
if response := get(t, remote.URL+"/v1/services/app", issued.Token); response.StatusCode != http.StatusUnauthorized {
t.Fatalf("revoked token still accepted: %d", response.StatusCode)
}
// Local access needs no credential at all.
if response := get(t, local.URL+"/v1/services/app", ""); response.StatusCode != http.StatusOK {
t.Fatalf("local read = %d", response.StatusCode)
}
}
func TestReadOnlyDashboardAndAssetsDoNotRequireServiceToken(t *testing.T) {
manager, _, _ := newHarness(t)
manager.ThemeCSS = []byte(":root { --test-theme: yes; }")
server := httptest.NewServer(manager.Handler())
defer server.Close()
response, err := http.Get(server.URL + "/?service=app")
if err != nil {
t.Fatal(err)
}
body, _ := io.ReadAll(response.Body)
response.Body.Close()
text := string(body)
for _, expected := range []string{"Toad deployment control", "Configuration", "Systemd", "Operations", "Readiness status", "Raw Quadlets and environment values are excluded"} {
if !strings.Contains(text, expected) {
t.Fatalf("dashboard missing %q", expected)
}
}
for _, excluded := range []string{"readiness.internal", "/private/ready", "token=not-for-dashboard"} {
if strings.Contains(text, excluded) {
t.Fatalf("dashboard exposed readiness detail %q", excluded)
}
}
if response.StatusCode != http.StatusOK {
t.Fatalf("dashboard status = %d", response.StatusCode)
}
theme, err := http.Get(server.URL + "/assets/theme.css")
if err != nil {
t.Fatal(err)
}
themeBody, _ := io.ReadAll(theme.Body)
theme.Body.Close()
if string(themeBody) != string(manager.ThemeCSS) {
t.Fatalf("theme = %q", themeBody)
}
if response := get(t, server.URL+"/v1/services/app", ""); response.StatusCode != http.StatusUnauthorized {
t.Fatalf("API without token = %d", response.StatusCode)
}
}
func TestDeploymentAssetsKeepDashboardBehindIngress(t *testing.T) {
quadlet, err := os.ReadFile(filepath.Join("..", "..", "deploy", "toad.container"))
if err != nil {
t.Fatal(err)
}
if strings.Contains(string(quadlet), "PublishPort=") || !strings.Contains(string(quadlet), "Network=toad.network") {
t.Fatalf("toad Quadlet exposes host TCP or misses private network:\n%s", quadlet)
}
gateway, err := os.ReadFile(filepath.Join("..", "..", "deploy", "gateway.Caddyfile"))
if err != nil {
t.Fatal(err)
}
for _, expected := range []string{"bind unix//run/toad-ingress/http.sock|0660", "reverse_proxy toad:8090", "path / /assets/*"} {
if !strings.Contains(string(gateway), expected) {
t.Fatalf("gateway missing %q", expected)
}
}
}
func TestTCPAPIRoutesRequireServiceScopedTokens(t *testing.T) {
manager, _, _ := newHarness(t)
server := httptest.NewServer(manager.Handler())
defer server.Close()
otherToken := issueToken(t, manager, "other")
paths := []struct {
method string
path string
body string
}{
{http.MethodPost, "/v1/services/app/deployments", `{"digest":"` + digestA + `","expectedGeneration":0,"idempotencyKey":"unauthorized"}`},
{http.MethodPost, "/v1/services/app/rollback", ""},
{http.MethodGet, "/v1/services/app", ""},
{http.MethodGet, "/v1/services/app/operations/unknown", ""},
{http.MethodGet, "/v1/services/app/operations/unknown/log", ""},
}
for _, route := range paths {
response := send(t, route.method, server.URL+route.path, otherToken, route.body)
if response.StatusCode != http.StatusUnauthorized {
t.Fatalf("%s %s with another service token = %d", route.method, route.path, response.StatusCode)
}
}
}
func TestHTTPAuthorizationAndReporting(t *testing.T) {
manager, _, _ := newHarness(t)
server := httptest.NewServer(manager.Handler())
defer server.Close()
token := issueToken(t, manager, "app")
body := `{"digest":"` + digestA + `","expectedGeneration":0,"idempotencyKey":"http-1"}`
response := call(t, server.URL+"/v1/services/app/deployments", "0123456789abcdef.wrong", body)
if response.StatusCode != http.StatusUnauthorized {
t.Fatalf("wrong token accepted: %d", response.StatusCode)
}
response = call(t, server.URL+"/v1/services/absent/deployments", token, body)
if response.StatusCode != http.StatusUnauthorized {
t.Fatalf("unknown service is distinguishable: %d", response.StatusCode)
}
response = call(t, server.URL+"/v1/services/app/deployments", token, body)
if response.StatusCode != http.StatusAccepted {
t.Fatalf("deploy = %d", response.StatusCode)
}
var accepted store.Operation
if err := json.NewDecoder(response.Body).Decode(&accepted); err != nil {
t.Fatal(err)
}
manager.Drain()
logResponse := get(t, server.URL+"/v1/services/app/operations/"+accepted.ID+"/log", token)
if logResponse.StatusCode != http.StatusOK {
t.Fatalf("log = %d", logResponse.StatusCode)
}
stateResponse := get(t, server.URL+"/v1/services/app", token)
var reported map[string]any
if err := json.NewDecoder(stateResponse.Body).Decode(&reported); err != nil {
t.Fatal(err)
}
if reported["generation"].(float64) != 1 {
t.Fatalf("reported = %v", reported)
}
}
// harness
func newHarness(t *testing.T) (*Manager, *fakeHost, *fakeClock) {
t.Helper()
root := t.TempDir()
quadlets := filepath.Join(root, "quadlets")
if err := os.MkdirAll(quadlets, 0o755); err != nil {
t.Fatal(err)
}
template := filepath.Join(root, "container.template")
write(t, template, "[Container]\nImage=${image}\n")
enrollmentFile := filepath.Join(root, "enrollment.json")
write(t, enrollmentFile, fmt.Sprintf(`[{
"name": "app",
"unit": "app",
"imageRepository": "registry.invalid/app",
"unitTemplateFile": %q,
"containerName": "app",
"readinessURL": "http://readiness.internal:8181/private/ready?token=not-for-dashboard",
"probation": {"deadline": "60s", "stableFor": "10s", "restartBudget": 1}
}, {
"name": "other",
"unit": "other",
"imageRepository": "registry.invalid/other",
"unitTemplateFile": %q,
"containerName": "other",
"readinessURL": "http://127.0.0.1:9/ready-other",
"probation": {"deadline": "60s", "stableFor": "10s", "restartBudget": 1}
}]`, template, template))
enrollment, err := enroll.Load(enrollmentFile)
if err != nil {
t.Fatal(err)
}
clock := &fakeClock{now: time.Unix(1700000000, 0).UTC()}
host := &fakeHost{}
data := store.Store{Root: filepath.Join(root, "data")}
manager := &Manager{
Store: data,
Enrollment: enrollment,
Credentials: credential.New(filepath.Join(root, "data")),
Now: clock.Now,
Engine: rollout.Engine{
Store: data,
Host: host,
Poll: 2 * time.Second,
StopTimeout: 30 * time.Second,
Now: clock.Now,
Sleep: clock.Sleep,
},
}
return manager, host, clock
}
func issueToken(t *testing.T, manager *Manager, service string) string {
t.Helper()
value, _, err := manager.Credentials.Issue(service, "test")
if err != nil {
t.Fatal(err)
}
return value
}
func service(t *testing.T, manager *Manager) enroll.Service {
t.Helper()
return serviceNamed(t, manager, "app")
}
func serviceNamed(t *testing.T, manager *Manager, name string) enroll.Service {
t.Helper()
found, ok := manager.Enrollment.Lookup(name)
if !ok {
t.Fatalf("service %s not enrolled", name)
}
return found
}
func submit(t *testing.T, manager *Manager, digest string, generation uint64) store.Operation {
t.Helper()
return submitFor(t, manager, "app", digest, generation)
}
func submitFor(t *testing.T, manager *Manager, name, digest string, generation uint64) store.Operation {
t.Helper()
key := fmt.Sprintf("deploy-%s-%d", name, generation+1)
operation, err := manager.Submit(serviceNamed(t, manager, name), DeployRequest{Digest: digest, ExpectedGeneration: generation, IdempotencyKey: key})
if err != nil {
t.Fatal(err)
}
return operation
}
func reload(t *testing.T, manager *Manager, operation store.Operation) store.Operation {
t.Helper()
reloaded, err := manager.Store.ReadOperation(operation.Service, operation.ID)
if err != nil {
t.Fatal(err)
}
return reloaded
}
// interruptedOperation writes the record a crashed rollout would leave behind.
func interruptedOperation(t *testing.T, manager *Manager, digest string, deadline time.Time, rollback *store.Revision) store.Operation {
t.Helper()
found := service(t, manager)
reference := found.Reference(digest)
operation := store.Operation{
ID: store.NewOperationID(manager.Now()),
Service: found.Name,
Phase: store.PhaseProbation,
ExpectedGeneration: currentGeneration(t, manager),
IdempotencyKey: "interrupted-" + digest,
Candidate: store.Revision{Digest: digest, Reference: reference, Unit: found.RenderUnit(reference)},
Rollback: rollback,
StartedAt: manager.Now(),
Deadline: deadline,
}
if err := manager.Store.WriteOperation(operation); err != nil {
t.Fatal(err)
}
return operation
}
func currentGeneration(t *testing.T, manager *Manager) uint64 {
t.Helper()
state, err := manager.Store.ReadState("app")
if err != nil {
t.Fatal(err)
}
return state.Generation
}
func currentRevision(t *testing.T, manager *Manager) *store.Revision {
t.Helper()
state, err := manager.Store.ReadState("app")
if err != nil || state.Current == nil {
t.Fatalf("state = %+v err = %v", state, err)
}
return state.Current
}
func write(t *testing.T, path, content string) {
t.Helper()
if err := os.WriteFile(path, []byte(content), 0o600); err != nil {
t.Fatal(err)
}
}
func call(t *testing.T, url, token, body string) *http.Response {
t.Helper()
return send(t, http.MethodPost, url, token, body)
}
func get(t *testing.T, url, token string) *http.Response {
t.Helper()
return send(t, http.MethodGet, url, token, "")
}
func remove(t *testing.T, url string) {
t.Helper()
if response := send(t, http.MethodDelete, url, "", ""); response.StatusCode != http.StatusOK {
t.Fatalf("delete %s = %d", url, response.StatusCode)
}
}
func send(t *testing.T, method, url, token, body string) *http.Response {
t.Helper()
var reader io.Reader
if body != "" {
reader = strings.NewReader(body)
}
request, err := http.NewRequest(method, url, reader)
if err != nil {
t.Fatal(err)
}
if token != "" {
request.Header.Set("Authorization", "Bearer "+token)
}
response, err := http.DefaultClient.Do(request)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { response.Body.Close() })
return response
}
type fakeClock struct {
mu sync.Mutex
now time.Time
}
func (c *fakeClock) Now() time.Time {
c.mu.Lock()
defer c.mu.Unlock()
return c.now
}
func (c *fakeClock) Sleep(ctx context.Context, duration time.Duration) error {
if err := ctx.Err(); err != nil {
return err
}
c.mu.Lock()
defer c.mu.Unlock()
c.now = c.now.Add(duration)
return nil
}
type workload struct {
installed string
running bool
digest string
restarts int
}
type fakeHost struct {
mu sync.Mutex
workloads map[string]*workload
stopRefused bool
stops int
starts int
reportDigest string
probe func(target string) error
pull func(reference string) (string, error)
onPoll func(*fakeHost, string)
}
// get resolves app.container, app.service, and app to the same workload.
func (f *fakeHost) get(name string) *workload {
f.mu.Lock()
defer f.mu.Unlock()
return f.locked(name)
}
func (f *fakeHost) locked(name string) *workload {
base, _, _ := strings.Cut(name, ".")
if f.workloads == nil {
f.workloads = map[string]*workload{}
}
if f.workloads[base] == nil {
f.workloads[base] = &workload{}
}
return f.workloads[base]
}
func (f *fakeHost) app() *workload { return f.get("app") }
func (f *fakeHost) Pull(_ context.Context, reference string) (string, error) {
f.mu.Lock()
pull := f.pull
f.mu.Unlock()
if pull != nil {
return pull(reference)
}
return digestOf(reference), nil
}
func (f *fakeHost) InstallUnit(fileName, content string) error {
f.mu.Lock()
defer f.mu.Unlock()
f.locked(fileName).installed = content
return nil
}
func (f *fakeHost) DaemonReload(context.Context) error { return nil }
func (f *fakeHost) Stop(_ context.Context, unit string) error {
f.mu.Lock()
defer f.mu.Unlock()
f.stops++
if !f.stopRefused {
f.locked(unit).running = false
}
return nil
}
func (f *fakeHost) Start(_ context.Context, unit string) error {
f.mu.Lock()
defer f.mu.Unlock()
f.starts++
started := f.locked(unit)
started.running = true
started.digest = digestOf(started.installed)
return nil
}
func (f *fakeHost) Container(_ context.Context, name string) (rollout.ContainerState, error) {
f.mu.Lock()
defer f.mu.Unlock()
observed := f.locked(name)
digest := observed.digest
if f.reportDigest != "" {
digest = f.reportDigest
}
return rollout.ContainerState{Running: observed.running, ImageDigest: digest}, nil
}
func (f *fakeHost) RestartCount(_ context.Context, unit string) (int, error) {
f.mu.Lock()
poll := f.onPoll
f.mu.Unlock()
if poll != nil {
poll(f, unit)
}
f.mu.Lock()
defer f.mu.Unlock()
return f.locked(unit).restarts, nil
}
func (f *fakeHost) Unit(_ context.Context, unit string) (rollout.UnitState, error) {
f.mu.Lock()
defer f.mu.Unlock()
return rollout.UnitState{LoadState: "loaded", ActiveState: "active", Result: "success", Restarts: f.locked(unit).restarts}, nil
}
func (f *fakeHost) Probe(_ context.Context, target string) error {
f.mu.Lock()
probe := f.probe
f.mu.Unlock()
if probe != nil {
return probe(target)
}
return nil
}
func digestOf(text string) string {
_, digest, found := strings.Cut(text, "@")
if !found {
return ""
}
return strings.TrimSpace(strings.SplitN(digest, "\n", 2)[0])
}