Function TestClusterAdminCreateAcls

admin_test.go:1404–1453

(t *testing.T)

}

func TestClusterAdminCreateAcls(t *testing.T) {
	resource := Resource{ResourceType: AclResourceTopic, ResourceName: "my_topic"}
	acl := Acl{Host: "localhost", Operation: AclOperationAlter, PermissionType: AclPermissionAny}
	createOK := func(req *request) encoderWithHeader { return NewMockCreateAclsResponse(t).For(req.body) }
	notController := func(req *request) encoderWithHeader {
		r := req.body.(*CreateAclsRequest)
		rsp := &CreateAclsResponse{Version: r.version()}
		for range r.AclCreations {
			rsp.AclCreationResponses = append(rsp.AclCreationResponses, &AclCreationResponse{Err: ErrNotController})
		}
		return rsp
	}

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

		rACLs := []*ResourceAcls{
			{
				Resource: resource,
				Acls:     []*Acl{&acl},
			},
			{
				Resource: Resource{ResourceType: AclResourceTopic, ResourceName: "your_topic"},
				Acls:     []*Acl{&acl},
			},
		}

		err := admin.CreateACLs(rACLs)
		require.NoError(t, err)
	})

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

		err := admin.CreateACL(resource, acl)
		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, retriedOnNewController := staleControllerAdmin(t, V1_0_0_0, "CreateAclsRequest", notController, notController)

		err := admin.CreateACL(resource, acl)
		require.Error(t, err)
		require.ErrorIs(t, err, ErrNotController)
		assert.True(t, retriedOnNewController(), "expected broker 2 to receive the retried request")
	})
}

// singleBrokerAdmin wires a single mock broker that names itself controller,
// installs the given per-request handlers, and returns a ready admin client.