generated from kyma-project/template-repository
-
Notifications
You must be signed in to change notification settings - Fork 16
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Added tests for Cloud Events and verification of reception of events …
…by the Sink (#93) * Added e2e event tests for CE * added code to fetch event from sink * updated * added verification for legacy events received by sink * cleanup code * fixed lint * addressed review comments * lint fix
- Loading branch information
Showing
13 changed files
with
560 additions
and
133 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
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,113 @@ | ||
| package eventing | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "context" | ||
| "encoding/json" | ||
| "fmt" | ||
| "io" | ||
| "net/http" | ||
| "time" | ||
|
|
||
| cloudevents "github.com/cloudevents/sdk-go/v2" | ||
| cev2event "github.com/cloudevents/sdk-go/v2/event" | ||
| "github.com/kyma-project/eventing-manager/hack/e2e/common" | ||
| "github.com/pkg/errors" | ||
| "go.uber.org/zap" | ||
| ) | ||
|
|
||
| const ( | ||
| EventsEndpointFormat = "%s/events/%s" | ||
| ) | ||
|
|
||
| type SinkClient struct { | ||
| ctx context.Context | ||
| clientHTTP *http.Client | ||
| sinkURL string | ||
| logger *zap.Logger | ||
| } | ||
|
|
||
| type SinkEvent struct { | ||
| // Header stores the non CE events, e.g. X-B3-Sampled and Traceparent | ||
| http.Header | ||
| cev2event.Event | ||
| } | ||
|
|
||
| func NewSinkClient(ctx context.Context, clientHTTP *http.Client, sinkURL string, logger *zap.Logger) *SinkClient { | ||
| return &SinkClient{ | ||
| ctx: ctx, | ||
| clientHTTP: clientHTTP, | ||
| sinkURL: sinkURL, | ||
| logger: logger, | ||
| } | ||
| } | ||
|
|
||
| func (sc *SinkClient) EventsEndpoint(eventId string) string { | ||
| return fmt.Sprintf(EventsEndpointFormat, sc.sinkURL, eventId) | ||
| } | ||
|
|
||
| func (sc *SinkClient) GetEventFromSinkWithRetries(eventId string, attempts int, interval time.Duration) (*SinkEvent, error) { | ||
| var gotEvent *SinkEvent | ||
| err := common.Retry(attempts, interval, func() error { | ||
| var err1 error | ||
| gotEvent, err1 = sc.GetEventFromSink(eventId) | ||
| return err1 | ||
| }) | ||
| return gotEvent, err | ||
| } | ||
|
|
||
| func (sc *SinkClient) GetEventFromSink(eventId string) (*SinkEvent, error) { | ||
| url := sc.EventsEndpoint(eventId) | ||
| sc.logger.Debug(fmt.Sprintf("Fetching event with ID: %s from the sink URL: %s", eventId, url)) | ||
|
|
||
| req, err := http.NewRequest(http.MethodGet, url, bytes.NewBuffer([]byte{})) | ||
| if err != nil { | ||
| err = errors.Wrap(err, "Failed to create HTTP request for fetching event from sink") | ||
| sc.logger.Debug(err.Error()) | ||
| return nil, err | ||
| } | ||
|
|
||
| req.Header.Set("Content-Type", "application/json") | ||
| resp, err := sc.clientHTTP.Do(req) | ||
| if err != nil { | ||
| err = errors.Wrap(err, "Failed to fetch event") | ||
| sc.logger.Debug(err.Error()) | ||
| return nil, err | ||
| } | ||
| defer func() { | ||
| err = resp.Body.Close() | ||
| if err != nil { | ||
| sc.logger.Error(err.Error()) | ||
| } | ||
| }() | ||
|
|
||
| // read body. | ||
| respBody, err := io.ReadAll(resp.Body) | ||
| if err != nil { | ||
| err = errors.Wrap(err, "Failed to read response body") | ||
| sc.logger.Debug(err.Error()) | ||
| return nil, err | ||
| } | ||
|
|
||
| // if not success, then return error. | ||
| if !Is2XX(resp.StatusCode) { | ||
| err = errors.New(fmt.Sprintf("Failed to fetch eventID:[%s] response:[%d] body:[%s]", eventId, | ||
| resp.StatusCode, string(respBody))) | ||
| sc.logger.Debug(err.Error()) | ||
| return nil, err | ||
| } | ||
|
|
||
| // success | ||
| // convert to cloud event object. | ||
| ceEvent := cloudevents.NewEvent() | ||
| err = json.Unmarshal(respBody, &ceEvent) | ||
| if err != nil { | ||
| err = errors.Wrap(err, "failed to convert JSON to CloudEvent") | ||
| sc.logger.Debug(err.Error()) | ||
| return nil, err | ||
| } | ||
|
|
||
| return &SinkEvent{ | ||
| Event: ceEvent, | ||
| }, 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
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
Oops, something went wrong.