From 09bd348c58d8d8bbb96e25a497fa7cb53887b41a Mon Sep 17 00:00:00 2001 From: Haven Xia Date: Wed, 2 Sep 2026 14:07:56 -0700 Subject: [PATCH] ateapi: drop unknown fields when decoding objects from the store (#1414) An old server replica doing Read-Modify-Write should drop unknown fields on read and serve the reduced object. Here we add `proto.UnmarshalOptions{DiscardUnknown: true}` because when we read an unknown field in storage layer, it must mean a newer replica has written such a field. Since we have declined requests with unknown fields with interceptors in #1216. Fixes #1205 - [ ] Tests pass - [x] Appropriate changes to documentation are included in the PR --- cmd/ateapi/internal/store/atepg/atepg.go | 52 +-- cmd/ateapi/internal/store/atepg/outbox.go | 4 +- .../internal/store/storecontract/contract.go | 307 ++++++++++++++++++ 3 files changed, 338 insertions(+), 25 deletions(-) diff --git a/cmd/ateapi/internal/store/atepg/atepg.go b/cmd/ateapi/internal/store/atepg/atepg.go index 8ffe88667..55e2295c0 100644 --- a/cmd/ateapi/internal/store/atepg/atepg.go +++ b/cmd/ateapi/internal/store/atepg/atepg.go @@ -214,6 +214,12 @@ type querier interface { Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) } +// unmarshalStored decodes a stored proto, dropping fields this binary has no +// descriptor for. This means a newer replica can have written such a field. +func unmarshalStored(b []byte, m proto.Message) error { + return proto.UnmarshalOptions{DiscardUnknown: true}.Unmarshal(b, m) +} + // TODO: EOL this in favor of setCreateMetadata func newCreateMetadata(atespace, name string) *ateapipb.ResourceMetadata { now := timestamppb.Now() @@ -328,7 +334,7 @@ func (p *Persistence) GetAtespace(ctx context.Context, name string) (*ateapipb.A return nil, fmt.Errorf("getting atespace %q: %w", name, err) } out := &ateapipb.Atespace{} - if err := proto.Unmarshal(protoBytes, out); err != nil { + if err := unmarshalStored(protoBytes, out); err != nil { return nil, fmt.Errorf("unmarshaling atespace: %w", err) } return out, nil @@ -368,7 +374,7 @@ func (p *Persistence) ListAtespaces(ctx context.Context, opts store.ListOptions) return store.ListResponse[*ateapipb.Atespace]{}, fmt.Errorf("scanning atespace row: %w", err) } a := &ateapipb.Atespace{} - if err := proto.Unmarshal(protoBytes, a); err != nil { + if err := unmarshalStored(protoBytes, a); err != nil { return store.ListResponse[*ateapipb.Atespace]{}, fmt.Errorf("unmarshaling atespace: %w", err) } result = append(result, a) @@ -399,7 +405,7 @@ func (p *Persistence) DeleteAtespace(ctx context.Context, name string) (*ateapip return nil, fmt.Errorf("deleting atespace %q: %w", name, err) } out := &ateapipb.Atespace{} - if err := proto.Unmarshal(protoBytes, out); err != nil { + if err := unmarshalStored(protoBytes, out); err != nil { return nil, fmt.Errorf("unmarshaling deleted atespace: %w", err) } return out, nil @@ -444,7 +450,7 @@ func (p *Persistence) GetActorTemplate(ctx context.Context, templateRef resource return nil, fmt.Errorf("getting actor template %s: %w", templateRef, err) } out := &ateapipb.ActorTemplate{} - if err := proto.Unmarshal(protoBytes, out); err != nil { + if err := unmarshalStored(protoBytes, out); err != nil { return nil, fmt.Errorf("unmarshaling actor template: %w", err) } return out, nil @@ -477,7 +483,7 @@ func (p *Persistence) UpdateActorTemplate(ctx context.Context, templateRef resou } dbTemplate := &ateapipb.ActorTemplate{} - if err := proto.Unmarshal(currentBytes, dbTemplate); err != nil { + if err := unmarshalStored(currentBytes, dbTemplate); err != nil { return nil, fmt.Errorf("unmarshaling actor template for update: %w", err) } if err := validateProtoMetadataMatchesColumns("actor template "+templateRef.String(), dbTemplate.GetMetadata(), currentUID, currentVersion); err != nil { @@ -567,7 +573,7 @@ func (p *Persistence) ListActorTemplates(ctx context.Context, atespace string, o return store.ListResponse[*ateapipb.ActorTemplate]{}, fmt.Errorf("scanning actor template row: %w", err) } template := &ateapipb.ActorTemplate{} - if err := proto.Unmarshal(protoBytes, template); err != nil { + if err := unmarshalStored(protoBytes, template); err != nil { return store.ListResponse[*ateapipb.ActorTemplate]{}, fmt.Errorf("unmarshaling actor template: %w", err) } keys = append(keys, k) @@ -602,7 +608,7 @@ func (p *Persistence) DeleteActorTemplate(ctx context.Context, templateRef resou return nil, fmt.Errorf("deleting actor template %s: %w", templateRef, err) } out := &ateapipb.ActorTemplate{} - if err := proto.Unmarshal(protoBytes, out); err != nil { + if err := unmarshalStored(protoBytes, out); err != nil { return nil, fmt.Errorf("unmarshaling deleted actor template: %w", err) } return out, nil @@ -654,7 +660,7 @@ func (p *Persistence) GetActor(ctx context.Context, actorRef resources.ActorRef) return nil, fmt.Errorf("getting actor %s/%s: %w", actorRef.Atespace, actorRef.Name, err) } out := &ateapipb.Actor{} - if err := proto.Unmarshal(protoBytes, out); err != nil { + if err := unmarshalStored(protoBytes, out); err != nil { return nil, fmt.Errorf("unmarshaling actor: %w", err) } return out, nil @@ -678,7 +684,7 @@ func (p *Persistence) UpdateActor(ctx context.Context, actorRef resources.ActorR } dbActor := &ateapipb.Actor{} - if err := proto.Unmarshal(currentBytes, dbActor); err != nil { + if err := unmarshalStored(currentBytes, dbActor); err != nil { return nil, fmt.Errorf("unmarshaling actor for update: %w", err) } if err := validateProtoMetadataMatchesColumns("actor "+actorRef.String(), dbActor.GetMetadata(), currentUID, currentVersion); err != nil { @@ -739,7 +745,7 @@ func (p *Persistence) DeleteActor(ctx context.Context, actorRef resources.ActorR } out := &ateapipb.Actor{} - if err := proto.Unmarshal(protoBytes, out); err != nil { + if err := unmarshalStored(protoBytes, out); err != nil { return nil, fmt.Errorf("unmarshaling actor for deletion: %w", err) } if out.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_DELETING { @@ -801,7 +807,7 @@ func (p *Persistence) listActorsScoped(ctx context.Context, atespace string, pag return nil, "", fmt.Errorf("scanning actor row: %w", err) } a := &ateapipb.Actor{} - if err := proto.Unmarshal(protoBytes, a); err != nil { + if err := unmarshalStored(protoBytes, a); err != nil { return nil, "", fmt.Errorf("unmarshaling actor: %w", err) } result = append(result, a) @@ -849,7 +855,7 @@ func (p *Persistence) listActorsGlobal(ctx context.Context, pageSize int32, page return nil, "", fmt.Errorf("scanning actor row: %w", err) } a := &ateapipb.Actor{} - if err := proto.Unmarshal(protoBytes, a); err != nil { + if err := unmarshalStored(protoBytes, a); err != nil { return nil, "", fmt.Errorf("unmarshaling actor: %w", err) } result = append(result, a) @@ -971,7 +977,7 @@ func getEgressPolicyRow(ctx context.Context, q querier, query string, args ...an func unmarshalEgressPolicy(uid string, version int64, protoBytes []byte) (*ateapipb.EgressPolicy, error) { policy := &ateapipb.EgressPolicy{} - if err := proto.Unmarshal(protoBytes, policy); err != nil { + if err := unmarshalStored(protoBytes, policy); err != nil { return nil, fmt.Errorf("unmarshaling egress policy: %w", err) } if err := validateProtoMetadataMatchesColumns("egress policy", policy.GetMetadata(), uid, version); err != nil { @@ -1016,7 +1022,7 @@ func (p *Persistence) GetActorSnapshot(ctx context.Context, snapshotRef resource return nil, fmt.Errorf("getting actor snapshot %s/%s: %w", atespace, name, err) } out := &ateapipb.ActorSnapshot{} - if err := proto.Unmarshal(protoBytes, out); err != nil { + if err := unmarshalStored(protoBytes, out); err != nil { return nil, fmt.Errorf("unmarshaling actor snapshot: %w", err) } return out, nil @@ -1034,7 +1040,7 @@ func (p *Persistence) GetActorSnapshotTag(ctx context.Context, tagRef resources. return nil, fmt.Errorf("getting actor snapshot tag %s/%s: %w", atespace, name, err) } tag := &ateapipb.ActorSnapshotTag{} - if err := proto.Unmarshal(protoBytes, tag); err != nil { + if err := unmarshalStored(protoBytes, tag); err != nil { return nil, fmt.Errorf("unmarshaling actor snapshot tag: %w", err) } return tag, nil @@ -1086,7 +1092,7 @@ func (p *Persistence) listActorSnapshotsScoped(ctx context.Context, atespace str return nil, "", fmt.Errorf("scanning actor snapshot row: %w", err) } snapshot := &ateapipb.ActorSnapshot{} - if err := proto.Unmarshal(protoBytes, snapshot); err != nil { + if err := unmarshalStored(protoBytes, snapshot); err != nil { return nil, "", fmt.Errorf("unmarshaling actor snapshot: %w", err) } result = append(result, snapshot) @@ -1132,7 +1138,7 @@ func (p *Persistence) listActorSnapshotsGlobal(ctx context.Context, pageSize int return nil, "", fmt.Errorf("scanning actor snapshot row: %w", err) } snapshot := &ateapipb.ActorSnapshot{} - if err := proto.Unmarshal(protoBytes, snapshot); err != nil { + if err := unmarshalStored(protoBytes, snapshot); err != nil { return nil, "", fmt.Errorf("unmarshaling actor snapshot: %w", err) } result = append(result, snapshot) @@ -1202,7 +1208,7 @@ func (p *Persistence) CreateActorSnapshotTag(ctx context.Context, snapshotRef re return nil, fmt.Errorf("getting existing actor snapshot tag %s/%s: %w", tagAtespace, tagName, err) } existing := &ateapipb.ActorSnapshotTag{} - if err := proto.Unmarshal(existingBytes, existing); err != nil { + if err := unmarshalStored(existingBytes, existing); err != nil { return nil, fmt.Errorf("unmarshaling actor snapshot tag: %w", err) } if existing.GetSnapshot().GetAtespace() != snapshotAtespace || existing.GetSnapshot().GetName() != snapshotName || existing.GetScope() != tag.GetScope() { @@ -1248,7 +1254,7 @@ func (p *Persistence) UpdateActorSnapshotTag(ctx context.Context, tagRef resourc } dbTag := &ateapipb.ActorSnapshotTag{} - if err := proto.Unmarshal(currentBytes, dbTag); err != nil { + if err := unmarshalStored(currentBytes, dbTag); err != nil { return nil, fmt.Errorf("unmarshaling actor snapshot tag: %w", err) } if err := validateProtoMetadataMatchesColumns(fmt.Sprintf("actor snapshot tag %s/%s", atespace, name), dbTag.GetMetadata(), currentUID, currentVersion); err != nil { @@ -1302,7 +1308,7 @@ func (p *Persistence) DeleteActorSnapshotTag(ctx context.Context, tagRef resourc return nil, fmt.Errorf("deleting actor snapshot tag %s/%s: %w", atespace, name, err) } tag := &ateapipb.ActorSnapshotTag{} - if err := proto.Unmarshal(protoBytes, tag); err != nil { + if err := unmarshalStored(protoBytes, tag); err != nil { return nil, fmt.Errorf("unmarshaling deleted actor snapshot tag: %w", err) } return tag, nil @@ -1351,7 +1357,7 @@ func getWorkerRow(ctx context.Context, q querier, name string) (*ateapipb.Worker return nil, fmt.Errorf("getting worker %s: %w", name, err) } out := &ateapipb.Worker{} - if err := proto.Unmarshal(protoBytes, out); err != nil { + if err := unmarshalStored(protoBytes, out); err != nil { return nil, fmt.Errorf("unmarshaling worker: %w", err) } return out, nil @@ -1373,7 +1379,7 @@ func getWorkerRowForUpdate(ctx context.Context, tx pgx.Tx, name string) (*ateapi return nil, fmt.Errorf("locking worker %s for update: %w", name, err) } out := &ateapipb.Worker{} - if err := proto.Unmarshal(protoBytes, out); err != nil { + if err := unmarshalStored(protoBytes, out); err != nil { return nil, fmt.Errorf("unmarshaling worker: %w", err) } return out, nil @@ -1484,7 +1490,7 @@ func (p *Persistence) ListWorkers(ctx context.Context, opts store.ListOptions) ( return store.ListResponse[*ateapipb.Worker]{}, fmt.Errorf("scanning worker row: %w", err) } w := &ateapipb.Worker{} - if err := proto.Unmarshal(protoBytes, w); err != nil { + if err := unmarshalStored(protoBytes, w); err != nil { return store.ListResponse[*ateapipb.Worker]{}, fmt.Errorf("unmarshaling worker: %w", err) } result = append(result, w) diff --git a/cmd/ateapi/internal/store/atepg/outbox.go b/cmd/ateapi/internal/store/atepg/outbox.go index 535b00762..8f43f4591 100644 --- a/cmd/ateapi/internal/store/atepg/outbox.go +++ b/cmd/ateapi/internal/store/atepg/outbox.go @@ -59,8 +59,8 @@ func unmarshalWorkerEvent(payload []byte) (store.WorkerEvent, error) { return store.WorkerEvent{}, fmt.Errorf("unknown worker event type byte %d", payload[0]) } worker := &ateapipb.Worker{} - if err := proto.Unmarshal(payload[1:], worker); err != nil { - return store.WorkerEvent{}, fmt.Errorf("in proto.Unmarshal: %w", err) + if err := unmarshalStored(payload[1:], worker); err != nil { + return store.WorkerEvent{}, fmt.Errorf("in unmarshalStored: %w", err) } if worker.GetMetadata().GetName() == "" { return store.WorkerEvent{}, fmt.Errorf("worker event payload has no worker name") diff --git a/cmd/ateapi/internal/store/storecontract/contract.go b/cmd/ateapi/internal/store/storecontract/contract.go index 321b9e508..9285dba2d 100644 --- a/cmd/ateapi/internal/store/storecontract/contract.go +++ b/cmd/ateapi/internal/store/storecontract/contract.go @@ -26,7 +26,11 @@ import ( "time" "github.com/google/go-cmp/cmp" + "google.golang.org/protobuf/encoding/protowire" "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/reflect/protopath" + "google.golang.org/protobuf/reflect/protorange" + "google.golang.org/protobuf/reflect/protoreflect" "google.golang.org/protobuf/testing/protocmp" "google.golang.org/protobuf/types/known/emptypb" "google.golang.org/protobuf/types/known/timestamppb" @@ -142,6 +146,36 @@ func receiveEvent(t *testing.T, ch <-chan store.WorkerEvent) store.WorkerEvent { } } +// withUnknownField adds a field number no message in this binary declares to m and returns m. +func withUnknownField[M proto.Message](m M) M { + b := protowire.AppendTag(nil, 9999, protowire.VarintType) + m.ProtoReflect().SetUnknown(protowire.AppendVarint(b, 42)) + return m +} + +// requireNoUnknownFields fails the test if m or any message nested in it carries unknown fields. +func requireNoUnknownFields(t *testing.T, op string, m proto.Message) { + t.Helper() + err := protorange.Range(m.ProtoReflect(), func(p protopath.Values) error { + if msg, ok := p.Index(-1).Value.Interface().(protoreflect.Message); ok && len(msg.GetUnknown()) > 0 { + return fmt.Errorf("%s returned unknown fields at %v", op, p.Path) + } + return nil + }) + if err != nil { + t.Error(err) + } +} + +// assertPruned checks that got is want with its unknown fields dropped, ignoring the server-owned version and timestamps. +func assertPruned(t *testing.T, op string, want, got proto.Message) { + t.Helper() + requireNoUnknownFields(t, op, got) + if diff := cmp.Diff(want, got, protocmp.Transform(), protocmp.IgnoreUnknown(), ignoreVersion, ignoreTimestamps); diff != "" { + t.Errorf("%s mismatch (-want +got):\n%s", op, diff) + } +} + // RunContractTests runs the backend-neutral store.Interface assertions // against a fresh store.Interface built by setup for each subtest. setup is // responsible for its own cleanup (e.g. via t.Cleanup). @@ -157,6 +191,7 @@ func RunContractTests(t *testing.T, setup func(t *testing.T) store.Interface) { runActorSnapshotContractTests(t, setup) runLeaseContractTests(t, setup) runListOptionsContractTests(t, setup) + runUnknownFieldContractTests(t, setup) } func runEgressPolicyContractTests(t *testing.T, setup func(t *testing.T) store.Interface) { @@ -1973,3 +2008,275 @@ func runLeaseContractTests(t *testing.T, setup func(t *testing.T) store.Interfac lease.Close() }) } + +// runUnknownFieldContractTests asserts that every read path, including the +// object handed to a read-modify-write mutation, drops fields this binary has +// no descriptor for at every nesting level. +func runUnknownFieldContractTests(t *testing.T, setup func(t *testing.T) store.Interface) { + t.Helper() + + t.Run("UnknownFields_Actor", func(t *testing.T) { + s := setup(t) + ctx := context.Background() + mustCreateAtespace(t, s, testAtespace) + created, err := s.CreateActor(ctx, withUnknownField(&ateapipb.Actor{ + Metadata: withUnknownField(&ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "actor-1"}), + ActorTemplate: withUnknownField(&ateapipb.ObjectRef{Atespace: testAtespace, Name: "tmpl-1"}), + Status: withUnknownField(&ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_DELETING}), + })) + if err != nil { + t.Fatalf("CreateActor failed: %v", err) + } + ref := resources.ActorRefFromActor(created) + + got, err := s.GetActor(ctx, ref) + if err != nil { + t.Fatalf("GetActor failed: %v", err) + } + assertPruned(t, "GetActor", created, got) + + list, err := s.ListActors(ctx, testAtespace, store.ListOptions{}) + if err != nil || len(list.Items) != 1 { + t.Fatalf("ListActors = %d items, %v; want 1", len(list.Items), err) + } + assertPruned(t, "ListActors", created, list.Items[0]) + + updated, err := s.UpdateActor(ctx, ref, store.PreconditionFrom(got), func(toUpdate *ateapipb.Actor) error { + requireNoUnknownFields(t, "UpdateActor mutation input", toUpdate) + return nil + }) + if err != nil { + t.Fatalf("UpdateActor failed: %v", err) + } + assertPruned(t, "UpdateActor", created, updated) + + deleted, err := s.DeleteActor(ctx, ref) + if err != nil { + t.Fatalf("DeleteActor failed: %v", err) + } + assertPruned(t, "DeleteActor", created, deleted) + }) + + t.Run("UnknownFields_EgressPolicy", func(t *testing.T) { + s := setup(t) + ctx := context.Background() + mustCreateAtespace(t, s, testAtespace) + actor, err := s.CreateActor(ctx, &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "actor-1"}, + Status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_RUNNING}, + }) + if err != nil { + t.Fatalf("CreateActor failed: %v", err) + } + ref := resources.ActorRefFromActor(actor) + created, err := s.CreateEgressPolicy(ctx, ref, withUnknownField(&ateapipb.EgressPolicy{ + Rules: []*ateapipb.EgressRule{withUnknownField(&ateapipb.EgressRule{ + Hostnames: &ateapipb.HostnameRule{Patterns: []string{"api.example.com"}}, + })}, + })) + if err != nil { + t.Fatalf("CreateEgressPolicy failed: %v", err) + } + + got, err := s.GetEgressPolicy(ctx, ref) + if err != nil { + t.Fatalf("GetEgressPolicy failed: %v", err) + } + assertPruned(t, "GetEgressPolicy", created, got) + + updated, err := s.UpdateEgressPolicy(ctx, ref, store.PreconditionFrom(got), func(toUpdate *ateapipb.EgressPolicy) error { + requireNoUnknownFields(t, "UpdateEgressPolicy mutation input", toUpdate) + return nil + }) + if err != nil { + t.Fatalf("UpdateEgressPolicy failed: %v", err) + } + assertPruned(t, "UpdateEgressPolicy", created, updated) + + deleted, err := s.DeleteEgressPolicy(ctx, ref) + if err != nil { + t.Fatalf("DeleteEgressPolicy failed: %v", err) + } + assertPruned(t, "DeleteEgressPolicy", created, deleted) + }) + + t.Run("UnknownFields_ActorSnapshotAndTag", func(t *testing.T) { + s := setup(t) + ctx := context.Background() + mustCreateAtespace(t, s, testAtespace) + snapshotRef := resources.ActorSnapshotRef{Atespace: testAtespace, Name: "snapshot-1"} + created, err := s.CreateActorSnapshot(ctx, withUnknownField(&ateapipb.ActorSnapshot{ + Metadata: withUnknownField(&ateapipb.ResourceMetadata{Atespace: testAtespace, Name: snapshotRef.Name}), + Status: withUnknownField(&ateapipb.ActorSnapshotStatus{ + SourceActor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "actor-1"}, + SnapshotUri: "gs://private/snapshot-1", + }), + })) + if err != nil { + t.Fatalf("CreateActorSnapshot failed: %v", err) + } + + got, err := s.GetActorSnapshot(ctx, snapshotRef) + if err != nil { + t.Fatalf("GetActorSnapshot failed: %v", err) + } + assertPruned(t, "GetActorSnapshot", created, got) + + list, err := s.ListActorSnapshots(ctx, testAtespace, store.ListOptions{}) + if err != nil || len(list.Items) != 1 { + t.Fatalf("ListActorSnapshots = %d items, %v; want 1", len(list.Items), err) + } + assertPruned(t, "ListActorSnapshots", created, list.Items[0]) + + tagRef := resources.ActorSnapshotTagRef{Atespace: testAtespace, Name: "production"} + createdTag, err := s.CreateActorSnapshotTag(ctx, snapshotRef, withUnknownField(&ateapipb.ActorSnapshotTag{ + Metadata: withUnknownField(&ateapipb.ResourceMetadata{Atespace: testAtespace, Name: tagRef.Name}), + Scope: ateapipb.ActorSnapshotTagScope_ACTOR_SNAPSHOT_TAG_SCOPE_ATESPACE, + })) + if err != nil { + t.Fatalf("CreateActorSnapshotTag failed: %v", err) + } + + gotTag, err := s.GetActorSnapshotTag(ctx, tagRef) + if err != nil { + t.Fatalf("GetActorSnapshotTag failed: %v", err) + } + assertPruned(t, "GetActorSnapshotTag", createdTag, gotTag) + + updatedTag, err := s.UpdateActorSnapshotTag(ctx, tagRef, store.PreconditionFrom(gotTag), func(toUpdate *ateapipb.ActorSnapshotTag) error { + requireNoUnknownFields(t, "UpdateActorSnapshotTag mutation input", toUpdate) + return nil + }) + if err != nil { + t.Fatalf("UpdateActorSnapshotTag failed: %v", err) + } + assertPruned(t, "UpdateActorSnapshotTag", createdTag, updatedTag) + + deletedTag, err := s.DeleteActorSnapshotTag(ctx, tagRef) + if err != nil { + t.Fatalf("DeleteActorSnapshotTag failed: %v", err) + } + assertPruned(t, "DeleteActorSnapshotTag", createdTag, deletedTag) + }) + + t.Run("UnknownFields_Atespace", func(t *testing.T) { + s := setup(t) + ctx := context.Background() + created, err := s.CreateAtespace(ctx, withUnknownField(&ateapipb.Atespace{ + Metadata: withUnknownField(&ateapipb.ResourceMetadata{Name: testAtespace}), + })) + if err != nil { + t.Fatalf("CreateAtespace failed: %v", err) + } + + got, err := s.GetAtespace(ctx, testAtespace) + if err != nil { + t.Fatalf("GetAtespace failed: %v", err) + } + assertPruned(t, "GetAtespace", created, got) + + list, err := s.ListAtespaces(ctx, store.ListOptions{}) + if err != nil || len(list.Items) != 1 { + t.Fatalf("ListAtespaces = %d items, %v; want 1", len(list.Items), err) + } + assertPruned(t, "ListAtespaces", created, list.Items[0]) + + deleted, err := s.DeleteAtespace(ctx, testAtespace) + if err != nil { + t.Fatalf("DeleteAtespace failed: %v", err) + } + assertPruned(t, "DeleteAtespace", created, deleted) + }) + + t.Run("UnknownFields_ActorTemplate", func(t *testing.T) { + s := setup(t) + ctx := context.Background() + mustCreateAtespace(t, s, testAtespace) + template := withUnknownField(newTestActorTemplate(testAtespace, "tmpl-1")) + withUnknownField(template.Metadata) + withUnknownField(template.Containers[0]) + created, err := s.CreateActorTemplate(ctx, template) + if err != nil { + t.Fatalf("CreateActorTemplate failed: %v", err) + } + ref := resources.ActorTemplateRef{Atespace: testAtespace, Name: "tmpl-1"} + + got, err := s.GetActorTemplate(ctx, ref) + if err != nil { + t.Fatalf("GetActorTemplate failed: %v", err) + } + assertPruned(t, "GetActorTemplate", created, got) + + list, err := s.ListActorTemplates(ctx, testAtespace, store.ListOptions{}) + if err != nil || len(list.Items) != 1 { + t.Fatalf("ListActorTemplates = %d items, %v; want 1", len(list.Items), err) + } + assertPruned(t, "ListActorTemplates", created, list.Items[0]) + + updated, err := s.UpdateActorTemplate(ctx, ref, store.PreconditionFrom(got), func(toUpdate *ateapipb.ActorTemplate) error { + requireNoUnknownFields(t, "UpdateActorTemplate mutation input", toUpdate) + return nil + }) + if err != nil { + t.Fatalf("UpdateActorTemplate failed: %v", err) + } + assertPruned(t, "UpdateActorTemplate", created, updated) + + deleted, err := s.DeleteActorTemplate(ctx, ref) + if err != nil { + t.Fatalf("DeleteActorTemplate failed: %v", err) + } + assertPruned(t, "DeleteActorTemplate", created, deleted) + }) + + t.Run("UnknownFields_Worker", func(t *testing.T) { + s := setup(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + watch, err := s.WatchWorkers(ctx) + if err != nil { + t.Fatalf("WatchWorkers failed: %v", err) + } + defer watch.Close() + + worker := withUnknownField(newTestWorker(testWorkerName, "pod-1")) + withUnknownField(worker.Metadata) + withUnknownField(worker.Capacity) + created, err := s.CreateWorker(ctx, worker) + if err != nil { + t.Fatalf("CreateWorker failed: %v", err) + } + event := receiveEvent(t, watch.Events) + if event.Type != store.WorkerEventCreated { + t.Fatalf("event type = %v, want WorkerEventCreated", event.Type) + } + assertPruned(t, "WatchWorkers created event", created, event.Worker) + + got, err := s.GetWorker(ctx, testWorkerName) + if err != nil { + t.Fatalf("GetWorker failed: %v", err) + } + assertPruned(t, "GetWorker", created, got) + + list, err := s.ListWorkers(ctx, store.ListOptions{}) + if err != nil || len(list.Items) != 1 { + t.Fatalf("ListWorkers = %d items, %v; want 1", len(list.Items), err) + } + assertPruned(t, "ListWorkers", created, list.Items[0]) + + updated, err := s.UpdateWorker(ctx, testWorkerName, store.PreconditionFrom(got), func(toUpdate *ateapipb.Worker) error { + requireNoUnknownFields(t, "UpdateWorker mutation input", toUpdate) + return nil + }) + if err != nil { + t.Fatalf("UpdateWorker failed: %v", err) + } + assertPruned(t, "UpdateWorker", created, updated) + + deleted, err := s.DeleteWorker(ctx, testWorkerName, store.DeletePreconditions{}) + if err != nil { + t.Fatalf("DeleteWorker failed: %v", err) + } + assertPruned(t, "DeleteWorker", created, deleted) + }) +}