DeforaNetworks/corteza
DeforaNetworks/corteza
3
0
Fork 0
Code

Envoy store encode improvements

Browse Source 
* Add default createdBy,
* improve ComposeRecord xreferencing,
* fix ComposeRecord self references.
This commit is contained in:
Tomaž Jerman 2021-04-12 17:16:28 +02:00
parent 419ebd8178
commit beca3c1e9c
18 changed files with 662 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
pkg/envoy/resource/compose_module.go
View File

@ -81,6 +81,22 @@ func FindComposeModule(rr InterfaceSet, ii Identifiers) (ns *types.Module) {
return nil
}
func FindComposeModuleResource(rr InterfaceSet, ii Identifiers) (mod *ComposeModule) {
rr.Walk(func(r Interface) error {
mr, ok := r.(*ComposeModule)
if !ok {
return nil
}
if mr.Identifiers().HasAny(ii) {
mod = mr
}
return nil
})
return mod
}
func ComposeModuleErrUnresolved(ii Identifiers) error {
return fmt.Errorf("compose module unresolved %v", ii.StringSlice())
}

24
pkg/envoy/resource/compose_record.go
View File

@ -39,8 +39,7 @@ type (
RefMod *Ref
RelMod *types.Module
IDMap map[string]uint64
RecMap map[string]*types.Record
IDMap map[string]uint64
// UserFlakes help the system by predefining a set of potential sys user references.
// This should make the operation cheaper for larger datasets.
UserFlakes UserstampIndex
@ -49,9 +48,8 @@ type (
func NewComposeRecordSet(w CrsWalker, nsRef, modRef string) *ComposeRecord {
r := &ComposeRecord{
base: &base{},
IDMap: make(map[string]uint64),
RecMap: make(map[string]*types.Record),
base: &base{},
IDMap: make(map[string]uint64),
}
r.SetResourceType(COMPOSE_RECORD_RESOURCE_TYPE)
@ -72,6 +70,22 @@ func (r *ComposeRecord) SetUserFlakes(uu UserstampIndex) {
r.AddRef(USER_RESOURCE_TYPE, "*")
}
func FindComposeRecordResource(rr InterfaceSet, ii Identifiers) (rec *ComposeRecord) {
rr.Walk(func(r Interface) error {
crr, ok := r.(*ComposeRecord)
if !ok {
return nil
}
if crr.Identifiers().HasAny(ii) {
rec = crr
}
return nil
})
return rec
}
func ComposeRecordErrUnresolved(ii Identifiers) error {
return fmt.Errorf("compose record unresolved %v", ii.StringSlice())
}

2
pkg/envoy/store/automation_workflow_marshal.go
View File

@ -99,6 +99,7 @@ func (n *automationWorkflow) encodeWorkflow(ctx context.Context, pl *payload) (e
}
}
res.CreatedBy = pl.invokerID
if us != nil {
if us.OwnedBy != nil {
res.OwnedBy = us.OwnedBy.UserID
@ -191,6 +192,7 @@ func (n *automationWorkflow) encodeTriggers(ctx context.Context, pl *payload) (e
res.DeletedAt = ts.DeletedAt.T
}
}
res.CreatedBy = pl.invokerID
if us != nil {
if us.OwnedBy != nil {
res.OwnedBy = us.OwnedBy.UserID

2
pkg/envoy/store/compose_module_marshal.go
View File

@ -29,6 +29,8 @@ func (n *composeModule) Prepare(ctx context.Context, pl *payload) (err error) {
return resource.ComposeNamespaceErrUnresolved(n.res.RefNs.Identifiers)
}
n.res.Res.NamespaceID = n.relNS.ID
// Get related record field modules
for _, refMod := range n.res.RefMods {
var mod *types.Module

5
pkg/envoy/store/compose_record.go
View File

@ -24,6 +24,11 @@ type (
relNS *types.Namespace
relMod *types.Module
fieldModRef map[string]resource.Identifiers
// module identifier -> record identifier -> recordID
externalRef map[string]map[string]uint64
recMap map[string]*types.Record
// Little helper flag for conditional encoding
missing bool
}

149
pkg/envoy/store/compose_record_marshal.go
View File

@ -2,7 +2,6 @@ package store
import (
"context"
"errors"
"fmt"
"strconv"
"time"
@ -24,7 +23,10 @@ var (
func NewComposeRecordFromResource(res *resource.ComposeRecord, cfg *EncoderConfig) resourceState {
return &composeRecord{
cfg: cfg,
cfg: cfg,
fieldModRef: make(map[string]resource.Identifiers),
externalRef: make(map[string]map[string]uint64),
recMap: make(map[string]*types.Record),
res: res,
}
@ -77,6 +79,7 @@ func (n *composeRecord) Prepare(ctx context.Context, pl *payload) (err error) {
}
// Add missing refs
preloadRefs := make(resource.RefSet, 0, int(len(n.relMod.Fields)/2)+1)
for _, f := range n.relMod.Fields {
switch f.Kind {
case "Record":
@ -86,49 +89,79 @@ func (n *composeRecord) Prepare(ctx context.Context, pl *payload) (err error) {
}
if refM != "" && refM != "0" {
// Make a reference with that module's records
n.res.AddRef(resource.COMPOSE_RECORD_RESOURCE_TYPE, refM).Constraint(n.res.RefNs)
ref := n.res.AddRef(resource.COMPOSE_RECORD_RESOURCE_TYPE, refM).Constraint(n.res.RefNs)
n.fieldModRef[f.Name] = ref.Identifiers
preloadRefs = append(preloadRefs, ref)
}
}
}
// Can't do anything else, since the NS doesn't yet exist
if n.relNS.ID <= 0 {
if n.relNS.ID == 0 {
return nil
}
// Check if empty
// Preload potential references
//
// This is a fairly primitive approach, try to think of something a bit nicer
for _, ref := range preloadRefs {
mod, err := findComposeModuleStore(ctx, pl.s, n.relNS.ID, makeGenericFilter(ref.Identifiers))
if err != nil && err != store.ErrNotFound {
return err
}
auxMap := make(map[string]uint64)
for i := range ref.Identifiers {
n.externalRef[i] = auxMap
}
// Preload all records
rr, _, err := store.SearchComposeRecords(ctx, pl.s, mod, types.RecordFilter{
ModuleID: mod.ID,
NamespaceID: mod.NamespaceID,
Paging: filter.Paging{
Limit: 0,
},
})
if err != nil {
return err
}
for _, r := range rr {
auxMap[strconv.FormatUint(r.ID, 10)] = r.ID
}
}
// Can't work with own record because the module doesn't yet exist
if n.relMod.ID == 0 {
return nil
}
// Preload own records
rr, _, err := store.SearchComposeRecords(ctx, pl.s, n.relMod, types.RecordFilter{
ModuleID: n.relMod.ID,
NamespaceID: n.relNS.ID,
Paging: filter.Paging{Limit: 1},
Paging: filter.Paging{
Limit: 0,
},
})
if err != nil && err != store.ErrNotFound {
return err
}
n.missing = len(rr) == 0
// Try to get existing records
//
// @todo handle large amounts of
for rID := range n.res.IDMap {
var r *types.Record
// @todo support for labels
if refy.MatchString(rID) {
id, _ := strconv.ParseUint(rID, 10, 64)
r, err = store.LookupComposeRecordByID(ctx, pl.s, n.relMod, id)
if err == store.ErrNotFound {
continue
} else if err != nil {
return err
}
if r != nil {
n.res.RecMap[rID] = r
}
} else {
continue
}
// Map existing records so we can perform updates
// Map to xref map for easier use later
auxMap := make(map[string]uint64)
for i := range n.res.RefMod.Identifiers {
n.externalRef[i] = auxMap
}
for _, r := range rr {
key := strconv.FormatUint(r.ID, 10)
n.recMap[key] = r
n.res.RecMap[rID] = r
// Map IDs to xref map
auxMap[key] = r.ID
}
return nil
@ -180,7 +213,7 @@ func (n *composeRecord) Encode(ctx context.Context, pl *payload) (err error) {
}
// Some pointing
rm := n.res.RecMap
rm := n.recMap
im := n.res.IDMap
createAcChecked := false
@ -194,6 +227,22 @@ func (n *composeRecord) Encode(ctx context.Context, pl *payload) (err error) {
return k
}
checkXRef := func(ii resource.Identifiers, ref string) (uint64, error) {
var auxMap map[string]uint64
for ri := range ii {
if mp, ok := n.externalRef[ri]; ok {
auxMap = mp
break
}
}
if auxMap == nil || len(auxMap) == 0 {
return 0, fmt.Errorf("referenced record not resolved: %s", resource.ComposeRecordErrUnresolved(resource.MakeIdentifiers(ref)))
}
return auxMap[ref], nil
}
i := -1
return n.res.Walker(func(r *resource.ComposeRecordRaw) error {
i++
@ -283,6 +332,7 @@ func (n *composeRecord) Encode(ctx context.Context, pl *payload) (err error) {
}
// Userstamps
rec.CreatedBy = pl.invokerID
if r.Us != nil {
if r.Us.CreatedBy != nil {
rec.CreatedBy = ux[r.Us.CreatedBy.Ref]
@ -308,7 +358,7 @@ func (n *composeRecord) Encode(ctx context.Context, pl *payload) (err error) {
}
f := mod.Fields.FindByName(k)
if f != nil {
if f != nil && v != "" {
switch f.Kind {
case "User":
uID := ux[v]
@ -319,9 +369,24 @@ func (n *composeRecord) Encode(ctx context.Context, pl *payload) (err error) {
rv.Ref = uID
case "Record":
refIdentifiers, ok := n.fieldModRef[f.Name]
if !ok {
return fmt.Errorf("module field record reference not resoled: %s", f.Name)
}
// if self...
if n.res.RefMod.Identifiers.HasAny(resource.MakeIdentifiers(f.Options.String("module"))) {
if n.res.RefMod.Identifiers.HasAny(refIdentifiers) {
rID := im[v]
// Check if its in the store
if rID == 0 {
// Check if we have an xref
rID, err = checkXRef(refIdentifiers, v)
if err != nil {
return err
}
}
if rID == 0 {
return resource.ComposeRecordErrUnresolved(resource.MakeIdentifiers(v))
}
@ -329,8 +394,28 @@ func (n *composeRecord) Encode(ctx context.Context, pl *payload) (err error) {
rv.Ref = rID
} else {
// not self...
// @todo...
return errors.New("record cross referencing not yet supported")
rID := uint64(0)
refRes := resource.FindComposeRecordResource(pl.state.ParentResources, refIdentifiers)
if refRes != nil {
// check if parent has it
rID = refRes.IDMap[v]
}
if rID == 0 {
// Check if we have an xref
rID, err = checkXRef(refIdentifiers, v)
if err != nil {
return err
}
}
if rID == 0 {
return fmt.Errorf("referenced record not resolved: %s", resource.ComposeRecordErrUnresolved(resource.MakeIdentifiers(v)))
}
rv.Value = strconv.FormatUint(rID, 10)
rv.Ref = rID
}
}
}

9
pkg/envoy/store/encoder.go
View File

@ -8,6 +8,7 @@ import (
"github.com/cortezaproject/corteza-server/compose/service"
"github.com/cortezaproject/corteza-server/compose/types"
"github.com/cortezaproject/corteza-server/pkg/auth"
"github.com/cortezaproject/corteza-server/pkg/envoy"
"github.com/cortezaproject/corteza-server/pkg/envoy/resource"
"github.com/cortezaproject/corteza-server/pkg/rbac"
@ -66,6 +67,7 @@ type (
composeAccessControl composeAccessController
systemAC accessControlRBACServicer
invokerID uint64
}
// resourceState allows each conforming struct to be initialized and encoded
@ -106,7 +108,7 @@ func NewStoreEncoder(s store.Storer, cfg *EncoderConfig) envoy.PrepareEncoder {
// It initializes and prepares the resource state for each provided resource
func (se *storeEncoder) Prepare(ctx context.Context, ee ...*envoy.ResourceState) (err error) {
f := func(rs resourceState, ers *envoy.ResourceState) error {
err = rs.Prepare(ctx, se.makePayload(ers))
err = rs.Prepare(ctx, se.makePayload(ctx, ers))
if err != nil {
return err
}
@ -176,7 +178,7 @@ func (se *storeEncoder) Encode(ctx context.Context, p envoy.Provider) error {
if state == nil {
err = ErrResourceStateUndefined
} else {
err = state.Encode(ctx, se.makePayload(ers))
err = state.Encode(ctx, se.makePayload(ctx, ers))
}
if err != nil {
@ -186,11 +188,12 @@ func (se *storeEncoder) Encode(ctx context.Context, p envoy.Provider) error {
})
}
func (se *storeEncoder) makePayload(ers *envoy.ResourceState) *payload {
func (se *storeEncoder) makePayload(ctx context.Context, ers *envoy.ResourceState) *payload {
return &payload{
s: se.s,
state: ers,
composeAccessControl: service.AccessControl(rbac.Global()),
invokerID: auth.GetIdentityFromContext(ctx).Identity(),
}
}

356
tests/envoy/data_shaping_test.go
View File

@ -105,7 +105,7 @@ func TestDataShaping_fieldTypes(t *testing.T) {
}
for _, c := range cases {
t.Run(fmt.Sprintf("record shaping; data_shaping_field types/%s", c), func(t *testing.T) {
t.Run(fmt.Sprintf("record shaping; data_shaping_field_types/%s", c), func(t *testing.T) {
var (
req = require.New(t)
)
@ -182,7 +182,7 @@ func TestDataShaping_refs(t *testing.T) {
}
for _, c := range cases {
t.Run(fmt.Sprintf("record shaping; data_shaping_field types/%s", c), func(t *testing.T) {
t.Run(fmt.Sprintf("record shaping; data_shaping_refs/%s", c), func(t *testing.T) {
var (
req = require.New(t)
)
@ -251,3 +251,355 @@ func TestDataShaping_refs(t *testing.T) {
})
}
}
func TestDataShaping_xrefsPeer(t *testing.T) {
var (
ctx = auth.SetSuperUserContext(context.Background())
s = initStore(ctx, t)
err error
cases = []string{
"csv_xrefs",
}
)
ni := uint64(10)
su.NextID = func() uint64 {
ni++
return ni
}
for _, c := range cases {
t.Run(fmt.Sprintf("record shaping; data_shaping_xrefs_peer/%s", c), func(t *testing.T) {
var (
req = require.New(t)
)
truncateStore(ctx, s, t)
err = collect(
err,
storeRole(ctx, s, 1, "everyone"),
storeRole(ctx, s, 2, "admins"),
)
if err != nil {
t.Fatal(err.Error())
}
nn, err := decodeDirectory(ctx, path.Join("data_shaping", c))
req.NoError(err)
crs := resource.ComposeRecordShaper()
nn, err = resource.Shape(nn, crs)
req.NoError(err)
req.NoError(encode(ctx, s, nn))
ns, err := store.LookupComposeNamespaceBySlug(ctx, s, "ns1")
req.NotNil(ns)
m, err := loadComposeModuleFull(ctx, s, req, ns.ID, "mod1")
req.NotNil(m)
refM, err := loadComposeModuleFull(ctx, s, req, ns.ID, "mod2")
req.NotNil(refM)
rr, _, err := store.SearchComposeRecords(ctx, s, m, types.RecordFilter{})
req.NoError(err)
req.Len(rr, 4)
refRR, _, err := store.SearchComposeRecords(ctx, s, refM, types.RecordFilter{})
req.NoError(err)
req.Len(refRR, 4)
r1 := rr[0]
r2 := rr[1]
r3 := rr[2]
r4 := rr[3]
refR1 := refRR[0]
refR2 := refRR[1]
refR3 := refRR[2]
refR4 := refRR[3]
req.Len(r1.Values, 1)
req.Equal(strconv.FormatUint(refR1.ID, 10), r1.Values.Get("f_record", 0).Value)
req.Equal(refR1.ID, r1.Values.Get("f_record", 0).Ref)
req.Len(r2.Values, 1)
req.Equal(strconv.FormatUint(refR2.ID, 10), r2.Values.Get("f_record", 0).Value)
req.Equal(refR2.ID, r2.Values.Get("f_record", 0).Ref)
req.Len(r3.Values, 1)
req.Equal(strconv.FormatUint(refR3.ID, 10), r3.Values.Get("f_record", 0).Value)
req.Equal(refR3.ID, r3.Values.Get("f_record", 0).Ref)
req.Len(r4.Values, 1)
req.Equal(strconv.FormatUint(refR4.ID, 10), r4.Values.Get("f_record", 0).Value)
req.Equal(refR4.ID, r4.Values.Get("f_record", 0).Ref)
s.TruncateComposeRecords(ctx, m)
})
}
}
func TestDataShaping_xrefsStore(t *testing.T) {
var (
ctx = auth.SetSuperUserContext(context.Background())
s = initStore(ctx, t)
err error
cases = []string{
"csv_xrefs_store",
}
)
ni := uint64(10)
su.NextID = func() uint64 {
ni++
return ni
}
for _, c := range cases {
t.Run(fmt.Sprintf("record shaping; data_shaping_xrefs_store/%s", c), func(t *testing.T) {
var (
req = require.New(t)
)
truncateStore(ctx, s, t)
err = collect(
err,
storeRole(ctx, s, 1, "everyone"),
storeRole(ctx, s, 2, "admins"),
storeComposeNamespace(ctx, s, 1001, "ns1"),
storeComposeModule(ctx, s, 1001, 2001, "mod_ref"),
storeComposeModuleField(ctx, s, 2001, 2101, "label"),
storeComposeRecord(ctx, s, 1001, 2001, 3001, "label"),
storeComposeRecord(ctx, s, 1001, 2001, 3002, "label"),
)
if err != nil {
t.Fatal(err.Error())
}
nn, err := decodeDirectory(ctx, path.Join("data_shaping", c))
req.NoError(err)
crs := resource.ComposeRecordShaper()
nn, err = resource.Shape(nn, crs)
req.NoError(err)
req.NoError(encode(ctx, s, nn))
ns, err := store.LookupComposeNamespaceBySlug(ctx, s, "ns1")
req.NotNil(ns)
m, err := loadComposeModuleFull(ctx, s, req, ns.ID, "mod1")
req.NotNil(m)
refM, err := loadComposeModuleFull(ctx, s, req, ns.ID, "mod_ref")
req.NotNil(refM)
rr, _, err := store.SearchComposeRecords(ctx, s, m, types.RecordFilter{})
req.NoError(err)
req.Len(rr, 4)
refRR, _, err := store.SearchComposeRecords(ctx, s, refM, types.RecordFilter{})
req.NoError(err)
req.Len(refRR, 2)
r1 := rr[0]
r2 := rr[1]
r3 := rr[2]
r4 := rr[3]
refR1 := refRR[0]
refR2 := refRR[1]
req.Len(r1.Values, 1)
req.Equal(strconv.FormatUint(refR1.ID, 10), r1.Values.Get("f_record", 0).Value)
req.Equal(refR1.ID, r1.Values.Get("f_record", 0).Ref)
req.Len(r2.Values, 1)
req.Equal(strconv.FormatUint(refR2.ID, 10), r2.Values.Get("f_record", 0).Value)
req.Equal(refR2.ID, r2.Values.Get("f_record", 0).Ref)
req.Len(r3.Values, 1)
req.Equal(strconv.FormatUint(refR1.ID, 10), r3.Values.Get("f_record", 0).Value)
req.Equal(refR1.ID, r3.Values.Get("f_record", 0).Ref)
req.Len(r4.Values, 1)
req.Equal(strconv.FormatUint(refR2.ID, 10), r4.Values.Get("f_record", 0).Value)
req.Equal(refR2.ID, r4.Values.Get("f_record", 0).Ref)
s.TruncateComposeRecords(ctx, m)
})
}
}
func TestDataShaping_xrefsMix(t *testing.T) {
var (
ctx = auth.SetSuperUserContext(context.Background())
s = initStore(ctx, t)
err error
cases = []string{
"csv_xrefs_mix",
}
)
ni := uint64(10)
su.NextID = func() uint64 {
ni++
return ni
}
for _, c := range cases {
t.Run(fmt.Sprintf("record shaping; data_shaping_xrefs_mix/%s", c), func(t *testing.T) {
var (
req = require.New(t)
)
truncateStore(ctx, s, t)
err = collect(
err,
storeRole(ctx, s, 1, "everyone"),
storeRole(ctx, s, 2, "admins"),
storeComposeNamespace(ctx, s, 1001, "ns1"),
storeComposeModule(ctx, s, 1001, 2001, "mod1"),
storeComposeModuleField(ctx, s, 2001, 2101, "f_label"),
storeComposeRecord(ctx, s, 1001, 2001, 3001, "f_label"),
storeComposeModule(ctx, s, 1001, 2002, "mod2"),
storeComposeModuleField(ctx, s, 2002, 2201, "f_label"),
storeComposeRecord(ctx, s, 1001, 2002, 3101, "f_label"),
storeComposeRecord(ctx, s, 1001, 2002, 3102, "f_label"),
)
if err != nil {
t.Fatal(err.Error())
}
nn, err := decodeDirectory(ctx, path.Join("data_shaping", c))
req.NoError(err)
crs := resource.ComposeRecordShaper()
nn, err = resource.Shape(nn, crs)
req.NoError(err)
req.NoError(encode(ctx, s, nn))
ns, err := store.LookupComposeNamespaceBySlug(ctx, s, "ns1")
req.NotNil(ns)
mod1, err := loadComposeModuleFull(ctx, s, req, ns.ID, "mod1")
req.NotNil(mod1)
mod2, err := loadComposeModuleFull(ctx, s, req, ns.ID, "mod2")
req.NotNil(mod2)
rr1, _, err := store.SearchComposeRecords(ctx, s, mod1, types.RecordFilter{})
req.NoError(err)
req.Len(rr1, 5)
rr2, _, err := store.SearchComposeRecords(ctx, s, mod2, types.RecordFilter{})
req.NoError(err)
req.Len(rr2, 4)
r1 := rr1[0]
r2 := rr1[1]
r3 := rr1[2]
r4 := rr1[3]
refStoreSelf := rr1[4]
refStoreR1 := rr2[2]
refStoreR2 := rr2[3]
refCSVR1 := rr2[0]
req.Len(r1.Values, 2)
req.Equal(strconv.FormatUint(refCSVR1.ID, 10), r1.Values.Get("f_record", 0).Value)
req.Equal(refCSVR1.ID, r1.Values.Get("f_record", 0).Ref)
req.Len(r2.Values, 2)
req.Equal(strconv.FormatUint(refStoreSelf.ID, 10), r2.Values.Get("f_record_self", 0).Value)
req.Equal(refStoreSelf.ID, r2.Values.Get("f_record_self", 0).Ref)
req.Len(r3.Values, 2)
req.Equal(strconv.FormatUint(refStoreR1.ID, 10), r3.Values.Get("f_record", 0).Value)
req.Equal(refStoreR1.ID, r3.Values.Get("f_record", 0).Ref)
req.Len(r4.Values, 2)
req.Equal(strconv.FormatUint(refStoreR2.ID, 10), r4.Values.Get("f_record", 0).Value)
req.Equal(refStoreR2.ID, r4.Values.Get("f_record", 0).Ref)
s.TruncateComposeRecords(ctx, mod1)
})
}
}
func TestDataShaping_update(t *testing.T) {
var (
ctx = auth.SetSuperUserContext(context.Background())
s = initStore(ctx, t)
err error
cases = []string{
"csv_update",
}
)
ni := uint64(10)
su.NextID = func() uint64 {
ni++
return ni
}
for _, c := range cases {
t.Run(fmt.Sprintf("record shaping; data_shaping_update/%s", c), func(t *testing.T) {
var (
req = require.New(t)
)
truncateStore(ctx, s, t)
err = collect(
err,
storeRole(ctx, s, 1, "everyone"),
storeRole(ctx, s, 2, "admins"),
storeComposeNamespace(ctx, s, 1001, "ns1"),
storeComposeModule(ctx, s, 1001, 2001, "mod1"),
storeComposeModuleField(ctx, s, 2001, 2101, "f_label"),
storeComposeRecord(ctx, s, 1001, 2001, 3001, "f_label"),
)
if err != nil {
t.Fatal(err.Error())
}
nn, err := decodeDirectory(ctx, path.Join("data_shaping", c))
req.NoError(err)
crs := resource.ComposeRecordShaper()
nn, err = resource.Shape(nn, crs)
req.NoError(err)
req.NoError(encode(ctx, s, nn))
ns, err := store.LookupComposeNamespaceBySlug(ctx, s, "ns1")
req.NotNil(ns)
mod1, err := loadComposeModuleFull(ctx, s, req, ns.ID, "mod1")
req.NotNil(mod1)
rr, _, err := store.SearchComposeRecords(ctx, s, mod1, types.RecordFilter{})
req.NoError(err)
req.Len(rr, 2)
r1 := rr[0]
r2 := rr[1]
req.Len(r1.Values, 1)
req.Equal("created", r1.Values.Get("f_label", 0).Value)
req.Len(r2.Values, 1)
req.Equal("updated", r2.Values.Get("f_label", 0).Value)
s.TruncateComposeRecords(ctx, mod1)
})
}
}

18
tests/envoy/testdata/data_shaping/csv_update/1100_modules.yaml vendored Normal file
View File

@ -0,0 +1,18 @@
namespaces:
ns1:
name: ns1 name
modules:
mod1:
fields:
f_label:
label: f_label label
kind: String
records:
source: mod1.csv
key: id
mapping:
id: /
c_label:
field: f_label

3
tests/envoy/testdata/data_shaping/csv_update/mod1.csv vendored Normal file
View File

@ -0,0 +1,3 @@
id,c_label
101,created
3001,updated
1 id c_label
2 101 created
3 3001 updated

33
tests/envoy/testdata/data_shaping/csv_xrefs/1100_modules.yaml vendored Normal file
View File

@ -0,0 +1,33 @@
namespaces:
ns1:
name: ns1 name
modules:
mod1:
fields:
f_record:
label: f_record label
kind: Record
options:
module: mod2
records:
source: mod1.csv
key: id
mapping:
id: /
c_record:
field: f_record
mod2:
fields:
f_label:
label: label label
kind: String
records:
source: mod2.csv
key: id
mapping:
id: /
c_label:
field: f_label

5
tests/envoy/testdata/data_shaping/csv_xrefs/mod1.csv vendored Normal file
View File

@ -0,0 +1,5 @@
id,c_record
101,201
102,202
103,103
104,104
1 id c_record
2 101 201
3 102 202
4 103 103
5 104 104

5
tests/envoy/testdata/data_shaping/csv_xrefs/mod2.csv vendored Normal file
View File

@ -0,0 +1,5 @@
id,c_label
201,record 1
202,record 2
103,record 3
104,record 4
1 id c_label
2 201 record 1
3 202 record 2
4 103 record 3
5 104 record 4

44
tests/envoy/testdata/data_shaping/csv_xrefs_mix/1100_modules.yaml vendored Normal file
View File

@ -0,0 +1,44 @@
namespaces:
ns1:
name: ns1 name
modules:
mod1:
fields:
f_record:
label: f_record label
kind: Record
options:
module: mod2
f_record_self:
label: f_record_self label
kind: Record
options:
module: mod1
f_label:
label: f_label label
kind: String
records:
source: mod1.csv
key: id
mapping:
id: /
c_record:
field: f_record
c_record_self:
field: f_record_self
c_label:
field: f_label
mod2:
fields:
f_label:
label: f_label label
kind: String
records:
source: mod2.csv
key: id
mapping:
id: /
c_label:
field: f_label

5
tests/envoy/testdata/data_shaping/csv_xrefs_mix/mod1.csv vendored Normal file
View File

@ -0,0 +1,5 @@
id,c_record,c_record_self
101,201,
102,,3001
103,3101,
104,3102,
1 id c_record c_record_self
2 101 201
3 102 3001
4 103 3101
5 104 3102

3
tests/envoy/testdata/data_shaping/csv_xrefs_mix/mod2.csv vendored Normal file
View File

@ -0,0 +1,3 @@
id,c_label
201,record 1
202,record 2
1 id c_label
2 201 record 1
3 202 record 2

20
tests/envoy/testdata/data_shaping/csv_xrefs_store/1100_modules.yaml vendored Normal file
View File

@ -0,0 +1,20 @@
namespaces:
ns1:
name: ns1 name
modules:
mod1:
fields:
f_record:
label: f_record label
kind: Record
options:
module: mod_ref
records:
source: mod1.csv
key: id
mapping:
id: /
c_record:
field: f_record

5
tests/envoy/testdata/data_shaping/csv_xrefs_store/mod1.csv vendored Normal file
View File

@ -0,0 +1,5 @@
id,c_record
101,3001
102,3002
103,3001
104,3002
1 id c_record
2 101 3001
3 102 3002
4 103 3001
5 104 3002