215 lines
9.3 KiB
Go
215 lines
9.3 KiB
Go
|
package devicefleet
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"crypto/sha256"
|
||
|
|
"encoding/hex"
|
||
|
|
"encoding/json"
|
||
|
|
"errors"
|
||
|
|
"fmt"
|
||
|
|
"os"
|
||
|
|
"strings"
|
||
|
|
"testing"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"github.com/google/uuid"
|
||
|
|
"github.com/jackc/pgx/v5"
|
||
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
||
|
|
)
|
||
|
|
|
||
|
|
type otaDeadlineFixture struct {
|
||
|
|
t *testing.T
|
||
|
|
ctx context.Context
|
||
|
|
service *Service
|
||
|
|
release Release
|
||
|
|
device string
|
||
|
|
}
|
||
|
|
|
||
|
|
func newOTADeadlineFixture(t *testing.T) *otaDeadlineFixture {
|
||
|
|
t.Helper()
|
||
|
|
dsn := os.Getenv("TEST_DATABASE_URL")
|
||
|
|
if dsn == "" {
|
||
|
|
t.Skip("isolated integration database required")
|
||
|
|
}
|
||
|
|
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
||
|
|
t.Cleanup(cancel)
|
||
|
|
pool, err := pgxpool.New(ctx, dsn)
|
||
|
|
must(t, err)
|
||
|
|
t.Cleanup(pool.Close)
|
||
|
|
const device = "10000000-0000-0000-0000-000000000501"
|
||
|
|
const tenant = "10000000-0000-0000-0000-000000000001"
|
||
|
|
_, err = pool.Exec(ctx, `UPDATE device SET factory_device_id=$2 WHERE id=$1`, device, "KP01-020000001235")
|
||
|
|
must(t, err)
|
||
|
|
store := &fakeStore{objects: map[string][]byte{}}
|
||
|
|
s := &Service{Pool: pool, Store: store, DownloadBaseURL: "https://fleet-fixture.invalid", Publish: func(string, []byte) error { return nil }}
|
||
|
|
content := []byte("synthetic deadline fixture firmware")
|
||
|
|
digest := sha256.Sum256(content)
|
||
|
|
r, err := s.Prepare(ctx, tenant, ReleaseInput{Model: "simulated-terminal", Version: "deadline-" + uuid.NewString(), Size: int64(len(content)), SHA256: hex.EncodeToString(digest[:])})
|
||
|
|
must(t, err)
|
||
|
|
t.Cleanup(func() {
|
||
|
|
_, _ = pool.Exec(context.Background(), `DELETE FROM fleet_ota WHERE release_id=$1`, r.ID)
|
||
|
|
_, _ = pool.Exec(context.Background(), `DELETE FROM fleet_firmware WHERE id=$1`, r.ID)
|
||
|
|
_, _ = pool.Exec(context.Background(), `DELETE FROM fleet_presence WHERE device_id=$1`, device)
|
||
|
|
_, _ = pool.Exec(context.Background(), `UPDATE device SET factory_device_id=NULL WHERE id=$1`, device)
|
||
|
|
})
|
||
|
|
store.objects[r.Object] = content
|
||
|
|
r, err = s.Finalize(ctx, tenant, r.ID)
|
||
|
|
must(t, err)
|
||
|
|
_, err = pool.Exec(ctx, `INSERT INTO fleet_presence(device_id,password_hash,connected,last_seen_at,ota_supported) VALUES($1,$2,true,now(),true) ON CONFLICT(device_id) DO UPDATE SET connected=true,last_seen_at=now(),ota_supported=true,version_code=0`, device, make([]byte, 32))
|
||
|
|
must(t, err)
|
||
|
|
return &otaDeadlineFixture{t: t, ctx: ctx, service: s, release: r, device: device}
|
||
|
|
}
|
||
|
|
|
||
|
|
func (f *otaDeadlineFixture) task(status string, age time.Duration) (string, string) {
|
||
|
|
f.t.Helper()
|
||
|
|
id := uuid.NewString()
|
||
|
|
_, err := f.service.Pool.Exec(f.ctx, `INSERT INTO fleet_ota(id,device_id,release_id,request_id,status,created_at) VALUES($1,$2,$3,$4,$5,now()-$6::interval)`, id, f.device, f.release.ID, uuid.NewString(), status, fmt.Sprintf("%f seconds", age.Seconds()))
|
||
|
|
must(f.t, err)
|
||
|
|
token, hash, _, expires, err := newDownloadTicket(f.service.DownloadBaseURL)
|
||
|
|
must(f.t, err)
|
||
|
|
_, err = f.service.Pool.Exec(f.ctx, `INSERT INTO fleet_firmware_download_ticket(token_hash,release_id,ota_id,expires_at) VALUES($1,$2,$3,$4)`, hash, f.release.ID, id, expires)
|
||
|
|
must(f.t, err)
|
||
|
|
return id, token
|
||
|
|
}
|
||
|
|
|
||
|
|
func (f *otaDeadlineFixture) expectState(task, status string) {
|
||
|
|
f.t.Helper()
|
||
|
|
state, err := f.service.State(f.ctx, f.device)
|
||
|
|
must(f.t, err)
|
||
|
|
if state.Task == nil || state.Task.ID != task || state.Task.Status != status {
|
||
|
|
f.t.Fatalf("read model expected %s/%s, got %+v", task, status, state.Task)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func expectOTAError(t *testing.T, err error, code string) {
|
||
|
|
t.Helper()
|
||
|
|
var fleetErr *Error
|
||
|
|
if !errors.As(err, &fleetErr) || fleetErr.Code != code {
|
||
|
|
t.Fatalf("expected %s, got %v", code, err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func TestFleetOTAAcceptedBeforeTwentyMinuteBoundary(t *testing.T) {
|
||
|
|
f := newOTADeadlineFixture(t)
|
||
|
|
id, _ := f.task("accepted", 20*time.Minute-time.Second)
|
||
|
|
manifest, err := f.service.Manifest(f.ctx, f.device, id)
|
||
|
|
must(t, err)
|
||
|
|
token := strings.TrimPrefix(manifest["download_url"].(string), "https://fleet-fixture.invalid/f/")
|
||
|
|
_, err = f.service.ResolveDownload(f.ctx, token)
|
||
|
|
must(t, err)
|
||
|
|
must(t, f.service.Expire(f.ctx))
|
||
|
|
f.expectState(id, "accepted")
|
||
|
|
// Retrying acceptance is idempotent and must not reset created_at or the budget.
|
||
|
|
must(t, f.service.Report(f.ctx, f.device, Report{TaskID: id, Status: "accepted"}))
|
||
|
|
must(t, f.service.Report(f.ctx, f.device, Report{TaskID: id, Status: "success", FirmwareVersion: f.release.Version, Code: f.release.Code}))
|
||
|
|
f.expectState(id, "success")
|
||
|
|
state, err := f.service.State(f.ctx, f.device)
|
||
|
|
must(t, err)
|
||
|
|
if state.CurrentCode != f.release.Code || state.CurrentVersion != f.release.Version {
|
||
|
|
t.Fatal("success did not update read model version")
|
||
|
|
}
|
||
|
|
_, err = f.service.ResolveDownload(f.ctx, token)
|
||
|
|
if !errors.Is(err, ErrDownloadTicketUnavailable) {
|
||
|
|
t.Fatalf("terminal ticket remained usable: %v", err)
|
||
|
|
}
|
||
|
|
_, err = f.service.Manifest(f.ctx, f.device, id)
|
||
|
|
expectOTAError(t, err, "OTA_TASK_UNAVAILABLE")
|
||
|
|
must(t, f.service.Report(f.ctx, f.device, Report{TaskID: id, Status: "success", FirmwareVersion: f.release.Version, Code: f.release.Code}))
|
||
|
|
expectOTAError(t, f.service.Report(f.ctx, f.device, Report{TaskID: id, Status: "failed"}), "OTA_FINISHED")
|
||
|
|
}
|
||
|
|
|
||
|
|
func TestFleetOTAAcceptedAfterTwentyMinuteBoundary(t *testing.T) {
|
||
|
|
for _, result := range []string{"success", "failed"} {
|
||
|
|
t.Run(result, func(t *testing.T) {
|
||
|
|
f := newOTADeadlineFixture(t)
|
||
|
|
id, token := f.task("accepted", 20*time.Minute+time.Second)
|
||
|
|
_, err := f.service.Manifest(f.ctx, f.device, id)
|
||
|
|
expectOTAError(t, err, "OTA_TASK_UNAVAILABLE")
|
||
|
|
// The periodic sweeper has not run: access still rejects an aged active row.
|
||
|
|
_, err = f.service.ResolveDownload(f.ctx, token)
|
||
|
|
if !errors.Is(err, ErrDownloadTicketUnavailable) {
|
||
|
|
t.Fatalf("expired active row authorized download: %v", err)
|
||
|
|
}
|
||
|
|
f.expectState(id, "accepted")
|
||
|
|
must(t, f.service.Expire(f.ctx))
|
||
|
|
f.expectState(id, "timeout")
|
||
|
|
var revoked bool
|
||
|
|
must(t, f.service.Pool.QueryRow(f.ctx, `SELECT revoked_at IS NOT NULL FROM fleet_firmware_download_ticket WHERE ota_id=$1`, id).Scan(&revoked))
|
||
|
|
if !revoked {
|
||
|
|
t.Fatal("expiry failed to revoke ticket")
|
||
|
|
}
|
||
|
|
expectOTAError(t, f.service.Report(f.ctx, f.device, Report{TaskID: id, Status: "accepted"}), "OTA_EXPIRED")
|
||
|
|
expectOTAError(t, f.service.Report(f.ctx, f.device, Report{TaskID: id, Status: "success", FirmwareVersion: "wrong", Code: f.release.Code}), "VERSION_MISMATCH")
|
||
|
|
// Preserve existing authenticated late-result correction after timeout.
|
||
|
|
must(t, f.service.Report(f.ctx, f.device, Report{TaskID: id, Status: result, FirmwareVersion: f.release.Version, Code: f.release.Code}))
|
||
|
|
f.expectState(id, result)
|
||
|
|
_, err = f.service.ResolveDownload(f.ctx, token)
|
||
|
|
if !errors.Is(err, ErrDownloadTicketUnavailable) {
|
||
|
|
t.Fatalf("late terminal result reenabled ticket: %v", err)
|
||
|
|
}
|
||
|
|
_, err = f.service.Manifest(f.ctx, f.device, id)
|
||
|
|
expectOTAError(t, err, "OTA_TASK_UNAVAILABLE")
|
||
|
|
})
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func TestFleetOTAPendingDeadlineRemainsThirtySeconds(t *testing.T) {
|
||
|
|
for _, age := range []time.Duration{29 * time.Second, 31 * time.Second} {
|
||
|
|
t.Run(age.String(), func(t *testing.T) {
|
||
|
|
f := newOTADeadlineFixture(t)
|
||
|
|
id, token := f.task("pending", age)
|
||
|
|
_, manifestErr := f.service.Manifest(f.ctx, f.device, id)
|
||
|
|
// Manifest rotates the ticket only for the unexpired task.
|
||
|
|
if age < 30*time.Second {
|
||
|
|
must(t, manifestErr)
|
||
|
|
must(t, f.service.Expire(f.ctx))
|
||
|
|
f.expectState(id, "pending")
|
||
|
|
must(t, f.service.Report(f.ctx, f.device, Report{TaskID: id, Status: "accepted"}))
|
||
|
|
f.expectState(id, "accepted")
|
||
|
|
} else {
|
||
|
|
expectOTAError(t, manifestErr, "OTA_TASK_UNAVAILABLE")
|
||
|
|
_, err := f.service.ResolveDownload(f.ctx, token)
|
||
|
|
if !errors.Is(err, ErrDownloadTicketUnavailable) {
|
||
|
|
t.Fatalf("expired pending task authorized download: %v", err)
|
||
|
|
}
|
||
|
|
expectOTAError(t, f.service.Report(f.ctx, f.device, Report{TaskID: id, Status: "accepted"}), "OTA_EXPIRED")
|
||
|
|
must(t, f.service.Expire(f.ctx))
|
||
|
|
f.expectState(id, "timeout")
|
||
|
|
}
|
||
|
|
})
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func TestFleetOTAWakeupAndDownloadTicketTTLsRemainUnchanged(t *testing.T) {
|
||
|
|
f := newOTADeadlineFixture(t)
|
||
|
|
var command map[string]any
|
||
|
|
f.service.Publish = func(_ string, payload []byte) error { return json.Unmarshal(payload, &command) }
|
||
|
|
task, err := f.service.Start(f.ctx, uuid.NewString(), f.release.Code, func(tx pgx.Tx) (string, error) {
|
||
|
|
var device string
|
||
|
|
err := tx.QueryRow(f.ctx, `SELECT device_id::text FROM device_twin WHERE device_id=$1 FOR UPDATE`, f.device).Scan(&device)
|
||
|
|
return device, err
|
||
|
|
})
|
||
|
|
must(t, err)
|
||
|
|
if int64(command["expires_at_ms"].(float64)) != task.Created.Add(30*time.Second).UnixMilli() {
|
||
|
|
t.Fatal("MQTT wakeup TTL changed")
|
||
|
|
}
|
||
|
|
var ttlSeconds float64
|
||
|
|
must(t, f.service.Pool.QueryRow(f.ctx, `SELECT EXTRACT(EPOCH FROM (expires_at-created_at)) FROM fleet_firmware_download_ticket WHERE ota_id=$1`, task.ID).Scan(&ttlSeconds))
|
||
|
|
if ttlSeconds < 1799 || ttlSeconds > 1801 {
|
||
|
|
t.Fatalf("OTA ticket TTL changed: %.3fs", ttlSeconds)
|
||
|
|
}
|
||
|
|
admin, err := f.service.Download(f.ctx, "10000000-0000-0000-0000-000000000001", f.release.ID)
|
||
|
|
must(t, err)
|
||
|
|
token := strings.TrimPrefix(admin.DownloadURL, "https://fleet-fixture.invalid/f/")
|
||
|
|
_, err = f.service.ResolveDownload(f.ctx, token)
|
||
|
|
must(t, err)
|
||
|
|
digest, err := downloadTokenHash(token)
|
||
|
|
must(t, err)
|
||
|
|
_, err = f.service.Pool.Exec(f.ctx, `UPDATE fleet_firmware_download_ticket SET expires_at=now()-interval '1 second' WHERE token_hash=$1`, digest)
|
||
|
|
must(t, err)
|
||
|
|
_, err = f.service.ResolveDownload(f.ctx, token)
|
||
|
|
if !errors.Is(err, ErrDownloadTicketUnavailable) {
|
||
|
|
t.Fatalf("expired administrator ticket was accepted: %v", err)
|
||
|
|
}
|
||
|
|
}
|