Codex Local
bf1233d5df
ci / verify (push) Canceled after 0s
Add customer and operator login, BLE provisioning, device fleet controls, call memory, memoir generation, and refreshed family and maintenance portals. Include API contracts, migrations, firmware sources, tests, assets, and deployment documentation.
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)
|
|
}
|
|
}
|