feat: test run event repository (#2270)

* add repository for test run events

* add index to test_id and run_id fields

server/migrations/23_add_test_run_events.down.sql

@@ -0,0 +1,5 @@
+BEGIN;
+
+DROP TABLE "test_run_events";
+
+COMMIT;

server/migrations/23_add_test_run_events.up.sql

@@ -0,0 +1,19 @@
+BEGIN;
+
+CREATE TABLE "test_run_events" (
+  "id" SERIAL PRIMARY KEY,
+  "test_id" varchar not null,
+  "run_id" int not null,
+  "type" varchar not null,
+  "stage" varchar not null,
+  "description" varchar not null,
+  "created_at" timestamp not null default now(),
+  "data_store_connection" JSONB,
+  "polling" JSONB,
+  "outputs" JSONB
+);
+
+CREATE INDEX test_run_event_test_id_idx ON test_run_events(test_id);
+CREATE INDEX test_run_event_run_id_idx ON test_run_events(run_id);
+
+COMMIT;

server/model/repository.go

@@ -70,6 +70,11 @@ type DataStoreRepository interface {
 	DataStoreIDExists(context.Context, string) (bool, error)
 }
 
+type TestRunEventRepository interface {
+	CreateTestRunEvent(context.Context, TestRunEvent) error
+	GetTestRunEvents(context.Context, id.ID, int) ([]TestRunEvent, error)
+}
+
 type Repository interface {
 	TestRepository
 	RunRepository
@@ -80,6 +85,8 @@ type Repository interface {
 
 	DataStoreRepository
 
+	TestRunEventRepository
+
 	ServerID() (id string, isNew bool, _ error)
 	Close() error
 	Drop() error

server/model/test_run_event.go

@@ -1,10 +1,17 @@
 package model
 
-import "time"
+import (
+	"time"
+
+	"github.com/kubeshop/tracetest/server/id"
+)
 
 type (
-	Protocol string
-	Status   string
+	Protocol          string
+	Status            string
+	TestRunEventStage string
+	PollingType       string
+	LogLevel          string
 )
 
 var (
@@ -18,32 +25,48 @@ var (
 	StatusFailed  Status = "failed"
 )
 
+var (
+	LogLevelWarn  LogLevel = "warning"
+	LogLevelError LogLevel = "error"
+)
+
+var (
+	StageTrigger TestRunEventStage = "trigger"
+	StageTrace   TestRunEventStage = "trace"
+	StageTest    TestRunEventStage = "test"
+)
+
+var (
+	PollingTypePeriodic PollingType = "periodic"
+)
+
 type TestRunEvent struct {
+	ID                  int64
 	Type                string
-	Stage               string
+	Stage               TestRunEventStage
 	Description         string
 	CreatedAt           time.Time
-	TestId              string
-	RunId               string
+	TestID              id.ID
+	RunID               int
 	DataStoreConnection ConnectionResult
 	Polling             PollingInfo
 	Outputs             []OutputInfo
 }
 
 type PollingInfo struct {
-	Type                string
+	Type                PollingType
 	ReasonNextIteration string
 	IsComplete          bool
 	Periodic            *PeriodicPollingConfig
 }
 
 type PeriodicPollingConfig struct {
-	NumberSpans      int32
-	NumberIterations int32
+	NumberSpans      int
+	NumberIterations int
 }
 
 type OutputInfo struct {
-	LogLevel   string
+	LogLevel   LogLevel
 	Message    string
 	OutputName string
 }

server/testdb/mock.go

@@ -281,3 +281,13 @@ func (m *MockRepository) GetDataStores(_ context.Context, take, skip int32, quer
 	}
 	return list, args.Error(1)
 }
+
+func (m *MockRepository) CreateTestRunEvent(_ context.Context, event model.TestRunEvent) error {
+	args := m.Called(event)
+	return args.Error(0)
+}
+
+func (m *MockRepository) GetTestRunEvents(_ context.Context, testID id.ID, runID int) ([]model.TestRunEvent, error) {
+	args := m.Called(testID, runID)
+	return args.Get(0).([]model.TestRunEvent), args.Error(1)
+}

server/testdb/test_run_event.go

@@ -0,0 +1,165 @@
+package testdb
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/kubeshop/tracetest/server/id"
+	"github.com/kubeshop/tracetest/server/model"
+)
+
+const insertTestRunEventQuery = `
+	INSERT INTO test_run_events (
+		"test_id",
+		"run_id",
+		"type",
+		"stage",
+		"description",
+		"created_at",
+		"data_store_connection",
+		"polling",
+		"outputs"
+	) VALUES (
+		$1, -- test_id
+		$2, -- run_id
+		$3, -- type
+		$4, -- stage
+		$5, -- description
+		$6, -- created_at
+		$7, -- data_store_connection
+		$8, -- polling
+		$9  -- outputs
+	)
+	RETURNING "id"
+`
+
+func (td *postgresDB) CreateTestRunEvent(ctx context.Context, event model.TestRunEvent) error {
+	dataStoreConnectionJSON, err := json.Marshal(event.DataStoreConnection)
+	if err != nil {
+		return fmt.Errorf("could not marshal data store connection into JSON: %w", err)
+	}
+
+	pollingJSON, err := json.Marshal(event.Polling)
+	if err != nil {
+		return fmt.Errorf("could not marshal polling into JSON: %w", err)
+	}
+
+	outputsJSON, err := json.Marshal(event.Outputs)
+	if err != nil {
+		return fmt.Errorf("could not marshal outputs into JSON: %w", err)
+	}
+
+	if event.CreatedAt.IsZero() {
+		event.CreatedAt = time.Now()
+	}
+
+	err = td.db.QueryRowContext(
+		ctx,
+		insertTestRunEventQuery,
+		event.TestID,
+		event.RunID,
+		event.Type,
+		event.Stage,
+		event.Description,
+		event.CreatedAt,
+		dataStoreConnectionJSON,
+		pollingJSON,
+		outputsJSON,
+	).Scan(&event.ID)
+
+	if err != nil {
+		return fmt.Errorf("could not insert event into database: %w", err)
+	}
+
+	return nil
+}
+
+const getTestRunEventsQuery = `
+	SELECT
+		"id",
+		"test_id",
+		"run_id",
+		"type",
+		"stage",
+		"description",
+		"created_at",
+		"data_store_connection",
+		"polling",
+		"outputs"
+	FROM test_run_events WHERE "test_id" = $1 AND "run_id" = $2 ORDER BY "created_at" ASC;
+`
+
+func (td *postgresDB) GetTestRunEvents(ctx context.Context, testID id.ID, runID int) ([]model.TestRunEvent, error) {
+	rows, err := td.db.QueryContext(ctx, getTestRunEventsQuery, testID, runID)
+	if err != nil {
+		return []model.TestRunEvent{}, fmt.Errorf("could not query test runs: %w", err)
+	}
+
+	events := make([]model.TestRunEvent, 0)
+
+	for rows.Next() {
+		event, err := readTestRunEventFromRows(rows)
+		if err != nil {
+			return []model.TestRunEvent{}, fmt.Errorf("could not parse row: %w", err)
+		}
+
+		events = append(events, event)
+	}
+
+	return events, nil
+}
+
+func readTestRunEventFromRows(rows *sql.Rows) (model.TestRunEvent, error) {
+	var dataStoreConnectionBytes, pollingBytes, outputsBytes []byte
+	event := model.TestRunEvent{}
+
+	err := rows.Scan(
+		&event.ID,
+		&event.TestID,
+		&event.RunID,
+		&event.Type,
+		&event.Stage,
+		&event.Description,
+		&event.CreatedAt,
+		&dataStoreConnectionBytes,
+		&pollingBytes,
+		&outputsBytes,
+	)
+
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return model.TestRunEvent{}, ErrNotFound
+		}
+
+		return model.TestRunEvent{}, fmt.Errorf("could not scan event: %w", err)
+	}
+
+	var dataStoreConnection model.ConnectionResult
+	var polling model.PollingInfo
+	outputs := make([]model.OutputInfo, 0)
+
+	err = json.Unmarshal(dataStoreConnectionBytes, &dataStoreConnection)
+	if err != nil {
+		return model.TestRunEvent{}, fmt.Errorf("could not unmarshal data store connection: %w", err)
+	}
+
+	err = json.Unmarshal(pollingBytes, &polling)
+	if err != nil {
+		return model.TestRunEvent{}, fmt.Errorf("could not unmarshal polling information: %w", err)
+	}
+
+	err = json.Unmarshal(outputsBytes, &outputs)
+	if err != nil {
+		return model.TestRunEvent{}, fmt.Errorf("could not unmarshal outputs: %w", err)
+	}
+
+	event.DataStoreConnection = dataStoreConnection
+	event.Polling = polling
+	event.Outputs = outputs
+
+	return event, nil
+}

server/testdb/test_run_event_test.go

@@ -0,0 +1,80 @@
+package testdb_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/kubeshop/tracetest/server/model"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRunEvents(t *testing.T) {
+	db, clean := getDB()
+	defer clean()
+
+	test1 := createTestWithName(t, db, "test 1")
+
+	run1 := createRun(t, db, test1)
+	run2 := createRun(t, db, test1)
+
+	events := []model.TestRunEvent{
+		{TestID: test1.ID, RunID: run1.ID, Type: "EVENT_1", Stage: model.StageTrigger, Description: "This happened"},
+		{TestID: test1.ID, RunID: run1.ID, Type: "EVENT_2", Stage: model.StageTrigger, Description: "That happened now"},
+
+		{TestID: test1.ID, RunID: run2.ID, Type: "EVENT_1", Stage: model.StageTrigger, Description: "That happened", DataStoreConnection: model.ConnectionResult{
+			PortCheck: model.ConnectionTestStep{
+				Passed:  true,
+				Status:  model.StatusPassed,
+				Message: "Should pass",
+				Error:   nil,
+			},
+		}},
+		{TestID: test1.ID, RunID: run2.ID, Type: "EVENT_2_FAILED", Stage: model.StageTrigger, Description: "That happened, but failed", Polling: model.PollingInfo{
+			Type: model.PollingTypePeriodic,
+			Periodic: &model.PeriodicPollingConfig{
+				NumberSpans:      3,
+				NumberIterations: 1,
+			},
+		}},
+		{TestID: test1.ID, RunID: run2.ID, Type: "ANOTHER_EVENT", Stage: model.StageTrigger, Description: "Clean up after error", Outputs: []model.OutputInfo{
+			{LogLevel: model.LogLevelWarn, Message: "INVALID SYNTAX", OutputName: "my_output"},
+		}},
+	}
+
+	for _, event := range events {
+		err := db.CreateTestRunEvent(context.Background(), event)
+		require.NoError(t, err)
+	}
+
+	events, err := db.GetTestRunEvents(context.Background(), test1.ID, run1.ID)
+	require.NoError(t, err)
+
+	assert.Len(t, events, 2)
+	assert.LessOrEqual(t, events[0].CreatedAt, events[1].CreatedAt)
+
+	eventsFromRun2, err := db.GetTestRunEvents(context.Background(), test1.ID, run2.ID)
+	require.NoError(t, err)
+
+	assert.Len(t, eventsFromRun2, 3)
+	assert.LessOrEqual(t, eventsFromRun2[0].CreatedAt, eventsFromRun2[1].CreatedAt)
+	assert.LessOrEqual(t, eventsFromRun2[1].CreatedAt, eventsFromRun2[2].CreatedAt)
+
+	// assert eevents from run 2 have fields that were stored as JSON
+	// data store connection
+	assert.Equal(t, true, eventsFromRun2[0].DataStoreConnection.PortCheck.Passed)
+	assert.Equal(t, model.StatusPassed, eventsFromRun2[0].DataStoreConnection.PortCheck.Status)
+	assert.Equal(t, "Should pass", eventsFromRun2[0].DataStoreConnection.PortCheck.Message)
+	assert.Nil(t, eventsFromRun2[0].DataStoreConnection.PortCheck.Error)
+
+	// polling
+	assert.Equal(t, model.PollingTypePeriodic, eventsFromRun2[1].Polling.Type)
+	assert.Equal(t, 3, eventsFromRun2[1].Polling.Periodic.NumberSpans)
+	assert.Equal(t, 1, eventsFromRun2[1].Polling.Periodic.NumberIterations)
+
+	// outputs
+	assert.Len(t, eventsFromRun2[2].Outputs, 1)
+	assert.Equal(t, model.LogLevelWarn, eventsFromRun2[2].Outputs[0].LogLevel)
+	assert.Equal(t, "INVALID SYNTAX", eventsFromRun2[2].Outputs[0].Message)
+	assert.Equal(t, "my_output", eventsFromRun2[2].Outputs[0].OutputName)
+}