consul/internal/resource/resourcetest/client.go

package resourcetest

import (
	"fmt"
	"math/rand"
	"time"

	"github.com/hashicorp/consul/internal/resource"
	"github.com/hashicorp/consul/proto-public/pbresource"
	"github.com/hashicorp/consul/sdk/testutil"
	"github.com/hashicorp/consul/sdk/testutil/retry"
	"github.com/stretchr/testify/require"
	"golang.org/x/exp/slices"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

type Client struct {
	pbresource.ResourceServiceClient

	timeout time.Duration
	wait    time.Duration
}

func NewClient(client pbresource.ResourceServiceClient) *Client {
	return &Client{
		ResourceServiceClient: client,
		timeout:               7 * time.Second,
		wait:                  25 * time.Millisecond,
	}
}

func (client *Client) SetRetryerConfig(timeout time.Duration, wait time.Duration) {
	client.timeout = timeout
	client.wait = wait
}

func (client *Client) retry(t T, fn func(r *retry.R)) {
	t.Helper()
	retryer := &retry.Timer{Timeout: client.timeout, Wait: client.wait}
	retry.RunWith(retryer, t, fn)
}

func (client *Client) PublishResources(t T, resources []*pbresource.Resource) {
	ctx := testutil.TestContext(t)

	// Randomize the order of insertion. Generally insertion order shouldn't matter as the
	// controllers should eventually converge on the desired state. The exception to this
	// is that you cannot insert resources with owner refs before the resource they are
	// owned by or insert a resource into a non-default tenant before that tenant exists.
	rand.Shuffle(len(resources), func(i, j int) {
		temp := resources[i]
		resources[i] = resources[j]
		resources[j] = temp
	})

	// This slice will be used to track the resources actually published each round. When
	// a resource with an owner ID is encountered we will not attempt to write but defer it
	// to the next round of publishing
	var written []*pbresource.ID

	for len(resources) > 0 {
		var left []*pbresource.Resource
		published := 0
		for _, res := range resources {

			// check that any owner references would be satisfied
			if res.Owner != nil {
				found := slices.ContainsFunc(written, func(id *pbresource.ID) bool {
					return resource.EqualID(res.Owner, id)
				})

				// the owner hasn't yet been published then we cannot publish this resource
				if !found {
					left = append(left, res)
					continue
				}
			}

			t.Logf("Writing resource %s with type %s", res.Id.Name, resource.ToGVK(res.Id.Type))
			rsp, err := client.Write(ctx, &pbresource.WriteRequest{
				Resource: res,
			})
			require.NoError(t, err)

			id := rsp.Resource.Id
			t.Cleanup(func() {
				client.MustDelete(t, id)
			})

			// track the number of resources published
			published += 1
			written = append(written, res.Id)
		}

		// the next round only has this subset of resources to attempt writing
		resources = left

		// if we didn't publish any resources this round then nothing would
		// enable us to do so by iterating again so we break to prevent infinite
		// loooping.
		if published == 0 {
			break
		}
	}

	require.Empty(t, resources, "Could not publish all resources - some resources have invalid owner references")
}

func (client *Client) RequireResourceNotFound(t T, id *pbresource.ID) {
	t.Helper()

	rsp, err := client.Read(testutil.TestContext(t), &pbresource.ReadRequest{Id: id})
	require.Error(t, err)
	require.Equal(t, codes.NotFound, status.Code(err))
	require.Nil(t, rsp)
}

func (client *Client) RequireResourceExists(t T, id *pbresource.ID) *pbresource.Resource {
	t.Helper()

	rsp, err := client.Read(testutil.TestContext(t), &pbresource.ReadRequest{Id: id})
	require.NoError(t, err, "error reading %s with type %s", id.Name, resource.ToGVK(id.Type))
	require.NotNil(t, rsp)
	return rsp.Resource
}

func (client *Client) RequireVersionUnchanged(t T, id *pbresource.ID, version string) *pbresource.Resource {
	t.Helper()

	res := client.RequireResourceExists(t, id)
	RequireVersionUnchanged(t, res, version)
	return res
}

func (client *Client) RequireVersionChanged(t T, id *pbresource.ID, version string) *pbresource.Resource {
	t.Helper()

	res := client.RequireResourceExists(t, id)
	RequireVersionChanged(t, res, version)
	return res
}

func (client *Client) RequireStatusCondition(t T, id *pbresource.ID, statusKey string, condition *pbresource.Condition) *pbresource.Resource {
	t.Helper()

	res := client.RequireResourceExists(t, id)
	RequireStatusCondition(t, res, statusKey, condition)
	return res
}

func (client *Client) RequireStatusConditionForCurrentGen(t T, id *pbresource.ID, statusKey string, condition *pbresource.Condition) *pbresource.Resource {
	t.Helper()

	res := client.RequireResourceExists(t, id)
	RequireStatusConditionForCurrentGen(t, res, statusKey, condition)
	return res
}

func (client *Client) RequireResourceMeta(t T, id *pbresource.ID, key string, value string) *pbresource.Resource {
	t.Helper()

	res := client.RequireResourceExists(t, id)
	RequireResourceMeta(t, res, key, value)
	return res
}

func (client *Client) RequireReconciledCurrentGen(t T, id *pbresource.ID, statusKey string) *pbresource.Resource {
	t.Helper()

	res := client.RequireResourceExists(t, id)
	RequireReconciledCurrentGen(t, res, statusKey)
	return res
}

func (client *Client) WaitForReconciliation(t T, id *pbresource.ID, statusKey string) *pbresource.Resource {
	t.Helper()

	var res *pbresource.Resource

	client.retry(t, func(r *retry.R) {
		res = client.RequireReconciledCurrentGen(r, id, statusKey)
	})

	return res
}

func (client *Client) WaitForStatusCondition(t T, id *pbresource.ID, statusKey string, condition *pbresource.Condition) *pbresource.Resource {
	t.Helper()

	var res *pbresource.Resource
	client.retry(t, func(r *retry.R) {
		res = client.RequireStatusConditionForCurrentGen(r, id, statusKey, condition)
	})

	return res
}

func (client *Client) WaitForNewVersion(t T, id *pbresource.ID, version string) *pbresource.Resource {
	t.Helper()

	var res *pbresource.Resource
	client.retry(t, func(r *retry.R) {
		res = client.RequireVersionChanged(r, id, version)
	})
	return res
}

func (client *Client) WaitForResourceState(t T, id *pbresource.ID, verify func(T, *pbresource.Resource)) *pbresource.Resource {
	t.Helper()

	var res *pbresource.Resource
	client.retry(t, func(r *retry.R) {
		res = client.RequireResourceExists(r, id)
		verify(r, res)
	})

	return res
}

func (client *Client) WaitForDeletion(t T, id *pbresource.ID) {
	t.Helper()

	client.retry(t, func(r *retry.R) {
		client.RequireResourceNotFound(r, id)
	})
}

// ResolveResourceID will read the specified resource and returns its full ID.
// This is mainly useful to get the ID with the Uid filled out.
func (client *Client) ResolveResourceID(t T, id *pbresource.ID) *pbresource.ID {
	t.Helper()

	return client.RequireResourceExists(t, id).Id
}

func (client *Client) MustDelete(t T, id *pbresource.ID) {
	t.Helper()
	ctx := testutil.TestContext(t)

	client.retry(t, func(r *retry.R) {
		_, err := client.Delete(ctx, &pbresource.DeleteRequest{Id: id})
		if status.Code(err) == codes.NotFound {
			return
		}

		// codes.Aborted indicates a CAS failure and that the delete request should
		// be retried. Anything else should be considered an unrecoverable error.
		if err != nil && status.Code(err) != codes.Aborted {
			r.Stop(fmt.Errorf("failed to delete the resource: %w", err))
			return
		}

		require.NoError(r, err)
	})
}