Handle wildcard parameter in namespace deletion API (#442)

michelin · Sep 24, 2024 · 5286f98 · 5286f98
1 parent 9bf1990
commit 5286f98
Show file tree

Hide file tree

Showing 2 changed files with 173 additions and 6 deletions.
diff --git a/src/main/java/com/michelin/ns4kafka/controller/NamespaceController.java b/src/main/java/com/michelin/ns4kafka/controller/NamespaceController.java
@@ -120,8 +120,10 @@ public HttpResponse<Namespace> apply(@Valid @Body Namespace namespace,
      * @param namespace The namespace
      * @param dryrun    Is dry run mode or not ?
      * @return An HTTP response
+     * @deprecated use bulkDelete instead.
      */
     @Delete("/{namespace}{?dryrun}")
+    @Deprecated(since = "1.13.0")
     public HttpResponse<Void> delete(String namespace, @QueryValue(defaultValue = "false") boolean dryrun) {
         Optional<Namespace> optionalNamespace = namespaceService.findByName(namespace);
         if (optionalNamespace.isEmpty()) {
@@ -141,17 +143,65 @@ public HttpResponse<Void> delete(String namespace, @QueryValue(defaultValue = "f
             return HttpResponse.noContent();
         }
 
-        var namespaceToDelete = optionalNamespace.get();
+        performDeletion(optionalNamespace.get());
+        return HttpResponse.noContent();
+    }
+
+    /**
+     * Delete namespaces.
+     *
+     * @param dryrun Is dry run mode or not ?
+     * @param name The name parameter
+     * @return An HTTP response
+     */
+    @Delete
+    public HttpResponse<Void> bulkDelete(@QueryValue(defaultValue = "*") String name,
+                                         @QueryValue(defaultValue = "false") boolean dryrun) {
+        List<Namespace> namespaces = namespaceService.findByWildcardName(name);
+        if (namespaces.isEmpty()) {
+            return HttpResponse.notFound();
+        }
+
+        List<String> namespaceResources = namespaces
+            .stream()
+            .flatMap(namespace -> namespaceService.findAllResourcesByNamespace(namespace)
+                .stream())
+            .toList();
+
+        if (!namespaceResources.isEmpty()) {
+            List<String> validationErrors = namespaceResources
+                .stream()
+                .map(FormatErrorUtils::invalidNamespaceDeleteOperation)
+                .toList();
+
+            throw new ResourceValidationException(
+                NAMESPACE,
+                String.join(",", namespaces.stream().map(namespace -> namespace.getMetadata().getName()).toList()),
+                validationErrors
+            );
+        }
 
+        if (dryrun) {
+            return HttpResponse.noContent();
+        }
+
+        namespaces.forEach(this::performDeletion);
+        return HttpResponse.noContent();
+    }
+
+    /**
+     * Perform the deletion of the namespace and send an event log.
+     *
+     * @param namespace The namespace to delete
+     */
+    private void performDeletion(Namespace namespace) {
         sendEventLog(
-            namespaceToDelete,
+            namespace,
             ApplyStatus.deleted,
-            namespaceToDelete.getSpec(),
+            namespace.getSpec(),
             null,
             EMPTY_STRING
         );
-
-        namespaceService.delete(optionalNamespace.get());
-        return HttpResponse.noContent();
+        namespaceService.delete(namespace);
     }
 }
diff --git a/src/test/java/com/michelin/ns4kafka/controller/NamespaceControllerTest.java b/src/test/java/com/michelin/ns4kafka/controller/NamespaceControllerTest.java
@@ -291,6 +291,7 @@ void shouldUpdateNamespaceInDryRunMode() {
     }
 
     @Test
+    @SuppressWarnings("deprecation")
     void shouldDeleteNamespace() {
         Namespace existing = Namespace.builder()
             .metadata(Metadata.builder()
@@ -317,6 +318,7 @@ void shouldDeleteNamespace() {
     }
 
     @Test
+    @SuppressWarnings("deprecation")
     void shouldDeleteNamespaceInDryRunMode() {
         Namespace existing = Namespace.builder()
             .metadata(Metadata.builder()
@@ -340,6 +342,7 @@ void shouldDeleteNamespaceInDryRunMode() {
     }
 
     @Test
+    @SuppressWarnings("deprecation")
     void shouldNotDeleteNamespaceWhenNotFound() {
         when(namespaceService.findByName("namespace"))
             .thenReturn(Optional.empty());
@@ -351,6 +354,7 @@ void shouldNotDeleteNamespaceWhenNotFound() {
     }
 
     @Test
+    @SuppressWarnings("deprecation")
     void shouldNotDeleteNamespaceWhenResourcesAreStillLinkedWithIt() {
         Namespace existing = Namespace.builder()
             .metadata(Metadata.builder()
@@ -371,4 +375,117 @@ void shouldNotDeleteNamespaceWhenResourcesAreStillLinkedWithIt() {
             () -> namespaceController.delete("namespace", false));
         verify(namespaceService, never()).delete(any());
     }
+
+    @Test
+    void shouldDeleteNamespaces() {
+        Namespace namespace1 = Namespace.builder()
+            .metadata(Metadata.builder()
+                .name("namespace1")
+                .cluster("local")
+                .build())
+            .spec(Namespace.NamespaceSpec.builder()
+                .kafkaUser("user")
+                .build())
+            .build();
+
+        Namespace namespace2 = Namespace.builder()
+            .metadata(Metadata.builder()
+                .name("namespace2")
+                .cluster("local")
+                .build())
+            .spec(Namespace.NamespaceSpec.builder()
+                .kafkaUser("user")
+                .build())
+            .build();
+
+        when(namespaceService.findByWildcardName("namespace*"))
+            .thenReturn(List.of(namespace1, namespace2));
+        when(namespaceService.findAllResourcesByNamespace(namespace1))
+            .thenReturn(List.of());
+        when(namespaceService.findAllResourcesByNamespace(namespace2))
+            .thenReturn(List.of());
+        when(securityService.username())
+            .thenReturn(Optional.of("test-user"));
+        when(securityService.hasRole(ResourceBasedSecurityRule.IS_ADMIN))
+            .thenReturn(false);
+
+        doNothing().when(applicationEventPublisher).publishEvent(any());
+        var result = namespaceController.bulkDelete("namespace*", false);
+        assertEquals(HttpResponse.noContent().getStatus(), result.getStatus());
+    }
+
+    @Test
+    void shouldDeleteNamespacesInDryRunMode() {
+        Namespace namespace1 = Namespace.builder()
+            .metadata(Metadata.builder()
+                .name("namespace1")
+                .cluster("local")
+                .build())
+            .spec(Namespace.NamespaceSpec.builder()
+                .kafkaUser("user")
+                .build())
+            .build();
+
+        Namespace namespace2 = Namespace.builder()
+            .metadata(Metadata.builder()
+                .name("namespace2")
+                .cluster("local")
+                .build())
+            .spec(Namespace.NamespaceSpec.builder()
+                .kafkaUser("user")
+                .build())
+            .build();
+
+        when(namespaceService.findByWildcardName("namespace*"))
+            .thenReturn(List.of(namespace1, namespace2));
+        when(namespaceService.findAllResourcesByNamespace(namespace1))
+            .thenReturn(List.of());
+        when(namespaceService.findAllResourcesByNamespace(namespace2))
+            .thenReturn(List.of());
+
+        var result = namespaceController.bulkDelete("namespace*", true);
+        verify(namespaceService, never()).delete(any());
+        assertEquals(HttpResponse.noContent().getStatus(), result.getStatus());
+    }
+
+    @Test
+    void shouldNotDeleteNamespacesWhenResourcesAreStillLinkedWithIt() {
+        Namespace namespace1 = Namespace.builder()
+            .metadata(Metadata.builder()
+                .name("namespace1")
+                .cluster("local")
+                .build())
+            .spec(Namespace.NamespaceSpec.builder()
+                .kafkaUser("user")
+                .build())
+            .build();
+
+        Namespace namespace2 = Namespace.builder()
+            .metadata(Metadata.builder()
+                .name("namespace2")
+                .cluster("local")
+                .build())
+            .spec(Namespace.NamespaceSpec.builder()
+                .kafkaUser("user")
+                .build())
+            .build();
+
+        when(namespaceService.findByWildcardName("namespace*"))
+            .thenReturn(List.of(namespace1, namespace2));
+        when(namespaceService.findAllResourcesByNamespace(namespace1))
+            .thenReturn(List.of("Topic/topic1"));
+        when(namespaceService.findAllResourcesByNamespace(namespace2))
+            .thenReturn(List.of());
+
+        assertThrows(ResourceValidationException.class,
+            () -> namespaceController.bulkDelete("namespace*", false));
+        verify(namespaceService, never()).delete(any());
+    }
+
+    @Test
+    void shouldNotDeleteNamespacesWhenPatternMatchesNothing() {
+        when(namespaceService.findByWildcardName("namespace*")).thenReturn(List.of());
+        var result = namespaceController.bulkDelete("namespace*", false);
+        assertEquals(HttpResponse.notFound().getStatus(), result.getStatus());
+    }
 }