Leverage new subscriber image endpoints in external-solution test (ky…

…ma-project#9354)

* Changed the endpoint the lambda function posts to

* Update readme

* Combine steps for sending and checking the event
Add a counter to differentiate each event by its id
Retrieve event via ID
If retrieving by ID fails get and print all received events to facilitate debugging

* Fix imports formatting

* Changes based on PR review

* Changes based on PR review

* Fix imports formatting

* Refactor send event struct

* Remove sendEvent field name

resources/core/values.yaml

@@ -33,7 +33,7 @@ global:
     version: d022112f
   e2e_external_solution:
     dir:
-    version: "1a5b13b8"
+    version: "PR-9354"
   e2e_external_solution_test_service:
     dir:
     version: 9913ea21

tests/end-to-end/external-solution-integration/chart/external-solution/values.yaml

@@ -3,7 +3,7 @@ containerRegistry:
 
 image:
   dir:
-  version: 1a5b13b8
+  version: PR-9354
   pullPolicy: "Always"
 
 testServiceImage:

tests/end-to-end/external-solution-integration/internal/scenario/event_mesh_evaluate/event_mesh_evaluate.go

@@ -45,17 +45,23 @@ func (s *Scenario) Steps(config *rest.Config) ([]step.Step, error) {
 	return []step.Step{
 		testsuite.NewLoadStoredCertificates(dataStore, state),
 		step.Retry(
-			testsuite.NewResetCounterPod(testService),
-			testsuite.NewSendEventToMesh(s.TestID, helpers.FunctionPayload, state),
-			testsuite.NewCheckCounterPod(testService, 1, retryOpts...),
+			testsuite.NewSendEventToMeshAndCheckEventId(
+				s.TestID,
+				helpers.FunctionPayload,
+				state,
+				testService,
+				retryOpts...),
 		).WithRetryOptions(
 			retry.Attempts(3),
 			retry.DelayType(retry.FixedDelay),
 			retry.Delay(500*time.Millisecond)),
 		step.Retry(
-			testsuite.NewResetCounterPod(testService),
-			testsuite.NewSendEventToCompatibilityLayer(s.TestID, helpers.FunctionPayload, state),
-			testsuite.NewCheckCounterPod(testService, 1, retryOpts...),
+			testsuite.NewSendEventToCompatibilityLayerAndCheckEventId(
+				s.TestID,
+				helpers.FunctionPayload,
+				state,
+				testService,
+				retryOpts...),
 		).WithRetryOptions(
 			retry.Attempts(3),
 			retry.DelayType(retry.FixedDelay),

tests/end-to-end/external-solution-integration/pkg/testkit/send_event.go

@@ -0,0 +1,12 @@
+package testkit
+
+type SendEvent struct {
+	State   SendEventState
+	AppName string
+	Payload string
+}
+
+// SendEventState represents SendEvent dependencies
+type SendEventState interface {
+	GetEventSender() *EventSender
+}

tests/end-to-end/external-solution-integration/pkg/testkit/testservice.go

@@ -7,6 +7,8 @@ import (
 	"net/http"
 	"strconv"
 
+	cloudevents "github.com/cloudevents/sdk-go"
+
 	"k8s.io/apimachinery/pkg/runtime"
 
 	"k8s.io/client-go/dynamic"
@@ -103,6 +105,68 @@ func (ts *TestService) checkValue() (int, error) {
 	return response.Counter, nil
 }
 
+// DumpAllReceivedEvents returns an error containing all the received CloudEvents
+func (ts *TestService) DumpAllReceivedEvents() error {
+
+	url := ts.GetTestServiceURL()
+	endpoint := url + "/ce"
+
+	resp, err := ts.HttpClient.Get(endpoint)
+
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		body, err := ioutil.ReadAll(resp.Body)
+		if err != nil {
+			return err
+		}
+		return errors.Errorf("error response: %s", body)
+	}
+
+	var response []cloudevents.Event
+
+	err = json.NewDecoder(resp.Body).Decode(&response)
+	if err != nil {
+		return err
+	}
+
+	var receivedCEs []string
+
+	for _, event := range response {
+		s := fmt.Sprintf("Event %s: %+v \n", event.Context.GetID(), event)
+		receivedCEs = append(receivedCEs, s)
+	}
+
+	return errors.Errorf("received cloudevents: %s", receivedCEs)
+}
+
+func (ts *TestService) CheckEventId(eventId string) error {
+
+	url := ts.GetTestServiceURL()
+	endpoint := url + "/ce/by-uuid/" + eventId
+
+	resp, err := ts.HttpClient.Get(endpoint)
+
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		body, err := ioutil.ReadAll(resp.Body)
+		if err != nil {
+			return err
+		}
+
+		return errors.Errorf("error response: %s", body)
+	}
+
+	return nil
+}
+
 func (ts *TestService) IsReady() error {
 
 	url := ts.getHealthEndpointURL()

tests/end-to-end/external-solution-integration/pkg/testsuite/deploy_function.go

@@ -32,13 +32,22 @@ function rejected(result) {
   console.log('Rejected:');	
   console.log(JSON.stringify(result, null, 2));	
 }
-function sendReq(url, resolve, reject) {	
-    request.post(url, { json: true }, (error, response, body) => {	
-        if (error) {	
-            reject(error);	
-        }	
-        resolve(response) ;	
-    });	
+function sendReq(event, url, resolve, reject) {
+	const options = {
+		body: event,
+		uri: url,
+		json: true,
+		headers: {
+			'content-type': 'application/cloudevents+json'
+		}
+	}
+	console.log(JSON.stringify(options));
+	request.post(options, (error, response, body) => {	
+		if (error) {	
+			reject(error);	
+		}	
+		resolve(response) ;	
+	});
 }
 function getGateway() {
 	if (legacy) {
@@ -48,6 +57,16 @@ function getGateway() {
 		return process.env[envKey]
 	}
 }
+function prepareEvent(event){
+	return {
+		"specversion": event.extensions.request.headers["ce-specversion"],
+		"source": event.extensions.request.headers["ce-source"],
+		"type": event.extensions.request.headers["ce-type"],
+		"eventtypeversion": event.extensions.request.headers["ce-eventtypeversion"],
+		"id": event.extensions.request.headers["ce-id"],
+		"data" : event.data 
+	}
+}
 
 module.exports = { main: function (event, context) {	
 	console.log("==============================");	
@@ -69,9 +88,10 @@ module.exports = { main: function (event, context) {
 		throw new Error("Payload not as expected")	
     }
 	return new Promise((resolve, reject) => {	
-		const url = gateway;	
+		const url = gateway + "/ce";
+		var preparedEvent = prepareEvent(event);
 		console.log("Counter URL: ", url);	
-		sendReq(url, resolve, reject);	
+		sendReq(preparedEvent, url, resolve, reject);	
 	}).then(resolved, rejected);	
 } };	
 `

tests/end-to-end/external-solution-integration/pkg/testsuite/send_event_to_compatibility_layer.go

@@ -10,16 +10,14 @@ import (
 
 // SendEventToCompatibilityLayer is a step which sends example event to the application gateway
 type SendEventToCompatibilityLayer struct {
-	state   SendEventState
-	appName string
-	payload string
+	testkit.SendEvent
 }
 
 var _ step.Step = &SendEventToCompatibilityLayer{}
 
 // NewSendEventToCompatibilityLayer returns new SendEventToCompatibilityLayer
-func NewSendEventToCompatibilityLayer(appName, payload string, state SendEventState) *SendEventToCompatibilityLayer {
-	return &SendEventToCompatibilityLayer{state: state, appName: appName, payload: payload}
+func NewSendEventToCompatibilityLayer(appName, payload string, state testkit.SendEventState) *SendEventToCompatibilityLayer {
+	return &SendEventToCompatibilityLayer{testkit.SendEvent{State: state, AppName: appName, Payload: payload}}
 }
 
 // Name returns name name of the step
@@ -30,7 +28,7 @@ func (s *SendEventToCompatibilityLayer) Name() string {
 // Run executes the step
 func (s *SendEventToCompatibilityLayer) Run() error {
 	event := s.prepareEvent()
-	return s.state.GetEventSender().SendEventToCompatibilityLayer(s.appName, event)
+	return s.State.GetEventSender().SendEventToCompatibilityLayer(s.AppName, event)
 }
 
 func (s *SendEventToCompatibilityLayer) prepareEvent() *testkit.ExampleEvent {
@@ -39,7 +37,7 @@ func (s *SendEventToCompatibilityLayer) prepareEvent() *testkit.ExampleEvent {
 		EventTypeVersion: example_schema.EventVersion,
 		EventID:          "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
 		EventTime:        time.Now(),
-		Data:             s.payload,
+		Data:             s.Payload,
 	}
 }
 

tests/end-to-end/external-solution-integration/pkg/testsuite/send_event_to_compatibility_layer_and_check_eventId.go

@@ -0,0 +1,91 @@
+package testsuite
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/pkg/errors"
+
+	retrygo "github.com/avast/retry-go"
+	"github.com/kyma-project/kyma/tests/end-to-end/external-solution-integration/internal/example_schema"
+	"github.com/kyma-project/kyma/tests/end-to-end/external-solution-integration/pkg/retry"
+	"github.com/kyma-project/kyma/tests/end-to-end/external-solution-integration/pkg/step"
+	"github.com/kyma-project/kyma/tests/end-to-end/external-solution-integration/pkg/testkit"
+	"github.com/sirupsen/logrus"
+)
+
+// SendEventToCompatibilityLayerAndCheckEventId is a step which sends an event and checks if the correct EventId has been received
+type SendEventToCompatibilityLayerAndCheckEventId struct {
+	testkit.SendEvent
+	counter     int
+	testService *testkit.TestService
+	retryOpts   []retrygo.Option
+}
+
+var _ step.Step = &SendEventToCompatibilityLayerAndCheckEventId{}
+
+// NewSendEventToCompatibilityLayerAndCheckEventId returns new SendEventToCompatibilityLayerAndCheckEventId
+func NewSendEventToCompatibilityLayerAndCheckEventId(appName, payload string, state testkit.SendEventState, testService *testkit.TestService,
+	opts ...retrygo.Option) *SendEventToCompatibilityLayerAndCheckEventId {
+	return &SendEventToCompatibilityLayerAndCheckEventId{
+		testkit.SendEvent{State: state, AppName: appName, Payload: payload},
+		0,
+		testService,
+		opts,
+	}
+}
+
+// Name returns name of the step
+func (s *SendEventToCompatibilityLayerAndCheckEventId) Name() string {
+	return "Send event to compatibility layer and check event id"
+}
+
+// Run executes the step
+func (s *SendEventToCompatibilityLayerAndCheckEventId) Run() error {
+	const basicId = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa"
+	eventId := fmt.Sprint(basicId, s.counter)
+
+	err := s.sendEventToCompatibilityLayer(eventId)
+	if err != nil {
+		return err
+	}
+
+	err = s.checkEventId(eventId)
+	if err != nil {
+		s.counter++
+		return errors.Wrap(err, s.testService.DumpAllReceivedEvents().Error())
+	}
+
+	return nil
+}
+
+// Cleanup removes all resources that may possibly created by the step
+func (s *SendEventToCompatibilityLayerAndCheckEventId) Cleanup() error {
+	return nil
+}
+
+func (s *SendEventToCompatibilityLayerAndCheckEventId) checkEventId(eventId string) error {
+	err := retry.Do(func() error {
+		return s.testService.CheckEventId(eventId)
+	}, s.retryOpts...)
+
+	return err
+}
+
+func (s *SendEventToCompatibilityLayerAndCheckEventId) sendEventToCompatibilityLayer(eventId string) error {
+	event := s.prepareEvent(eventId)
+	err := s.State.GetEventSender().SendEventToCompatibilityLayer(s.AppName, event)
+	logrus.WithField("component", "SendEventToCompatibilityLayer").Debugf("SendCloudEventToCompatibilityLayer: eventID: %v; error: %v", eventId, err)
+
+	return err
+}
+
+func (s *SendEventToCompatibilityLayerAndCheckEventId) prepareEvent(eventId string) *testkit.ExampleEvent {
+	return &testkit.ExampleEvent{
+		EventType:        example_schema.EventType,
+		EventTypeVersion: example_schema.EventVersion,
+		EventID:          eventId,
+		EventTime:        time.Now(),
+		Data:             s.Payload,
+	}
+}

tests/end-to-end/external-solution-integration/pkg/testsuite/send_event_to_mesh.go

@@ -13,23 +13,16 @@ import (
 	"github.com/kyma-project/kyma/tests/end-to-end/external-solution-integration/pkg/testkit"
 )
 
-// SendEvent is a step which sends an example event to the application gateway
+// SendEventToMesh is a step which sends an example event to the application gateway
 type SendEventToMesh struct {
-	state   SendEventState
-	appName string
-	payload string
-}
-
-// SendEventState represents SendEvent dependencies
-type SendEventState interface {
-	GetEventSender() *testkit.EventSender
+	testkit.SendEvent
 }
 
 var _ step.Step = &SendEventToMesh{}
 
 // NewSendEvent returns new SendEvent
-func NewSendEventToMesh(appName, payload string, state SendEventState) *SendEventToMesh {
-	return &SendEventToMesh{state: state, appName: appName, payload: payload}
+func NewSendEventToMesh(appName, payload string, state testkit.SendEventState) *SendEventToMesh {
+	return &SendEventToMesh{testkit.SendEvent{State: state, AppName: appName, Payload: payload}}
 }
 
 // Name returns name name of the step
@@ -45,7 +38,7 @@ func (s *SendEventToMesh) Run() error {
 		return err
 	}
 
-	_, _, err = s.state.GetEventSender().SendCloudEventToMesh(ctx, event)
+	_, _, err = s.State.GetEventSender().SendCloudEventToMesh(ctx, event)
 	logrus.WithField("component", "SendEventToMesh").Debugf("SendCloudEventToMesh: eventID: %v; error: %v", event.ID(), err)
 	return err
 }
@@ -57,7 +50,7 @@ func (s *SendEventToMesh) prepareEvent() (cloudevents.Event, error) {
 	event.SetSource("some source")
 	// TODO(k15r): infer mime type automatically
 	event.SetDataContentType("text/plain")
-	if err := event.SetData(s.payload); err != nil {
+	if err := event.SetData(s.Payload); err != nil {
 		return event, err
 	}