iotlcm/worker-svr/internal/fullsync/contract/validation_test.go

package contract_test

import (
	"errors"
	"math"
	"testing"

	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/contract"
)

func TestRunAccounting(t *testing.T) {
	valid := contract.RunMeta{
		RunID: "run", Status: contract.RunRunning, CurrentRound: 1, MaxRound: 5,
		EnqueuedCount: 10, ActiveCount: 3, InflightCount: 2, NextRoundCount: 2,
		SuccessCount: 4, FailedCount: 1, PendingFailureCount: 1,
	}
	if err := valid.ValidateAccounting(); err != nil {
		t.Fatal(err)
	}
	for name, mutate := range map[string]func(*contract.RunMeta){
		"unbalanced": func(m *contract.RunMeta) { m.EnqueuedCount++ },
		"negative":   func(m *contract.RunMeta) { m.BatchCount = -1 },
		"overflow": func(m *contract.RunMeta) {
			m.EnqueuedCount = math.MaxInt64
			m.SuccessCount = math.MaxInt64
			m.FailedCount = math.MaxInt64
		},
		"inflight double count":        func(m *contract.RunMeta) { m.InflightCount = m.ActiveCount + 1 },
		"pending exceeds history":      func(m *contract.RunMeta) { m.PendingFailureCount = 2 },
		"terminal with work":           func(m *contract.RunMeta) { m.Status = contract.RunCompletedWithFailures; m.ProducerDone = true },
		"unknown status":               func(m *contract.RunMeta) { m.Status = "succeeded" },
		"unknown sync type":            func(m *contract.RunMeta) { m.SyncType = "manual" },
		"action status in archive":     func(m *contract.RunMeta) { m.ArchiveStatus = "completed" },
		"action status in disposition": func(m *contract.RunMeta) { m.ReplayDisposition = "cleanup_pending" },
		"round out of range":           func(m *contract.RunMeta) { m.CurrentRound = 6 },
		"replay without source":        func(m *contract.RunMeta) { m.SyncType = contract.SyncReplay },
		"self replay": func(m *contract.RunMeta) {
			m.SyncType = contract.SyncReplay
			m.SourceRunID = m.RunID
			m.ReplayCycle = 1
		},
		"normal with provenance": func(m *contract.RunMeta) { m.SyncType = contract.SyncNormal; m.SourceRunID = "source" },
	} {
		t.Run(name, func(t *testing.T) {
			meta := valid
			mutate(&meta)
			if err := meta.ValidateAccounting(); !errors.Is(err, contract.ErrInvalidArgument) {
				t.Fatalf("invalid accounting returned %v", err)
			}
		})
	}
	// Cleaned source failures never subtract from the historical failed total.
	completed := contract.RunMeta{RunID: "source", Status: contract.RunCompletedWithFailures, ProducerDone: true, EnqueuedCount: 5, SuccessCount: 3, FailedCount: 2, PendingFailureCount: 0}
	if err := completed.ValidateAccounting(); err != nil {
		t.Fatal(err)
	}
	completed.ProducerDone = false
	if err := completed.ValidateAccounting(); err == nil {
		t.Fatal("accepted terminal run with unfinished producer")
	}
	if err := (contract.RunMeta{RunID: "old", Status: contract.RunRunning, SuccessCount: 42}).Validate(); err != nil {
		t.Fatalf("historical run without new counters rejected: %v", err)
	}
}

func TestRoundStatsValidation(t *testing.T) {
	valid := contract.RoundStats{RunID: "run", Round: 1, ErrorCounts: contract.ErrorCounts{"unrecognized": 1}}
	if err := valid.Validate(); err != nil {
		t.Fatal(err)
	}
	for _, stats := range []contract.RoundStats{
		{RunID: "run", Round: 0}, {RunID: "run", Round: 1, BatchCount: -1},
		{RunID: "run", Round: 1, ErrorCounts: contract.ErrorCounts{contract.ErrorOther: -1}},
	} {
		if err := stats.Validate(); !errors.Is(err, contract.ErrInvalidArgument) {
			t.Fatalf("invalid round stats returned %v", err)
		}
	}
}

func TestPageLimitsAndOpaqueCursor(t *testing.T) {
	for _, limit := range []int{0, 1, 100, 500} {
		request, err := (contract.PageRequest{Limit: limit, Cursor: "opaque+/=:%?next"}).Normalized()
		if err != nil || request.Cursor != "opaque+/=:%?next" || request.Limit < 1 || request.Limit > contract.MaxPageLimit {
			t.Fatalf("unexpected page: %+v, %v", request, err)
		}
		if limit == 0 && request.Limit != 100 {
			t.Fatal("default limit changed")
		}
	}
	for _, limit := range []int{-1, 501, math.MaxInt} {
		if _, err := (contract.PageRequest{Limit: limit}).Normalized(); !errors.Is(err, contract.ErrInvalidArgument) {
			t.Fatalf("limit %d returned %v", limit, err)
		}
	}
	page := contract.FailurePage{Items: []contract.Failure{}, NextCursor: "continue"}
	if got := string(mustJSON(t, page)); got != `{"items":[],"nextCursor":"continue"}` {
		t.Fatalf("empty scan page lost its continuation: %s", got)
	}
}

func TestStatusVocabulary(t *testing.T) {
	for _, status := range []contract.ActionStatus{"accepted", "processing", "cleanup_pending", "receipt_pending", "completed", "needs_attention"} {
		if !status.Valid() || status.Terminal() != (status == "completed" || status == "needs_attention") {
			t.Errorf("invalid action status %q", status)
		}
	}
	for _, status := range []contract.ArchiveStatus{"disabled", "pending", "processing", "succeeded", "failed"} {
		if !status.Valid() {
			t.Errorf("invalid archive status %q", status)
		}
	}
	for _, status := range []contract.ReplayDisposition{"none", "pending", "processing", "completed", "needs_attention"} {
		if !status.Valid() {
			t.Errorf("invalid disposition %q", status)
		}
	}
	if contract.ActionStatus("succeeded").Valid() || contract.ArchiveStatus("completed").Valid() || contract.ReplayDisposition("receipt_pending").Valid() {
		t.Fatal("status domains were conflated")
	}
}