package moves

import (
	"encoding/json"
	"fmt"
	"reflect"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/provision"
)

const (
	offTestHost = "a.go.example.com"
	offTestOp   = "mv_abcdefghijklmnopqrstuvwxyz"
)

// planManifestIDs are the rows of the plan's "The dependency list" table, in
// its order. A row added to or dropped from OfflineManifest without the plan
// (and this list) fails TestOfflineManifestIsCovered.
var planManifestIDs = []string{
	"db", "db-tail", "build-state", "lease", "files", "env", "key-adoption", "config", "env-pin", "release",
	"store", "timers", "next-due", "replication", "row", "routing", "source-retire", "sidecar", "restore-verify",
	"warm-cache", "receipts", "class",
}

// planManifestPairs pins each row's Owner (who proves it where it must end
// up) and Source (who reports its expected value on the source side). A
// reassignment fails here, so it is a decision and not a drift. Rows the
// sealer alone produces (db-tail, build-state, sidecar) are owned by the seal;
// every carried outage item is owned by the landing and sourced by the seal.
var planManifestPairs = map[string][2]string{
	"db": {OfflineOwnerLanding, OfflineSourceSeal}, "db-tail": {OfflineOwnerSeal, OfflineSourceSeal},
	"build-state": {OfflineOwnerSeal, OfflineSourceSeal}, "lease": {OfflineOwnerLanding, OfflineSourceSeal},
	"files": {OfflineOwnerLanding, OfflineSourceSeal}, "env": {OfflineOwnerLanding, OfflineSourceSeal},
	"key-adoption": {OfflineOwnerLanding, ""}, "config": {OfflineOwnerLanding, ""},
	"env-pin": {OfflineOwnerLanding, OfflineSourceSeal}, "release": {OfflineOwnerLanding, OfflineSourceSeal},
	"store": {OfflineOwnerNone, ""}, "timers": {OfflineOwnerLanding, OfflineSourceSeal},
	"next-due": {OfflineOwnerLanding, OfflineSourceSeal}, "replication": {OfflineOwnerLanding, OfflineSourceSeal},
	"row": {OfflineOwnerLanding, ""}, "routing": {OfflineOwnerNone, ""}, "source-retire": {OfflineOwnerNone, ""},
	"sidecar": {OfflineOwnerSeal, OfflineSourceSeal}, "restore-verify": {OfflineOwnerNone, ""},
	"warm-cache": {OfflineOwnerLanding, ""}, "receipts": {OfflineOwnerNone, ""}, "class": {OfflineOwnerDoor, ""},
}

var offlineDispositionKinds = []string{"carried", "rebuilt", "refused", "parked", "not carried"}

// validateOfflineManifest is the checklist rule: every row has an id, a thing
// it names, a disposition that begins with its kind and gives its reason, and
// exactly one owner; no id repeats.
func validateOfflineManifest(rows []OfflineItem) error {
	seen := map[string]bool{}
	owners := map[string]bool{OfflineOwnerDoor: true, OfflineOwnerSeal: true, OfflineOwnerLanding: true, OfflineOwnerNone: true}
	for i, it := range rows {
		switch {
		case it.ID == "":
			return fmt.Errorf("row %d has no id", i)
		case seen[it.ID]:
			return fmt.Errorf("row %q is listed twice", it.ID)
		case it.Item == "":
			return fmt.Errorf("row %q names no item", it.ID)
		case !owners[it.Owner]:
			return fmt.Errorf("row %q has owner %q, which is not door, seal, landing or none", it.ID, it.Owner)
		case it.Source != "" && it.Source != OfflineSourceSeal:
			return fmt.Errorf("row %q has source %q, which is neither seal nor empty", it.ID, it.Source)
		}
		seen[it.ID] = true
		kind := ""
		for _, k := range offlineDispositionKinds {
			if strings.HasPrefix(it.Disposition, k+": ") && len(it.Disposition) > len(k)+2 {
				kind = k
			}
		}
		if kind == "" {
			return fmt.Errorf("row %q has disposition %q, which does not begin with a kind and a reason", it.ID, it.Disposition)
		}
		if strings.ContainsRune(it.Disposition, '\u2014') || strings.ContainsRune(it.Item, '\u2014') {
			return fmt.Errorf("row %q has an em dash", it.ID)
		}
	}
	return nil
}

func TestOfflineManifestIsCovered(t *testing.T) {
	if err := validateOfflineManifest(OfflineManifest); err != nil {
		t.Fatal(err)
	}
	var ids []string
	for _, it := range OfflineManifest {
		ids = append(ids, it.ID)
	}
	if !reflect.DeepEqual(ids, planManifestIDs) {
		t.Fatalf("the manifest's rows are not the plan's:\n got %v\nwant %v", ids, planManifestIDs)
	}

	// Owner and Source of every row are the pinned pair, and the pins cover
	// every row.
	if len(planManifestPairs) != len(OfflineManifest) {
		t.Fatalf("%d pinned pairs for %d rows", len(planManifestPairs), len(OfflineManifest))
	}
	for _, it := range OfflineManifest {
		want, ok := planManifestPairs[it.ID]
		if !ok || it.Owner != want[0] || it.Source != want[1] {
			t.Errorf("%s: owner %q source %q, pinned %v", it.ID, it.Owner, it.Source, want)
		}
	}
	// The outage items: every carried item the target must have is required
	// at open (owner landing) and reported at flip (source seal).
	for _, id := range []string{"db", "lease", "files", "env", "timers", "next-due", "replication"} {
		it, _ := offlineItem(id)
		if it.Owner != OfflineOwnerLanding || it.Source != OfflineSourceSeal {
			t.Errorf("%s is not proven on the target and reported by the seal: %+v", id, it)
		}
	}
	var sourced []string
	for _, it := range OfflineSourced() {
		sourced = append(sourced, it.ID)
	}
	wantSourced := []string{"db", "db-tail", "build-state", "lease", "files", "env", "env-pin", "release", "timers", "next-due", "replication", "sidecar"}
	if !reflect.DeepEqual(sourced, wantSourced) {
		t.Fatalf("OfflineSourced: %v, want %v", sourced, wantSourced)
	}

	// Every row is owned by exactly one of the four: OfflineOwned partitions
	// the manifest.
	total := 0
	for _, owner := range []string{OfflineOwnerDoor, OfflineOwnerSeal, OfflineOwnerLanding, OfflineOwnerNone} {
		rows := OfflineOwned(owner)
		total += len(rows)
		for _, it := range rows {
			if got, ok := OfflineItemOwner(it.ID); !ok || got != owner {
				t.Errorf("OfflineItemOwner(%q) = %q, %v; OfflineOwned(%q) lists it", it.ID, got, ok, owner)
			}
		}
	}
	if total != len(OfflineManifest) {
		t.Fatalf("the owners cover %d of %d rows", total, len(OfflineManifest))
	}
	if len(OfflineOwned(OfflineOwnerSeal)) == 0 || len(OfflineOwned(OfflineOwnerLanding)) == 0 || len(OfflineOwned(OfflineOwnerDoor)) == 0 {
		t.Fatal("a stage owns nothing: the doors would have nothing to wait for")
	}
	if _, ok := OfflineItemOwner("nope"); ok {
		t.Fatal("an unlisted item has an owner")
	}
	// An item nothing checks must say why it is not checked.
	for _, it := range OfflineOwned(OfflineOwnerNone) {
		if !strings.Contains(it.Disposition, ": ") {
			t.Errorf("%s: owner none with no reason", it.ID)
		}
	}
}

// The checklist rule must bite: a row added with no owner, a duplicate, an
// unknown owner or a disposition with no reason fails.
func TestOfflineManifestRuleBites(t *testing.T) {
	good := OfflineItem{"extra", "something", "carried: because", OfflineOwnerSeal, OfflineSourceSeal}
	if err := validateOfflineManifest(append(append([]OfflineItem(nil), OfflineManifest...), good)); err != nil {
		t.Fatalf("a well-formed extra row: %v", err)
	}
	for name, bad := range map[string]OfflineItem{
		"no owner":         {"extra", "something", "carried: because", "", ""},
		"unknown owner":    {"extra", "something", "carried: because", "operator", ""},
		"no id":            {"", "something", "carried: because", OfflineOwnerSeal, ""},
		"duplicate id":     {"db", "something", "carried: because", OfflineOwnerSeal, ""},
		"no item":          {"extra", "", "carried: because", OfflineOwnerSeal, ""},
		"no reason":        {"extra", "something", "carried", OfflineOwnerSeal, ""},
		"empty reason":     {"extra", "something", "carried: ", OfflineOwnerSeal, ""},
		"unknown kind":     {"extra", "something", "forgotten: because", OfflineOwnerSeal, ""},
		"unknown source":   {"extra", "something", "carried: because", OfflineOwnerSeal, "operator"},
		"em dash in a row": {"extra", "something", "carried: because \u2014 reasons", OfflineOwnerSeal, ""},
	} {
		if err := validateOfflineManifest(append(append([]OfflineItem(nil), OfflineManifest...), bad)); err == nil {
			t.Errorf("%s: the rule accepted the row", name)
		}
	}
}

// A report may carry only items the manifest lists, and only the ones its
// role carries: the source side carries what the seal reports, the target
// what the landing owns.
func TestOfflineReportItemsAreTheManifests(t *testing.T) {
	base := func(role string) OfflineReport {
		return OfflineReport{V: OfflineReportVersion, Op: offTestOp, Host: offTestHost, Box: "box-a", Role: role, Phase: OfflinePhaseSealed,
			Items: map[string]Check{}, At: time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)}
	}
	sealer := base(OfflineRoleSealer)
	for _, it := range OfflineSourced() {
		sealer.Items[it.ID] = Check{OK: true}
	}
	if err := ValidateOfflineReport(sealer); err != nil {
		t.Fatalf("a full sealer report: %v", err)
	}
	if ok, bad := sealer.SourceItemsOK(); !ok || len(bad) != 0 {
		t.Fatalf("source items: %v %v", ok, bad)
	}
	// A sealer report does not satisfy the open door, whatever it carries.
	if ok, bad := sealer.ItemsOK(OfflineOwnerLanding); ok || !reflect.DeepEqual(bad, []string{"key-adoption", "config", "row", "warm-cache"}) {
		t.Fatalf("a sealer report satisfies the landing's items: %v %v", ok, bad)
	}
	// A not-OK seal item is named, in manifest order.
	sealer.Items["files"] = Check{OK: false, Reason: "no archive"}
	delete(sealer.Items, "db")
	if ok, bad := sealer.SourceItemsOK(); ok || !reflect.DeepEqual(bad, []string{"db", "files"}) {
		t.Fatalf("SourceItemsOK: %v %v", ok, bad)
	}

	// The open door's set includes every outage item, so a target with no
	// files directory cannot pass it.
	target := base(OfflineRoleTarget)
	target.Phase = OfflinePhaseLanded
	for _, it := range OfflineOwned(OfflineOwnerLanding) {
		target.Items[it.ID] = Check{OK: true}
	}
	if err := ValidateOfflineReport(target); err != nil {
		t.Fatalf("a full target report: %v", err)
	}
	if ok, bad := target.ItemsOK(OfflineOwnerLanding); !ok || len(bad) != 0 {
		t.Fatalf("a full target report: %v %v", ok, bad)
	}
	for _, id := range []string{"db", "lease", "files", "env", "timers", "next-due", "replication"} {
		missing := target
		missing.Items = map[string]Check{}
		for k, v := range target.Items {
			missing.Items[k] = v
		}
		delete(missing.Items, id)
		if ok, bad := missing.ItemsOK(OfflineOwnerLanding); ok || !reflect.DeepEqual(bad, []string{id}) {
			t.Errorf("open passes a target report without %s: %v %v", id, ok, bad)
		}
		failed := missing
		failed.Items = map[string]Check{}
		for k, v := range target.Items {
			failed.Items[k] = v
		}
		failed.Items[id] = Check{OK: false, Reason: "not there"}
		if ok, bad := failed.ItemsOK(OfflineOwnerLanding); ok || !reflect.DeepEqual(bad, []string{id}) {
			t.Errorf("open passes a target report with %s failed: %v %v", id, ok, bad)
		}
	}

	// An item the manifest does not list fails the report.
	rogue := base(OfflineRoleSealer)
	rogue.Items["made-up"] = Check{OK: true}
	if err := ValidateOfflineReport(rogue); err == nil || !strings.Contains(err.Error(), "made-up") {
		t.Fatalf("an unlisted item: %v", err)
	}
	// An item the role does not carry: landing-only items on the source
	// side, source-only items on the target.
	for _, id := range []string{"row", "config", "key-adoption", "warm-cache"} {
		r := base(OfflineRoleSealer)
		r.Items[id] = Check{OK: true}
		if err := ValidateOfflineReport(r); err == nil {
			t.Errorf("a sealer report carried %s", id)
		}
	}
	for _, id := range []string{"db-tail", "build-state", "sidecar"} {
		r := base(OfflineRoleTarget)
		r.Items[id] = Check{OK: true}
		if err := ValidateOfflineReport(r); err == nil {
			t.Errorf("a target report carried %s", id)
		}
	}
	for _, owner := range []string{OfflineOwnerDoor, OfflineOwnerNone} {
		for _, it := range OfflineOwned(owner) {
			for _, role := range []string{OfflineRoleSource, OfflineRoleSealer, OfflineRoleTarget} {
				r := base(role)
				r.Items[it.ID] = Check{OK: true}
				if err := ValidateOfflineReport(r); err == nil {
					t.Errorf("a %s report carried %s, which is %s-owned and checked by no report", role, it.ID, owner)
				}
			}
		}
	}
	// Version, phase, role and names.
	for name, mut := range map[string]func(*OfflineReport){
		"version":  func(r *OfflineReport) { r.V = 0 },
		"phase":    func(r *OfflineReport) { r.Phase = "dancing" },
		"role":     func(r *OfflineReport) { r.Role = "bystander" },
		"host":     func(r *OfflineReport) { r.Host = "../x" },
		"no op":    func(r *OfflineReport) { r.Op = "" },
		"box path": func(r *OfflineReport) { r.Box = "a/b" },
	} {
		r := base(OfflineRoleSealer)
		mut(&r)
		if err := ValidateOfflineReport(r); err == nil {
			t.Errorf("%s: accepted", name)
		}
	}
}

// Every field a report carries survives the wire.
func TestOfflineReportRoundTrips(t *testing.T) {
	at := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
	in := OfflineReport{
		V: 1, Op: offTestOp, Host: offTestHost, Box: "box-a", Role: OfflineRoleSealer, Phase: OfflinePhaseSealed,
		Items: map[string]Check{"db": {OK: true}, "files": {OK: false, Reason: "no archive"}}, At: at,
		Held: true, Fenced: true, Quiesced: true, LocalDB: true, Files: true, Env: true, Sidecar: true, RowPresent: true, LocalTXID: 41,
		FinalTXID: 42, Staged: map[string]StagedObject{"replica/0001.ltx": {Size: 9, ETag: "e1"}},
		FilesPlain: "p", FilesSealed: "s", FilesBytes: 7, FilesEntries: 2,
		EnvSHA256: "e", KeySHA256: "k", EnvDigest: "d", Release: "v0.28.4", TimersDigest: "t", Replication: "cold", NextDue: 99,
		TombstoneOwner: TombstoneOwner(offTestOp), TombstoneUntil: at.Add(TombstoneTTL),
		Procs: 2, LayoutJournal: true,
		Retired: []RetiredPath{{Path: "/data/a.go.example.com.db", Action: RetiredParked}, {Path: "/opt/carlos/live/a.go.example.com", Action: RetiredLeft, Reason: "root-owned"}},
	}
	raw, err := json.Marshal(in)
	if err != nil {
		t.Fatal(err)
	}
	var out OfflineReport
	if err := json.Unmarshal(raw, &out); err != nil {
		t.Fatal(err)
	}
	if !reflect.DeepEqual(in, out) {
		t.Fatalf("round trip:\n in %+v\nout %+v", in, out)
	}
	// A report's booleans are always written: an absent held means not held,
	// and a reader can see the box said so.
	empty, _ := json.Marshal(OfflineReport{})
	for _, k := range []string{`"held":false`, `"fenced":false`, `"quiesced":false`} {
		if !strings.Contains(string(empty), k) {
			t.Errorf("a zero report omits %s: %s", k, empty)
		}
	}
	// The tombstone's time is written only when there is one.
	if strings.Contains(string(empty), "tombstone_until") {
		t.Errorf("a zero report writes a tombstone time: %s", empty)
	}
}

func TestOfflineKeyHelpers(t *testing.T) {
	if got := OfflineReportKey(offTestHost, offTestOp, OfflineRoleSealer, "box-a"); got != "control/offline-moves/"+offTestHost+"/"+offTestOp+"/sealer/box-a.json" {
		t.Fatalf("report key %q", got)
	}
	if got := OfflineReportPrefix(offTestHost, offTestOp); got != "control/offline-moves/"+offTestHost+"/"+offTestOp+"/" {
		t.Fatalf("report prefix %q", got)
	}
	if got := OfflineArchivePrefix(offTestHost, offTestOp); got != "moves/@offline/"+offTestHost+"/"+offTestOp+"/" {
		t.Fatalf("archive prefix %q", got)
	}
	if got := TombstoneOwner(offTestOp); got != "moved:"+offTestOp {
		t.Fatalf("tombstone owner %q", got)
	}
	if got := OfflineParkDir("/data", offTestHost, offTestOp, OfflineRoleSource); got != "/data/.offline-moves/"+offTestHost+"/"+offTestOp+"/source" {
		t.Fatalf("park dir %q", got)
	}
	for _, role := range []string{OfflineParkTargetBefore, OfflineParkVerify} {
		if got := OfflineParkDir("/data", offTestHost, offTestOp, role); got != "/data/.offline-moves/"+offTestHost+"/"+offTestOp+"/"+role {
			t.Errorf("park dir for %s: %q", role, got)
		}
		// A park role is not a report role: the report key does not take it.
		if OfflineReportKey(offTestHost, offTestOp, role, "box-a") != "" {
			t.Errorf("a report key for park role %s", role)
		}
	}
	if OfflineParkTargetBefore != "target-before" || OfflineParkVerify != "verify" {
		t.Fatal("the park role names are the plan's")
	}
	// Nothing else is accepted, and a numbered suffix is the caller's.
	for _, role := range []string{"target-before-2", "scratch", "verify/x"} {
		if OfflineParkDir("/data", offTestHost, offTestOp, role) != "" {
			t.Errorf("park role %q accepted", role)
		}
	}
	if TombstoneTTL != 7*24*time.Hour {
		t.Fatalf("tombstone ttl %v", TombstoneTTL)
	}

	// Hosts that fail ValidHostShape are rejected by every helper that takes
	// one. The precondition keeps this honest if the shape rule moves.
	for _, host := range []string{"", "../x", "a/b.example.com", "a b.example.com", "UPPER.EXAMPLE.COM", ".leading.example.com", "a.b.example.com/", "a..b"} {
		if provision.ValidHostShape(host) {
			t.Fatalf("test precondition: %q is a valid host shape", host)
		}
		if k := OfflineReportKey(host, offTestOp, OfflineRoleSealer, "box-a"); k != "" {
			t.Errorf("report key for %q: %q", host, k)
		}
		if k := OfflineReportPrefix(host, offTestOp); k != "" {
			t.Errorf("report prefix for %q: %q", host, k)
		}
		if k := OfflineArchivePrefix(host, offTestOp); k != "" {
			t.Errorf("archive prefix for %q: %q", host, k)
		}
		if k := OfflineParkDir("/data", host, offTestOp, OfflineRoleSource); k != "" {
			t.Errorf("park dir for %q: %q", host, k)
		}
		if OfflineNamesOK(host, offTestOp, OfflineRoleSealer, "box-a") {
			t.Errorf("names ok for %q", host)
		}
	}
	// And the other parts cannot climb out of the prefix they are joined to.
	for _, op := range []string{"", ".", "..", "a/b", "../mv_x", "mv x"} {
		if OfflineReportKey(offTestHost, op, OfflineRoleSealer, "box-a") != "" || OfflineArchivePrefix(offTestHost, op) != "" ||
			OfflineParkDir("/data", offTestHost, op, OfflineRoleSource) != "" || OfflineReportPrefix(offTestHost, op) != "" {
			t.Errorf("op %q accepted", op)
		}
	}
	for _, box := range []string{"", "..", "a/b", "box a"} {
		if OfflineReportKey(offTestHost, offTestOp, OfflineRoleSealer, box) != "" {
			t.Errorf("box %q accepted", box)
		}
	}
	for _, role := range []string{"", "Source", "operator", "../source"} {
		if OfflineReportKey(offTestHost, offTestOp, role, "box-a") != "" || OfflineParkDir("/data", offTestHost, offTestOp, role) != "" {
			t.Errorf("role %q accepted", role)
		}
	}
	if OfflineParkDir("", offTestHost, offTestOp, OfflineRoleSource) != "" {
		t.Error("a park dir with no data dir")
	}
}

// The report key sits outside every prefix the bucket makes create-only or
// no-delete, so a box may rewrite its report.
func TestOfflineReportKeyIsOutsideTheProtectedPrefixes(t *testing.T) {
	key := OfflineReportKey(offTestHost, offTestOp, OfflineRoleTarget, "box-a")
	for _, p := range []string{"control/boxes", "control/box-volumes", "control/box-enrollment", "control/box-fences"} {
		if strings.HasPrefix(key, p) {
			t.Errorf("%s is under %s", key, p)
		}
	}
}

// Sanity for the inventory the other tasks pin their wiring to: the exported
// state constants of the record side are the ones a report's phases pair with.
func TestOfflineStateAndPhaseNamesAreDistinct(t *testing.T) {
	states := []string{provision.OfflineStopping, provision.OfflineSealing, provision.OfflineSealed, provision.OfflineLanding,
		provision.OfflineLanded, provision.OfflineOpen, provision.OfflineFinishing, provision.OfflineReverting}
	sort.Strings(states)
	for i := 1; i < len(states); i++ {
		if states[i] == states[i-1] {
			t.Fatalf("state %q is declared twice", states[i])
		}
	}
	for _, s := range states {
		if !strings.HasPrefix(s, "offline-") {
			t.Errorf("state %q lacks the offline- prefix", s)
		}
	}
	if provision.MoveKindOffline != "offline" {
		t.Fatalf("kind %q", provision.MoveKindOffline)
	}
}

func TestOfflineSealKeyAndValidation(t *testing.T) {
	const host, op = "acme.go.example.com", "mv_aaaaaaaaaaaaaaaaaaaaaaaaaa"
	if got := OfflineSealKey(host, op); got != "control/offline-moves/"+host+"/"+op+"/seal.json" {
		t.Fatalf("seal key = %q", got)
	}
	for _, bad := range [][2]string{{"", op}, {"not a host", op}, {host, ""}, {host, "../x"}, {host, "a/b"}} {
		if OfflineSealKey(bad[0], bad[1]) != "" || OfflineSealEvidence(bad[0], bad[1], []byte("x")) != "" {
			t.Errorf("key for %q accepted", bad)
		}
	}
	// The seal key is not a report key and sits under the report prefix.
	if !strings.HasPrefix(OfflineSealKey(host, op), OfflineReportPrefix(host, op)) {
		t.Error("the seal is outside the move's prefix")
	}
	items := map[string]Check{}
	for _, it := range OfflineSourced() {
		items[it.ID] = Check{OK: true}
	}
	good := OfflineReport{V: OfflineReportVersion, Op: op, Host: host, Box: "ie-box", Role: OfflineRoleSealer, Phase: OfflinePhaseSealed,
		Items: items, TombstoneOwner: TombstoneOwner(op), At: time.Now()}
	if err := ValidateOfflineSeal(good, host, op); err != nil {
		t.Fatal(err)
	}
	for name, edit := range map[string]func(*OfflineReport){
		"another host":  func(r *OfflineReport) { r.Host = "other.go.example.com" },
		"another op":    func(r *OfflineReport) { r.Op = "mv_bbbbbbbbbbbbbbbbbbbbbbbbbb" },
		"a source role": func(r *OfflineReport) { r.Role = OfflineRoleSource },
		"not sealed":    func(r *OfflineReport) { r.Phase = OfflinePhaseStopped },
		"no tombstone":  func(r *OfflineReport) { r.TombstoneOwner = "" },
		"an item not OK": func(r *OfflineReport) {
			c := map[string]Check{}
			for k, v := range items {
				c[k] = v
			}
			c["files"] = Check{}
			r.Items = c
		},
		"a missing item": func(r *OfflineReport) { r.Items = map[string]Check{} },
	} {
		r := good
		edit(&r)
		if ValidateOfflineSeal(r, host, op) == nil {
			t.Errorf("%s: accepted", name)
		}
	}
	// The evidence is the key and the hash of the bytes.
	ev := OfflineSealEvidence(host, op, []byte("x"))
	if !strings.HasPrefix(ev, OfflineSealKey(host, op)+"@") || len(ev) != len(OfflineSealKey(host, op))+1+64 {
		t.Errorf("evidence = %q", ev)
	}
}

// The stager (offline-staging's pre-copy): its report is a report like any
// other, under its own role, and carries no manifest item; the stager is the
// lexically first box, whatever order the listing gave.
func TestOfflineStagerReport(t *testing.T) {
	r := OfflineReport{V: OfflineReportVersion, Op: offTestOp, Host: offTestHost, Box: "box-a", Role: OfflineRoleStager, Phase: OfflinePhaseStaged,
		ReplicaObjects: 10, ReplicaBytes: 100, StagedObjects: 9, StagedBytes: 90, RoundObjects: 3, RoundBytes: 30}
	if err := ValidateOfflineReport(r); err != nil {
		t.Fatal(err)
	}
	if got := OfflineReportKey(offTestHost, offTestOp, OfflineRoleStager, "box-a"); got != "control/offline-moves/"+offTestHost+"/"+offTestOp+"/stager/box-a.json" {
		t.Fatalf("key = %q", got)
	}
	r.Items = map[string]Check{"db": {OK: true}}
	if ValidateOfflineReport(r) == nil {
		t.Fatal("a stager report carrying an item was accepted")
	}
	body, err := json.Marshal(OfflineReport{V: 1})
	if err != nil || strings.Contains(string(body), "replica_objects") || strings.Contains(string(body), "stage_error") {
		t.Fatalf("the stager's fields are not omitempty: %s", body)
	}
	for _, c := range []struct {
		boxes []string
		want  string
	}{{nil, ""}, {[]string{"box-c", "box-a", "box-b"}, "box-a"}, {[]string{"", "box-z"}, "box-z"}} {
		if got := OfflineStager(c.boxes); got != c.want {
			t.Errorf("OfflineStager(%v) = %q, want %q", c.boxes, got, c.want)
		}
	}
}