fix(#3372): Fix time range selector test

streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java

@@ -97,7 +97,7 @@ private void createDataStreamForAdapter(
     var storedDescription = new SourcesManagement()
         .createAdapterDataStream(adapterDescription, streamId);
     storedDescription.setCorrespondingAdapterId(adapterId);
-    installDataSource(storedDescription, principalSid, true);
+    installDataSource(storedDescription, principalSid);
     LOG.info("Install source (source URL: {} in backend", adapterDescription.getElementId());
   }
 
@@ -141,15 +141,8 @@ public void deleteAdapter(String elementId) throws AdapterException {
     LOG.info("Successfully deleted data stream: " + adapter.getCorrespondingDataStreamElementId());
   }
 
-  public List<AdapterDescription> getAllAdapterInstances() throws AdapterException {
-
-    List<AdapterDescription> allAdapters = adapterInstanceStorage.findAll();
-
-    if (allAdapters == null) {
-      throw new AdapterException("Could not get all adapters");
-    }
-
-    return allAdapters;
+  public List<AdapterDescription> getAllAdapterInstances() {
+    return adapterInstanceStorage.findAll();
   }
 
   public void stopStreamAdapter(String elementId) throws AdapterException {
@@ -198,11 +191,10 @@ public void startStreamAdapter(String elementId) throws AdapterException {
 
   private void installDataSource(
       SpDataStream stream,
-      String principalSid,
-      boolean publicElement
+      String principalSid
   ) throws AdapterException {
     try {
-      new DataStreamVerifier(stream).verifyAndAdd(principalSid, publicElement);
+      new DataStreamVerifier(stream).verifyAndAdd(principalSid, false);
     } catch (SepaParseException e) {
       LOG.error("Error while installing data source: {}", stream.getElementId(), e);
       throw new AdapterException();

streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterMasterManagementTest.java

@@ -71,7 +71,7 @@ public void getAdapter_Fail() {
   }
 
   @Test
-  public void getAllAdapters_Success() throws AdapterException {
+  public void getAllAdapters_Success() {
     var adapterDescriptions = List.of(new AdapterDescription());
     var adapterStorage = mock(AdapterInstanceStorageImpl.class);
     var resourceManager = mock(AdapterResourceManager.class);
@@ -90,21 +90,4 @@ public void getAllAdapters_Success() throws AdapterException {
     Assertions.assertEquals(1, result.size());
   }
 
-  @Test
-  public void getAllAdapters_Fail() {
-    var adapterStorage = mock(AdapterInstanceStorageImpl.class);
-    var resourceManager = mock(AdapterResourceManager.class);
-    when(adapterStorage.findAll()).thenReturn(null);
-
-    var adapterMasterManagement =
-        new AdapterMasterManagement(
-            adapterStorage,
-            resourceManager,
-            null,
-            AdapterMetricsManager.INSTANCE.getAdapterMetrics()
-        );
-
-    assertThrows(AdapterException.class, adapterMasterManagement::getAllAdapterInstances);
-
-  }
 }

streampipes-rest/src/main/java/org/apache/streampipes/rest/ResetManagement.java

@@ -108,24 +108,20 @@ private static void stopAndDeleteAllPipelines() {
   private static void stopAndDeleteAllAdapters() {
     AdapterMasterManagement adapterMasterManagement = new AdapterMasterManagement(
         StorageDispatcher.INSTANCE.getNoSqlStore()
-            .getAdapterInstanceStorage(),
+                                  .getAdapterInstanceStorage(),
         new SpResourceManager().manageAdapters(),
         new SpResourceManager().manageDataStreams(),
         AdapterMetricsManager.INSTANCE.getAdapterMetrics()
     );
 
-    try {
-      List<AdapterDescription> allAdapters = adapterMasterManagement.getAllAdapterInstances();
-      allAdapters.forEach(adapterDescription -> {
-        try {
-          adapterMasterManagement.deleteAdapter(adapterDescription.getElementId());
-        } catch (AdapterException e) {
-          logger.error("Failed to delete adapter with id: " + adapterDescription.getElementId(), e);
-        }
-      });
-    } catch (AdapterException e) {
-      logger.error("Failed to load all adapter descriptions", e);
-    }
+    List<AdapterDescription> allAdapters = adapterMasterManagement.getAllAdapterInstances();
+    allAdapters.forEach(adapterDescription -> {
+      try {
+        adapterMasterManagement.deleteAdapter(adapterDescription.getElementId());
+      } catch (AdapterException e) {
+        logger.error("Failed to delete adapter with id: " + adapterDescription.getElementId(), e);
+      }
+    });
   }
 
   private static void deleteAllFiles() {
@@ -154,37 +150,37 @@ private static void removeAllDataInDataLake() {
   private static void removeAllDataViewWidgets() {
     var widgetStorage =
         StorageDispatcher.INSTANCE.getNoSqlStore()
-            .getDataExplorerWidgetStorage();
+                                  .getDataExplorerWidgetStorage();
     widgetStorage.findAll()
-        .forEach(widget -> widgetStorage.deleteElementById(widget.getElementId()));
+                 .forEach(widget -> widgetStorage.deleteElementById(widget.getElementId()));
   }
 
   private static void removeAllDataViews() {
     var dataLakeDashboardStorage =
         StorageDispatcher.INSTANCE.getNoSqlStore()
-            .getDataExplorerDashboardStorage();
+                                  .getDataExplorerDashboardStorage();
     dataLakeDashboardStorage.findAll()
-        .forEach(dashboard -> dataLakeDashboardStorage.deleteElementById(dashboard.getElementId()));
+                            .forEach(dashboard -> dataLakeDashboardStorage.deleteElementById(dashboard.getElementId()));
   }
 
   private static void removeAllDashboardWidgets() {
     var dashboardWidgetStorage =
         StorageDispatcher.INSTANCE.getNoSqlStore()
-            .getDashboardWidgetStorage();
+                                  .getDashboardWidgetStorage();
     dashboardWidgetStorage.findAll()
-        .forEach(widget -> dashboardWidgetStorage.deleteElementById(widget.getElementId()));
+                          .forEach(widget -> dashboardWidgetStorage.deleteElementById(widget.getElementId()));
   }
 
   private static void removeAllDashboards() {
     var dashboardStorage = StorageDispatcher.INSTANCE.getNoSqlStore()
-        .getDashboardStorage();
+                                                     .getDashboardStorage();
     dashboardStorage.findAll()
-        .forEach(dashboard -> dashboardStorage.deleteElementById(dashboard.getElementId()));
+                    .forEach(dashboard -> dashboardStorage.deleteElementById(dashboard.getElementId()));
   }
 
   private static void removeAllAssets(String username) {
     IGenericStorage genericStorage = StorageDispatcher.INSTANCE.getNoSqlStore()
-        .getGenericStorage();
+                                                               .getGenericStorage();
     try {
       for (Map<String, Object> asset : genericStorage.findAll("asset-management")) {
         genericStorage.delete((String) asset.get("_id"), (String) asset.get("_rev"));
@@ -211,13 +207,19 @@ private static void clearGenericStorage() {
         "asset-management",
         "asset-sites"
     );
-    var genericStorage = StorageDispatcher.INSTANCE.getNoSqlStore().getGenericStorage();
+    var genericStorage = StorageDispatcher.INSTANCE.getNoSqlStore()
+                                                   .getGenericStorage();
 
     appDocTypesToDelete.forEach(docType -> {
       try {
         var allDocs = genericStorage.findAll(docType);
         for (var doc : allDocs) {
-          genericStorage.delete(doc.get("_id").toString(), doc.get("_rev").toString());
+          genericStorage.delete(
+              doc.get("_id")
+                 .toString(),
+              doc.get("_rev")
+                 .toString()
+          );
         }
       } catch (IOException e) {
         throw new RuntimeException(e);

streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/AdapterMonitoringResource.java

@@ -21,11 +21,13 @@
 
 import org.apache.streampipes.manager.monitoring.pipeline.ExtensionsLogProvider;
 import org.apache.streampipes.manager.monitoring.pipeline.ExtensionsServiceLogExecutor;
+import org.apache.streampipes.model.client.user.DefaultPrivilege;
 import org.apache.streampipes.model.monitoring.SpLogEntry;
 import org.apache.streampipes.model.monitoring.SpMetricsEntry;
 
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
+import org.springframework.security.access.prepost.PreAuthorize;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -39,21 +41,51 @@
 @RequestMapping("/api/v2/adapter-monitoring")
 public class AdapterMonitoringResource extends AbstractMonitoringResource {
 
-  @GetMapping(path = "adapter/{elementId}/logs", produces = MediaType.APPLICATION_JSON_VALUE)
-  public ResponseEntity<List<SpLogEntry>> getLogInfoForAdapter(@PathVariable("elementId") String elementId) {
+  @GetMapping(
+      path = "adapter/{elementId}/logs",
+      produces = MediaType.APPLICATION_JSON_VALUE
+  )
+  @PreAuthorize("this.hasReadAuthority() and hasPermission('#elementId', 'READ')")
+  public ResponseEntity<List<SpLogEntry>> getLogInfoForAdapter(
+      @PathVariable("elementId") String elementId
+  ) {
     return ok(ExtensionsLogProvider.INSTANCE.getLogInfosForResource(elementId));
   }
 
-  @GetMapping(path = "adapter/{elementId}/metrics", produces = MediaType.APPLICATION_JSON_VALUE)
-  public ResponseEntity<SpMetricsEntry> getMetricsInfoForAdapter(@PathVariable("elementId") String elementId) {
+  @GetMapping(
+      path = "adapter/{elementId}/metrics",
+      produces = MediaType.APPLICATION_JSON_VALUE
+  )
+  @PreAuthorize("this.hasReadAuthority() and hasPermission('#elementId', 'READ')")
+  public ResponseEntity<SpMetricsEntry> getMetricsInfoForAdapter(
+      @PathVariable("elementId") String elementId
+  ) {
     return ok(ExtensionsLogProvider.INSTANCE.getMetricInfosForResource(elementId));
   }
 
-  @GetMapping(path = "metrics", produces = MediaType.APPLICATION_JSON_VALUE)
+  @GetMapping(
+      path = "metrics",
+      produces = MediaType.APPLICATION_JSON_VALUE
+  )
+  @PreAuthorize("this.hasReadAuthority() and hasPermission('#elementId', 'READ')")
   public ResponseEntity<Map<String, SpMetricsEntry>> getMetricsInfos(
       @RequestParam(value = "filter") List<String> elementIds
   ) {
     new ExtensionsServiceLogExecutor().triggerUpdate();
     return ok(ExtensionsLogProvider.INSTANCE.getMetricsInfoForResources(elementIds));
   }
+
+  /**
+   * required by Spring expression
+   */
+  public boolean hasReadAuthority() {
+    return isAdminOrHasAnyAuthority(DefaultPrivilege.Constants.PRIVILEGE_READ_ADAPTER_VALUE);
+  }
+
+  /**
+   * required by Spring expression
+   */
+  public boolean hasWriteAuthority() {
+    return isAdminOrHasAnyAuthority(DefaultPrivilege.Constants.PRIVILEGE_WRITE_ADAPTER_VALUE);
+  }
 }

streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineMonitoring.java

@@ -19,11 +19,13 @@
 
 import org.apache.streampipes.manager.monitoring.pipeline.ExtensionsLogProvider;
 import org.apache.streampipes.manager.monitoring.pipeline.ExtensionsServiceLogExecutor;
+import org.apache.streampipes.model.client.user.DefaultPrivilege;
 import org.apache.streampipes.model.monitoring.SpLogEntry;
 import org.apache.streampipes.model.monitoring.SpMetricsEntry;
 
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
+import org.springframework.security.access.prepost.PreAuthorize;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -37,20 +39,43 @@
 @RequestMapping("/api/v2/pipeline-monitoring")
 public class PipelineMonitoring extends AbstractMonitoringResource {
 
-  @GetMapping(value = "/pipeline/{pipelineId}/logs", produces = MediaType.APPLICATION_JSON_VALUE)
+  @GetMapping(
+      value = "/pipeline/{pipelineId}/logs",
+      produces = MediaType.APPLICATION_JSON_VALUE
+  )
+  @PreAuthorize("this.hasReadAuthority() and hasPermission('#pipelineId', 'READ')")
   public ResponseEntity<Map<String, List<SpLogEntry>>> getLogInfoForPipeline(
-      @PathVariable("pipelineId") String pipelineId) {
+      @PathVariable("pipelineId") String pipelineId
+  ) {
     return ok(ExtensionsLogProvider.INSTANCE.getLogInfosForPipeline(pipelineId));
   }
 
-  @GetMapping(value = "/pipeline/{pipelineId}/metrics", produces = MediaType.APPLICATION_JSON_VALUE)
+  @GetMapping(
+      value = "/pipeline/{pipelineId}/metrics",
+      produces = MediaType.APPLICATION_JSON_VALUE
+  )
+  @PreAuthorize("this.hasReadAuthority() and hasPermission('#pipelineId', 'READ')")
   public ResponseEntity<Map<String, SpMetricsEntry>> getMetricsInfoForPipeline(
       @PathVariable("pipelineId") String pipelineId,
-      @RequestParam(value = "forceUpdate", required = false, defaultValue = "false") boolean forceUpdate) {
+      @RequestParam(value = "forceUpdate", required = false, defaultValue = "false") boolean forceUpdate
+  ) {
     if (forceUpdate) {
       new ExtensionsServiceLogExecutor().triggerUpdate();
     }
     return ok(ExtensionsLogProvider.INSTANCE.getMetricInfosForPipeline(pipelineId));
   }
 
+  /**
+   * required by Spring expression
+   */
+  public boolean hasReadAuthority() {
+    return isAdminOrHasAnyAuthority(DefaultPrivilege.Constants.PRIVILEGE_READ_PIPELINE_VALUE);
+  }
+
+  /**
+   * required by Spring expression
+   */
+  public boolean hasWriteAuthority() {
+    return isAdminOrHasAnyAuthority(DefaultPrivilege.Constants.PRIVILEGE_WRITE_PIPELINE_VALUE);
+  }
 }