DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Restructure compose/record/report service for easier tests

Browse Source
This commit is contained in:
Tomaž Jerman
2022-09-19 16:22:37 +02:00
parent cf769c2485
commit 95f836f248
4 changed files with 108 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
compose/service/record.go
+63 -59
View File
@@ -304,65 +304,7 @@ func (svc record) Report(ctx context.Context, namespaceID, moduleID uint64, metr
return RecordErrNotAllowedToSearch()
}
// Map dimension to the aggregate group
// @note we only ever used a single dimension so this is ok
dim := []dal.AttributeMapping{
dal.SimpleAttr{
Ident: "dimension_0",
Expr: dimensions,
},
}
// Map metrics to the aggregate attrs
// - count is always present
mms := []dal.AttributeMapping{
dal.SimpleAttr{
Ident: "count",
Expr: "count(ID)",
Props: dal.MapProperties{
Type: dal.TypeNumber{},
},
},
}
// - other requested metrices
if len(metrics) > 0 {
pts := strings.Split(metrics, " AS ")
expr := strings.TrimSpace(pts[0])
ident := expr
if len(pts) > 1 {
ident = strings.TrimSpace(pts[1])
}
mms = append(mms, dal.SimpleAttr{
Ident: ident,
Expr: expr,
Props: dal.MapProperties{
Type: dal.TypeNumber{},
},
})
}
// Build the pipeline
pp := dal.Pipeline{
&dal.Datasource{
Ident: "ds",
Filter: filter.Generic(filter.WithExpression(f)),
ModelRef: dal.ModelRef{
ConnectionID: m.Config.DAL.ConnectionID,
ResourceID: m.ID,
ResourceType: types.ModuleResourceType,
},
},
&dal.Aggregate{
Ident: "agg",
RelSource: "ds",
Group: dim,
OutAttributes: mms,
},
}
err = pp.LinkSteps()
pp, err := recordReportToDalPipeline(m, metrics, dimensions, f)
if err != nil {
return err
}
@@ -1753,6 +1695,68 @@ func loadRecord(ctx context.Context, s store.Storer, namespaceID, moduleID, reco
return
}
func recordReportToDalPipeline(m *types.Module, metrics, dimensions, f string) (pp dal.Pipeline, err error) {
// Map dimension to the aggregate group
// @note we only ever used a single dimension so this is ok
dim := []dal.AttributeMapping{
dal.SimpleAttr{
Ident: "dimension_0",
Expr: dimensions,
},
}
// Map metrics to the aggregate attrs
// - count is always present
mms := []dal.AttributeMapping{
dal.SimpleAttr{
Ident: "count",
Expr: "count(ID)",
Props: dal.MapProperties{
Type: dal.TypeNumber{},
},
},
}
// - other requested metrices
if len(metrics) > 0 {
pts := strings.Split(metrics, " AS ")
expr := strings.TrimSpace(pts[0])
ident := expr
if len(pts) > 1 {
ident = strings.TrimSpace(pts[1])
}
mms = append(mms, dal.SimpleAttr{
Ident: ident,
Expr: expr,
Props: dal.MapProperties{
Type: dal.TypeNumber{},
},
})
}
// Build the pipeline
pp = dal.Pipeline{
&dal.Datasource{
Ident: "ds",
Filter: filter.Generic(filter.WithExpression(f)),
ModelRef: dal.ModelRef{
ConnectionID: m.Config.DAL.ConnectionID,
ResourceID: m.ID,
ResourceType: types.ModuleResourceType,
},
},
&dal.Aggregate{
Ident: "agg",
RelSource: "ds",
Group: dim,
OutAttributes: mms,
},
}
return pp, pp.LinkSteps()
}
func (ei ErrorIndex) Add(err string) {
if _, has := ei[err]; has {
ei[err]++
compose/service/record_test.go
+42
View File
@@ -1014,3 +1014,45 @@ func TestSetRecordOwner(t *testing.T) {
req.Equal(upd.OwnedBy, invoker.ID)
})
}
func TestRecordReportToDalPipeline(t *testing.T) {
mod := &types.Module{
ID: 10,
NamespaceID: 42,
}
t.Run("no additional metrics", func(t *testing.T) {
pp, err := recordReportToDalPipeline(mod, "", "created_at", "")
require.NoError(t, err)
require.Len(t, pp, 2)
agg := pp[1].(*dal.Aggregate)
require.Len(t, agg.OutAttributes, 1)
require.Equal(t, "count", agg.OutAttributes[0].Identifier())
})
t.Run("additional metrics with alias", func(t *testing.T) {
pp, err := recordReportToDalPipeline(mod, "MAX(numbers) AS something", "created_at", "")
require.NoError(t, err)
require.Len(t, pp, 2)
agg := pp[1].(*dal.Aggregate)
require.Len(t, agg.OutAttributes, 2)
require.Equal(t, "something", agg.OutAttributes[1].Identifier())
require.Equal(t, "MAX(numbers)", agg.OutAttributes[1].Expression())
})
t.Run("additional metrics without alias", func(t *testing.T) {
pp, err := recordReportToDalPipeline(mod, "MAX(numbers)", "created_at", "")
require.NoError(t, err)
require.Len(t, pp, 2)
agg := pp[1].(*dal.Aggregate)
require.Len(t, agg.OutAttributes, 2)
require.Equal(t, "MAX(numbers)", agg.OutAttributes[1].Identifier())
require.Equal(t, "MAX(numbers)", agg.OutAttributes[1].Expression())
})
}
system/rest/report.go
+1 -1
View File
@@ -28,7 +28,7 @@ type (
Delete(ctx context.Context, ID uint64) (err error)
Undelete(ctx context.Context, ID uint64) (err error)
Run(ctx context.Context, ID uint64, dd reporting.FrameDefinitionSet) (rr []*reporting.Frame, err error)
Describe(ctx context.Context, src types.ReportDataSourceSet, st types.ReportStepSet, sources ...string) (out reporting.[]FrameDescription, err error)
Describe(ctx context.Context, src types.ReportDataSourceSet, st types.ReportStepSet, sources ...string) (out []reporting.FrameDescription, err error)
}
reportAccessController interface {
system/service/report.go
+2 -2
View File
@@ -293,8 +293,8 @@ func (svc *report) Undelete(ctx context.Context, ID uint64) (err error) {
}
// @todo actionlog?
func (svc *report) Describe(ctx context.Context, src types.ReportDataSourceSet, st types.ReportStepSet, sources ...string) (out reporting.[]FrameDescription, err error) {
out = make(reporting.[]FrameDescription, 0, len(sources)*2)
func (svc *report) Describe(ctx context.Context, src types.ReportDataSourceSet, st types.ReportStepSet, sources ...string) (out []reporting.FrameDescription, err error) {
out = make([]reporting.FrameDescription, 0, len(sources)*2)
err = func() (err error) {
if !svc.ac.CanCreateReport(ctx) {