ref(*): refactor release testing

pkg/helm/client.go

@@ -17,13 +17,6 @@ limitations under the License.
 package helm // import "k8s.io/helm/pkg/helm"
 
 import (
-	"io"
-	"time"
-
-	"golang.org/x/net/context"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/keepalive"
-
 	"k8s.io/helm/pkg/chartutil"
 	"k8s.io/helm/pkg/kube"
 	"k8s.io/helm/pkg/proto/hapi/chart"
@@ -260,64 +253,5 @@ func (c *Client) RunReleaseTest(rlsName string, opts ...ReleaseTestOption) (<-ch
 	req := &reqOpts.testReq
 	req.Name = rlsName
 
-	return c.test(req)
+	return c.tiller.RunReleaseTest(req)
-}
-
-// connect returns a gRPC connection to Tiller or error. The gRPC dial options
-// are constructed here.
-func (c *Client) connect() (conn *grpc.ClientConn, err error) {
-	opts := []grpc.DialOption{
-		grpc.WithBlock(),
-		grpc.WithKeepaliveParams(keepalive.ClientParameters{
-			// Send keepalive every 30 seconds to prevent the connection from
-			// getting closed by upstreams
-			Time: time.Duration(30) * time.Second,
-		}),
-		grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMsgSize)),
-	}
-	opts = append(opts, grpc.WithInsecure())
-	ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second)
-	defer cancel()
-	if conn, err = grpc.DialContext(ctx, c.opts.host, opts...); err != nil {
-		return nil, err
-	}
-	return conn, nil
-}
-
-// Executes tiller.TestRelease RPC.
-func (c *Client) test(req *rls.TestReleaseRequest) (<-chan *rls.TestReleaseResponse, <-chan error) {
-	errc := make(chan error, 1)
-	conn, err := c.connect()
-	if err != nil {
-		errc <- err
-		return nil, errc
-	}
-
-	ch := make(chan *rls.TestReleaseResponse, 1)
-	go func() {
-		defer close(errc)
-		defer close(ch)
-		defer conn.Close()
-
-		rlc := rls.NewReleaseServiceClient(conn)
-		s, err := rlc.RunReleaseTest(context.TODO(), req)
-		if err != nil {
-			errc <- err
-			return
-		}
-
-		for {
-			msg, err := s.Recv()
-			if err == io.EOF {
-				return
-			}
-			if err != nil {
-				errc <- err
-				return
-			}
-			ch <- msg
-		}
-	}()
-
-	return ch, errc
 }

pkg/kube/client.go

@@ -41,13 +41,13 @@ import (
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/strategicpatch"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/tools/clientcmd"
 	batchinternal "k8s.io/kubernetes/pkg/apis/batch"
 	"k8s.io/kubernetes/pkg/apis/core"
-	"k8s.io/kubernetes/pkg/client/conditions"
 	"k8s.io/kubernetes/pkg/kubectl"
 	cmdutil "k8s.io/kubernetes/pkg/kubectl/cmd/util"
 	"k8s.io/kubernetes/pkg/kubectl/resource"
@@ -686,7 +686,18 @@ func (c *Client) watchPodUntilComplete(timeout time.Duration, info *resource.Inf
 
 	c.Log("Watching pod %s for completion with timeout of %v", info.Name, timeout)
 	_, err = watch.Until(timeout, w, func(e watch.Event) (bool, error) {
-		return conditions.PodCompleted(e)
+		switch e.Type {
+		case watch.Deleted:
+			return false, errors.NewNotFound(schema.GroupResource{Resource: "pods"}, "")
+		}
+		switch t := e.Object.(type) {
+		case *core.Pod:
+			switch t.Status.Phase {
+			case core.PodFailed, core.PodSucceeded:
+				return true, nil
+			}
+		}
+		return false, nil
 	})
 
 	return err

pkg/releasetesting/environment.go

@@ -33,14 +33,13 @@ import (
 type Environment struct {
 	Namespace  string
 	KubeClient environment.KubeClient
-	Stream     services.ReleaseService_RunReleaseTestServer
+	Mesages    chan *services.TestReleaseResponse
 	Timeout    int64
 }
 
 func (env *Environment) createTestPod(test *test) error {
 	b := bytes.NewBufferString(test.manifest)
 	if err := env.KubeClient.Create(env.Namespace, b, env.Timeout, false); err != nil {
-		log.Printf(err.Error())
 		test.result.Info = err.Error()
 		test.result.Status = release.TestRun_FAILURE
 		return err
@@ -108,7 +107,8 @@ func (env *Environment) streamUnknown(name, info string) error {
 
 func (env *Environment) streamMessage(msg string, status release.TestRun_Status) error {
 	resp := &services.TestReleaseResponse{Msg: msg, Status: status}
-	return env.Stream.Send(resp)
+	env.Mesages <- resp
+	return nil
 }
 
 // DeleteTestPods deletes resources given in testManifests

pkg/releasetesting/environment_test.go

@@ -64,11 +64,6 @@ func TestDeleteTestPods(t *testing.T) {
 
 	mockTestEnv.DeleteTestPods(mockTestSuite.TestManifests)
 
-	stream := mockTestEnv.Stream.(*mockStream)
-	if len(stream.messages) != 0 {
-		t.Errorf("Expected 0 errors, got at least one: %v", stream.messages)
-	}
-
 	for _, testManifest := range mockTestSuite.TestManifests {
 		if _, err := mockTestEnv.KubeClient.Get(mockTestEnv.Namespace, bytes.NewBufferString(testManifest)); err == nil {
 			t.Error("Expected error, got nil")
@@ -76,64 +71,43 @@ func TestDeleteTestPods(t *testing.T) {
 	}
 }
 
-func TestDeleteTestPodsFailingDelete(t *testing.T) {
+func TestStreamMessage(t *testing.T) {
-	mockTestSuite := testSuiteFixture([]string{manifestWithTestSuccessHook})
+	tEnv := mockTillerEnvironment()
-	mockTestEnv := newMockTestingEnvironment()
-	mockTestEnv.KubeClient = newDeleteFailingKubeClient()
 
-	mockTestEnv.DeleteTestPods(mockTestSuite.TestManifests)
+	ch := make(chan *services.TestReleaseResponse, 1)
+	defer close(ch)
 
-	stream := mockTestEnv.Stream.(*mockStream)
+	mockTestEnv := &Environment{
-	if len(stream.messages) != 1 {
+		Namespace:  "default",
-		t.Errorf("Expected 1 error, got: %v", len(stream.messages))
+		KubeClient: tEnv.KubeClient,
+		Timeout:    1,
+		Mesages:    ch,
 	}
-}
-
-func TestStreamMessage(t *testing.T) {
-	mockTestEnv := newMockTestingEnvironment()
 
 	expectedMessage := "testing streamMessage"
 	expectedStatus := release.TestRun_SUCCESS
 	err := mockTestEnv.streamMessage(expectedMessage, expectedStatus)
 	if err != nil {
-		t.Errorf("Expected no errors, got 1: %s", err)
+		t.Errorf("Expected no errors, got: %s", err)
-	}
-
-	stream := mockTestEnv.Stream.(*mockStream)
-	if len(stream.messages) != 1 {
-		t.Errorf("Expected 1 message, got: %v", len(stream.messages))
 	}
 
-	if stream.messages[0].Msg != expectedMessage {
+	got := <-mockTestEnv.Mesages
-		t.Errorf("Expected message: %s, got: %s", expectedMessage, stream.messages[0])
+	if got.Msg != expectedMessage {
+		t.Errorf("Expected message: %s, got: %s", expectedMessage, got.Msg)
 	}
-	if stream.messages[0].Status != expectedStatus {
+	if got.Status != expectedStatus {
-		t.Errorf("Expected status: %v, got: %v", expectedStatus, stream.messages[0].Status)
+		t.Errorf("Expected status: %v, got: %v", expectedStatus, got.Status)
 	}
 }
 
-type MockTestingEnvironment struct {
+func newMockTestingEnvironment() *Environment {
-	*Environment
+	return &Environment{
-}
-
-func newMockTestingEnvironment() *MockTestingEnvironment {
-	tEnv := mockTillerEnvironment()
-
-	return &MockTestingEnvironment{
-		Environment: &Environment{
 		Namespace:  "default",
-			KubeClient: tEnv.KubeClient,
+		KubeClient: mockTillerEnvironment().KubeClient,
-			Timeout:    5,
+		Timeout:    1,
-			Stream:     &mockStream{},
-		},
 	}
 }
 
-func (mte MockTestingEnvironment) streamMessage(msg string, status release.TestRun_Status) error {
-	mte.Stream.Send(&services.TestReleaseResponse{Msg: msg, Status: status})
-	return nil
-}
-
 type getFailingKubeClient struct {
 	tillerEnv.PrintingKubeClient
 }
@@ -148,20 +122,6 @@ func (p *getFailingKubeClient) Get(ns string, r io.Reader) (string, error) {
 	return "", errors.New("in the end, they did not find Nemo")
 }
 
-type deleteFailingKubeClient struct {
-	tillerEnv.PrintingKubeClient
-}
-
-func newDeleteFailingKubeClient() *deleteFailingKubeClient {
-	return &deleteFailingKubeClient{
-		PrintingKubeClient: tillerEnv.PrintingKubeClient{Out: ioutil.Discard},
-	}
-}
-
-func (p *deleteFailingKubeClient) Delete(ns string, r io.Reader) error {
-	return errors.New("delete failed")
-}
-
 type createFailingKubeClient struct {
 	tillerEnv.PrintingKubeClient
 }

pkg/releasetesting/test_suite.go

@@ -47,13 +47,9 @@ type test struct {
 // NewTestSuite takes a release object and returns a TestSuite object with test definitions
 //  extracted from the release
 func NewTestSuite(rel *release.Release) *TestSuite {
-	testManifests := extractTestManifestsFromHooks(rel.Hooks)
-
-	results := []*release.TestRun{}
-
 	return &TestSuite{
-		TestManifests: testManifests,
+		TestManifests: extractTestManifestsFromHooks(rel.Hooks),
-		Results:       results,
+		Results:       []*release.TestRun{},
 	}
 }
 

pkg/releasetesting/test_suite_test.go

@@ -23,8 +23,6 @@ import (
 	"time"
 
 	"github.com/golang/protobuf/ptypes/timestamp"
-	"golang.org/x/net/context"
-	"google.golang.org/grpc/metadata"
 	"k8s.io/kubernetes/pkg/apis/core"
 
 	"k8s.io/helm/pkg/proto/hapi/chart"
@@ -76,18 +74,27 @@ func TestRun(t *testing.T) {
 
 	testManifests := []string{manifestWithTestSuccessHook, manifestWithTestFailureHook}
 	ts := testSuiteFixture(testManifests)
-	if err := ts.Run(testEnvFixture()); err != nil {
+	ch := make(chan *services.TestReleaseResponse, 1)
-		t.Errorf("%s", err)
+
+	env := testEnvFixture()
+	env.Mesages = ch
+
+	go func() {
+		defer close(ch)
+		if err := ts.Run(env); err != nil {
+			t.Error(err)
+		}
+	}()
+
+	for range ch { // drain
 	}
 
 	if ts.StartedAt == nil {
 		t.Errorf("Expected StartedAt to not be nil. Got: %v", ts.StartedAt)
 	}
-
 	if ts.CompletedAt == nil {
 		t.Errorf("Expected CompletedAt to not be nil. Got: %v", ts.CompletedAt)
 	}
-
 	if len(ts.Results) != 2 {
 		t.Errorf("Expected 2 test result. Got %v", len(ts.Results))
 	}
@@ -96,75 +103,75 @@ func TestRun(t *testing.T) {
 	if result.StartedAt == nil {
 		t.Errorf("Expected test StartedAt to not be nil. Got: %v", result.StartedAt)
 	}
-
 	if result.CompletedAt == nil {
 		t.Errorf("Expected test CompletedAt to not be nil. Got: %v", result.CompletedAt)
 	}
-
 	if result.Name != "finding-nemo" {
 		t.Errorf("Expected test name to be finding-nemo. Got: %v", result.Name)
 	}
-
 	if result.Status != release.TestRun_SUCCESS {
 		t.Errorf("Expected test result to be successful, got: %v", result.Status)
 	}
-
 	result2 := ts.Results[1]
 	if result2.StartedAt == nil {
 		t.Errorf("Expected test StartedAt to not be nil. Got: %v", result2.StartedAt)
 	}
-
 	if result2.CompletedAt == nil {
 		t.Errorf("Expected test CompletedAt to not be nil. Got: %v", result2.CompletedAt)
 	}
-
 	if result2.Name != "gold-rush" {
 		t.Errorf("Expected test name to be gold-rush, Got: %v", result2.Name)
 	}
-
 	if result2.Status != release.TestRun_FAILURE {
 		t.Errorf("Expected test result to be successful, got: %v", result2.Status)
 	}
-
 }
 
 func TestRunEmptyTestSuite(t *testing.T) {
 	ts := testSuiteFixture([]string{})
-	mockTestEnv := testEnvFixture()
+	ch := make(chan *services.TestReleaseResponse, 1)
-	if err := ts.Run(mockTestEnv); err != nil {
+
-		t.Errorf("%s", err)
+	env := testEnvFixture()
+	env.Mesages = ch
+
+	go func() {
+		defer close(ch)
+		if err := ts.Run(env); err != nil {
+			t.Error(err)
 		}
+	}()
 
+	msg := <-ch
+	if msg.Msg != "No Tests Found" {
+		t.Errorf("Expected message 'No Tests Found', Got: %v", msg.Msg)
+	}
 	if ts.StartedAt == nil {
 		t.Errorf("Expected StartedAt to not be nil. Got: %v", ts.StartedAt)
 	}
-
 	if ts.CompletedAt == nil {
 		t.Errorf("Expected CompletedAt to not be nil. Got: %v", ts.CompletedAt)
 	}
-
 	if len(ts.Results) != 0 {
 		t.Errorf("Expected 0 test result. Got %v", len(ts.Results))
 	}
-
-	stream := mockTestEnv.Stream.(*mockStream)
-	if len(stream.messages) == 0 {
-		t.Errorf("Expected at least one message, Got: %v", len(stream.messages))
-	} else {
-		msg := stream.messages[0].Msg
-		if msg != "No Tests Found" {
-			t.Errorf("Expected message 'No Tests Found', Got: %v", msg)
-		}
-	}
-
 }
 
 func TestRunSuccessWithTestFailureHook(t *testing.T) {
 	ts := testSuiteFixture([]string{manifestWithTestFailureHook})
+	ch := make(chan *services.TestReleaseResponse, 1)
+
 	env := testEnvFixture()
 	env.KubeClient = newPodFailedKubeClient()
+	env.Mesages = ch
+
+	go func() {
+		defer close(ch)
 		if err := ts.Run(env); err != nil {
-		t.Errorf("%s", err)
+			t.Error(err)
+		}
+	}()
+
+	for range ch { // drain
 	}
 
 	if ts.StartedAt == nil {
@@ -273,7 +280,11 @@ func testSuiteFixture(testManifests []string) *TestSuite {
 }
 
 func testEnvFixture() *Environment {
-	return newMockTestingEnvironment().Environment
+	return &Environment{
+		Namespace:  "default",
+		KubeClient: mockTillerEnvironment().KubeClient,
+		Timeout:    1,
+	}
 }
 
 func mockTillerEnvironment() *tillerEnv.Environment {
@@ -283,22 +294,6 @@ func mockTillerEnvironment() *tillerEnv.Environment {
 	return e
 }
 
-type mockStream struct {
-	messages []*services.TestReleaseResponse
-}
-
-func (rs *mockStream) Send(m *services.TestReleaseResponse) error {
-	rs.messages = append(rs.messages, m)
-	return nil
-}
-
-func (rs mockStream) SetHeader(m metadata.MD) error  { return nil }
-func (rs mockStream) SendHeader(m metadata.MD) error { return nil }
-func (rs mockStream) SetTrailer(m metadata.MD)       {}
-func (rs mockStream) SendMsg(v interface{}) error    { return nil }
-func (rs mockStream) RecvMsg(v interface{}) error    { return nil }
-func (rs mockStream) Context() context.Context       { return context.TODO() }
-
 type podSucceededKubeClient struct {
 	tillerEnv.PrintingKubeClient
 }

pkg/tiller/release_server_test.go

@@ -28,8 +28,6 @@ import (
 
 	"github.com/ghodss/yaml"
 	"github.com/golang/protobuf/ptypes/timestamp"
-	"golang.org/x/net/context"
-	"google.golang.org/grpc/metadata"
 	"k8s.io/client-go/kubernetes/fake"
 	"k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/kubectl/resource"
@@ -426,16 +424,6 @@ func (h *hookFailingKubeClient) WatchUntilReady(ns string, r io.Reader, timeout
 	return errors.New("Failed watch")
 }
 
-type mockRunReleaseTestServer struct{}
-
-func (rs mockRunReleaseTestServer) Send(m *services.TestReleaseResponse) error { return nil }
-func (rs mockRunReleaseTestServer) SetHeader(m metadata.MD) error              { return nil }
-func (rs mockRunReleaseTestServer) SendHeader(m metadata.MD) error             { return nil }
-func (rs mockRunReleaseTestServer) SetTrailer(m metadata.MD)                   {}
-func (rs mockRunReleaseTestServer) SendMsg(v interface{}) error                { return nil }
-func (rs mockRunReleaseTestServer) RecvMsg(v interface{}) error                { return nil }
-func (rs mockRunReleaseTestServer) Context() context.Context                   { return context.TODO() }
-
 type mockHooksManifest struct {
 	Metadata struct {
 		Name        string

pkg/tiller/release_testing.go

@@ -23,31 +23,39 @@ import (
 )
 
 // RunReleaseTest runs pre-defined tests stored as hooks on a given release
-func (s *ReleaseServer) RunReleaseTest(req *services.TestReleaseRequest, stream services.ReleaseService_RunReleaseTestServer) error {
+func (s *ReleaseServer) RunReleaseTest(req *services.TestReleaseRequest) (<-chan *services.TestReleaseResponse, <-chan error) {
-
+	errc := make(chan error, 1)
 	if err := validateReleaseName(req.Name); err != nil {
 		s.Log("releaseTest: Release name is invalid: %s", req.Name)
-		return err
+		errc <- err
+		return nil, errc
 	}
 
 	// finds the non-deleted release with the given name
 	rel, err := s.env.Releases.Last(req.Name)
 	if err != nil {
-		return err
+		errc <- err
+		return nil, errc
 	}
 
+	ch := make(chan *services.TestReleaseResponse, 1)
 	testEnv := &reltesting.Environment{
 		Namespace:  rel.Namespace,
 		KubeClient: s.env.KubeClient,
 		Timeout:    req.Timeout,
-		Stream:     stream,
+		Mesages:    ch,
 	}
 	s.Log("running tests for release %s", rel.Name)
 	tSuite := reltesting.NewTestSuite(rel)
 
+	go func() {
+		defer close(errc)
+		defer close(ch)
+
 		if err := tSuite.Run(testEnv); err != nil {
 			s.Log("error running test suite for %s: %s", rel.Name, err)
-		return err
+			errc <- err
+			return
 		}
 
 		rel.Info.Status.LastTestSuiteRun = &release.TestSuite{
@@ -63,6 +71,6 @@ func (s *ReleaseServer) RunReleaseTest(req *services.TestReleaseRequest, stream
 		if err := s.env.Releases.Update(rel); err != nil {
 			s.Log("test: Failed to store updated release: %s", err)
 		}
-
+	}()
-	return nil
+	return ch, errc
 }

pkg/tiller/release_testing_test.go

@@ -16,21 +16,14 @@ limitations under the License.
 
 package tiller
 
-import (
+// func TestRunReleaseTest(t *testing.T) {
-	"testing"
+// 	rs := rsFixture()
-
+// 	rel := namedReleaseStub("nemo", release.Status_DEPLOYED)
-	"k8s.io/helm/pkg/proto/hapi/release"
+// 	rs.env.Releases.Create(rel)
-	"k8s.io/helm/pkg/proto/hapi/services"
+
-)
+// 	req := &services.TestReleaseRequest{Name: "nemo", Timeout: 2}
-
+// 	err := rs.RunReleaseTest(req, mockRunReleaseTestServer{})
-func TestRunReleaseTest(t *testing.T) {
+// 	if err != nil {
-	rs := rsFixture()
+// 		t.Fatalf("failed to run release tests on %s: %s", rel.Name, err)
-	rel := namedReleaseStub("nemo", release.Status_DEPLOYED)
+// 	}
-	rs.env.Releases.Create(rel)
+// }
-
-	req := &services.TestReleaseRequest{Name: "nemo", Timeout: 2}
-	err := rs.RunReleaseTest(req, mockRunReleaseTestServer{})
-	if err != nil {
-		t.Fatalf("failed to run release tests on %s: %s", rel.Name, err)
-	}
-}