Function TestClusterAdminDeleteAcl

admin_test.go:1567–1602

(t *testing.T)

}

func TestClusterAdminDeleteAcl(t *testing.T) {
	resourceName := "my_topic"
	filter := AclFilter{ResourceType: AclResourceTopic, Operation: AclOperationAlter, ResourceName: &resourceName}
	deleteOK := func(req *request) encoderWithHeader { return NewMockDeleteAclsResponse(t).For(req.body) }
	notController := func(req *request) encoderWithHeader {
		return &DeleteAclsResponse{
			Version:         req.body.version(),
			FilterResponses: []*FilterResponse{{Err: ErrNotController}},
		}
	}

	t.Run("deletes acls", func(t *testing.T) {
		admin := singleBrokerAdmin(t, V1_0_0_0, map[string]requestHandlerFunc{
			"DeleteAclsRequest": deleteOK,
		})

		_, err := admin.DeleteACL(filter, false)
		require.NoError(t, err)
	})

	t.Run("retries on stale controller", func(t *testing.T) {
		admin, retriedOnNewController := staleControllerAdmin(t, V1_0_0_0, "DeleteAclsRequest", notController, deleteOK)

		_, err := admin.DeleteACL(filter, false)
		require.NoError(t, err)
		assert.True(t, retriedOnNewController(), "expected broker 2 to receive the retried request")
	})

	t.Run("returns error when retries exhausted", func(t *testing.T) {
		admin, _ := staleControllerAdmin(t, V1_0_0_0, "DeleteAclsRequest", notController, notController)

		_, err := admin.DeleteACL(filter, false)
		require.Error(t, err)
		assert.ErrorIs(t, err, ErrNotController)
	})
}

func TestClusterAdminDescribeUserScramCredentials(t *testing.T) {
	users := []string{"my_user"}