feat: emit assertion errors (#2309)

* move assertions code to the assertion executor component

* fix test suite

* generate TEST_SPECS_ASSERTION_ERROR event when assertion fails

TEST_RUN_EVENTS.csv

@@ -31,9 +31,7 @@ Trace,FETCHING_SUCCESS,The trace was successfully processed by the backend,
 Trace,FETCHING_ERROR,The trace was not able to be fetched,
 Trace,STOPPED_INFO,The test run was stopped during its execution,
 Test,OUTPUT_GENERATION_WARNING,The value for output <output_name> could not be generated,
-Test,RESOLVE_START,Resolving test specs details start,
-Test,RESOLVE_SUCCESS,Resolving test specs details success,
-Test,RESOLVE_ERROR,An error ocurred while parsing the test specs,
 Test,TEST_SPECS_RUN_SUCCESS,Test Specs were successfully executed,
 Test,TEST_SPECS_RUN_ERROR,Test specs execution error,
-Test,TEST_SPECS_RUN_START,Test specs execution start,
+Test,TEST_SPECS_RUN_START,Test specs execution start,
+Test,TEST_SPECS_ASSERTION_ERROR,An assertion in the test spec failed

server/app/app.go

@@ -366,7 +366,7 @@ func httpRouter(
 	mappers mappings.Mappings,
 	triggerRegistry *trigger.Registry,
 ) openapi.Router {
-	controller := httpServer.NewController(testDB, tracedb.Factory(testDB), rf, mappers, triggerRegistry)
+	controller := httpServer.NewController(testDB, tracedb.Factory(testDB), rf, mappers, triggerRegistry, tracer)
 	apiApiController := openapi.NewApiApiController(controller)
 	customController := httpServer.NewCustomController(controller, apiApiController, openapi.DefaultErrorHandler, tracer)
 	httpRouter := customController

server/executor/assertion_executor.go

@@ -3,7 +3,7 @@ package executor
 import (
 	"context"
 
-	"github.com/kubeshop/tracetest/server/assertions"
+	"github.com/kubeshop/tracetest/server/assertions/selectors"
 	"github.com/kubeshop/tracetest/server/expression"
 	"github.com/kubeshop/tracetest/server/model"
 	"go.opentelemetry.io/otel/attribute"
@@ -17,7 +17,73 @@ type AssertionExecutor interface {
 type defaultAssertionExecutor struct{}
 
 func (e defaultAssertionExecutor) Assert(ctx context.Context, defs model.OrderedMap[model.SpanQuery, model.NamedAssertions], trace model.Trace, ds []expression.DataStore) (model.OrderedMap[model.SpanQuery, []model.AssertionResult], bool) {
-	return assertions.Assert(defs, trace, ds)
+	testResult := model.OrderedMap[model.SpanQuery, []model.AssertionResult]{}
+	allPassed := true
+	defs.ForEach(func(spanQuery model.SpanQuery, asserts model.NamedAssertions) error {
+		spans := selector(spanQuery).Filter(trace)
+		assertionResults := make([]model.AssertionResult, 0)
+		for _, assertion := range asserts.Assertions {
+			res := e.assert(assertion, spans, ds)
+			if !res.AllPassed {
+				allPassed = false
+			}
+			assertionResults = append(assertionResults, res)
+		}
+		testResult, _ = testResult.Add(spanQuery, assertionResults)
+		return nil
+	})
+
+	return testResult, allPassed
+}
+
+func (e defaultAssertionExecutor) assert(assertion model.Assertion, spans model.Spans, ds []expression.DataStore) model.AssertionResult {
+	ds = append([]expression.DataStore{
+		expression.MetaAttributesDataStore{SelectedSpans: spans},
+		expression.VariableDataStore{},
+	}, ds...)
+
+	allPassed := true
+	spanResults := make([]model.SpanAssertionResult, 0, len(spans))
+	spans.
+		ForEach(func(_ int, span model.Span) bool {
+			res := e.assertSpan(span, ds, string(assertion))
+			spanResults = append(spanResults, res)
+
+			if res.CompareErr != nil {
+				allPassed = false
+			}
+
+			return true
+		}).
+		OrEmpty(func() {
+			res := e.assertSpan(model.Span{}, ds, string(assertion))
+			spanResults = append(spanResults, res)
+			allPassed = res.CompareErr == nil
+		})
+
+	return model.AssertionResult{
+		Assertion: assertion,
+		AllPassed: allPassed,
+		Results:   spanResults,
+	}
+}
+
+func (e defaultAssertionExecutor) assertSpan(span model.Span, ds []expression.DataStore, assertion string) model.SpanAssertionResult {
+	ds = append([]expression.DataStore{expression.AttributeDataStore{Span: span}}, ds...)
+	expressionExecutor := expression.NewExecutor(ds...)
+
+	actualValue, _, err := expressionExecutor.Statement(assertion)
+
+	sar := model.SpanAssertionResult{
+		ObservedValue: actualValue,
+		CompareErr:    err,
+	}
+
+	if span.ID.IsValid() {
+		sar.SpanID = &span.ID
+	}
+
+	return sar
 }
 
 type instrumentedAssertionExecutor struct {
@@ -43,3 +109,8 @@ func NewAssertionExecutor(tracer trace.Tracer) AssertionExecutor {
 		tracer:            tracer,
 	}
 }
+
+func selector(sq model.SpanQuery) selectors.Selector {
+	sel, _ := selectors.New(string(sq))
+	return sel
+}

server/executor/assertion_runner.go

@@ -2,6 +2,7 @@ package executor
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log"
 	"time"
@@ -167,6 +168,8 @@ func (e *defaultAssertionRunner) executeAssertions(ctx context.Context, req Asse
 
 	assertionResult, allPassed := e.assertionExecutor.Assert(ctx, req.Test.Specs, *run.Trace, ds)
 
+	e.emitFailedAssertions(ctx, req, assertionResult)
+
 	run = run.SuccessfullyAsserted(
 		outputs,
 		newEnvironment,
@@ -177,6 +180,27 @@ func (e *defaultAssertionRunner) executeAssertions(ctx context.Context, req Asse
 	return run, nil
 }
 
+func (e *defaultAssertionRunner) emitFailedAssertions(ctx context.Context, req AssertionRequest, result model.OrderedMap[model.SpanQuery, []model.AssertionResult]) {
+	for _, assertionResults := range result.Unordered() {
+		for _, assertionResult := range assertionResults {
+			for _, spanAssertionResult := range assertionResult.Results {
+
+				if errors.Is(spanAssertionResult.CompareErr, expression.ErrExpressionResolution) {
+					unwrappedError := errors.Unwrap(spanAssertionResult.CompareErr)
+					e.eventEmitter.Emit(ctx, events.TestSpecsAssertionError(
+						req.Run.TestID,
+						req.Run.ID,
+						unwrappedError,
+						spanAssertionResult.SpanID.String(),
+						string(assertionResult.Assertion),
+					))
+				}
+
+			}
+		}
+	}
+}
+
 func createEnvironment(environment model.Environment, outputs model.OrderedMap[string, model.RunOutput]) model.Environment {
 	outputVariables := make([]model.EnvironmentValue, 0)
 	outputs.ForEach(func(key string, val model.RunOutput) error {

server/assertions/assetions_test.go → server/executor/assetion_executor_test.go

@@ -1,15 +1,17 @@
-package assertions_test
+package executor_test
 
 import (
+	"context"
 	"testing"
 
-	"github.com/kubeshop/tracetest/server/assertions"
 	"github.com/kubeshop/tracetest/server/assertions/comparator"
+	"github.com/kubeshop/tracetest/server/executor"
 	"github.com/kubeshop/tracetest/server/expression"
 	"github.com/kubeshop/tracetest/server/id"
 	"github.com/kubeshop/tracetest/server/model"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/trace"
 )
 
 func TestAssertion(t *testing.T) {
@@ -205,9 +207,9 @@ func TestAssertion(t *testing.T) {
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
 			cl := c
-			// t.Parallel()
 
-			actual, allPassed := assertions.Assert(cl.testDef, cl.trace, []expression.DataStore{})
+			executor := executor.NewAssertionExecutor(trace.NewNoopTracerProvider().Tracer("tracer"))
+			actual, allPassed := executor.Assert(context.Background(), cl.testDef, cl.trace, []expression.DataStore{})
 
 			assert.Equal(t, cl.expectedAllPassed, allPassed)
 

server/http/controller.go

@@ -8,7 +8,6 @@ import (
 	"net/http"
 	"strconv"
 
-	"github.com/kubeshop/tracetest/server/assertions"
 	"github.com/kubeshop/tracetest/server/assertions/selectors"
 	"github.com/kubeshop/tracetest/server/executor"
 	"github.com/kubeshop/tracetest/server/executor/trigger"
@@ -29,6 +28,7 @@ import (
 var IDGen = id.NewRandGenerator()
 
 type controller struct {
+	tracer          trace.Tracer
 	testDB          model.Repository
 	runner          runner
 	newTraceDBFn    func(ds model.DataStore) (tracedb.TraceDB, error)
@@ -48,8 +48,10 @@ func NewController(
 	runner runner,
 	mappers mappings.Mappings,
 	triggerRegistry *trigger.Registry,
+	tracer trace.Tracer,
 ) openapi.ApiApiServicer {
 	return &controller{
+		tracer:          tracer,
 		testDB:          testDB,
 		runner:          runner,
 		newTraceDBFn:    newTraceDBFn,
@@ -372,7 +374,9 @@ func (c *controller) DryRunAssertion(ctx context.Context, testID, runID string,
 		Values: run.Environment.Values,
 	}}
 
-	results, allPassed := assertions.Assert(definition, *run.Trace, ds)
+	assertionExecutor := executor.NewAssertionExecutor(c.tracer)
+
+	results, allPassed := assertionExecutor.Assert(ctx, definition, *run.Trace, ds)
 	res := c.mappers.Out.Result(&model.RunResults{
 		AllPassed: allPassed,
 		Results:   results,

server/http/controller_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/kubeshop/tracetest/server/traces"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/trace"
 )
 
 var (
@@ -121,6 +122,7 @@ func setupController(t *testing.T) controllerFixture {
 			nil,
 			mappings.New(traces.NewConversionConfig(), comparator.DefaultRegistry(), mdb),
 			&trigger.Registry{},
+			trace.NewNoopTracerProvider().Tracer("tracer"),
 		),
 	}
 }

server/model/events/events.go

@@ -323,51 +323,6 @@ func TestOutputGenerationWarning(testID id.ID, runID int, output string) model.T
 	}
 }
 
-func TestResolveStart(testID id.ID, runID int) model.TestRunEvent {
-	return model.TestRunEvent{
-		TestID:              testID,
-		RunID:               runID,
-		Stage:               model.StageTest,
-		Type:                "RESOLVE_START",
-		Title:               "Resolving test specs details start",
-		Description:         "Resolving test specs details start",
-		CreatedAt:           time.Now(),
-		DataStoreConnection: model.ConnectionResult{},
-		Polling:             model.PollingInfo{},
-		Outputs:             []model.OutputInfo{},
-	}
-}
-
-func TestResolveSuccess(testID id.ID, runID int) model.TestRunEvent {
-	return model.TestRunEvent{
-		TestID:              testID,
-		RunID:               runID,
-		Stage:               model.StageTest,
-		Type:                "RESOLVE_SUCCESS",
-		Title:               "Resolving test specs details success",
-		Description:         "Resolving test specs details success",
-		CreatedAt:           time.Now(),
-		DataStoreConnection: model.ConnectionResult{},
-		Polling:             model.PollingInfo{},
-		Outputs:             []model.OutputInfo{},
-	}
-}
-
-func TestResolveError(testID id.ID, runID int) model.TestRunEvent {
-	return model.TestRunEvent{
-		TestID:              testID,
-		RunID:               runID,
-		Stage:               model.StageTest,
-		Type:                "RESOLVE_ERROR",
-		Title:               "An error ocurred while parsing the test specs",
-		Description:         "An error ocurred while parsing the test specs",
-		CreatedAt:           time.Now(),
-		DataStoreConnection: model.ConnectionResult{},
-		Polling:             model.PollingInfo{},
-		Outputs:             []model.OutputInfo{},
-	}
-}
-
 func TestSpecsRunSuccess(testID id.ID, runID int) model.TestRunEvent {
 	return model.TestRunEvent{
 		TestID:              testID,
@@ -427,3 +382,18 @@ func TestSpecsRunStart(testID id.ID, runID int) model.TestRunEvent {
 		Outputs:             []model.OutputInfo{},
 	}
 }
+
+func TestSpecsAssertionError(testID id.ID, runID int, err error, spanID string, assertion string) model.TestRunEvent {
+	return model.TestRunEvent{
+		TestID:              testID,
+		RunID:               runID,
+		Stage:               model.StageTest,
+		Type:                "TEST_SPECS_ASSERTION_ERROR",
+		Title:               "Assertion execution failed",
+		Description:         fmt.Sprintf(`Assertion '%s' returned an error on span %s: %s`, assertion, spanID, err.Error()),
+		CreatedAt:           time.Now(),
+		DataStoreConnection: model.ConnectionResult{},
+		Polling:             model.PollingInfo{},
+		Outputs:             []model.OutputInfo{},
+	}
+}