create functional zero-change watch

Author: deads2k

pkg/manifestclient/io.go

@@ -0,0 +1,37 @@
+package manifestclient
+
+import (
+	"io"
+	"sync/atomic"
+	"time"
+)
+
+func newDelayedNothingReader(timeout time.Duration) *delayedNothingReaderCloser {
+	return &delayedNothingReaderCloser{timeout: timeout}
+}
+
+type delayedNothingReaderCloser struct {
+	timeout time.Duration
+	closed  atomic.Bool
+}
+
+func (d *delayedNothingReaderCloser) Read(p []byte) (n int, err error) {
+	if d.closed.Load() {
+		return 0, io.EOF
+	}
+	select {
+	case <-time.After(d.timeout):
+		d.Close()
+	}
+	if d.closed.Load() {
+		return 0, io.EOF
+	}
+	return 0, nil
+}
+
+func (d *delayedNothingReaderCloser) Close() error {
+	d.closed.Store(true)
+	return nil
+}
+
+var _ io.ReadCloser = &delayedNothingReaderCloser{}

pkg/manifestclient/roundtripper.go

@@ -9,6 +9,8 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
+	"strconv"
+	"time"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -123,6 +125,24 @@ func (mrt *manifestRoundTripper) RoundTrip(req *http.Request) (*http.Response, e
 		// TODO handle label and field selectors
 		returnBody, returnErr = mrt.list(requestInfo)
 
+	case "watch":
+		// our watches do nothing.  We keep the connection alive (I  think), but nothing else.
+		timeoutSecondsString := req.URL.Query().Get("timeoutSeconds")
+		timeoutDuration := 10 * time.Minute
+		if len(timeoutSecondsString) > 0 {
+			currSeconds, err := strconv.ParseInt(timeoutSecondsString, 10, 32)
+			if err != nil {
+				returnErr = err
+				break
+			}
+			timeoutDuration = time.Duration(currSeconds) * time.Second
+		}
+		resp := &http.Response{}
+		resp.StatusCode = http.StatusOK
+		resp.Status = http.StatusText(resp.StatusCode)
+		resp.Body = newDelayedNothingReader(timeoutDuration)
+		return resp, nil
+
 	default:
 		return nil, fmt.Errorf("verb %v is not supported by this implementation", requestInfo.Verb)
 	}

pkg/manifestclienttest/client_test.go

@@ -6,6 +6,7 @@ import (
 	"net/http"
 	"reflect"
 	"testing"
+	"time"
 
 	"github.com/davecgh/go-spew/spew"
 	configv1 "github.com/openshift/api/config/v1"
@@ -22,29 +23,6 @@ import (
 var mustGather01 embed.FS
 
 func TestSimpleChecks(t *testing.T) {
-	mustGatherRoundTripper, err := manifestclient.NewRoundTripper("testdata/must-gather-01")
-	if err != nil {
-		t.Fatal(err)
-	}
-	testRoundTripper, err := manifestclient.NewTestingRoundTripper(mustGather01, "testdata/must-gather-01")
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	roundTrippers := []struct {
-		name         string
-		roundTripper http.RoundTripper
-	}{
-		{
-			name:         "directory read",
-			roundTripper: mustGatherRoundTripper,
-		},
-		{
-			name:         "embed read",
-			roundTripper: testRoundTripper,
-		},
-	}
-
 	tests := []struct {
 		name   string
 		testFn func(*testing.T, *http.Client)
@@ -178,15 +156,91 @@ func TestSimpleChecks(t *testing.T) {
 			},
 		},
 	}
-	for _, roundTripperTest := range roundTrippers {
-		httpClient := &http.Client{
-			Transport: roundTripperTest.roundTripper,
-		}
 
+	for _, roundTripperTest := range defaultRoundTrippers(t) {
+		t.Run(roundTripperTest.name, func(t *testing.T) {
+			for _, test := range tests {
+				t.Run(test.name, func(t *testing.T) {
+					test.testFn(t, roundTripperTest.getClient())
+				})
+			}
+		})
+	}
+}
+
+func defaultRoundTrippers(t *testing.T) []*testRoundTrippers {
+	t.Helper()
+
+	mustGatherRoundTripper, err := manifestclient.NewRoundTripper("testdata/must-gather-01")
+	if err != nil {
+		t.Fatal(err)
+	}
+	testRoundTripper, err := manifestclient.NewTestingRoundTripper(mustGather01, "testdata/must-gather-01")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	return []*testRoundTrippers{
+		{
+			name:         "directory read",
+			roundTripper: mustGatherRoundTripper,
+		},
+		{
+			name:         "embed read",
+			roundTripper: testRoundTripper,
+		},
+	}
+}
+
+type testRoundTrippers struct {
+	name         string
+	roundTripper http.RoundTripper
+}
+
+func (r *testRoundTrippers) getClient() *http.Client {
+	return &http.Client{
+		Transport: r.roundTripper,
+	}
+}
+
+func TestWatchChecks(t *testing.T) {
+	tests := []struct {
+		name   string
+		testFn func(*testing.T, *http.Client)
+	}{
+		{
+			name: "WATCH-from-individual-file-success-server-close",
+			testFn: func(t *testing.T, httpClient *http.Client) {
+				timeout := int64(4)
+				configClient, err := configclient.NewForConfigAndClient(&rest.Config{}, httpClient)
+				if err != nil {
+					t.Fatal(err)
+				}
+				watcher, err := configClient.ConfigV1().FeatureGates().Watch(context.TODO(), metav1.ListOptions{
+					TimeoutSeconds: &timeout,
+				})
+				if err != nil {
+					t.Fatal(err)
+				}
+				select {
+				case <-watcher.ResultChan():
+					t.Fatal("closed early!")
+				case <-time.After(500 * time.Millisecond):
+				}
+
+				select {
+				case <-watcher.ResultChan():
+				case <-time.After(5 * time.Second):
+					t.Fatal("closed late!")
+				}
+			},
+		},
+	}
+	for _, roundTripperTest := range defaultRoundTrippers(t) {
 		t.Run(roundTripperTest.name, func(t *testing.T) {
 			for _, test := range tests {
 				t.Run(test.name, func(t *testing.T) {
-					test.testFn(t, httpClient)
+					test.testFn(t, roundTripperTest.getClient())
 				})
 			}
 		})