Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(server): adding event emitter (#2269)
- Loading branch information
1 parent
b907493
commit f56d33e
Showing
7 changed files
with
274 additions
and
69 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,2 +1,6 @@ | ||
| dataStore: | ||
| --- | ||
| type: DataStore | ||
| spec: | ||
| name: OpenTelemetry Collector | ||
| type: otlp | ||
| isdefault: true |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,38 @@ | ||
| package executor | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "github.com/kubeshop/tracetest/server/model" | ||
| ) | ||
|
|
||
| type EventEmitter interface { | ||
| Emit(ctx context.Context, event model.TestRunEvent) error | ||
| } | ||
|
|
||
| type publisher interface { | ||
| Publish(eventID string, message any) | ||
| } | ||
|
|
||
| type internalEventEmitter struct { | ||
| repository model.TestRunEventRepository | ||
| publisher publisher | ||
| } | ||
|
|
||
| func NewEventEmitter(repository model.TestRunEventRepository, publisher publisher) EventEmitter { | ||
| return &internalEventEmitter{ | ||
| repository: repository, | ||
| publisher: publisher, | ||
| } | ||
| } | ||
|
|
||
| func (em *internalEventEmitter) Emit(ctx context.Context, event model.TestRunEvent) error { | ||
| err := em.repository.CreateTestRunEvent(ctx, event) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| em.publisher.Publish(event.ResourceID(), event) | ||
|
|
||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,153 @@ | ||
| package executor_test | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "testing" | ||
|
|
||
| "github.com/kubeshop/tracetest/server/executor" | ||
| "github.com/kubeshop/tracetest/server/id" | ||
| "github.com/kubeshop/tracetest/server/model" | ||
| "github.com/kubeshop/tracetest/server/subscription" | ||
| "github.com/kubeshop/tracetest/server/testdb" | ||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TestEventEmitter_SuccessfulScenario(t *testing.T) { | ||
| // Given I have a test run event | ||
|
|
||
| run := model.NewRun() | ||
|
|
||
| test := model.Test{ | ||
| ID: id.ID("some-test"), | ||
| ServiceUnderTest: model.Trigger{ | ||
| Type: model.TriggerTypeHTTP, | ||
| }, | ||
| } | ||
|
|
||
| testRunEvent := model.TestRunEvent{ | ||
| TestID: test.ID, | ||
| RunID: run.ID, | ||
| Type: "EVENT_1", | ||
| Stage: model.StageTrigger, | ||
| Title: "OP 1", | ||
| Description: "This happened", | ||
| } | ||
|
|
||
| // When I emit this event successfully | ||
| repository := getTestRunEventRepositoryMock(t, false) | ||
| subscriptionManager, subscriber := getSubscriptionManagerMock(t, testRunEvent) | ||
|
|
||
| eventEmitter := executor.NewEventEmitter(repository, subscriptionManager) | ||
|
|
||
| err := eventEmitter.Emit(context.Background(), testRunEvent) | ||
| require.NoError(t, err) | ||
|
|
||
| // Then I expect that it was persisted | ||
| assert.Len(t, repository.events, 1) | ||
| assert.Equal(t, testRunEvent.Title, repository.events[0].Title) | ||
| assert.Equal(t, testRunEvent.Stage, repository.events[0].Stage) | ||
| assert.Equal(t, testRunEvent.Description, repository.events[0].Description) | ||
|
|
||
| // And that it was sent to subscribers | ||
| assert.Len(t, subscriber.events, 1) | ||
| assert.Equal(t, testRunEvent.Title, subscriber.events[0].Title) | ||
| assert.Equal(t, testRunEvent.Stage, subscriber.events[0].Stage) | ||
| assert.Equal(t, testRunEvent.Description, subscriber.events[0].Description) | ||
| } | ||
|
|
||
| func TestEventEmitter_FailedScenario(t *testing.T) { | ||
| // Given I have a test run event | ||
|
|
||
| run := model.NewRun() | ||
|
|
||
| test := model.Test{ | ||
| ID: id.ID("some-test"), | ||
| ServiceUnderTest: model.Trigger{ | ||
| Type: model.TriggerTypeHTTP, | ||
| }, | ||
| } | ||
|
|
||
| testRunEvent := model.TestRunEvent{ | ||
| TestID: test.ID, | ||
| RunID: run.ID, | ||
| Type: "EVENT_1", | ||
| Stage: model.StageTrigger, | ||
| Title: "OP 1", | ||
| Description: "This happened", | ||
| } | ||
|
|
||
| // When I emit this event and it fails | ||
| repository := getTestRunEventRepositoryMock(t, true) | ||
| subscriptionManager, subscriber := getSubscriptionManagerMock(t, testRunEvent) | ||
|
|
||
| eventEmitter := executor.NewEventEmitter(repository, subscriptionManager) | ||
|
|
||
| err := eventEmitter.Emit(context.Background(), testRunEvent) | ||
| require.Error(t, err) | ||
|
|
||
| // Then I expect that it was not persisted | ||
| assert.Len(t, repository.events, 0) | ||
|
|
||
| // And that it was not sent to subscribers | ||
| assert.Len(t, subscriber.events, 0) | ||
| } | ||
|
|
||
| // TestRunEventRepository | ||
| type testRunEventRepositoryMock struct { | ||
| testdb.MockRepository | ||
| events []model.TestRunEvent | ||
| returnError bool | ||
| // ... | ||
| } | ||
|
|
||
| func (m *testRunEventRepositoryMock) CreateTestRunEvent(ctx context.Context, event model.TestRunEvent) error { | ||
| if m.returnError { | ||
| return errors.New("error on persistence") | ||
| } | ||
|
|
||
| m.events = append(m.events, event) | ||
| return nil | ||
| } | ||
|
|
||
| func getTestRunEventRepositoryMock(t *testing.T, returnError bool) *testRunEventRepositoryMock { | ||
| t.Helper() | ||
|
|
||
| mock := new(testRunEventRepositoryMock) | ||
| mock.T = t | ||
| mock.Test(t) | ||
|
|
||
| mock.events = []model.TestRunEvent{} | ||
| mock.returnError = returnError | ||
|
|
||
| return mock | ||
| } | ||
|
|
||
| // TestRunEventSubscriber | ||
| type testRunEventSubscriber struct { | ||
| events []model.TestRunEvent | ||
| } | ||
|
|
||
| func (s *testRunEventSubscriber) ID() string { | ||
| return "some-id" | ||
| } | ||
|
|
||
| func (s *testRunEventSubscriber) Notify(message subscription.Message) error { | ||
| event := message.Content.(model.TestRunEvent) | ||
| s.events = append(s.events, event) | ||
| return nil | ||
| } | ||
|
|
||
| func getSubscriptionManagerMock(t *testing.T, event model.TestRunEvent) (*subscription.Manager, *testRunEventSubscriber) { | ||
| t.Helper() | ||
|
|
||
| subscriptionManager := subscription.NewManager() | ||
| subscriber := &testRunEventSubscriber{ | ||
| events: []model.TestRunEvent{}, | ||
| } | ||
|
|
||
| subscriptionManager.Subscribe(event.ResourceID(), subscriber) | ||
|
|
||
| return subscriptionManager, subscriber | ||
| } |
Oops, something went wrong.