Add topic name query param on LIST namespace API (#491)

src/main/java/com/michelin/ns4kafka/controller/NamespaceController.java

@@ -24,6 +24,7 @@
 import jakarta.validation.Valid;
 import java.time.Instant;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Date;
 import java.util.List;
 import java.util.Optional;
@@ -39,14 +40,21 @@ public class NamespaceController extends NonNamespacedResourceController {
     NamespaceService namespaceService;
 
     /**
-     * List namespaces, filtered by name parameter.
+     * List namespaces, filtered by namespace name and topic name parameters.
      *
-     * @param name The name parameter
+     * @param name  The namespace name parameter
+     * @param topic The topic name parameter
      * @return A list of namespaces
      */
     @Get
-    public List<Namespace> list(@QueryValue(defaultValue = "*") String name) {
-        return namespaceService.findByWildcardName(name);
+    public List<Namespace> list(@QueryValue(defaultValue = "*") String name,
+                                @QueryValue(defaultValue = "") String topic) {
+        List<Namespace> namespaces = namespaceService.findByWildcardName(name);
+
+        return topic.isEmpty() ? namespaces :
+            namespaceService.findByTopicName(namespaces, topic)
+                .map(Collections::singletonList)
+                .orElse(List.of());
     }
 
     /**

src/main/java/com/michelin/ns4kafka/service/AclService.java

@@ -366,7 +366,7 @@ public List<AccessControlEntry> findAllRelatedToNamespaceByWildcardName(Namespac
     }
 
     /**
-     * Find all owner-ACLs on a resource for a given namespace.
+     * Find all owner-ACLs on a resource granted to a given namespace.
      *
      * @param namespace    The namespace
      * @param resourceType The resource
@@ -376,10 +376,9 @@ public List<AccessControlEntry> findResourceOwnerGrantedToNamespace(Namespace na
                                                                         AccessControlEntry.ResourceType resourceType) {
         return accessControlEntryRepository.findAll()
             .stream()
-            .filter(accessControlEntry ->
-                accessControlEntry.getSpec().getGrantedTo().equals(namespace.getMetadata().getName())
-                    && accessControlEntry.getSpec().getPermission() == AccessControlEntry.Permission.OWNER
-                    && accessControlEntry.getSpec().getResourceType() == resourceType)
+            .filter(acl -> acl.getSpec().getGrantedTo().equals(namespace.getMetadata().getName())
+                && acl.getSpec().getPermission() == AccessControlEntry.Permission.OWNER
+                && acl.getSpec().getResourceType() == resourceType)
             .toList();
     }
 

src/main/java/com/michelin/ns4kafka/service/NamespaceService.java

@@ -9,6 +9,7 @@
 import static com.michelin.ns4kafka.util.enumation.Kind.ROLE_BINDING;
 import static com.michelin.ns4kafka.util.enumation.Kind.TOPIC;
 
+import com.michelin.ns4kafka.model.AccessControlEntry;
 import com.michelin.ns4kafka.model.Namespace;
 import com.michelin.ns4kafka.property.ManagedClusterProperties;
 import com.michelin.ns4kafka.repository.NamespaceRepository;
@@ -80,6 +81,21 @@ public List<Namespace> findByWildcardName(String name) {
             .toList();
     }
 
+    /**
+     * Find the namespace which are owner of the given topic name, out of the given list.
+     *
+     * @param namespaces The namespaces list
+     * @param topic      The topic name to search
+     * @return The namespace which is owner of the given topic name
+     */
+    public Optional<Namespace> findByTopicName(List<Namespace> namespaces, String topic) {
+        return namespaces
+            .stream()
+            .filter(ns -> aclService.isResourceCoveredByAcls(
+                aclService.findResourceOwnerGrantedToNamespace(ns, AccessControlEntry.ResourceType.TOPIC), topic))
+            .findFirst();
+    }
+
     /**
      * Find a namespace by name.
      *

src/test/java/com/michelin/ns4kafka/controller/NamespaceControllerTest.java

@@ -60,7 +60,7 @@ void shouldListNamespacesWithWildcardParameter() {
         when(namespaceService.findByWildcardName("*"))
             .thenReturn(List.of(ns1, ns2));
 
-        assertEquals(List.of(ns1, ns2), namespaceController.list("*"));
+        assertEquals(List.of(ns1, ns2), namespaceController.list("*", ""));
     }
 
     @Test
@@ -74,13 +74,45 @@ void shouldListNamespacesWithNameParameter() {
         when(namespaceService.findByWildcardName("ns"))
             .thenReturn(List.of(ns));
 
-        assertEquals(List.of(ns), namespaceController.list("ns"));
+        assertEquals(List.of(ns), namespaceController.list("ns", ""));
+    }
+
+    @Test
+    void shouldListNamespaceFilteredByTopic() {
+        Namespace ns = Namespace.builder()
+            .metadata(Metadata.builder()
+                .name("ns")
+                .build())
+            .build();
+
+        when(namespaceService.findByWildcardName("*"))
+            .thenReturn(List.of(ns));
+        when(namespaceService.findByTopicName(List.of(ns), "topic"))
+            .thenReturn(Optional.of(ns));
+
+        assertEquals(List.of(ns), namespaceController.list("*", "topic"));
+    }
+
+    @Test
+    void shouldListNoNamespaceFilteredByTopic() {
+        Namespace ns = Namespace.builder()
+            .metadata(Metadata.builder()
+                .name("ns")
+                .build())
+            .build();
+
+        when(namespaceService.findByWildcardName("*"))
+            .thenReturn(List.of(ns));
+        when(namespaceService.findByTopicName(List.of(ns), "topic"))
+            .thenReturn(Optional.empty());
+
+        assertEquals(List.of(), namespaceController.list("*", "topic"));
     }
 
     @Test
     void shouldListNamespacesWhenEmpty() {
         when(namespaceService.findByWildcardName("*")).thenReturn(List.of());
-        assertEquals(List.of(), namespaceController.list("*"));
+        assertEquals(List.of(), namespaceController.list("*", ""));
     }
 
     @Test

src/test/java/com/michelin/ns4kafka/service/NamespaceServiceTest.java

@@ -516,6 +516,76 @@ void shouldListNamespacesWithSuffixWildcardNameParameter() {
         assertEquals(List.of(ns2, ns5), namespaceService.findByWildcardName("*2"));
     }
 
+    @Test
+    void shouldFindNamespaceByTopicName() {
+        Namespace ns1 = Namespace.builder()
+            .metadata(Metadata.builder()
+                .name("ns1")
+                .build())
+            .build();
+
+        Namespace ns2 = Namespace.builder()
+            .metadata(Metadata.builder()
+                .name("ns2")
+                .build())
+            .build();
+
+        Namespace ns3 = Namespace.builder()
+            .metadata(Metadata.builder()
+                .name("namespace1")
+                .build())
+            .build();
+
+        AccessControlEntry acl3 = AccessControlEntry.builder()
+            .spec(AccessControlEntry.AccessControlEntrySpec.builder()
+                .resourceType(AccessControlEntry.ResourceType.TOPIC)
+                .resourcePatternType(AccessControlEntry.ResourcePatternType.PREFIXED)
+                .permission(AccessControlEntry.Permission.OWNER)
+                .resource("abc.")
+                .grantedTo("namespace1")
+                .build())
+            .build();
+
+        when(aclService.findResourceOwnerGrantedToNamespace(ns1, AccessControlEntry.ResourceType.TOPIC))
+            .thenReturn(List.of());
+        when(aclService.findResourceOwnerGrantedToNamespace(ns2, AccessControlEntry.ResourceType.TOPIC))
+            .thenReturn(List.of());
+        when(aclService.findResourceOwnerGrantedToNamespace(ns3, AccessControlEntry.ResourceType.TOPIC))
+            .thenReturn(List.of(acl3));
+        when(aclService.isResourceCoveredByAcls(List.of(), "abc.topic"))
+            .thenReturn(false);
+        when(aclService.isResourceCoveredByAcls(List.of(acl3), "abc.topic"))
+            .thenReturn(true);
+
+        assertEquals(Optional.of(ns3), namespaceService.findByTopicName(List.of(ns1, ns2, ns3), "abc.topic"));
+    }
+
+    @Test
+    void shouldFindNoNamespaceByTopicName() {
+        Namespace ns = Namespace.builder()
+            .metadata(Metadata.builder()
+                .name("ns")
+                .build())
+            .build();
+
+        AccessControlEntry acl = AccessControlEntry.builder()
+            .spec(AccessControlEntry.AccessControlEntrySpec.builder()
+                .resourceType(AccessControlEntry.ResourceType.TOPIC)
+                .resourcePatternType(AccessControlEntry.ResourcePatternType.PREFIXED)
+                .permission(AccessControlEntry.Permission.OWNER)
+                .resource("abc.")
+                .grantedTo("ns")
+                .build())
+            .build();
+
+        when(aclService.findResourceOwnerGrantedToNamespace(ns, AccessControlEntry.ResourceType.TOPIC))
+            .thenReturn(List.of(acl));
+        when(aclService.isResourceCoveredByAcls(List.of(acl), "xyz.topic"))
+            .thenReturn(false);
+
+        assertEquals(Optional.empty(), namespaceService.findByTopicName(List.of(ns), "xyz.topic"));
+    }
+
     @Test
     void shouldListAllNamespaceResourcesWhenEmpty() {
         Namespace ns = Namespace.builder()