fix pulsar admin revoke subscription permission endpoint (#1251)

* fix pulsar admin revoke subscription permission endpoint

* feat: support get subscription permissions and add test case

* feat: remove fmt package

---------

Co-authored-by: haotao chen <haotao.chen@shopee.com>
(cherry picked from commit df6a15c)

pulsaradmin/pkg/admin/namespace.go

@@ -200,6 +200,9 @@ type Namespaces interface {
  // RevokeSubPermission revoke permissions on a subscription's admin-api access
  RevokeSubPermission(namespace utils.NameSpaceName, sName, role string) error
 
+ // GetSubPermissions returns subscription permissions on a namespace
+ GetSubPermissions(namespace utils.NameSpaceName) (map[string][]string, error)
+
  // SetSubscriptionAuthMode sets the given subscription auth mode on all topics on a namespace
  SetSubscriptionAuthMode(namespace utils.NameSpaceName, mode utils.SubscriptionAuthMode) error
 
@@ -750,10 +753,17 @@ func (n *namespaces) GrantSubPermission(namespace utils.NameSpaceName, sName str
 
 func (n *namespaces) RevokeSubPermission(namespace utils.NameSpaceName, sName, role string) error {
  endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "permissions",
- "subscription", sName, role)
+ sName, role)
  return n.pulsar.Client.Delete(endpoint)
 }
 
+func (n *namespaces) GetSubPermissions(namespace utils.NameSpaceName) (map[string][]string, error) {
+ endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "permissions", "subscription")
+ var permissions map[string][]string
+ err := n.pulsar.Client.Get(endpoint, &permissions)
+ return permissions, err
+}
+
 func (n *namespaces) SetSubscriptionAuthMode(namespace utils.NameSpaceName, mode utils.SubscriptionAuthMode) error {
  endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "subscriptionAuthMode")
  return n.pulsar.Client.Post(endpoint, mode.String())

pulsaradmin/pkg/admin/namespace_test.go

@@ -173,3 +173,31 @@ func TestGetTopicAutoCreation(t *testing.T) {
  }
  assert.Equal(t, expected, *topicAutoCreation)
 }
+
+func TestRevokeSubPermission(t *testing.T) {
+ config := &config.Config{}
+ admin, err := New(config)
+ require.NoError(t, err)
+ require.NotNil(t, admin)
+
+ namespace, err := utils.GetNamespaceName("public/default")
+ require.NoError(t, err)
+ require.NotNil(t, namespace)
+
+ sub := "subscription"
+ roles := []string{"user"}
+
+ // grant subscription permission and get it
+ err = admin.Namespaces().GrantSubPermission(*namespace, sub, roles)
+ require.NoError(t, err)
+ permissions, err := admin.Namespaces().GetSubPermissions(*namespace)
+ require.NoError(t, err)
+ assert.Equal(t, roles, permissions[sub])
+
+ // revoke subscription permission and get it
+ err = admin.Namespaces().RevokeSubPermission(*namespace, sub, roles[0])
+ require.NoError(t, err)
+ permissions, err = admin.Namespaces().GetSubPermissions(*namespace)
+ require.NoError(t, err)
+ assert.Equal(t, 0, len(permissions[sub]))
+}