Abort tests in case of go-vcr failure (Azure#5067)

* Abort tests if go-vcr fails

* Limit the number of response replays

* Only replay GETs for terminal states

* Don't require sequence match

* Update tests

* Tweak decisions about cachability

* Add documentation

* Refactor handling of resource status

* Refactor terminal state detection

* Increase limit for GET replay

* Effectively unlimited GET replays

* Fix lint issue

Author: theunrepentantgeek

v2/internal/testcommon/vcr/v3/replay_roundtripper.go

@@ -8,10 +8,12 @@ package v3
 import (
 	"bytes"
 	"crypto/sha256"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
 	"net/http"
+	"strings"
 	"sync"
 
 	"github.com/go-logr/logr"
@@ -30,13 +32,34 @@ import (
 // the original, which a GET then shows is complete.
 type replayRoundTripper struct {
 	inner    http.RoundTripper
-	gets     map[string]*http.Response
-	puts     map[string]*http.Response
+	gets     map[string]*replayResponse
+	puts     map[string]*replayResponse
 	log      logr.Logger
 	padlock  sync.Mutex
 	redactor *vcr.Redactor
 }
 
+type replayResponse struct {
+	response         *http.Response
+	remainingReplays int
+}
+
+const (
+	// Timing variations during test replay may result in a resource being reconciled an additional time - we don't
+	// want to fail the test in this situation because we've already successfully achieved our goal state.
+	// Thus we allow the last PUT for each resource to be replayed one extra time.
+	maxPutReplays = 1 // Maximum number of times to replay a PUT request
+
+	// GET requests may be replayed multiple times to allow multiple reconciles to observe the same stable final state.
+	// We set this to accommodate timing variations during test replay, while avoiding unbounded replays as they might
+	// result in a test getting stuck and continuing to run until the entire test suite times out.
+	//
+	// Currently we need to set this to an apparently comical limit because some of our tests requiring many many
+	// repetitions to work due to changes in timing used during test replay. Once we've addressed other issues causing
+	// test instability, we should be able to reduce this limit significantly.
+	maxGetReplays = 1000 // Maximum number of times to replay a GET request (effectively unlimited for now)
+)
+
 var _ http.RoundTripper = &replayRoundTripper{}
 
 // newReplayRoundTripper creates a new replayRoundTripper that will replay selected requests to improve test resilience.
@@ -47,8 +70,8 @@ func NewReplayRoundTripper(
 ) http.RoundTripper {
 	return &replayRoundTripper{
 		inner:    inner,
-		gets:     make(map[string]*http.Response),
-		puts:     make(map[string]*http.Response),
+		gets:     make(map[string]*replayResponse),
+		puts:     make(map[string]*replayResponse),
 		log:      log,
 		redactor: redactor,
 	}
@@ -69,6 +92,8 @@ func (replayer *replayRoundTripper) RoundTrip(request *http.Request) (*http.Resp
 }
 
 func (replayer *replayRoundTripper) roundTripGet(request *http.Request) (*http.Response, error) {
+	requestURL := request.URL.RequestURI()
+
 	// First use our inner round tripper to get the response.
 	response, err := replayer.inner.RoundTrip(request)
 	if err != nil {
@@ -81,20 +106,37 @@ func (replayer *replayRoundTripper) roundTripGet(request *http.Request) (*http.R
 		defer replayer.padlock.Unlock()
 
 		// We didn't find an interaction, see if we have a cached response to return
-		if cachedResponse, ok := replayer.gets[request.URL.String()]; ok {
-			replayer.log.Info("Replaying GET request", "url", request.URL.String())
-			return cachedResponse, nil
+		if cachedResponse, ok := replayer.gets[requestURL]; ok {
+			if cachedResponse.remainingReplays > 0 {
+				cachedResponse.remainingReplays--
+				replayer.log.Info("Replaying GET request", "url", requestURL)
+				return cachedResponse.response, nil
+			}
+
+			// It's expired, remove it from the cache to ensure we don't replay it again
+			delete(replayer.gets, requestURL)
 		}
 
 		// No cached response, return the original response and error
 		return response, err
 	}
 
-	replayer.padlock.Lock()
-	defer replayer.padlock.Unlock()
+	// We have a response; if it has a status, cache only if that represents a terminal state
+	cacheable := true
+	if state, ok := replayer.resourceStateFromBody(response); ok {
+		cacheable = replayer.isTerminalProvisioningState(state)
+	}
+	if status, ok := replayer.operationStatusFromBody(response); ok {
+		cacheable = replayer.isTerminalOperationStatus(status)
+	}
+
+	if cacheable {
+		replayer.padlock.Lock()
+		defer replayer.padlock.Unlock()
+
+		replayer.gets[requestURL] = newReplayResponse(response, maxGetReplays)
+	}
 
-	// We have a response, cache it and return it
-	replayer.gets[request.URL.String()] = response
 	return response, nil
 }
 
@@ -115,10 +157,15 @@ func (replayer *replayRoundTripper) roundTripPut(request *http.Request) (*http.R
 
 		// We didn't find an interaction, see if we have a cached response to return
 		if cachedResponse, ok := replayer.puts[hash]; ok {
-			// Remove it from the cache to ensure we only replay it once
+			if cachedResponse.remainingReplays > 0 {
+				replayer.log.Info("Replaying PUT request", "url", request.URL.String(), "hash", hash)
+				cachedResponse.remainingReplays--
+				return cachedResponse.response, nil
+			}
+
+			// It's expired, remove it from the cache to ensure we don't replay it again
 			delete(replayer.puts, hash)
-			replayer.log.Info("Replaying PUT request", "url", request.URL.String(), "hash", hash)
-			return cachedResponse, nil
+
 		}
 
 		// No cached response, return the original response and error
@@ -129,7 +176,7 @@ func (replayer *replayRoundTripper) roundTripPut(request *http.Request) (*http.R
 	defer replayer.padlock.Unlock()
 
 	// We have a response, cache it and return it
-	replayer.puts[hash] = response
+	replayer.puts[hash] = newReplayResponse(response, maxPutReplays)
 	return response, nil
 }
 
@@ -155,3 +202,77 @@ func (replayer *replayRoundTripper) hashOfBody(request *http.Request) string {
 
 	return fmt.Sprintf("%x", hash)
 }
+
+type operation struct {
+	Status string `json:"status"`
+}
+
+// operationStatusFromBody extracts the operation status from the response body, if present.
+func (replayer *replayRoundTripper) operationStatusFromBody(response *http.Response) (string, bool) {
+	body := replayer.bodyOfResponse(response)
+
+	var op operation
+	if err := json.Unmarshal([]byte(body), &op); err == nil {
+		if op.Status != "" {
+			return op.Status, true
+		}
+	}
+
+	return "", false
+}
+
+// isTerminalOperationStatus returns true if the specified status represents a terminal state for a long-running operation
+func (replayer *replayRoundTripper) isTerminalOperationStatus(status string) bool {
+	return !strings.EqualFold(status, "InProgress")
+}
+
+type resource struct {
+	Properties struct {
+		ProvisioningState string `json:"provisioningState"`
+	} `json:"properties"`
+}
+
+// resourceStateFromBody extracts the provisioning state from the response body, if present.
+func (replayer *replayRoundTripper) resourceStateFromBody(response *http.Response) (string, bool) {
+	body := replayer.bodyOfResponse(response)
+
+	// Treat the body as an operation and deserialize it
+
+	// Treat the body as a resource and deserialize it
+	var res resource
+	if err := json.Unmarshal([]byte(body), &res); err == nil {
+		if res.Properties.ProvisioningState != "" {
+			return res.Properties.ProvisioningState, true
+		}
+	}
+
+	return "", false
+}
+
+// isTerminalStatus returns true if the specified status represents a terminal state a resource
+func (*replayRoundTripper) isTerminalProvisioningState(status string) bool {
+	return strings.EqualFold(status, "Succeeded") ||
+		strings.EqualFold(status, "Failed") ||
+		strings.EqualFold(status, "Canceled")
+}
+
+func (replayer *replayRoundTripper) bodyOfResponse(response *http.Response) string {
+	// Read all the content of the response body
+	var body bytes.Buffer
+	_, err := body.ReadFrom(response.Body)
+	if err != nil {
+		// Should never fail
+		panic(fmt.Sprintf("reading response.Body failed: %s", err))
+	}
+
+	// Reset the body so it can be read again
+	response.Body = io.NopCloser(&body)
+	return body.String()
+}
+
+func newReplayResponse(resp *http.Response, maxReplays int) *replayResponse {
+	return &replayResponse{
+		response:         resp,
+		remainingReplays: maxReplays,
+	}
+}

v2/internal/testcommon/vcr/v3/replay_roundtripper_test.go

@@ -23,7 +23,7 @@ import (
 	"github.com/Azure/azure-service-operator/v2/internal/testcommon/vcr"
 )
 
-func TestReplayRoundTripperRoundTrip_GivenSingleGET_ReturnsMultipleTimes(t *testing.T) {
+func TestReplayRoundTripperRoundTrip_GivenSingleGETReturningTerminalState_ReturnsMultipleTimes(t *testing.T) {
 	t.Parallel()
 
 	// Arrange
@@ -35,7 +35,7 @@ func TestReplayRoundTripperRoundTrip_GivenSingleGET_ReturnsMultipleTimes(t *test
 
 	resp := &http.Response{
 		StatusCode: 200,
-		Body:       io.NopCloser(strings.NewReader("GET response goes here")),
+		Body:       io.NopCloser(strings.NewReader(`{"properties":{"provisioningState": "Succeeded"}}`)),
 	}
 
 	fake := vcr.NewFakeRoundTripper()
@@ -47,13 +47,46 @@ func TestReplayRoundTripperRoundTrip_GivenSingleGET_ReturnsMultipleTimes(t *test
 	replayer := NewReplayRoundTripper(fake, logr.Discard(), redactor)
 
 	// Assert - first request works
-	assertExpectedResponse(t, replayer, req, 200, "GET response goes here")
+	assertExpectedResponse(t, replayer, req, 200, `{"provisioningState": "Succeeded"`)
 
 	// Assert - second request works by replaying the first
-	assertExpectedResponse(t, replayer, req, 200, "GET response goes here")
+	assertExpectedResponse(t, replayer, req, 200, `{"provisioningState": "Succeeded"`)
 
 	// Assert - third request works by replaying the first
-	assertExpectedResponse(t, replayer, req, 200, "GET response goes here")
+	assertExpectedResponse(t, replayer, req, 200, `{"provisioningState": "Succeeded"`)
+}
+
+func TestReplayRoundTripperRoundTrip_GivenSingleGETReturningNonterminalState_ReturnsOnce(t *testing.T) {
+	t.Parallel()
+	g := NewGomegaWithT(t)
+
+	// Arrange
+	req := &http.Request{
+		URL:    &url.URL{Path: "/foo"},
+		Method: http.MethodGet,
+		Body:   io.NopCloser(strings.NewReader("GET body goes here")),
+	}
+
+	resp := &http.Response{
+		StatusCode: 200,
+		Body:       io.NopCloser(strings.NewReader(`{"properties":{"provisioningState": "Deleting"}}`)),
+	}
+
+	fake := vcr.NewFakeRoundTripper()
+	fake.AddResponse(req, resp)
+
+	redactor := vcr.NewRedactor(creds.DummyAzureIDs())
+
+	// Act
+	replayer := NewReplayRoundTripper(fake, logr.Discard(), redactor)
+
+	// Assert - first request works
+	assertExpectedResponse(t, replayer, req, 200, `{"provisioningState": "Deleting"`)
+
+	// Assert - second request fails
+	//nolint:bodyclose // there's no actual body in this response to close
+	_, err := fake.RoundTrip(req)
+	g.Expect(err).To(MatchError(ContainSubstring("requested interaction not found")))
 }
 
 func TestReplayRoundTripperRoundTrip_GivenSinglePut_ReturnsOnceExtra(t *testing.T) {

v2/internal/testcommon/vcr/v3/test_recorder.go

@@ -84,15 +84,6 @@ func NewTestRecorder(
 			return false
 		}
 
-		// verify custom request count header matches, if present
-		if header := r.Header.Get(CountHeader); header != "" {
-			interactionHeader := i.Headers.Get(CountHeader)
-			if header != interactionHeader {
-				log.Info("Request count header mismatch", CountHeader, header, "interaction", interactionHeader)
-				return false
-			}
-		}
-
 		// verify custom body hash header matches, if present
 		if header := r.Header.Get(HashHeader); header != "" {
 			interactionHeader := i.Headers.Get(HashHeader)

v2/internal/testcommon/verify.go

@@ -7,7 +7,9 @@ package testcommon
 
 import (
 	"context"
+	"strings"
 
+	"github.com/onsi/gomega"
 	"github.com/rotisserie/eris"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -71,7 +73,17 @@ func (e *Verify) HasState(
 
 	inGoalState := ready.Status == desiredState && ready.Severity == desiredSeverity
 	hasObservedGenerationChanged := oldGeneration != ready.ObservedGeneration
-	return inGoalState && hasObservedGenerationChanged, nil
+	if inGoalState && hasObservedGenerationChanged {
+		return true, nil
+	}
+
+	// If we ever get an error about a missing recording, the test has failed and we
+	// can abort early without waiting for a timeout
+	if strings.Contains(ready.Message, "cannot find go-vcr recording") {
+		return false, gomega.StopTrying(ready.Message)
+	}
+
+	return false, nil
 }
 
 // Deleted verifies that the object specified has been deleted
@@ -88,6 +100,25 @@ func (e *Verify) Deleted(ctx context.Context, obj client.Object) (bool, error) {
 		return false, err
 	}
 
+	// Object hasn't been deleted, check its Condition for a terminal problem
+	conditioner, ok := obj.(conditions.Conditioner)
+	if !ok {
+		return false, eris.Errorf("result of get was not conditions.Conditioner, was: %T", obj)
+	}
+
+	ready, ok := conditions.GetCondition(conditioner, conditions.ConditionTypeReady)
+	if !ok {
+		return false, nil
+	}
+
+	// If we ever get an error about a missing recording, the test has failed and we
+	// can abort early without waiting for a timeout
+	if ready.Reason == "DeletionNotSupportedInAzure" {
+		if strings.Contains(ready.Message, "cannot find go-vcr recording") {
+			return false, gomega.StopTrying(ready.Message)
+		}
+	}
+
 	return false, nil
 }