diff --git a/common/util.go b/common/util.go index 66c6f6168f..7dea2bf5c9 100644 --- a/common/util.go +++ b/common/util.go @@ -44,8 +44,8 @@ func DefaultServiceName(serviceName string) string { return fmt.Sprintf("%s-svc", serviceName) } -// DefaultGatewayConfigurationName returns a formulated name for a gateway configuration -func DefaultGatewayConfigurationName(gatewayName string, configurationName string) string { +// DefaultEventSourceName returns a formulated name for a gateway configuration +func DefaultEventSourceName(gatewayName string, configurationName string) string { return fmt.Sprintf("%s:%s", gatewayName, configurationName) } @@ -88,6 +88,12 @@ func SendErrorResponse(writer http.ResponseWriter, response string) { writer.Write([]byte(response)) } +// SendInternalErrorResponse sends http internal error response +func SendInternalErrorResponse(writer http.ResponseWriter, response string) { + writer.WriteHeader(http.StatusInternalServerError) + writer.Write([]byte(response)) +} + // LoggerConf returns standard logging configuration func LoggerConf() zerolog.ConsoleWriter { output := zerolog.ConsoleWriter{Out: os.Stdout, TimeFormat: time.RFC3339} diff --git a/common/util_test.go b/common/util_test.go index 622d959fb8..7d3592b875 100644 --- a/common/util_test.go +++ b/common/util_test.go @@ -17,9 +17,10 @@ limitations under the License. package common import ( - "testing" - + "fmt" "github.com/smartystreets/goconvey/convey" + "net/http" + "testing" "github.com/stretchr/testify/assert" corev1 "k8s.io/api/core/v1" @@ -27,6 +28,24 @@ import ( "k8s.io/client-go/kubernetes/fake" ) +type fakeHttpWriter struct { + header int + payload []byte +} + +func (f *fakeHttpWriter) Header() http.Header { + return http.Header{} +} + +func (f *fakeHttpWriter) Write(body []byte) (int, error) { + f.payload = body + return len(body), nil +} + +func (f *fakeHttpWriter) WriteHeader(status int) { + f.header = status +} + func TestGetObjectHash(t *testing.T) { convey.Convey("Given a value, hash it", t, func() { hash, err := GetObjectHash(&corev1.Pod{}) @@ -47,6 +66,36 @@ func TestDefaultConfigMapName(t *testing.T) { assert.Equal(t, "sensor-controller-configmap", res) } +func TestDefaultServiceName(t *testing.T) { + convey.Convey("Given a service, get the default name", t, func() { + convey.So(DefaultServiceName("default"), convey.ShouldEqual, fmt.Sprintf("%s-svc", "default")) + }) +} + +func TestDefaultNatsQueueName(t *testing.T) { + convey.Convey("Given a nats queue, get the default name", t, func() { + convey.So(DefaultNatsQueueName("default"), convey.ShouldEqual, "default-queue") + }) +} + +func TestHTTPMethods(t *testing.T) { + convey.Convey("Given a http write", t, func() { + convey.Convey("Write a success response", func() { + f := &fakeHttpWriter{} + SendSuccessResponse(f, "hello") + convey.So(string(f.payload), convey.ShouldEqual, "hello") + convey.So(f.header, convey.ShouldEqual, http.StatusOK) + }) + + convey.Convey("Write a failure response", func() { + f := &fakeHttpWriter{} + SendErrorResponse(f, "failure") + convey.So(string(f.payload), convey.ShouldEqual, "failure") + convey.So(f.header, convey.ShouldEqual, http.StatusBadRequest) + }) + }) +} + func TestServerResourceForGroupVersionKind(t *testing.T) { convey.Convey("Given a k8s client", t, func() { fakeClient := fake.NewSimpleClientset() diff --git a/controllers/gateway/state_test.go b/controllers/gateway/state_test.go new file mode 100644 index 0000000000..cced645eb9 --- /dev/null +++ b/controllers/gateway/state_test.go @@ -0,0 +1,44 @@ +package gateway + +import ( + "testing" + + "github.com/argoproj/argo-events/common" + "github.com/argoproj/argo-events/pkg/client/gateway/clientset/versioned/fake" + "github.com/smartystreets/goconvey/convey" +) + +func TestPersistUpdates(t *testing.T) { + convey.Convey("Given a gateway resource", t, func() { + namespace := "argo-events" + client := fake.NewSimpleClientset() + logger := common.GetLoggerContext(common.LoggerConf()).Logger() + gw, err := getGateway() + convey.So(err, convey.ShouldBeNil) + + convey.Convey("Create the gateway", func() { + gw, err = client.ArgoprojV1alpha1().Gateways(namespace).Create(gw) + convey.So(err, convey.ShouldBeNil) + convey.So(gw, convey.ShouldNotBeNil) + + gw.ObjectMeta.Labels = map[string]string{ + "default": "default", + } + + convey.Convey("Update the gateway", func() { + updatedGw, err := PersistUpdates(client, gw, &logger) + convey.So(err, convey.ShouldBeNil) + convey.So(updatedGw, convey.ShouldNotEqual, gw) + convey.So(updatedGw.Labels, convey.ShouldResemble, gw.Labels) + + updatedGw.Labels["new"] = "new" + + convey.Convey("Reapply the gateway", func() { + err := ReapplyUpdates(client, updatedGw) + convey.So(err, convey.ShouldBeNil) + convey.So(len(updatedGw.Labels), convey.ShouldEqual, 2) + }) + }) + }) + }) +} diff --git a/docs/gateway-protocol.md b/docs/gateway-protocol.md index b6869dd737..81933fdf58 100644 --- a/docs/gateway-protocol.md +++ b/docs/gateway-protocol.md @@ -3,7 +3,7 @@ ## Table of Contents -- [github.com/argoproj/argo-events/pkg/apis/gateway/v1alpha1/generated.proto](#github.com/argoproj/argo-events/pkg/apis/gateway/v1alpha1/generated.proto) +- [pkg/apis/gateway/v1alpha1/generated.proto](#pkg/apis/gateway/v1alpha1/generated.proto) - [Gateway](#github.com.argoproj.argo_events.pkg.apis.gateway.v1alpha1.Gateway) - [GatewayList](#github.com.argoproj.argo_events.pkg.apis.gateway.v1alpha1.GatewayList) - [GatewayNotificationWatcher](#github.com.argoproj.argo_events.pkg.apis.gateway.v1alpha1.GatewayNotificationWatcher) @@ -22,10 +22,10 @@ - +
-## github.com/argoproj/argo-events/pkg/apis/gateway/v1alpha1/generated.proto +## pkg/apis/gateway/v1alpha1/generated.proto diff --git a/docs/sensor-protocol.md b/docs/sensor-protocol.md index ce1015269d..8ca4d083d8 100644 --- a/docs/sensor-protocol.md +++ b/docs/sensor-protocol.md @@ -3,7 +3,7 @@ ## Table of Contents -- [github.com/argoproj/argo-events/pkg/apis/sensor/v1alpha1/generated.proto](#github.com/argoproj/argo-events/pkg/apis/sensor/v1alpha1/generated.proto) +- [pkg/apis/sensor/v1alpha1/generated.proto](#pkg/apis/sensor/v1alpha1/generated.proto) - [ArtifactLocation](#github.com.argoproj.argo_events.pkg.apis.sensor.v1alpha1.ArtifactLocation) - [ConfigmapArtifact](#github.com.argoproj.argo_events.pkg.apis.sensor.v1alpha1.ConfigmapArtifact) - [DataFilter](#github.com.argoproj.argo_events.pkg.apis.sensor.v1alpha1.DataFilter) @@ -11,6 +11,9 @@ - [EventDependency](#github.com.argoproj.argo_events.pkg.apis.sensor.v1alpha1.EventDependency) - [EventDependencyFilter](#github.com.argoproj.argo_events.pkg.apis.sensor.v1alpha1.EventDependencyFilter) - [FileArtifact](#github.com.argoproj.argo_events.pkg.apis.sensor.v1alpha1.FileArtifact) + - [GitArtifact](#github.com.argoproj.argo_events.pkg.apis.sensor.v1alpha1.GitArtifact) + - [GitCreds](#github.com.argoproj.argo_events.pkg.apis.sensor.v1alpha1.GitCreds) + - [GitRemoteConfig](#github.com.argoproj.argo_events.pkg.apis.sensor.v1alpha1.GitRemoteConfig) - [GroupVersionKind](#github.com.argoproj.argo_events.pkg.apis.sensor.v1alpha1.GroupVersionKind) - [NodeStatus](#github.com.argoproj.argo_events.pkg.apis.sensor.v1alpha1.NodeStatus) - [ResourceObject](#github.com.argoproj.argo_events.pkg.apis.sensor.v1alpha1.ResourceObject) @@ -36,10 +39,10 @@ - + -## github.com/argoproj/argo-events/pkg/apis/sensor/v1alpha1/generated.proto +## pkg/apis/sensor/v1alpha1/generated.proto @@ -56,6 +59,7 @@ ArtifactLocation describes the source location for an external artifact | file | [FileArtifact](#github.com.argoproj.argo_events.pkg.apis.sensor.v1alpha1.FileArtifact) | optional | | | url | [URLArtifact](#github.com.argoproj.argo_events.pkg.apis.sensor.v1alpha1.URLArtifact) | optional | | | configmap | [ConfigmapArtifact](#github.com.argoproj.argo_events.pkg.apis.sensor.v1alpha1.ConfigmapArtifact) | optional | | +| git | [GitArtifact](#github.com.argoproj.argo_events.pkg.apis.sensor.v1alpha1.GitArtifact) | optional | | @@ -165,6 +169,61 @@ FileArtifact contains information about an artifact in a filesystem + + +### GitArtifact +GitArtifact contains information about an artifact stored in git + + +| Field | Type | Label | Description | +| ----- | ---- | ----- | ----------- | +| url | [string](#string) | optional | Git URL | +| cloneDirectory | [string](#string) | optional | Directory to clone the repository. We clone complete directory because GitArtifact is not limited to any specific Git service providers. Hence we don't use any specific git provider client. | +| creds | [GitCreds](#github.com.argoproj.argo_events.pkg.apis.sensor.v1alpha1.GitCreds) | optional | Creds contain reference to git username and password +optional | +| namespace | [string](#string) | optional | Namespace where creds are stored. +optional | +| sshKeyPath | [string](#string) | optional | SSHKeyPath is path to your ssh key path. Use this if you don't want to provide username and password. ssh key path must be mounted in sensor pod. +optional | +| filePath | [string](#string) | optional | Path to file that contains trigger resource definition | +| branch | [string](#string) | optional | Branch to use to pull trigger resource +optional | +| tag | [string](#string) | optional | Tag to use to pull trigger resource +optional | +| remote | [GitRemoteConfig](#github.com.argoproj.argo_events.pkg.apis.sensor.v1alpha1.GitRemoteConfig) | optional | Git remote to manage set of tracked repositories. Defaults to "origin". Refer https://git-scm.com/docs/git-remote +optional | + + + + + + + + +### GitCreds +GitCreds contain reference to git username and password + + +| Field | Type | Label | Description | +| ----- | ---- | ----- | ----------- | +| username | [k8s.io.api.core.v1.SecretKeySelector](#k8s.io.api.core.v1.SecretKeySelector) | optional | | +| password | [k8s.io.api.core.v1.SecretKeySelector](#k8s.io.api.core.v1.SecretKeySelector) | optional | | + + + + + + + + +### GitRemoteConfig + + + +| Field | Type | Label | Description | +| ----- | ---- | ----- | ----------- | +| name | [string](#string) | optional | Name of the remote to fetch from. | +| urls | [string](#string) | repeated | URLs the URLs of a remote repository. It must be non-empty. Fetch will always use the first URL, while push will use all of them. | + + + + + + ### GroupVersionKind diff --git a/gateways/common/aws.go b/gateways/common/aws.go new file mode 100644 index 0000000000..e9b6fd8769 --- /dev/null +++ b/gateways/common/aws.go @@ -0,0 +1,50 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package common + +import ( + "github.com/argoproj/argo-events/store" + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/aws/credentials" + "github.com/aws/aws-sdk-go/aws/session" + corev1 "k8s.io/api/core/v1" + "k8s.io/client-go/kubernetes" +) + +// GetAWSCreds reads credential stored in Kubernetes secrets and return it. +func GetAWSCreds(client kubernetes.Interface, namespace string, access *corev1.SecretKeySelector, secret *corev1.SecretKeySelector) (*credentials.Credentials, error) { + accessKey, err := store.GetSecrets(client, namespace, access.Name, access.Key) + if err != nil { + return nil, err + } + secretKey, err := store.GetSecrets(client, namespace, secret.Name, secret.Key) + if err != nil { + return nil, err + } + + return credentials.NewStaticCredentialsFromCreds(credentials.Value{ + AccessKeyID: accessKey, + SecretAccessKey: secretKey, + }), nil +} + +func GetAWSSession(creds *credentials.Credentials, region string) (*session.Session, error) { + return session.NewSession(&aws.Config{ + Region: ®ion, + Credentials: creds, + }) +} diff --git a/gateways/common/aws_test.go b/gateways/common/aws_test.go new file mode 100644 index 0000000000..d57e6ba75d --- /dev/null +++ b/gateways/common/aws_test.go @@ -0,0 +1,78 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package common + +import ( + "testing" + + "github.com/smartystreets/goconvey/convey" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes/fake" +) + +func TestAWS(t *testing.T) { + client := fake.NewSimpleClientset() + namespace := "test" + secretName := "test-secret" + accessKey := "YWNjZXNz" + secretKey := "c2VjcmV0" + LabelAccessKey := "access" + LabelSecretKey := "secret" + + convey.Convey("Given kubernetes secret that hold credentials, create AWS credential", t, func() { + secret, err := client.CoreV1().Secrets(namespace).Create(&corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: secretName, + Namespace: namespace, + }, + Data: map[string][]byte{ + LabelAccessKey: []byte(accessKey), + LabelSecretKey: []byte(secretKey), + }, + }) + + convey.So(err, convey.ShouldBeNil) + convey.So(secret, convey.ShouldNotBeNil) + + creds, err := GetAWSCreds(client, namespace, &corev1.SecretKeySelector{ + Key: LabelAccessKey, + LocalObjectReference: corev1.LocalObjectReference{ + Name: secretName, + }, + }, &corev1.SecretKeySelector{ + Key: LabelSecretKey, + LocalObjectReference: corev1.LocalObjectReference{ + Name: secretName, + }, + }) + + convey.So(err, convey.ShouldBeNil) + convey.So(creds, convey.ShouldNotBeNil) + + value, err := creds.Get() + convey.So(err, convey.ShouldBeNil) + convey.So(value.AccessKeyID, convey.ShouldEqual, accessKey) + convey.So(value.SecretAccessKey, convey.ShouldEqual, secretKey) + + convey.Convey("Get a new aws session", func() { + session, err := GetAWSSession(creds, "mock-region") + convey.So(err, convey.ShouldBeNil) + convey.So(session, convey.ShouldNotBeNil) + }) + }) +} diff --git a/gateways/common/fake.go b/gateways/common/fake.go new file mode 100644 index 0000000000..e57811ec3a --- /dev/null +++ b/gateways/common/fake.go @@ -0,0 +1,81 @@ +package common + +import ( + "context" + "github.com/argoproj/argo-events/common" + "github.com/argoproj/argo-events/gateways" + "google.golang.org/grpc/metadata" + "net/http" +) + +var Hook = &Webhook{ + Endpoint: "/fake", + Port: "12000", + URL: "test-url", +} + +type FakeHttpWriter struct { + HeaderStatus int + Payload []byte +} + +func (f *FakeHttpWriter) Header() http.Header { + return http.Header{} +} + +func (f *FakeHttpWriter) Write(body []byte) (int, error) { + f.Payload = body + return len(body), nil +} + +func (f *FakeHttpWriter) WriteHeader(status int) { + f.HeaderStatus = status +} + +func GetFakeRouteConfig() *RouteConfig { + return &RouteConfig{ + Webhook: Hook, + EventSource: &gateways.EventSource{ + Name: "fake-event-source", + Data: "hello", + Id: "123", + }, + Log: common.GetLoggerContext(common.LoggerConf()).Logger(), + Configs: make(map[string]interface{}), + StartCh: make(chan struct{}), + } +} + +type FakeGRPCStream struct { + SentData *gateways.Event + Ctx context.Context +} + +func (f *FakeGRPCStream) Send(event *gateways.Event) error { + f.SentData = event + return nil +} + +func (f *FakeGRPCStream) SetHeader(metadata.MD) error { + return nil +} + +func (f *FakeGRPCStream) SendHeader(metadata.MD) error { + return nil +} + +func (f *FakeGRPCStream) SetTrailer(metadata.MD) { + return +} + +func (f *FakeGRPCStream) Context() context.Context { + return f.Ctx +} + +func (f *FakeGRPCStream) SendMsg(m interface{}) error { + return nil +} + +func (f *FakeGRPCStream) RecvMsg(m interface{}) error { + return nil +} diff --git a/gateways/common/validate_test.go b/gateways/common/validate_test.go new file mode 100644 index 0000000000..5f7f2e5271 --- /dev/null +++ b/gateways/common/validate_test.go @@ -0,0 +1,41 @@ +package common + +import ( + "fmt" + "github.com/ghodss/yaml" + "github.com/smartystreets/goconvey/convey" + "testing" +) + +type fakeEventSource struct { + Msg string `json:"msg"` +} + +func TestValidateGatewayEventSource(t *testing.T) { + convey.Convey("Given an event source, validate it", t, func() { + esStr := ` +msg: "hello" +` + es, err := ValidateGatewayEventSource(esStr, func(s string) (i interface{}, e error) { + t := &fakeEventSource{} + err := yaml.Unmarshal([]byte(esStr), t) + return t, err + }, func(i interface{}) error { + t, ok := i.(*fakeEventSource) + if !ok { + return fmt.Errorf("failed to cast to fake event source") + } + if t == nil { + return fmt.Errorf("event source cant be nil") + } + if t.Msg == "" { + return fmt.Errorf("msg can't be empty") + } + return nil + }) + + convey.So(err, convey.ShouldBeNil) + convey.So(es, convey.ShouldNotBeNil) + convey.So(es.IsValid, convey.ShouldEqual, true) + }) +} diff --git a/gateways/common/webhook.go b/gateways/common/webhook.go index 580cc6dbe9..b8f7578fcc 100644 --- a/gateways/common/webhook.go +++ b/gateways/common/webhook.go @@ -155,7 +155,7 @@ func (rc *RouteConfig) startHttpServer(helper *WebhookHelper) { } else { err = rc.Webhook.srv.ListenAndServeTLS(rc.Webhook.ServerCertPath, rc.Webhook.ServerKeyPath) } - rc.Log.Info().Str("event-source", rc.EventSource.Name).Str("port", rc.Webhook.Port).Msg("http server stopped") + rc.Log.Error().Err(err).Str("event-source", rc.EventSource.Name).Str("port", rc.Webhook.Port).Msg("http server stopped") if err != nil { errChan <- err } diff --git a/gateways/common/webhook_test.go b/gateways/common/webhook_test.go new file mode 100644 index 0000000000..d0c79c0b3a --- /dev/null +++ b/gateways/common/webhook_test.go @@ -0,0 +1,168 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package common + +import ( + "context" + "fmt" + "net/http" + "testing" + "time" + + "github.com/smartystreets/goconvey/convey" +) + +func TestDefaultPostActivate(t *testing.T) { + convey.Convey("Given a route configuration, default post activate should be a no-op", t, func() { + rc := GetFakeRouteConfig() + err := DefaultPostActivate(rc) + convey.So(err, convey.ShouldBeNil) + }) +} + +func TestDefaultPostStop(t *testing.T) { + convey.Convey("Given a route configuration, default post stop should be a no-op", t, func() { + rc := GetFakeRouteConfig() + err := DefaultPostStop(rc) + convey.So(err, convey.ShouldBeNil) + }) +} + +func TestProcessRoute(t *testing.T) { + convey.Convey("Given a route configuration", t, func() { + convey.Convey("Activate the route configuration", func() { + rc := GetFakeRouteConfig() + rc.Webhook.mux = http.NewServeMux() + + rc.PostActivate = DefaultPostActivate + rc.PostStop = DefaultPostStop + + ctx, cancel := context.WithCancel(context.Background()) + fgs := &FakeGRPCStream{ + Ctx: ctx, + } + + helper := NewWebhookHelper() + helper.ActiveEndpoints[rc.Webhook.Endpoint] = &Endpoint{ + DataCh: make(chan []byte), + } + helper.ActiveServers[rc.Webhook.Port] = &activeServer{ + errChan: make(chan error), + } + + errCh := make(chan error) + go func() { + <-helper.RouteDeactivateChan + }() + + go func() { + <-helper.RouteActivateChan + }() + go func() { + rc.StartCh <- struct{}{} + }() + go func() { + time.Sleep(3 * time.Second) + cancel() + }() + + go func() { + errCh <- ProcessRoute(rc, helper, fgs) + }() + + err := <-errCh + convey.So(err, convey.ShouldBeNil) + }) + }) +} + +func TestProcessRouteChannels(t *testing.T) { + convey.Convey("Given a route configuration", t, func() { + convey.Convey("Stop server stream", func() { + rc := GetFakeRouteConfig() + ctx, cancel := context.WithCancel(context.Background()) + fgs := &FakeGRPCStream{ + Ctx: ctx, + } + helper := NewWebhookHelper() + helper.ActiveEndpoints[rc.Webhook.Endpoint] = &Endpoint{ + DataCh: make(chan []byte), + } + helper.ActiveServers[rc.Webhook.Port] = &activeServer{ + errChan: make(chan error), + } + errCh := make(chan error) + go func() { + <-helper.RouteDeactivateChan + }() + go func() { + errCh <- rc.processChannels(helper, fgs) + }() + cancel() + err := <-errCh + convey.So(err, convey.ShouldBeNil) + }) + convey.Convey("Handle error", func() { + rc := GetFakeRouteConfig() + fgs := &FakeGRPCStream{ + Ctx: context.Background(), + } + helper := NewWebhookHelper() + helper.ActiveEndpoints[rc.Webhook.Endpoint] = &Endpoint{ + DataCh: make(chan []byte), + } + helper.ActiveServers[rc.Webhook.Port] = &activeServer{ + errChan: make(chan error), + } + errCh := make(chan error) + err := fmt.Errorf("error") + go func() { + helper.ActiveServers[rc.Webhook.Port].errChan <- err + }() + go func() { + errCh <- rc.processChannels(helper, fgs) + }() + newErr := <-errCh + convey.So(newErr.Error(), convey.ShouldEqual, err.Error()) + }) + }) +} + +func TestFormatWebhookEndpoint(t *testing.T) { + convey.Convey("Given a webhook endpoint, format it", t, func() { + convey.So(FormatWebhookEndpoint("hello"), convey.ShouldEqual, "/hello") + }) +} + +func TestValidateWebhook(t *testing.T) { + convey.Convey("Given a webhook, validate it", t, func() { + convey.So(ValidateWebhook(Hook), convey.ShouldBeNil) + }) +} + +func TestGenerateFormattedURL(t *testing.T) { + convey.Convey("Given a webhook, generate formatted URL", t, func() { + convey.So(GenerateFormattedURL(Hook), convey.ShouldEqual, "test-url/fake") + }) +} + +func TestNewWebhookHelper(t *testing.T) { + convey.Convey("Make sure webhook helper is not empty", t, func() { + helper := NewWebhookHelper() + convey.So(helper, convey.ShouldNotBeNil) + }) +} diff --git a/gateways/community/README.md b/gateways/community/README.md deleted file mode 100644 index 32ea89b366..0000000000 --- a/gateways/community/README.md +++ /dev/null @@ -1,34 +0,0 @@ -# Community Gateways - -## How to set up gateways that run web server internally? -For any gateway that run web server internally, you need to create the gateway resource before creating gateway configmap. -Why? Because the `webhook` configuration in gateway configmap contains the `url` which the URL of the service that exposes the gateway pod, you need to have -the gateway pod and service running before you can create the configmap. - - ### Example - - Lets setup AWS SNS gateway, - - 1. Create the gateway resource, - ```bash - kubectl -n argo-events create -f https://raw.githubusercontent.com/argoproj/argo-events/master/examples/gateways/aws-sns.yaml - ``` - - Once the gateway resource is created, you should see the corresponding gateway pod and service being created. - Make sure the gateway service is being backed by the gateway pod. - - How to get the URL for that resolves to your service is up to you. You can use Ingress controller, Route (if your installation is on Openshift) or other means. - - 2. Create the gateway configmap. The gateway configmap basically contains the event source configurations. You need to put the URL of your service. - - ```bash - kubectl -n argo-events create -f ../../examples/gateway/aws-sns-gateway-configmap.yaml - ``` - - 3. Create the sensor. - - ```bash - kubectl -n argo-events create -f https://raw.githubusercontent.com/argoproj/argo-events/master/examples/sensors/aws-sns.yaml - ``` - - Now your setup is complete. diff --git a/gateways/community/aws-sns/config_test.go b/gateways/community/aws-sns/config_test.go new file mode 100644 index 0000000000..e1c07c570f --- /dev/null +++ b/gateways/community/aws-sns/config_test.go @@ -0,0 +1,47 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package aws_sns + +import ( + "github.com/smartystreets/goconvey/convey" + "testing" +) + +var es = ` +hook: + endpoint: "/test" + port: "8080" + url: "myurl/test" +topicArn: "test-arn" +region: "us-east-1" +accessKey: + key: accesskey + name: sns +secretKey: + key: secretkey + name: sns +` + +func TestParseConfig(t *testing.T) { + convey.Convey("Given a aws-sns event source, parse it", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + _, ok := ps.(*snsConfig) + convey.So(ok, convey.ShouldEqual, true) + }) +} diff --git a/gateways/community/aws-sns/start.go b/gateways/community/aws-sns/start.go index 6cc067b5d0..7d1723e69b 100644 --- a/gateways/community/aws-sns/start.go +++ b/gateways/community/aws-sns/start.go @@ -24,10 +24,6 @@ import ( "github.com/argoproj/argo-events/common" "github.com/argoproj/argo-events/gateways" gwcommon "github.com/argoproj/argo-events/gateways/common" - "github.com/argoproj/argo-events/store" - "github.com/aws/aws-sdk-go/aws" - "github.com/aws/aws-sdk-go/aws/credentials" - "github.com/aws/aws-sdk-go/aws/session" snslib "github.com/aws/aws-sdk-go/service/sns" "github.com/ghodss/yaml" ) @@ -65,7 +61,7 @@ func RouteActiveHandler(writer http.ResponseWriter, request *http.Request, rc *g body, err := ioutil.ReadAll(request.Body) if err != nil { logger.Error().Err(err).Msg("failed to parse request body") - common.SendErrorResponse(writer, fmt.Sprintf("failed to parse request. err: %+v", err)) + common.SendErrorResponse(writer, "failed to parse request") return } @@ -73,6 +69,7 @@ func RouteActiveHandler(writer http.ResponseWriter, request *http.Request, rc *g err = yaml.Unmarshal(body, &snspayload) if err != nil { logger.Error().Err(err).Msg("failed to convert request payload into snsConfig payload") + common.SendErrorResponse(writer, "failed to marshal request") return } @@ -87,6 +84,7 @@ func RouteActiveHandler(writer http.ResponseWriter, request *http.Request, rc *g }) if err != nil { logger.Error().Err(err).Msg("failed to send confirmation response to amazon") + common.SendErrorResponse(writer, "failed to confirm subscription") return } rc.Configs[labelSubscriptionArn] = out.SubscriptionArn @@ -106,39 +104,24 @@ func (ese *SNSEventSourceExecutor) PostActivate(rc *gwcommon.RouteConfig) error Str("port", rc.Webhook.Port).Logger() sc := rc.Configs[labelSNSConfig].(*snsConfig) - // retrieve access key id and secret access key - accessKey, err := store.GetSecrets(ese.Clientset, ese.Namespace, sc.AccessKey.Name, sc.AccessKey.Key) - if err != nil { - logger.Error().Err(err).Msg("failed to retrieve access key") - return err - } - secretKey, err := store.GetSecrets(ese.Clientset, ese.Namespace, sc.SecretKey.Name, sc.SecretKey.Key) + + creds, err := gwcommon.GetAWSCreds(ese.Clientset, ese.Namespace, sc.AccessKey, sc.SecretKey) if err != nil { - logger.Error().Err(err).Msg("failed to retrieve secret key") - return err + logger.Error().Err(err).Msg("failed to get aws credentials") } - creds := credentials.NewStaticCredentialsFromCreds(credentials.Value{ - AccessKeyID: accessKey, - SecretAccessKey: secretKey, - }) - - formattedUrl := gwcommon.GenerateFormattedURL(sc.Hook) - - awsSession, err := session.NewSession(&aws.Config{ - Region: &sc.Region, - Credentials: creds, - HTTPClient: &http.Client{}, - }) + awsSession, err := gwcommon.GetAWSSession(creds, sc.Region) if err != nil { logger.Error().Err(err).Msg("failed to create new session") return err } + logger.Info().Msg("subscribing to sns topic") + snsSession := snslib.New(awsSession) rc.Configs[labelSNSSession] = snsSession + formattedUrl := gwcommon.GenerateFormattedURL(sc.Hook) - logger.Info().Msg("subscribing to snsConfig topic") if _, err := snsSession.Subscribe(&snslib.SubscribeInput{ Endpoint: &formattedUrl, Protocol: &snsProtocol, diff --git a/gateways/community/aws-sns/start_test.go b/gateways/community/aws-sns/start_test.go new file mode 100644 index 0000000000..46dd2f0c97 --- /dev/null +++ b/gateways/community/aws-sns/start_test.go @@ -0,0 +1,106 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package aws_sns + +import ( + "bytes" + "github.com/argoproj/argo-events/common" + gwcommon "github.com/argoproj/argo-events/gateways/common" + "github.com/aws/aws-sdk-go/aws/credentials" + snslib "github.com/aws/aws-sdk-go/service/sns" + "github.com/ghodss/yaml" + "github.com/smartystreets/goconvey/convey" + "io/ioutil" + "k8s.io/client-go/kubernetes/fake" + "net/http" + "testing" +) + +func TestAWSSNS(t *testing.T) { + convey.Convey("Given an route configuration", t, func() { + rc := gwcommon.GetFakeRouteConfig() + helper.ActiveEndpoints[rc.Webhook.Endpoint] = &gwcommon.Endpoint{ + DataCh: make(chan []byte), + } + writer := &gwcommon.FakeHttpWriter{} + subscriptionArn := "arn://fake" + awsSession, err := gwcommon.GetAWSSession(credentials.NewStaticCredentialsFromCreds(credentials.Value{ + AccessKeyID: "access", + SecretAccessKey: "secret", + }), "mock-region") + + convey.So(err, convey.ShouldBeNil) + + snsSession := snslib.New(awsSession) + rc.Configs[labelSNSSession] = snsSession + rc.Configs[labelSubscriptionArn] = &subscriptionArn + + convey.Convey("handle the inactive route", func() { + RouteActiveHandler(writer, &http.Request{}, rc) + convey.So(writer.HeaderStatus, convey.ShouldEqual, http.StatusBadRequest) + }) + + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + + helper.ActiveEndpoints[rc.Webhook.Endpoint].Active = true + rc.Configs[labelSNSConfig] = ps.(*snsConfig) + + convey.Convey("handle the active route", func() { + payload := httpNotification{ + TopicArn: "arn://fake", + Token: "faketoken", + Type: messageTypeSubscriptionConfirmation, + } + + payloadBytes, err := yaml.Marshal(payload) + convey.So(err, convey.ShouldBeNil) + RouteActiveHandler(writer, &http.Request{ + Body: ioutil.NopCloser(bytes.NewBuffer(payloadBytes)), + }, rc) + convey.So(writer.HeaderStatus, convey.ShouldEqual, http.StatusBadRequest) + convey.So(string(writer.Payload), convey.ShouldEqual, "failed to confirm subscription") + + go func() { + <-helper.ActiveEndpoints[rc.Webhook.Endpoint].DataCh + }() + + payload.Type = messageTypeNotification + payloadBytes, err = yaml.Marshal(payload) + convey.So(err, convey.ShouldBeNil) + RouteActiveHandler(writer, &http.Request{ + Body: ioutil.NopCloser(bytes.NewBuffer(payloadBytes)), + }, rc) + convey.So(writer.HeaderStatus, convey.ShouldEqual, http.StatusOK) + }) + + convey.Convey("Run post activate", func() { + ese := SNSEventSourceExecutor{ + Namespace: "fake", + Clientset: fake.NewSimpleClientset(), + Log: common.GetLoggerContext(common.LoggerConf()).Logger(), + } + err := ese.PostActivate(rc) + convey.So(err, convey.ShouldNotBeNil) + }) + + convey.Convey("Run post stop", func() { + err = PostStop(rc) + convey.So(err, convey.ShouldNotBeNil) + }) + }) +} diff --git a/gateways/community/aws-sns/validate_test.go b/gateways/community/aws-sns/validate_test.go index 9fb2377895..85bfb39295 100644 --- a/gateways/community/aws-sns/validate_test.go +++ b/gateways/community/aws-sns/validate_test.go @@ -39,7 +39,6 @@ accessKey: secretKey: key: secretkey name: sns - ` invalidConfig = ` diff --git a/gateways/community/aws-sqs/config_test.go b/gateways/community/aws-sqs/config_test.go new file mode 100644 index 0000000000..5d6929dfe8 --- /dev/null +++ b/gateways/community/aws-sqs/config_test.go @@ -0,0 +1,45 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package aws_sqs + +import ( + "testing" + + "github.com/smartystreets/goconvey/convey" +) + +var es = ` +region: "us-east-1" +accessKey: + key: accesskey + name: sns +secretKey: + key: secretkey + name: sns +queue: "test-queue" +waitTimeSeconds: 10 +` + +func TestParseConfig(t *testing.T) { + convey.Convey("Given a aws-sqs event source, parse it", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + _, ok := ps.(*sqs) + convey.So(ok, convey.ShouldEqual, true) + }) +} diff --git a/gateways/community/aws-sqs/start.go b/gateways/community/aws-sqs/start.go index 68280e4c8a..df44dfdee4 100644 --- a/gateways/community/aws-sqs/start.go +++ b/gateways/community/aws-sqs/start.go @@ -18,10 +18,8 @@ package aws_sqs import ( "github.com/argoproj/argo-events/gateways" - "github.com/argoproj/argo-events/store" + gwcommon "github.com/argoproj/argo-events/gateways/common" "github.com/aws/aws-sdk-go/aws" - "github.com/aws/aws-sdk-go/aws/credentials" - "github.com/aws/aws-sdk-go/aws/session" sqslib "github.com/aws/aws-sdk-go/service/sqs" ) @@ -48,26 +46,13 @@ func (ese *SQSEventSourceExecutor) StartEventSource(eventSource *gateways.EventS func (ese *SQSEventSourceExecutor) listenEvents(s *sqs, eventSource *gateways.EventSource, dataCh chan []byte, errorCh chan error, doneCh chan struct{}) { defer gateways.Recover(eventSource.Name) - // retrieve access key id and secret access key - accessKey, err := store.GetSecrets(ese.Clientset, ese.Namespace, s.AccessKey.Name, s.AccessKey.Key) - if err != nil { - errorCh <- err - return - } - secretKey, err := store.GetSecrets(ese.Clientset, ese.Namespace, s.SecretKey.Name, s.SecretKey.Key) + creds, err := gwcommon.GetAWSCreds(ese.Clientset, ese.Namespace, s.AccessKey, s.SecretKey) if err != nil { errorCh <- err return } - creds := credentials.NewStaticCredentialsFromCreds(credentials.Value{ - AccessKeyID: accessKey, - SecretAccessKey: secretKey, - }) - awsSession, err := session.NewSession(&aws.Config{ - Region: &s.Region, - Credentials: creds, - }) + awsSession, err := gwcommon.GetAWSSession(creds, s.Region) if err != nil { errorCh <- err return diff --git a/gateways/community/aws-sqs/start_test.go b/gateways/community/aws-sqs/start_test.go new file mode 100644 index 0000000000..499b311f5d --- /dev/null +++ b/gateways/community/aws-sqs/start_test.go @@ -0,0 +1,59 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package aws_sqs + +import ( + "testing" + + "github.com/argoproj/argo-events/common" + "github.com/argoproj/argo-events/gateways" + "github.com/smartystreets/goconvey/convey" + "k8s.io/client-go/kubernetes/fake" +) + +func TestListenEvents(t *testing.T) { + convey.Convey("Given an event source, listen to events", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + + ese := &SQSEventSourceExecutor{ + Clientset: fake.NewSimpleClientset(), + Namespace: "fake", + Log: common.GetLoggerContext(common.LoggerConf()).Logger(), + } + + dataCh := make(chan []byte) + errorCh := make(chan error) + doneCh := make(chan struct{}, 1) + errorCh2 := make(chan error) + + go func() { + err := <-errorCh + errorCh2 <- err + }() + + ese.listenEvents(ps.(*sqs), &gateways.EventSource{ + Name: "fake", + Data: es, + Id: "1234", + }, dataCh, errorCh, doneCh) + + err = <-errorCh2 + convey.So(err, convey.ShouldNotBeNil) + }) +} diff --git a/gateways/community/gcp-pubsub/config_test.go b/gateways/community/gcp-pubsub/config_test.go new file mode 100644 index 0000000000..2eb280bf48 --- /dev/null +++ b/gateways/community/gcp-pubsub/config_test.go @@ -0,0 +1,37 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package pubsub + +import ( + "github.com/smartystreets/goconvey/convey" + "testing" +) + +var es = ` +projectID: "1234" +topic: "test" +` + +func TestParseConfig(t *testing.T) { + convey.Convey("Given a gcp-pubsub event source, parse it", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + _, ok := ps.(*pubSubConfig) + convey.So(ok, convey.ShouldEqual, true) + }) +} diff --git a/gateways/community/gcp-pubsub/start_test.go b/gateways/community/gcp-pubsub/start_test.go new file mode 100644 index 0000000000..ea21d8c23c --- /dev/null +++ b/gateways/community/gcp-pubsub/start_test.go @@ -0,0 +1,57 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package pubsub + +import ( + "context" + "github.com/argoproj/argo-events/common" + "github.com/argoproj/argo-events/gateways" + "github.com/smartystreets/goconvey/convey" + "testing" +) + +func TestListenEvents(t *testing.T) { + convey.Convey("Given a pubsub event source, listen to events", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + psc := ps.(*pubSubConfig) + + ese := &GcpPubSubEventSourceExecutor{ + Log: common.GetLoggerContext(common.LoggerConf()).Logger(), + } + + dataCh := make(chan []byte) + errorCh := make(chan error) + doneCh := make(chan struct{}, 1) + errCh2 := make(chan error) + + go func() { + err := <-errorCh + errCh2 <- err + }() + + ese.listenEvents(context.Background(), psc, &gateways.EventSource{ + Name: "fake", + Data: es, + Id: "1234", + }, dataCh, errorCh, doneCh) + + err = <-errCh2 + convey.So(err, convey.ShouldNotBeNil) + }) +} diff --git a/gateways/community/github/config.go b/gateways/community/github/config.go index f38ee9e029..22a5610089 100644 --- a/gateways/community/github/config.go +++ b/gateways/community/github/config.go @@ -19,7 +19,6 @@ package github import ( "github.com/argoproj/argo-events/gateways/common" "github.com/ghodss/yaml" - "github.com/google/go-github/github" "github.com/rs/zerolog" corev1 "k8s.io/api/core/v1" "k8s.io/client-go/kubernetes" @@ -28,8 +27,6 @@ import ( // GithubEventSourceExecutor implements ConfigExecutor type GithubEventSourceExecutor struct { Log zerolog.Logger - // GitlabClient is client for gitlab api - GithubClient *github.Client // Clientset is kubernetes client Clientset kubernetes.Interface // Namespace where gateway is deployed diff --git a/gateways/community/github/config_test.go b/gateways/community/github/config_test.go new file mode 100644 index 0000000000..dde443535d --- /dev/null +++ b/gateways/community/github/config_test.go @@ -0,0 +1,46 @@ +/* +Copyright 2018 KompiTech GmbH + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package github + +import ( + "github.com/smartystreets/goconvey/convey" + "testing" +) + +var es = ` +hook: + endpoint: "/push" + port: "12000" + url: "http://webhook-gateway-svc" +owner: "asd" +repository: "dsa" +events: +- PushEvents +apiToken: + key: accesskey + name: githab-access +` + +func TestParseConfig(t *testing.T) { + convey.Convey("Given a github event source, parse it", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + _, ok := ps.(*githubConfig) + convey.So(ok, convey.ShouldEqual, true) + }) +} diff --git a/gateways/community/github/start_test.go b/gateways/community/github/start_test.go new file mode 100644 index 0000000000..628de12ca2 --- /dev/null +++ b/gateways/community/github/start_test.go @@ -0,0 +1,105 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package github + +import ( + "bytes" + "io/ioutil" + "net/http" + "testing" + + "github.com/argoproj/argo-events/common" + gwcommon "github.com/argoproj/argo-events/gateways/common" + "github.com/ghodss/yaml" + "github.com/google/go-github/github" + "github.com/smartystreets/goconvey/convey" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes/fake" +) + +var ( + ese = &GithubEventSourceExecutor{ + Clientset: fake.NewSimpleClientset(), + Namespace: "fake", + Log: common.GetLoggerContext(common.LoggerConf()).Logger(), + } + + secretName = "githab-access" + accessKey = "YWNjZXNz" + LabelAccessKey = "accesskey" +) + +func TestGetCredentials(t *testing.T) { + convey.Convey("Given a kubernetes secret, get credentials", t, func() { + secret, err := ese.Clientset.CoreV1().Secrets(ese.Namespace).Create(&corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: secretName, + Namespace: ese.Namespace, + }, + Data: map[string][]byte{ + LabelAccessKey: []byte(accessKey), + }, + }) + convey.So(err, convey.ShouldBeNil) + convey.So(secret, convey.ShouldNotBeNil) + + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + creds, err := ese.getCredentials(ps.(*githubConfig).APIToken) + convey.So(err, convey.ShouldBeNil) + convey.So(creds, convey.ShouldNotBeNil) + convey.So(creds.secret, convey.ShouldEqual, "YWNjZXNz") + }) +} + +func TestRouteActiveHandler(t *testing.T) { + convey.Convey("Given a route configuration", t, func() { + rc := gwcommon.GetFakeRouteConfig() + helper.ActiveEndpoints[rc.Webhook.Endpoint] = &gwcommon.Endpoint{ + DataCh: make(chan []byte), + } + + convey.Convey("Inactive route should return error", func() { + writer := &gwcommon.FakeHttpWriter{} + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + pbytes, err := yaml.Marshal(ps.(*githubConfig)) + convey.So(err, convey.ShouldBeNil) + RouteActiveHandler(writer, &http.Request{ + Body: ioutil.NopCloser(bytes.NewReader(pbytes)), + }, rc) + convey.So(writer.HeaderStatus, convey.ShouldEqual, http.StatusBadRequest) + + convey.Convey("Active route should return success", func() { + helper.ActiveEndpoints[rc.Webhook.Endpoint].Active = true + rc.Configs[labelWebhook] = &github.Hook{ + Config: make(map[string]interface{}), + } + + RouteActiveHandler(writer, &http.Request{ + Body: ioutil.NopCloser(bytes.NewReader(pbytes)), + }, rc) + + convey.So(writer.HeaderStatus, convey.ShouldEqual, http.StatusBadRequest) + rc.Configs[labelGithubConfig] = ps.(*githubConfig) + err = ese.PostActivate(rc) + convey.So(err, convey.ShouldNotBeNil) + }) + }) + }) +} diff --git a/gateways/community/gitlab/config_test.go b/gateways/community/gitlab/config_test.go new file mode 100644 index 0000000000..6f88cdef76 --- /dev/null +++ b/gateways/community/gitlab/config_test.go @@ -0,0 +1,47 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package gitlab + +import ( + "testing" + + "github.com/smartystreets/goconvey/convey" +) + +var es = ` +hook: + endpoint: "/push" + port: "12000" + url: "http://webhook-gateway-gateway-svc/push" +projectId: "28" +event: "PushEvents" +accessToken: + key: accesskey + name: gitlab-access +enableSSLVerification: false +gitlabBaseUrl: "http://gitlab.com/" +` + +func TestParseConfig(t *testing.T) { + convey.Convey("Given a gitlab event source, parse it", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + _, ok := ps.(*glab) + convey.So(ok, convey.ShouldEqual, true) + }) +} diff --git a/gateways/community/gitlab/start_test.go b/gateways/community/gitlab/start_test.go new file mode 100644 index 0000000000..b75716ffdf --- /dev/null +++ b/gateways/community/gitlab/start_test.go @@ -0,0 +1,112 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package gitlab + +import ( + "bytes" + "io/ioutil" + "net/http" + "testing" + + "github.com/argoproj/argo-events/common" + gwcommon "github.com/argoproj/argo-events/gateways/common" + "github.com/ghodss/yaml" + "github.com/google/go-github/github" + "github.com/smartystreets/goconvey/convey" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes/fake" +) + +var ( + ese = &GitlabEventSourceExecutor{ + Clientset: fake.NewSimpleClientset(), + Namespace: "fake", + Log: common.GetLoggerContext(common.LoggerConf()).Logger(), + } + + secretName = "gitlab-access" + accessKey = "YWNjZXNz" + LabelAccessKey = "accesskey" +) + +func TestGetCredentials(t *testing.T) { + convey.Convey("Given a kubernetes secret, get credentials", t, func() { + secret, err := ese.Clientset.CoreV1().Secrets(ese.Namespace).Create(&corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: secretName, + Namespace: ese.Namespace, + }, + Data: map[string][]byte{ + LabelAccessKey: []byte(accessKey), + }, + }) + convey.So(err, convey.ShouldBeNil) + convey.So(secret, convey.ShouldNotBeNil) + + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + creds, err := ese.getCredentials(ps.(*glab).AccessToken) + convey.So(err, convey.ShouldBeNil) + convey.So(creds, convey.ShouldNotBeNil) + convey.So(creds.token, convey.ShouldEqual, "YWNjZXNz") + }) +} + +func TestRouteActiveHandler(t *testing.T) { + convey.Convey("Given a route configuration", t, func() { + rc := gwcommon.GetFakeRouteConfig() + helper.ActiveEndpoints[rc.Webhook.Endpoint] = &gwcommon.Endpoint{ + DataCh: make(chan []byte), + } + + convey.Convey("Inactive route should return error", func() { + writer := &gwcommon.FakeHttpWriter{} + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + pbytes, err := yaml.Marshal(ps.(*glab)) + convey.So(err, convey.ShouldBeNil) + RouteActiveHandler(writer, &http.Request{ + Body: ioutil.NopCloser(bytes.NewReader(pbytes)), + }, rc) + convey.So(writer.HeaderStatus, convey.ShouldEqual, http.StatusBadRequest) + + convey.Convey("Active route should return success", func() { + helper.ActiveEndpoints[rc.Webhook.Endpoint].Active = true + rc.Configs[labelWebhook] = &github.Hook{ + Config: make(map[string]interface{}), + } + dataCh := make(chan []byte) + go func() { + resp := <-helper.ActiveEndpoints[rc.Webhook.Endpoint].DataCh + dataCh <- resp + }() + + RouteActiveHandler(writer, &http.Request{ + Body: ioutil.NopCloser(bytes.NewReader(pbytes)), + }, rc) + + data := <-dataCh + convey.So(writer.HeaderStatus, convey.ShouldEqual, http.StatusOK) + convey.So(string(data), convey.ShouldEqual, string(pbytes)) + rc.Configs[labelGitlabConfig] = ps.(*glab) + err = ese.PostActivate(rc) + convey.So(err, convey.ShouldNotBeNil) + }) + }) + }) +} diff --git a/gateways/community/slack/config_test.go b/gateways/community/slack/config_test.go new file mode 100644 index 0000000000..d61fe38be9 --- /dev/null +++ b/gateways/community/slack/config_test.go @@ -0,0 +1,42 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package slack + +import ( + "github.com/smartystreets/goconvey/convey" + "testing" +) + +var es = ` +hook: + endpoint: "/" + port: "8080" + url: "testurl" +token: + name: fake-token + key: fake +` + +func TestParseConfig(t *testing.T) { + convey.Convey("Given a slack event source, parse it", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + _, ok := ps.(*slackConfig) + convey.So(ok, convey.ShouldEqual, true) + }) +} diff --git a/gateways/community/slack/start.go b/gateways/community/slack/start.go index af614a726d..a1048b66c4 100644 --- a/gateways/community/slack/start.go +++ b/gateways/community/slack/start.go @@ -60,7 +60,7 @@ func RouteActiveHandler(writer http.ResponseWriter, request *http.Request, rc *g var buf bytes.Buffer if _, err := buf.ReadFrom(request.Body); err != nil { logger.Error().Err(err).Msg("failed to parse request body") - common.SendErrorResponse(writer, fmt.Sprintf("failed to parse request. err: %+v", err)) + common.SendInternalErrorResponse(writer, fmt.Sprintf("failed to parse request. err: %+v", err)) return } @@ -68,14 +68,20 @@ func RouteActiveHandler(writer http.ResponseWriter, request *http.Request, rc *g token := rc.Configs[labelSlackToken] eventsAPIEvent, e := slackevents.ParseEvent(json.RawMessage(body), slackevents.OptionVerifyToken(&slackevents.TokenComparator{VerificationToken: token.(string)})) if e != nil { - writer.WriteHeader(http.StatusInternalServerError) + response = "failed to extract event" + logger.Error().Msg(response) + common.SendInternalErrorResponse(writer, "failed to extract event") + return } if eventsAPIEvent.Type == slackevents.URLVerification { var r *slackevents.ChallengeResponse err := json.Unmarshal([]byte(body), &r) if err != nil { - writer.WriteHeader(http.StatusInternalServerError) + response = "failed to verify the challenge" + logger.Error().Msg(response) + common.SendInternalErrorResponse(writer, response) + return } writer.Header().Set("Content-Type", "text") writer.Write([]byte(r.Challenge)) @@ -84,8 +90,9 @@ func RouteActiveHandler(writer http.ResponseWriter, request *http.Request, rc *g if eventsAPIEvent.Type == slackevents.CallbackEvent { data, err := json.Marshal(eventsAPIEvent.InnerEvent.Data) if err != nil { - logger.Error().Err(err).Msg("failed to ") - common.SendErrorResponse(writer, "failed to marshal event data") + response = "failed to marshal event data" + logger.Error().Err(err).Msg(response) + common.SendInternalErrorResponse(writer, response) return } helper.ActiveEndpoints[rc.Webhook.Endpoint].DataCh <- data @@ -93,6 +100,7 @@ func RouteActiveHandler(writer http.ResponseWriter, request *http.Request, rc *g response = "request successfully processed" logger.Info().Msg(response) + common.SendSuccessResponse(writer, response) } // StartEventSource starts a event source diff --git a/gateways/community/slack/start_test.go b/gateways/community/slack/start_test.go new file mode 100644 index 0000000000..1443654074 --- /dev/null +++ b/gateways/community/slack/start_test.go @@ -0,0 +1,103 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package slack + +import ( + "bytes" + "encoding/json" + "github.com/ghodss/yaml" + "github.com/nlopes/slack/slackevents" + "io/ioutil" + "net/http" + "testing" + + gwcommon "github.com/argoproj/argo-events/gateways/common" + "github.com/smartystreets/goconvey/convey" +) + +func TestRouteActiveHandler(t *testing.T) { + convey.Convey("Given a route configuration", t, func() { + rc := gwcommon.GetFakeRouteConfig() + + helper.ActiveEndpoints[rc.Webhook.Endpoint] = &gwcommon.Endpoint{ + DataCh: make(chan []byte), + } + + convey.Convey("Inactive route should return 404", func() { + writer := &gwcommon.FakeHttpWriter{} + RouteActiveHandler(writer, &http.Request{}, rc) + convey.So(writer.HeaderStatus, convey.ShouldEqual, http.StatusBadRequest) + }) + + rc.Configs[labelSlackToken] = "Jhj5dZrVaK7ZwHHjRyZWjbDl" + helper.ActiveEndpoints[rc.Webhook.Endpoint].Active = true + + convey.Convey("Test url verification request", func() { + writer := &gwcommon.FakeHttpWriter{} + urlVer := slackevents.EventsAPIURLVerificationEvent{ + Type: slackevents.URLVerification, + Token: "Jhj5dZrVaK7ZwHHjRyZWjbDl", + Challenge: "3eZbrw1aBm2rZgRNFdxV2595E9CY3gmdALWMmHkvFXO7tYXAYM8P", + } + payload, err := yaml.Marshal(urlVer) + convey.So(err, convey.ShouldBeNil) + convey.So(payload, convey.ShouldNotBeNil) + RouteActiveHandler(writer, &http.Request{ + Body: ioutil.NopCloser(bytes.NewReader(payload)), + }, rc) + convey.So(writer.HeaderStatus, convey.ShouldEqual, http.StatusInternalServerError) + }) + + convey.Convey("Test an event notification", func() { + writer := &gwcommon.FakeHttpWriter{} + event := []byte(` +{ +"type": "name_of_event", +"event_ts": "1234567890.123456", +"user": "UXXXXXXX1" +} +`) + + var j json.RawMessage + j = event + ce := slackevents.EventsAPICallbackEvent{ + Token: "Jhj5dZrVaK7ZwHHjRyZWjbDl", + Type: slackevents.CallbackEvent, + EventTime: 1234567890, + APIAppID: "AXXXXXXXXX", + AuthedUsers: []string{ + "UXXXXXXX1", + "UXXXXXXX2", + }, + EventID: "Ev08MFMKH6", + InnerEvent: &j, + } + payload, err := yaml.Marshal(ce) + convey.So(err, convey.ShouldBeNil) + + go func() { + <-helper.ActiveEndpoints[rc.Webhook.Endpoint].DataCh + }() + + RouteActiveHandler(writer, &http.Request{ + Body: ioutil.NopCloser(bytes.NewBuffer(payload)), + }, rc) + convey.So(writer.HeaderStatus, convey.ShouldEqual, http.StatusInternalServerError) + }) + + }) +} diff --git a/gateways/community/storagegrid/config_test.go b/gateways/community/storagegrid/config_test.go new file mode 100644 index 0000000000..e8c88b27e2 --- /dev/null +++ b/gateways/community/storagegrid/config_test.go @@ -0,0 +1,44 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package storagegrid + +import ( + "github.com/smartystreets/goconvey/convey" + "testing" +) + +var es = ` +hook: + endpoint: "/" + port: "8080" + url: "testurl" +events: + - "ObjectCreated:Put" +filter: + suffix: ".txt" + prefix: "hello-" +` + +func TestParseConfig(t *testing.T) { + convey.Convey("Given a storage grid event source, parse it", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + _, ok := ps.(*storageGrid) + convey.So(ok, convey.ShouldEqual, true) + }) +} diff --git a/gateways/community/storagegrid/start_test.go b/gateways/community/storagegrid/start_test.go new file mode 100644 index 0000000000..9a6cec6a43 --- /dev/null +++ b/gateways/community/storagegrid/start_test.go @@ -0,0 +1,146 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package storagegrid + +import ( + "bytes" + "encoding/json" + gwcommon "github.com/argoproj/argo-events/gateways/common" + "github.com/ghodss/yaml" + "github.com/smartystreets/goconvey/convey" + "io/ioutil" + "net/http" + "testing" +) + +var notification = ` +{ + "Action": "Publish", + "Message": { + "Records": [ + { + "eventName": "ObjectCreated:Put", + "eventSource": "sgws:s3", + "eventTime": "2019-02-27T21:15:09Z", + "eventVersion": "2.0", + "requestParameters": { + "sourceIPAddress": "1.1.1.1" + }, + "responseElements": { + "x-amz-request-id": "12345678" + }, + "s3": { + "bucket": { + "arn": "urn:sgfs:s3:::my_bucket", + "name": "my_bucket", + "ownerIdentity": { + "principalId": "55555555555555555" + } + }, + "configurationId": "Object-Event", + "object": { + "eTag": "4444444444444444", + "key": "hello-world.txt", + "sequencer": "AAAAAA", + "size": 6 + }, + "s3SchemaVersion": "1.0" + }, + "userIdentity": { + "principalId": "1111111111111111" + } + } + ] + }, + "TopicArn": "urn:h:sns:us-east::my_topic_1", + "Version": "2010-03-31" +} +` + +func TestRouteActiveHandler(t *testing.T) { + convey.Convey("Given a route configuration", t, func() { + rc := gwcommon.GetFakeRouteConfig() + helper.ActiveEndpoints[rc.Webhook.Endpoint] = &gwcommon.Endpoint{ + DataCh: make(chan []byte), + } + + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + writer := &gwcommon.FakeHttpWriter{} + + convey.Convey("Inactive route should return error", func() { + pbytes, err := yaml.Marshal(ps.(*storageGrid)) + convey.So(err, convey.ShouldBeNil) + RouteActiveHandler(writer, &http.Request{ + Body: ioutil.NopCloser(bytes.NewReader(pbytes)), + }, rc) + convey.So(writer.HeaderStatus, convey.ShouldEqual, http.StatusBadRequest) + }) + + convey.Convey("Active route should return success", func() { + helper.ActiveEndpoints[rc.Webhook.Endpoint].Active = true + rc.Configs[LabelStorageGridConfig] = ps.(*storageGrid) + dataCh := make(chan []byte) + go func() { + resp := <-helper.ActiveEndpoints[rc.Webhook.Endpoint].DataCh + dataCh <- resp + }() + + RouteActiveHandler(writer, &http.Request{ + Body: ioutil.NopCloser(bytes.NewReader([]byte(notification))), + }, rc) + convey.So(writer.HeaderStatus, convey.ShouldEqual, http.StatusOK) + }) + + }) +} + +func TestGenerateUUID(t *testing.T) { + convey.Convey("Make sure generated UUIDs are unique", t, func() { + u1 := generateUUID() + u2 := generateUUID() + convey.So(u1.String(), convey.ShouldNotEqual, u2.String()) + }) +} + +func TestFilterEvent(t *testing.T) { + convey.Convey("Given a storage grid event, test whether it passes the filter", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + var sg *storageGridNotification + err = json.Unmarshal([]byte(notification), &sg) + convey.So(err, convey.ShouldBeNil) + convey.So(sg, convey.ShouldNotBeNil) + + ok := filterEvent(sg, ps.(*storageGrid)) + convey.So(ok, convey.ShouldEqual, true) + }) +} + +func TestFilterName(t *testing.T) { + convey.Convey("Given a storage grid event, test whether the object key passes the filter", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + var sg *storageGridNotification + err = json.Unmarshal([]byte(notification), &sg) + convey.So(err, convey.ShouldBeNil) + convey.So(sg, convey.ShouldNotBeNil) + + ok := filterName(sg, ps.(*storageGrid)) + convey.So(ok, convey.ShouldEqual, true) + }) +} diff --git a/gateways/community/trello/config_test.go b/gateways/community/trello/config_test.go new file mode 100644 index 0000000000..26d63464df --- /dev/null +++ b/gateways/community/trello/config_test.go @@ -0,0 +1,47 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package trello + +import ( + "github.com/smartystreets/goconvey/convey" + "testing" +) + +var es = ` +hook: + endpoint: "/" + port: "8080" + url: "testurl" +apiKey: + key: api + name: trello +token: + key: token + name: trello +url: "URL to register on trello" +description: "test hook" +` + +func TestParseConfig(t *testing.T) { + convey.Convey("Given a trello event source, parse it", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + _, ok := ps.(*trello) + convey.So(ok, convey.ShouldEqual, true) + }) +} diff --git a/gateways/community/trello/start_test.go b/gateways/community/trello/start_test.go new file mode 100644 index 0000000000..488caae8c0 --- /dev/null +++ b/gateways/community/trello/start_test.go @@ -0,0 +1,95 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package trello + +import ( + "bytes" + "github.com/argoproj/argo-events/common" + "io/ioutil" + "k8s.io/client-go/kubernetes/fake" + "net/http" + "testing" + + gwcommon "github.com/argoproj/argo-events/gateways/common" + "github.com/ghodss/yaml" + "github.com/smartystreets/goconvey/convey" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +func TestRouteActiveHandler(t *testing.T) { + convey.Convey("Given a route configuration", t, func() { + rc := gwcommon.GetFakeRouteConfig() + helper.ActiveEndpoints[rc.Webhook.Endpoint] = &gwcommon.Endpoint{ + DataCh: make(chan []byte), + } + + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + rc.Configs[LabelTrelloConfig] = ps.(*trello) + writer := &gwcommon.FakeHttpWriter{} + + convey.Convey("Inactive route should return error", func() { + pbytes, err := yaml.Marshal(ps.(*trello)) + convey.So(err, convey.ShouldBeNil) + RouteActiveHandler(writer, &http.Request{ + Body: ioutil.NopCloser(bytes.NewReader(pbytes)), + }, rc) + convey.So(writer.HeaderStatus, convey.ShouldEqual, http.StatusBadRequest) + }) + + convey.Convey("Active route should return success", func() { + helper.ActiveEndpoints[rc.Webhook.Endpoint].Active = true + dataCh := make(chan []byte) + go func() { + resp := <-helper.ActiveEndpoints[rc.Webhook.Endpoint].DataCh + dataCh <- resp + }() + + RouteActiveHandler(writer, &http.Request{ + Body: ioutil.NopCloser(bytes.NewReader([]byte("fake notification"))), + }, rc) + convey.So(writer.HeaderStatus, convey.ShouldEqual, http.StatusOK) + data := <-dataCh + convey.So(string(data), convey.ShouldEqual, "fake notification") + }) + + convey.Convey("Run post activate", func() { + ese := &TrelloEventSourceExecutor{ + Clientset: fake.NewSimpleClientset(), + Log: common.GetLoggerContext(common.LoggerConf()).Logger(), + Namespace: "fake", + } + secret, err := ese.Clientset.CoreV1().Secrets(ese.Namespace).Create(&corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: "trello", + Namespace: ese.Namespace, + }, + Data: map[string][]byte{ + "api": []byte("api"), + "token": []byte("token"), + }, + }) + convey.So(err, convey.ShouldBeNil) + convey.So(secret, convey.ShouldNotBeNil) + + err = ese.PostActivate(rc) + convey.So(err, convey.ShouldNotBeNil) + }) + + }) +} diff --git a/gateways/core/artifact/config_test.go b/gateways/core/artifact/config_test.go new file mode 100644 index 0000000000..ac73c722aa --- /dev/null +++ b/gateways/core/artifact/config_test.go @@ -0,0 +1,51 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package artifact + +import ( + "testing" + + apicommon "github.com/argoproj/argo-events/pkg/apis/common" + "github.com/smartystreets/goconvey/convey" +) + +var es = ` +bucket: + name: input +endpoint: minio-service.argo-events:9000 +event: s3:ObjectCreated:Put +filter: + prefix: "" + suffix: "" +insecure: true +accessKey: + key: accesskey + name: artifacts-minio +secretKey: + key: secretkey + name: artifacts-minio +` + +func TestParseConfig(t *testing.T) { + convey.Convey("Given a artifact event source, parse it", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + _, ok := ps.(*apicommon.S3Artifact) + convey.So(ok, convey.ShouldEqual, true) + }) +} diff --git a/gateways/core/artifact/start_test.go b/gateways/core/artifact/start_test.go new file mode 100644 index 0000000000..ecb5dbafd6 --- /dev/null +++ b/gateways/core/artifact/start_test.go @@ -0,0 +1,72 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package artifact + +import ( + "testing" + + "github.com/argoproj/argo-events/common" + "github.com/argoproj/argo-events/gateways" + apicommon "github.com/argoproj/argo-events/pkg/apis/common" + "github.com/smartystreets/goconvey/convey" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes/fake" +) + +func TestListeEvents(t *testing.T) { + convey.Convey("Given an event source, listen to events", t, func() { + ese := &S3EventSourceExecutor{ + Clientset: fake.NewSimpleClientset(), + Log: common.GetLoggerContext(common.LoggerConf()).Logger(), + Namespace: "fake", + } + secret, err := ese.Clientset.CoreV1().Secrets(ese.Namespace).Create(&corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: "artifacts-minio", + Namespace: ese.Namespace, + }, + Data: map[string][]byte{ + "accesskey": []byte("access"), + "secretkey": []byte("secret"), + }, + }) + convey.So(err, convey.ShouldBeNil) + convey.So(secret, convey.ShouldNotBeNil) + + dataCh := make(chan []byte) + errorCh := make(chan error) + doneCh := make(chan struct{}, 1) + errCh2 := make(chan error) + + go func() { + err := <-errorCh + errCh2 <- err + }() + + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + ese.listenEvents(ps.(*apicommon.S3Artifact), &gateways.EventSource{ + Id: "1234", + Data: es, + Name: "fake", + }, dataCh, errorCh, doneCh) + + err = <-errCh2 + convey.So(err, convey.ShouldNotBeNil) + }) +} diff --git a/gateways/core/calendar/config_test.go b/gateways/core/calendar/config_test.go new file mode 100644 index 0000000000..0d94761cc5 --- /dev/null +++ b/gateways/core/calendar/config_test.go @@ -0,0 +1,37 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package calendar + +import ( + "github.com/smartystreets/goconvey/convey" + "testing" +) + +var es = ` +interval: 2s +userPayload: "{\r\n\"hello\": \"world\"\r\n}" +` + +func TestParseConfig(t *testing.T) { + convey.Convey("Given a calendar event source, parse it", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + _, ok := ps.(*calSchedule) + convey.So(ok, convey.ShouldEqual, true) + }) +} diff --git a/gateways/core/calendar/start_test.go b/gateways/core/calendar/start_test.go new file mode 100644 index 0000000000..1d5ea565e1 --- /dev/null +++ b/gateways/core/calendar/start_test.go @@ -0,0 +1,73 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package calendar + +import ( + "github.com/argoproj/argo-events/common" + "github.com/argoproj/argo-events/gateways" + "github.com/ghodss/yaml" + "github.com/smartystreets/goconvey/convey" + "testing" +) + +func TestResolveSchedule(t *testing.T) { + convey.Convey("Given a calendar schedule, resolve it", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + + schedule, err := resolveSchedule(ps.(*calSchedule)) + convey.So(err, convey.ShouldBeNil) + convey.So(schedule, convey.ShouldNotBeNil) + }) +} + +func TestListenEvents(t *testing.T) { + convey.Convey("Given a calendar schedule, listen events", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + + ese := &CalendarEventSourceExecutor{ + Log: common.GetLoggerContext(common.LoggerConf()).Logger(), + } + dataCh := make(chan []byte) + errorCh := make(chan error) + doneCh := make(chan struct{}, 1) + dataCh2 := make(chan []byte) + + go func() { + data := <-dataCh + dataCh2 <- data + }() + + go ese.listenEvents(ps.(*calSchedule), &gateways.EventSource{ + Name: "fake", + Data: es, + Id: "1234", + }, dataCh, errorCh, doneCh) + + data := <-dataCh2 + doneCh <- struct{}{} + + var cal *calResponse + err = yaml.Unmarshal(data, &cal) + convey.So(err, convey.ShouldBeNil) + + convey.So(cal.UserPayload, convey.ShouldEqual, "{\r\n\"hello\": \"world\"\r\n}") + }) +} diff --git a/gateways/core/file/config_test.go b/gateways/core/file/config_test.go new file mode 100644 index 0000000000..b4be30ff2c --- /dev/null +++ b/gateways/core/file/config_test.go @@ -0,0 +1,38 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package file + +import ( + "github.com/smartystreets/goconvey/convey" + "testing" +) + +var es = ` +directory: "/bin/" +type: CREATE +path: x.txt +` + +func TestParseConfig(t *testing.T) { + convey.Convey("Given a file event source, parse it", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + _, ok := ps.(*fileWatcher) + convey.So(ok, convey.ShouldEqual, true) + }) +} diff --git a/gateways/core/resource/config_test.go b/gateways/core/resource/config_test.go new file mode 100644 index 0000000000..f62282866d --- /dev/null +++ b/gateways/core/resource/config_test.go @@ -0,0 +1,43 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package resource + +import ( + "github.com/smartystreets/goconvey/convey" + "testing" +) + +var es = ` +namespace: "argo-events" +group: "" +version: "v1" +kind: "Pod" +filter: + labels: + workflows.argoproj.io/phase: Succeeded + name: "my-workflow" +` + +func TestParseConfig(t *testing.T) { + convey.Convey("Given a resource event source, parse it", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + _, ok := ps.(*resource) + convey.So(ok, convey.ShouldEqual, true) + }) +} diff --git a/gateways/core/resource/start_test.go b/gateways/core/resource/start_test.go new file mode 100644 index 0000000000..e8e940d54a --- /dev/null +++ b/gateways/core/resource/start_test.go @@ -0,0 +1,109 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package resource + +import ( + "github.com/mitchellh/mapstructure" + "github.com/smartystreets/goconvey/convey" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/client-go/kubernetes/fake" + "testing" +) + +var ( + ese = &ResourceEventSourceExecutor{} + resourceList = metav1.APIResourceList{ + GroupVersion: "v1", + APIResources: []metav1.APIResource{ + { + Group: "", + Version: "v1", + Name: "fake-1", + Kind: "Fake", + Namespaced: true, + ShortNames: []string{"f1"}, + SingularName: "fake-1", + Verbs: []string{"watch"}, + }, + }, + } +) + +func TestServerResourceForGVK(t *testing.T) { + convey.Convey("Given a resource gvk, discover the resource", t, func() { + result, err := ese.serverResourceForGVK(&resourceList, "Fake") + convey.So(err, convey.ShouldBeNil) + convey.So(result.Kind, convey.ShouldEqual, "Fake") + }) +} + +func TestCanWatchResource(t *testing.T) { + convey.Convey("Given a resource, test if it can watched", t, func() { + result, err := ese.serverResourceForGVK(&resourceList, "Fake") + convey.So(err, convey.ShouldBeNil) + ok := ese.canWatchResource(result) + convey.So(ok, convey.ShouldEqual, true) + }) +} + +func TestResolveGroupVersion(t *testing.T) { + convey.Convey("Given a resource, resolve the group and version", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + gv := ese.resolveGroupVersion(ps.(*resource)) + convey.So(gv, convey.ShouldEqual, "v1") + gv = ese.resolveGroupVersion(&resource{ + GroupVersionKind: metav1.GroupVersionKind{ + Group: "argoproj.io", + Kind: "Fake", + Version: "v1alpha1", + }, + Version: "v1alpha1", + }) + convey.So(gv, convey.ShouldEqual, "argoproj.io/v1alpha1") + }) +} + +func TestFilter(t *testing.T) { + convey.Convey("Given a resource object, apply filter on it", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "fake", + Namespace: "fake", + Labels: map[string]string{ + "workflows.argoproj.io/phase": "Succeeded", + "name": "my-workflow", + }, + }, + } + pod, err = fake.NewSimpleClientset().CoreV1().Pods("fake").Create(pod) + convey.So(err, convey.ShouldBeNil) + + outmap := make(map[string]interface{}) + err = mapstructure.Decode(pod, &outmap) + convey.So(err, convey.ShouldBeNil) + + ok := ese.passFilters("fake", &unstructured.Unstructured{ + Object: outmap, + }, ps.(*resource).Filter) + convey.So(ok, convey.ShouldEqual, true) + }) +} diff --git a/gateways/core/stream/amqp/config_test.go b/gateways/core/stream/amqp/config_test.go new file mode 100644 index 0000000000..890604fd75 --- /dev/null +++ b/gateways/core/stream/amqp/config_test.go @@ -0,0 +1,39 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package amqp + +import ( + "github.com/smartystreets/goconvey/convey" + "testing" +) + +var es = ` +url: amqp://amqp.argo-events:5672 +exchangeName: fooExchangeName +exchangeType: fanout +routingKey: fooRoutingKey +` + +func TestParseConfig(t *testing.T) { + convey.Convey("Given a amqp event source, parse it", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + _, ok := ps.(*amqp) + convey.So(ok, convey.ShouldEqual, true) + }) +} diff --git a/gateways/core/stream/kafka/config_test.go b/gateways/core/stream/kafka/config_test.go new file mode 100644 index 0000000000..c8d6a26a43 --- /dev/null +++ b/gateways/core/stream/kafka/config_test.go @@ -0,0 +1,38 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package kafka + +import ( + "github.com/smartystreets/goconvey/convey" + "testing" +) + +var es = ` +url: kafka.argo-events:9092 +topic: foo +partition: "0" +` + +func TestParseConfig(t *testing.T) { + convey.Convey("Given a kafka event source, parse it", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + _, ok := ps.(*kafka) + convey.So(ok, convey.ShouldEqual, true) + }) +} diff --git a/gateways/core/stream/mqtt/config_test.go b/gateways/core/stream/mqtt/config_test.go new file mode 100644 index 0000000000..f4282092c0 --- /dev/null +++ b/gateways/core/stream/mqtt/config_test.go @@ -0,0 +1,38 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package mqtt + +import ( + "github.com/smartystreets/goconvey/convey" + "testing" +) + +var es = ` +url: tcp://mqtt.argo-events:1883 +topic: foo +clientId: 1 +` + +func TestParseConfig(t *testing.T) { + convey.Convey("Given a mqtt event source, parse it", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + _, ok := ps.(*mqtt) + convey.So(ok, convey.ShouldEqual, true) + }) +} diff --git a/gateways/core/stream/nats/config_test.go b/gateways/core/stream/nats/config_test.go new file mode 100644 index 0000000000..907edb4adf --- /dev/null +++ b/gateways/core/stream/nats/config_test.go @@ -0,0 +1,37 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package nats + +import ( + "github.com/smartystreets/goconvey/convey" + "testing" +) + +var es = ` +url: natsConfig://natsConfig.argo-events:4222 +subject: foo +` + +func TestParseConfig(t *testing.T) { + convey.Convey("Given a nats event source, parse it", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + _, ok := ps.(*natsConfig) + convey.So(ok, convey.ShouldEqual, true) + }) +} diff --git a/gateways/core/webhook/config_test.go b/gateways/core/webhook/config_test.go new file mode 100644 index 0000000000..044285ae78 --- /dev/null +++ b/gateways/core/webhook/config_test.go @@ -0,0 +1,40 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package webhook + +import ( + "testing" + + "github.com/argoproj/argo-events/gateways/common" + "github.com/smartystreets/goconvey/convey" +) + +var es = ` +endpoint: "/bar" +port: "10000" +method: "POST" +` + +func TestParseConfig(t *testing.T) { + convey.Convey("Given a webhook event source, parse it", t, func() { + ps, err := parseEventSource(es) + convey.So(err, convey.ShouldBeNil) + convey.So(ps, convey.ShouldNotBeNil) + _, ok := ps.(*common.Webhook) + convey.So(ok, convey.ShouldEqual, true) + }) +} diff --git a/gateways/core/webhook/start_test.go b/gateways/core/webhook/start_test.go new file mode 100644 index 0000000000..2758a1e485 --- /dev/null +++ b/gateways/core/webhook/start_test.go @@ -0,0 +1,71 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package webhook + +import ( + "bytes" + gwcommon "github.com/argoproj/argo-events/gateways/common" + "github.com/smartystreets/goconvey/convey" + "io/ioutil" + "net/http" + "testing" +) + +func TestRouteActiveHandler(t *testing.T) { + convey.Convey("Given a route configuration", t, func() { + rc := gwcommon.GetFakeRouteConfig() + rc.Webhook.Method = http.MethodGet + helper.ActiveEndpoints[rc.Webhook.Endpoint] = &gwcommon.Endpoint{ + DataCh: make(chan []byte), + } + + writer := &gwcommon.FakeHttpWriter{} + + convey.Convey("Inactive route should return error", func() { + RouteActiveHandler(writer, &http.Request{ + Body: ioutil.NopCloser(bytes.NewReader([]byte("hello"))), + }, rc) + convey.So(writer.HeaderStatus, convey.ShouldEqual, http.StatusBadRequest) + }) + + helper.ActiveEndpoints[rc.Webhook.Endpoint].Active = true + + convey.Convey("Active route with correct method should return success", func() { + dataCh := make(chan []byte) + go func() { + resp := <-helper.ActiveEndpoints[rc.Webhook.Endpoint].DataCh + dataCh <- resp + }() + + RouteActiveHandler(writer, &http.Request{ + Body: ioutil.NopCloser(bytes.NewReader([]byte("fake notification"))), + Method: http.MethodGet, + }, rc) + convey.So(writer.HeaderStatus, convey.ShouldEqual, http.StatusOK) + data := <-dataCh + convey.So(string(data), convey.ShouldEqual, "fake notification") + }) + + convey.Convey("Active route with incorrect method should return failure", func() { + RouteActiveHandler(writer, &http.Request{ + Body: ioutil.NopCloser(bytes.NewReader([]byte("fake notification"))), + Method: http.MethodHead, + }, rc) + convey.So(writer.HeaderStatus, convey.ShouldEqual, http.StatusBadRequest) + }) + }) +} diff --git a/gateways/gateway_test.go b/gateways/gateway_test.go new file mode 100644 index 0000000000..9cbb0f565e --- /dev/null +++ b/gateways/gateway_test.go @@ -0,0 +1,105 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package gateways + +import ( + "context" + "fmt" + "github.com/rs/zerolog" + "github.com/smartystreets/goconvey/convey" + "google.golang.org/grpc/metadata" + "os" + "testing" + "time" +) + +type FakeGRPCStream struct { + SentData *Event + Ctx context.Context +} + +func (f *FakeGRPCStream) Send(event *Event) error { + f.SentData = event + return nil +} + +func (f *FakeGRPCStream) SetHeader(metadata.MD) error { + return nil +} + +func (f *FakeGRPCStream) SendHeader(metadata.MD) error { + return nil +} + +func (f *FakeGRPCStream) SetTrailer(metadata.MD) { + return +} + +func (f *FakeGRPCStream) Context() context.Context { + return f.Ctx +} + +func (f *FakeGRPCStream) SendMsg(m interface{}) error { + return nil +} + +func (f *FakeGRPCStream) RecvMsg(m interface{}) error { + return nil +} + +func TestRecover(t *testing.T) { + convey.Convey("Recover from panic", t, func() { + defer Recover("fake") + panic("fake panic") + }) +} + +func TestHandleEventsFromEventSource(t *testing.T) { + convey.Convey("Given a gateway server, handle events from an event source", t, func() { + dataCh := make(chan []byte) + errorCh := make(chan error) + doneCh := make(chan struct{}) + logger := zerolog.New(os.Stdout) + + ctx, cancel := context.WithCancel(context.Background()) + + convey.Convey("handle data", func() { + es := &FakeGRPCStream{ + Ctx: ctx, + } + go HandleEventsFromEventSource("fake", es, dataCh, errorCh, doneCh, &logger) + dataCh <- []byte("hello") + time.Sleep(1 * time.Second) + cancel() + convey.So(string(es.SentData.Payload), convey.ShouldEqual, "hello") + }) + + convey.Convey("handle error", func() { + es := &FakeGRPCStream{ + Ctx: ctx, + } + errorCh2 := make(chan error) + go func() { + err := HandleEventsFromEventSource("fake", es, dataCh, errorCh, doneCh, &logger) + errorCh2 <- err + }() + errorCh <- fmt.Errorf("fake error") + err := <-errorCh2 + convey.So(err.Error(), convey.ShouldEqual, "fake error") + }) + }) +} diff --git a/gateways/state_test.go b/gateways/state_test.go index f1c7290133..82fb3545d5 100644 --- a/gateways/state_test.go +++ b/gateways/state_test.go @@ -17,10 +17,9 @@ limitations under the License. package gateways import ( - "testing" - "github.com/argoproj/argo-events/pkg/apis/gateway/v1alpha1" "github.com/smartystreets/goconvey/convey" + "testing" ) func TestGatewayState(t *testing.T) { @@ -84,3 +83,69 @@ func TestGatewayState(t *testing.T) { }) }) } + +func TestMarkGatewayNodePhase(t *testing.T) { + convey.Convey("Given a node status, mark node state", t, func() { + gc := getGatewayConfig() + nodeStatus := &EventSourceStatus{ + Name: "fake", + Id: "1234", + Message: "running", + Phase: v1alpha1.NodePhaseRunning, + Gw: gc.gw, + } + gc.gw.Status.Nodes = map[string]v1alpha1.NodeStatus{ + "1234": v1alpha1.NodeStatus{ + Phase: v1alpha1.NodePhaseNew, + Message: "init", + Name: "fake", + ID: "1234", + }, + } + + resultStatus := gc.markGatewayNodePhase(nodeStatus) + convey.So(resultStatus, convey.ShouldNotBeNil) + convey.So(resultStatus.Name, convey.ShouldEqual, nodeStatus.Name) + + gc.gw.Status.Nodes = map[string]v1alpha1.NodeStatus{ + "4567": v1alpha1.NodeStatus{ + Phase: v1alpha1.NodePhaseNew, + Message: "init", + Name: "fake", + ID: "1234", + }, + } + + resultStatus = gc.markGatewayNodePhase(nodeStatus) + convey.So(resultStatus, convey.ShouldBeNil) + }) +} + +func TestGetNodeByID(t *testing.T) { + convey.Convey("Given a node id, retrieve the node", t, func() { + gc := getGatewayConfig() + gc.gw.Status.Nodes = map[string]v1alpha1.NodeStatus{ + "1234": v1alpha1.NodeStatus{ + Phase: v1alpha1.NodePhaseNew, + Message: "init", + Name: "fake", + ID: "1234", + }, + } + status := gc.getNodeByID("1234") + convey.So(status, convey.ShouldNotBeNil) + convey.So(status.ID, convey.ShouldEqual, "1234") + }) +} + +func TestInitializeNode(t *testing.T) { + convey.Convey("Given a node, initialize it", t, func() { + gc := getGatewayConfig() + status := gc.initializeNode("1234", "fake", "init") + convey.So(status, convey.ShouldNotBeNil) + convey.So(status.ID, convey.ShouldEqual, "1234") + convey.So(status.Name, convey.ShouldEqual, "fake") + convey.So(status.Message, convey.ShouldEqual, "init") + convey.So(len(gc.gw.Status.Nodes), convey.ShouldEqual, 1) + }) +} diff --git a/gateways/transformer.go b/gateways/transformer.go index f52bc4cb9a..43977c7fa8 100644 --- a/gateways/transformer.go +++ b/gateways/transformer.go @@ -84,7 +84,7 @@ func (gc *GatewayConfig) transformEvent(gatewayEvent *Event) (*apicommon.Event, EventType: gc.gw.Spec.Type, EventTypeVersion: gc.gw.Spec.EventVersion, Source: &apicommon.URI{ - Host: common.DefaultGatewayConfigurationName(gc.gw.Name, gatewayEvent.Name), + Host: common.DefaultEventSourceName(gc.gw.Name, gatewayEvent.Name), }, }, Payload: gatewayEvent.Payload, diff --git a/gateways/transformer_test.go b/gateways/transformer_test.go new file mode 100644 index 0000000000..e0bd2f761d --- /dev/null +++ b/gateways/transformer_test.go @@ -0,0 +1,80 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package gateways + +import ( + "fmt" + "github.com/argoproj/argo-events/pkg/apis/common" + "github.com/smartystreets/goconvey/convey" + "github.com/stretchr/testify/mock" + "testing" +) + +type MyMockedObject struct { + mock.Mock +} + +func (m *MyMockedObject) dispatchEventOverHttp(source string, eventPayload []byte) error { + return nil +} + +func (m *MyMockedObject) dispatchEventOverNats(source string, eventPayload []byte) error { + return nil +} + +func TestDispatchEvent(t *testing.T) { + convey.Convey("Given an event, dispatch it to sensor", t, func() { + gc := getGatewayConfig() + event := &Event{ + Name: "fake", + Payload: []byte("fake"), + } + testObj := new(MyMockedObject) + testObj.On("dispatchEventOverHttp", "fake", []byte("fake")).Return(nil) + testObj.On("dispatchEventOverNats", "fake", []byte("fake")).Return(nil) + + err := gc.DispatchEvent(event) + convey.So(err, convey.ShouldBeNil) + + gc.gw.Spec.EventProtocol.Type = common.NATS + + err = gc.DispatchEvent(event) + convey.So(err, convey.ShouldBeNil) + + gc.gw.Spec.EventProtocol.Type = common.NATS + + err = gc.DispatchEvent(event) + convey.So(err, convey.ShouldBeNil) + + gc.gw.Spec.EventProtocol.Type = common.EventProtocolType("fake") + err = gc.DispatchEvent(event) + convey.So(err, convey.ShouldNotBeNil) + }) +} + +func TestTransformEvent(t *testing.T) { + convey.Convey("Given a gateway event, convert it into cloud event", t, func() { + gc := getGatewayConfig() + ce, err := gc.transformEvent(&Event{ + Name: "fake", + Payload: []byte("fake"), + }) + convey.So(err, convey.ShouldBeNil) + convey.So(ce, convey.ShouldNotBeNil) + convey.So(ce.Context.Source.Host, convey.ShouldEqual, fmt.Sprintf("%s:%s", gc.gw.Name, "fake")) + }) +} diff --git a/gateways/utils.go b/gateways/utils.go index 8dd0387529..2e70adea15 100644 --- a/gateways/utils.go +++ b/gateways/utils.go @@ -17,31 +17,15 @@ limitations under the License. package gateways import ( - "time" - + "github.com/argoproj/argo-events/common" "k8s.io/apimachinery/pkg/util/wait" ) -// SetValidateReason set the result of event source validation -func SetValidEventSource(v *ValidEventSource, reason string, valid bool) { - v.Reason = reason - v.IsValid = valid -} - -// InitBackoff initializes backoff -func InitBackoff(backoff *wait.Backoff) { - if backoff == nil { - backoff = &wait.Backoff{ - Steps: 1, - Duration: 1, - } - } - backoff.Duration = backoff.Duration * time.Second -} - // General connection helper func Connect(backoff *wait.Backoff, conn func() error) error { - InitBackoff(backoff) + if backoff == nil { + backoff = &common.DefaultRetry + } err := wait.ExponentialBackoff(*backoff, func() (bool, error) { if err := conn(); err != nil { return false, nil diff --git a/gateways/utils_test.go b/gateways/utils_test.go index 8017d4e043..926f26e162 100644 --- a/gateways/utils_test.go +++ b/gateways/utils_test.go @@ -18,14 +18,16 @@ package gateways import ( "github.com/smartystreets/goconvey/convey" + "k8s.io/apimachinery/pkg/util/wait" "testing" ) -func TestGatewayUtil(t *testing.T) { - convey.Convey("Given a event source, set the validation message", t, func() { - v := ValidEventSource{} - SetValidEventSource(&v, "event source is valid", true) - convey.So(v.IsValid, convey.ShouldBeTrue) - convey.So(v.Reason, convey.ShouldEqual, "event source is valid") +func TestConnect(t *testing.T) { + convey.Convey("Given a backoff option, test connection", t, func() { + var backoff *wait.Backoff + err := Connect(backoff, func() error { + return nil + }) + convey.So(err, convey.ShouldBeNil) }) } diff --git a/hack/generate-proto.sh b/hack/generate-proto.sh index 2080dafcf9..f05ad7bab4 100755 --- a/hack/generate-proto.sh +++ b/hack/generate-proto.sh @@ -42,6 +42,3 @@ dist/go-to-protobuf \ --apimachinery-packages=$(IFS=, ; echo "${APIMACHINERY_PKGS[*]}") \ --proto-import=${PROJECT_ROOT}/vendor -# generate protocol documentation -protoc -I=${GOPATH}/src --doc_out=../docs --doc_opt=markdown,gateway-protocol.md github.com/argoproj/argo-events/pkg/apis/gateway/v1alpha1/generated.proto -protoc -I=${GOPATH}/src --doc_out=../docs --doc_opt=markdown,sensor-protocol.md github.com/argoproj/argo-events/pkg/apis/sensor/v1alpha1/generated.proto diff --git a/pkg/apis/common/deepcopy_generated.go b/pkg/apis/common/deepcopy_generated.go index a84fcce7b0..97df0ef8f2 100644 --- a/pkg/apis/common/deepcopy_generated.go +++ b/pkg/apis/common/deepcopy_generated.go @@ -21,7 +21,7 @@ limitations under the License. package common import ( - v1 "k8s.io/api/core/v1" + "k8s.io/api/core/v1" ) // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. diff --git a/pkg/apis/common/generated.pb.go b/pkg/apis/common/generated.pb.go index 19b9ad1544..647712c744 100644 --- a/pkg/apis/common/generated.pb.go +++ b/pkg/apis/common/generated.pb.go @@ -18,18 +18,18 @@ limitations under the License. package common import ( - fmt "fmt" + "fmt" - io "io" + "io" - proto "github.com/gogo/protobuf/proto" + "github.com/gogo/protobuf/proto" github_com_gogo_protobuf_sortkeys "github.com/gogo/protobuf/sortkeys" github_com_minio_minio_go "github.com/minio/minio-go" v11 "k8s.io/api/core/v1" - math "math" - reflect "reflect" - strings "strings" + "math" + "reflect" + "strings" ) // Reference imports to suppress errors if they are not otherwise used. diff --git a/pkg/apis/sensor/v1alpha1/generated.proto b/pkg/apis/sensor/v1alpha1/generated.proto index 7d645d9a32..e24a1aa161 100644 --- a/pkg/apis/sensor/v1alpha1/generated.proto +++ b/pkg/apis/sensor/v1alpha1/generated.proto @@ -229,7 +229,7 @@ message ResourceObject { optional GroupVersionKind groupVersionKind = 5; // Namespace in which to create this object - // defaults to the service account namespace + // defaults to the sensor namespace optional string namespace = 1; // Source of the K8 resource file(s) diff --git a/sensors/event-handler_test.go b/sensors/event-handler_test.go index 3aa6ea19de..720dec7493 100644 --- a/sensors/event-handler_test.go +++ b/sensors/event-handler_test.go @@ -17,8 +17,12 @@ limitations under the License. package sensors import ( + "bytes" + "encoding/json" "fmt" + "io/ioutil" "net/http" + "strings" "testing" "time" @@ -30,6 +34,8 @@ import ( "github.com/ghodss/yaml" "github.com/smartystreets/goconvey/convey" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/selection" discoveryFake "k8s.io/client-go/discovery/fake" "k8s.io/client-go/kubernetes/fake" ) @@ -37,7 +43,6 @@ import ( var sensorStr = `apiVersion: argoproj.io/v1alpha1 kind: Sensor metadata: - namespace: argo-events name: test-sensor labels: sensors.argoproj.io/sensor-controller-instanceid: argo-events @@ -92,18 +97,22 @@ func getSensor() (*v1alpha1.Sensor, error) { return &sensor, err } -type mockHttpWriter struct{} +type mockHttpWriter struct { + Status int + Payload []byte +} func (m *mockHttpWriter) Header() http.Header { return http.Header{} } -func (m *mockHttpWriter) Write([]byte) (int, error) { +func (m *mockHttpWriter) Write(p []byte) (int, error) { + m.Payload = p return 0, nil } func (m *mockHttpWriter) WriteHeader(statusCode int) { - + m.Status = statusCode } func getsensorExecutionCtx(sensor *v1alpha1.Sensor) *sensorExecutionCtx { @@ -121,6 +130,7 @@ func getsensorExecutionCtx(sensor *v1alpha1.Sensor) *sensorExecutionCtx { sensorClient: sensorFake.NewSimpleClientset(), sensor: sensor, controllerInstanceID: "test-1", + queue: make(chan *updateNotification), } } @@ -134,7 +144,7 @@ func getCloudEvent() *apicommon.Event { EventType: "test", EventTypeVersion: common.CloudEventsVersion, Source: &apicommon.URI{ - Host: common.DefaultGatewayConfigurationName("test-gateway", "test"), + Host: common.DefaultEventSourceName("test-gateway", "test"), }, }, Payload: []byte(`{ @@ -225,3 +235,205 @@ func TestDeleteStaleStatusNodes(t *testing.T) { convey.So(ok, convey.ShouldEqual, false) }) } + +func TestValidateEvent(t *testing.T) { + convey.Convey("Given an event, validate it", t, func() { + s, _ := getSensor() + sec := getsensorExecutionCtx(s) + dep, valid := sec.validateEvent(&apicommon.Event{ + Context: apicommon.EventContext{ + Source: &apicommon.URI{ + Host: "test-gateway:test", + }, + }, + }) + convey.So(valid, convey.ShouldEqual, true) + convey.So(dep, convey.ShouldNotBeNil) + }) +} + +func TestParseEvent(t *testing.T) { + convey.Convey("Given an event payload, parse event", t, func() { + s, _ := getSensor() + sec := getsensorExecutionCtx(s) + e := &apicommon.Event{ + Payload: []byte("hello"), + Context: apicommon.EventContext{ + Source: &apicommon.URI{ + Host: "test-gateway:test", + }, + }, + } + payload, err := json.Marshal(e) + convey.So(err, convey.ShouldBeNil) + + event, err := sec.parseEvent(payload) + convey.So(err, convey.ShouldBeNil) + convey.So(string(event.Payload), convey.ShouldEqual, "hello") + }) +} + +func TestSendToInternalQueue(t *testing.T) { + convey.Convey("Given an event, send it on internal queue", t, func() { + s, _ := getSensor() + sec := getsensorExecutionCtx(s) + e := &apicommon.Event{ + Payload: []byte("hello"), + Context: apicommon.EventContext{ + Source: &apicommon.URI{ + Host: "test-gateway:test", + }, + }, + } + go func() { + <-sec.queue + }() + ok := sec.sendEventToInternalQueue(e, &mockHttpWriter{}) + convey.So(ok, convey.ShouldEqual, true) + }) +} + +func TestHandleHttpEventHandler(t *testing.T) { + convey.Convey("Test http handler", t, func() { + s, _ := getSensor() + sec := getsensorExecutionCtx(s) + e := &apicommon.Event{ + Payload: []byte("hello"), + Context: apicommon.EventContext{ + Source: &apicommon.URI{ + Host: "test-gateway:test", + }, + }, + } + go func() { + <-sec.queue + }() + payload, err := json.Marshal(e) + convey.So(err, convey.ShouldBeNil) + writer := &mockHttpWriter{} + sec.httpEventHandler(writer, &http.Request{ + Body: ioutil.NopCloser(bytes.NewReader(payload)), + }) + convey.So(writer.Status, convey.ShouldEqual, http.StatusOK) + }) +} + +func TestSuccessNatsConnection(t *testing.T) { + convey.Convey("Given a successful nats connection, generate K8s event", t, func() { + s, _ := getSensor() + sec := getsensorExecutionCtx(s) + sec.successNatsConnection() + req1, err := labels.NewRequirement(common.LabelOperation, selection.Equals, []string{"nats_connection_setup"}) + convey.So(err, convey.ShouldBeNil) + req2, err := labels.NewRequirement(common.LabelEventType, selection.Equals, []string{string(common.OperationSuccessEventType)}) + convey.So(err, convey.ShouldBeNil) + req3, err := labels.NewRequirement(common.LabelSensorName, selection.Equals, []string{string(sec.sensor.Name)}) + convey.So(err, convey.ShouldBeNil) + + eventList, err := sec.kubeClient.CoreV1().Events(sec.sensor.Namespace).List(metav1.ListOptions{ + LabelSelector: labels.NewSelector().Add([]labels.Requirement{*req1, *req2, *req3}...).String(), + }) + convey.So(err, convey.ShouldBeNil) + convey.So(len(eventList.Items), convey.ShouldEqual, 1) + event := eventList.Items[0] + convey.So(event.Reason, convey.ShouldEqual, "connection setup successfully") + }) +} + +func TestEscalateNatsConnectionFailure(t *testing.T) { + convey.Convey("Given a failed nats connection, escalate through K8s event", t, func() { + s, _ := getSensor() + sec := getsensorExecutionCtx(s) + sec.escalateNatsConnectionFailure() + req1, err := labels.NewRequirement(common.LabelOperation, selection.Equals, []string{"nats_connection_setup"}) + convey.So(err, convey.ShouldBeNil) + req2, err := labels.NewRequirement(common.LabelEventType, selection.Equals, []string{string(common.OperationFailureEventType)}) + convey.So(err, convey.ShouldBeNil) + req3, err := labels.NewRequirement(common.LabelSensorName, selection.Equals, []string{string(sec.sensor.Name)}) + convey.So(err, convey.ShouldBeNil) + + eventList, err := sec.kubeClient.CoreV1().Events(sec.sensor.Namespace).List(metav1.ListOptions{ + LabelSelector: labels.NewSelector().Add([]labels.Requirement{*req1, *req2, *req3}...).String(), + }) + convey.So(err, convey.ShouldBeNil) + convey.So(len(eventList.Items), convey.ShouldEqual, 1) + event := eventList.Items[0] + convey.So(event.Reason, convey.ShouldEqual, "connection setup failed") + }) +} + +func TestSuccessNatsSubscription(t *testing.T) { + convey.Convey("Given a successful nats subscription, generate K8s event", t, func() { + s, _ := getSensor() + eventSource := "fake" + sec := getsensorExecutionCtx(s) + sec.successNatsSubscription(eventSource) + req1, err := labels.NewRequirement(common.LabelOperation, selection.Equals, []string{"nats_subscription_success"}) + convey.So(err, convey.ShouldBeNil) + req2, err := labels.NewRequirement(common.LabelEventType, selection.Equals, []string{string(common.OperationSuccessEventType)}) + convey.So(err, convey.ShouldBeNil) + req3, err := labels.NewRequirement(common.LabelSensorName, selection.Equals, []string{string(sec.sensor.Name)}) + convey.So(err, convey.ShouldBeNil) + req4, err := labels.NewRequirement(common.LabelEventSource, selection.Equals, []string{strings.Replace(eventSource, ":", "_", -1)}) + convey.So(err, convey.ShouldBeNil) + + eventList, err := sec.kubeClient.CoreV1().Events(sec.sensor.Namespace).List(metav1.ListOptions{ + LabelSelector: labels.NewSelector().Add([]labels.Requirement{*req1, *req2, *req3, *req4}...).String(), + }) + convey.So(err, convey.ShouldBeNil) + convey.So(len(eventList.Items), convey.ShouldEqual, 1) + event := eventList.Items[0] + convey.So(event.Reason, convey.ShouldEqual, "nats subscription success") + }) +} + +func TestEscalateNatsSubscriptionFailure(t *testing.T) { + convey.Convey("Given a failed nats subscription, escalate K8s event", t, func() { + s, _ := getSensor() + eventSource := "fake" + sec := getsensorExecutionCtx(s) + sec.escalateNatsSubscriptionFailure(eventSource) + req1, err := labels.NewRequirement(common.LabelOperation, selection.Equals, []string{"nats_subscription_failure"}) + convey.So(err, convey.ShouldBeNil) + req2, err := labels.NewRequirement(common.LabelEventType, selection.Equals, []string{string(common.OperationFailureEventType)}) + convey.So(err, convey.ShouldBeNil) + req3, err := labels.NewRequirement(common.LabelSensorName, selection.Equals, []string{string(sec.sensor.Name)}) + convey.So(err, convey.ShouldBeNil) + req4, err := labels.NewRequirement(common.LabelEventSource, selection.Equals, []string{strings.Replace(eventSource, ":", "_", -1)}) + convey.So(err, convey.ShouldBeNil) + + eventList, err := sec.kubeClient.CoreV1().Events(sec.sensor.Namespace).List(metav1.ListOptions{ + LabelSelector: labels.NewSelector().Add([]labels.Requirement{*req1, *req2, *req3, *req4}...).String(), + }) + convey.So(err, convey.ShouldBeNil) + convey.So(len(eventList.Items), convey.ShouldEqual, 1) + event := eventList.Items[0] + convey.So(event.Reason, convey.ShouldEqual, "nats subscription failed") + }) +} + +func TestProcessNatsMessage(t *testing.T) { + convey.Convey("Given nats message, process it", t, func() { + s, _ := getSensor() + sec := getsensorExecutionCtx(s) + e := &apicommon.Event{ + Payload: []byte("hello"), + Context: apicommon.EventContext{ + Source: &apicommon.URI{ + Host: "test-gateway:test", + }, + }, + } + dataCh := make(chan []byte) + go func() { + data := <-sec.queue + dataCh <- data.event.Payload + }() + payload, err := json.Marshal(e) + convey.So(err, convey.ShouldBeNil) + sec.processNatsMessage(payload, "fake") + data := <-dataCh + convey.So(data, convey.ShouldNotBeNil) + convey.So(string(data), convey.ShouldEqual, "hello") + }) +} diff --git a/sensors/nats-events-handler.go b/sensors/nats-events-handler.go index dfc8a04a2a..69de54659e 100644 --- a/sensors/nats-events-handler.go +++ b/sensors/nats-events-handler.go @@ -44,11 +44,11 @@ func (sec *sensorExecutionCtx) successNatsConnection() { func (sec *sensorExecutionCtx) escalateNatsConnectionFailure() { // escalate error labels := map[string]string{ - common.LabelEventType: string(common.EscalationEventType), + common.LabelEventType: string(common.OperationFailureEventType), common.LabelSensorName: sec.sensor.Name, common.LabelOperation: "nats_connection_setup", } - if err := common.GenerateK8sEvent(sec.kubeClient, "connection setup failed", common.EscalationEventType, "connection setup", sec.sensor.Name, sec.sensor.Namespace, sec.controllerInstanceID, sensor.Kind, labels); err != nil { + if err := common.GenerateK8sEvent(sec.kubeClient, "connection setup failed", common.OperationFailureEventType, "connection setup", sec.sensor.Name, sec.sensor.Namespace, sec.controllerInstanceID, sensor.Kind, labels); err != nil { sec.log.Error().Err(err).Msg("failed to create K8s event to log nats connection setup error") return } diff --git a/sensors/trigger_test.go b/sensors/trigger_test.go index b52fd6dd8d..8a3b93f153 100644 --- a/sensors/trigger_test.go +++ b/sensors/trigger_test.go @@ -19,6 +19,7 @@ package sensors import ( "testing" + apicommon "github.com/argoproj/argo-events/pkg/apis/common" "github.com/argoproj/argo-events/pkg/apis/sensor/v1alpha1" "github.com/smartystreets/goconvey/convey" corev1 "k8s.io/api/core/v1" @@ -288,3 +289,33 @@ func getUnstructuredPod(pod *corev1.Pod) (*unstructured.Unstructured, error) { } return &unstructured.Unstructured{Object: obj}, nil } + +func TestExtractEvents(t *testing.T) { + convey.Convey("Given a sensor, extract events", t, func() { + sensor, _ := getSensor() + sec := getsensorExecutionCtx(sensor) + id := sensor.NodeID("test-gateway:test") + sensor.Status.Nodes = map[string]v1alpha1.NodeStatus{ + id: { + Type: v1alpha1.NodeTypeEventDependency, + Event: &apicommon.Event{ + Payload: []byte("hello"), + Context: apicommon.EventContext{ + Source: &apicommon.URI{ + Host: "test-gateway:test", + }, + }, + }, + }, + } + extractedEvents := sec.extractEvents([]v1alpha1.ResourceParameter{ + { + Src: &v1alpha1.ResourceParameterSource{ + Event: "test-gateway:test", + }, + Dest: "fake-dest", + }, + }) + convey.So(len(extractedEvents), convey.ShouldEqual, 1) + }) +} diff --git a/store/git_test.go b/store/git_test.go new file mode 100644 index 0000000000..49244ef010 --- /dev/null +++ b/store/git_test.go @@ -0,0 +1,97 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package store + +import ( + "github.com/argoproj/argo-events/pkg/apis/sensor/v1alpha1" + "github.com/smartystreets/goconvey/convey" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes/fake" + "testing" +) + +var gar = &GitArtifactReader{ + kubeClientset: fake.NewSimpleClientset(), + artifact: &v1alpha1.GitArtifact{ + URL: "fake", + Namespace: "fake", + Creds: &v1alpha1.GitCreds{ + Username: &corev1.SecretKeySelector{ + Key: "username", + LocalObjectReference: corev1.LocalObjectReference{ + Name: "git-secret", + }, + }, + Password: &corev1.SecretKeySelector{ + Key: "password", + LocalObjectReference: corev1.LocalObjectReference{ + Name: "git-secret", + }, + }, + }, + }, +} + +func TestNewGitReader(t *testing.T) { + convey.Convey("Given configuration, get new git reader", t, func() { + reader, err := NewGitReader(fake.NewSimpleClientset(), &v1alpha1.GitArtifact{}) + convey.So(err, convey.ShouldBeNil) + convey.So(reader, convey.ShouldNotBeNil) + }) +} + +func TestGetRemote(t *testing.T) { + convey.Convey("Test git remote", t, func() { + remote := gar.getRemote() + convey.So(remote, convey.ShouldEqual, DefaultRemote) + }) +} + +func TestGetGitAuth(t *testing.T) { + convey.Convey("Given auth secret, get git auth tokens", t, func() { + secret, err := gar.kubeClientset.CoreV1().Secrets("fake").Create(&corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: gar.artifact.Creds.Username.Name, + Namespace: gar.artifact.Namespace, + }, + Data: map[string][]byte{ + "username": []byte("username"), + "password": []byte("password"), + }, + }) + convey.So(err, convey.ShouldBeNil) + convey.So(secret, convey.ShouldNotBeNil) + + auth, err := gar.getGitAuth() + convey.So(err, convey.ShouldBeNil) + convey.So(auth, convey.ShouldNotBeNil) + }) +} + +func TestGetBranchOrTag(t *testing.T) { + convey.Convey("Given a git artifact, get the branch or tag", t, func() { + br := gar.getBranchOrTag() + convey.So(br.Branch, convey.ShouldEqual, "refs/heads/master") + gar.artifact.Branch = "br" + br = gar.getBranchOrTag() + convey.So(br.Branch, convey.ShouldNotEqual, "refs/heads/master") + gar.artifact.Tag = "t" + tag := gar.getBranchOrTag() + convey.So(tag.Branch, convey.ShouldNotEqual, "refs/heads/master") + }) +} diff --git a/store/s3_test.go b/store/s3_test.go new file mode 100644 index 0000000000..6fa5445d00 --- /dev/null +++ b/store/s3_test.go @@ -0,0 +1,52 @@ +/* +Copyright 2018 BlackRock, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package store + +import ( + "testing" + + apicommon "github.com/argoproj/argo-events/pkg/apis/common" + "github.com/smartystreets/goconvey/convey" +) + +func TestNewMinioClient(t *testing.T) { + convey.Convey("Given a configuration, get minio client", t, func() { + client, err := NewMinioClient(&apicommon.S3Artifact{ + Endpoint: "fake", + Region: "us-east-1", + }, Credentials{ + accessKey: "access", + secretKey: "secret", + }) + convey.So(err, convey.ShouldBeNil) + convey.So(client, convey.ShouldNotBeNil) + }) +} + +func TestNewS3Reader(t *testing.T) { + convey.Convey("Given a minio client, get a reader", t, func() { + reader, err := NewS3Reader(&apicommon.S3Artifact{ + Endpoint: "fake", + Region: "us-east-1", + }, &Credentials{ + accessKey: "access", + secretKey: "secret", + }) + convey.So(err, convey.ShouldBeNil) + convey.So(reader, convey.ShouldNotBeNil) + }) +}