mirror of
https://github.com/agent-substrate/substrate.git
synced 2026-11-12 10:35:06 +08:00
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
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user