This is an automated email from the ASF dual-hosted git repository.
dominikriemer pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 09e55cec33 fix(#5193): Missing icons and documentation during
extension migration (#5197)
09e55cec33 is described below
commit 09e55cec33698b8f9245e2d7efe463f39468b0bc
Author: Sven Oehler <[email protected]>
AuthorDate: Tue Nov 3 16:16:28 2026 +0100
fix(#5193): Missing icons and documentation during extension migration
(#5197)
---
.../management/AdapterMasterManagement.java | 3 +-
.../management/WorkerAdministrationManagement.java | 6 +-
.../streampipes/manager/assets/AssetFetcher.java | 13 +-
.../streampipes/manager/assets/AssetManager.java | 16 +--
.../manager/extensions/ExtensionItemInstaller.java | 4 +-
.../migration/AbstractMigrationManager.java | 6 +-
.../manager/verification/ElementVerifier.java | 26 ++--
.../manager/verification/TypedElementVerifier.java | 13 +-
.../migration/MigrationAssetUpdateTest.java | 132 +++++++++++++++++++++
.../manager/verification/TypeExtractorTest.java | 38 +++++-
10 files changed, 212 insertions(+), 45 deletions(-)
diff --git
a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java
b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java
index 74b30550ef..e433254851 100644
---
a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java
+++
b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java
@@ -279,6 +279,7 @@ public class AdapterMasterManagement {
requestManager,
resourceManager
);
- return verifier.verifyAndAdd(principalSid, false).isSuccess();
+ // Adapter-created streams are local descriptions without a selected
extension service.
+ return verifier.verifyAndAdd(principalSid, false, null).isSuccess();
}
}
diff --git
a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerAdministrationManagement.java
b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerAdministrationManagement.java
index 1d50d8f8f7..f5d95980ff 100644
---
a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerAdministrationManagement.java
+++
b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerAdministrationManagement.java
@@ -21,6 +21,7 @@ package org.apache.streampipes.connect.management.management;
import
org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
import org.apache.streampipes.manager.assets.AssetManager;
+import
org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointGenerator;
import org.apache.streampipes.model.connect.adapter.AdapterDescription;
import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag;
import org.apache.streampipes.resource.management.PermissionResourceManager;
@@ -36,6 +37,7 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.List;
+import java.util.Set;
public class WorkerAdministrationManagement {
@@ -71,7 +73,9 @@ public class WorkerAdministrationManagement {
if (!assetManager.existsAssetDir(adapter.getAppId())) {
try {
LOG.info("Updating assets for adapter {}", adapter.getAppId());
- assetManager.storeAsset(SpServiceUrlProvider.ADAPTER,
adapter.getAppId(), requestManager, serviceDiscovery);
+ var service = new
ExtensionsServiceEndpointGenerator(serviceDiscovery)
+ .selectService(adapter.getAppId(),
SpServiceUrlProvider.ADAPTER, Set.of());
+ assetManager.storeAsset(SpServiceUrlProvider.ADAPTER,
adapter.getAppId(), requestManager, service);
} catch (IOException | NoServiceEndpointsAvailableException e) {
LOG.error(
"Could not fetch asset for adapter {}, please try to
manually update this adapter.",
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetFetcher.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetFetcher.java
index 08e28c0ede..51cab4e187 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetFetcher.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetFetcher.java
@@ -17,38 +17,31 @@
*/
package org.apache.streampipes.manager.assets;
-import
org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestTargets;
import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequests;
-import
org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointGenerator;
-import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery;
+import
org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
-import java.util.Set;
public class AssetFetcher {
- private final ISpServiceDiscovery serviceDiscovery;
private final SpServiceUrlProvider spServiceUrlProvider;
private final String appId;
private final ExtensionServiceRequestManager requestManager;
- public AssetFetcher(ISpServiceDiscovery serviceDiscovery,
- SpServiceUrlProvider spServiceUrlProvider,
+ public AssetFetcher(SpServiceUrlProvider spServiceUrlProvider,
String appId,
ExtensionServiceRequestManager requestManager) {
- this.serviceDiscovery = serviceDiscovery;
this.spServiceUrlProvider = spServiceUrlProvider;
this.appId = appId;
this.requestManager = requestManager;
}
- public InputStream fetchPipelineElementAssets() throws IOException,
NoServiceEndpointsAvailableException {
- var service = new
ExtensionsServiceEndpointGenerator(serviceDiscovery).selectService(appId,
spServiceUrlProvider, Set.of());
+ public InputStream fetchPipelineElementAssets(SpServiceRegistration service)
throws IOException {
var requestTarget =
ExtensionServiceRequestTargets.pipelineElementAssets(service,
spServiceUrlProvider, appId);
var response =
requestManager.request(ExtensionServiceRequests.pipelineElementAssets(requestTarget));
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetManager.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetManager.java
index 525aa67a73..ca00c77252 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetManager.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetManager.java
@@ -18,10 +18,9 @@
package org.apache.streampipes.manager.assets;
import org.apache.streampipes.commons.constants.GlobalStreamPipesConstants;
-import
org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
+import
org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
import org.apache.streampipes.storage.api.system.ISpCoreConfigurationStorage;
-import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery;
import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
import org.apache.commons.io.FileUtils;
@@ -59,12 +58,13 @@ public class AssetManager {
}
public void storeAsset(SpServiceUrlProvider spServiceUrlProvider,
- String appId,
- ExtensionServiceRequestManager requestManager,
- ISpServiceDiscovery serviceDiscovery) throws
IOException, NoServiceEndpointsAvailableException {
- InputStream assetStream = new AssetFetcher(serviceDiscovery,
spServiceUrlProvider, appId, requestManager)
- .fetchPipelineElementAssets();
- new AssetExtractor(assetStream, appId,
coreConfigurationStorage).extractAssetContents();
+ String appId,
+ ExtensionServiceRequestManager requestManager,
+ SpServiceRegistration service) throws IOException {
+ var fetcher = new AssetFetcher(spServiceUrlProvider, appId,
requestManager);
+ try (InputStream assetStream =
fetcher.fetchPipelineElementAssets(service)) {
+ new AssetExtractor(assetStream, appId,
coreConfigurationStorage).extractAssetContents();
+ }
}
public void deleteAsset(String appId) throws IOException {
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/extensions/ExtensionItemInstaller.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/extensions/ExtensionItemInstaller.java
index 4671a8cae7..a9b29dfb70 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/extensions/ExtensionItemInstaller.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/extensions/ExtensionItemInstaller.java
@@ -54,14 +54,14 @@ public class ExtensionItemInstaller {
var requestTarget = getDescriptionRequestTarget(req);
var description = fetchDescription(requestTarget);
return new TypeExtractor(description, descriptionStorage, requestManager,
- resourceManager).getTypeVerifier().verifyAndAdd(principalSid,
req.publicElement());
+ resourceManager).getTypeVerifier().verifyAndAdd(principalSid,
req.publicElement(), service);
}
public Message updateExtension(ExtensionItemInstallationRequest req) throws
IOException, SepaParseException {
var requestTarget = getDescriptionRequestTarget(req);
var description = fetchDescription(requestTarget);
return new TypeExtractor(description, descriptionStorage, requestManager,
- resourceManager).getTypeVerifier().verifyAndUpdate();
+ resourceManager).getTypeVerifier().verifyAndUpdate(service);
}
private ExtensionServiceRequestTarget
getDescriptionRequestTarget(ExtensionItemInstallationRequest req) {
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/AbstractMigrationManager.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/AbstractMigrationManager.java
index c933c3aefd..5afbdbaf4f 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/AbstractMigrationManager.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/AbstractMigrationManager.java
@@ -148,21 +148,21 @@ public abstract class AbstractMigrationManager {
.forEach(config -> {
if (isInstalled(config.modelType(), config.targetAppId())) {
var requestTarget = getRequestTarget(config.modelType(),
config.targetAppId(), service);
- performUpdate(requestTarget);
+ performUpdate(requestTarget, service);
}
});
}
protected abstract boolean isInstalled(SpServiceTagPrefix modelType, String
appId);
- protected void performUpdate(ExtensionServiceRequestTarget requestTarget) {
+ protected void performUpdate(ExtensionServiceRequestTarget requestTarget,
SpServiceRegistration service) {
try {
var entityPayload = requestManager
.request(ExtensionServiceRequests.descriptionUpdate(requestTarget,
resourceManager))
.responseBody();
var updateResult = new TypeExtractor(entityPayload, descriptionStorage,
requestManager, resourceManager)
- .getTypeVerifier().verifyAndUpdate();
+ .getTypeVerifier().verifyAndUpdate(service);
if (!updateResult.isSuccess()) {
LOG.error(
"Updating the pipeline element description failed: {}",
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/ElementVerifier.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/ElementVerifier.java
index b18fdb3c84..50db874da2 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/ElementVerifier.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/ElementVerifier.java
@@ -23,6 +23,7 @@ import org.apache.streampipes.manager.assets.AssetManager;
import org.apache.streampipes.model.base.NamedStreamPipesEntity;
import org.apache.streampipes.model.client.user.Permission;
import org.apache.streampipes.model.client.user.PermissionBuilder;
+import
org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
import org.apache.streampipes.model.message.ErrorMessage;
import org.apache.streampipes.model.message.Message;
import org.apache.streampipes.model.message.Notification;
@@ -86,7 +87,7 @@ public abstract class ElementVerifier<T extends
NamedStreamPipesEntity> {
protected abstract void update();
- public Message verifyAndAdd(String principalSid, boolean publicElement) {
+ public Message verifyAndAdd(String principalSid, boolean publicElement,
SpServiceRegistration service) {
var transformError = transformEntity();
if (transformError != null) {
return transformError;
@@ -96,9 +97,10 @@ public abstract class ElementVerifier<T extends
NamedStreamPipesEntity> {
if (state == StorageState.STORED) {
createAndStorePermission(principalSid, publicElement);
try {
- storeAssets();
+ storeAssets(service);
} catch (IOException | NoServiceEndpointsAvailableException e) {
LOG.error("Could not store assets for app id '{}'",
elementDescription.getAppId(), e);
+ return assetErrorMessage();
}
return successMessage();
} else {
@@ -107,7 +109,7 @@ public abstract class ElementVerifier<T extends
NamedStreamPipesEntity> {
}
- public Message verifyAndUpdate() {
+ public Message verifyAndUpdate(SpServiceRegistration service) {
var transformError = transformEntity();
if (transformError != null) {
return transformError;
@@ -115,23 +117,31 @@ public abstract class ElementVerifier<T extends
NamedStreamPipesEntity> {
update();
try {
- updateAssets();
+ updateAssets(service);
} catch (IOException | NoServiceEndpointsAvailableException e) {
LOG.error("Could not update assets for app id '{}'",
elementDescription.getAppId(), e);
+ return assetErrorMessage();
}
return successMessage();
}
- protected abstract void storeAssets() throws IOException,
NoServiceEndpointsAvailableException;
+ protected abstract void storeAssets(SpServiceRegistration service)
+ throws IOException, NoServiceEndpointsAvailableException;
- protected void updateAssets() throws IOException,
NoServiceEndpointsAvailableException {
+ protected void updateAssets(SpServiceRegistration service) throws
IOException, NoServiceEndpointsAvailableException {
if (elementDescription.isIncludesAssets()) {
- assetManager.deleteAsset(elementDescription.getAppId());
- storeAssets();
+ storeAssets(service);
}
}
+ private Message assetErrorMessage() {
+ return new ErrorMessage(elementDescription.getName(), List.of(
+ NotificationType.STORAGE_SUCCESS.uiNotification(),
+ new Notification("Asset update failed", "The element description was
stored, but its assets could not be updated. "
+ + "Please retry the update.")));
+ }
+
private Message successMessage() {
List<Notification> notifications = new ArrayList<>();
notifications.add(NotificationType.STORAGE_SUCCESS.uiNotification());
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/TypedElementVerifier.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/TypedElementVerifier.java
index b8bb1ac341..5f1a12dae7 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/TypedElementVerifier.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/TypedElementVerifier.java
@@ -18,12 +18,11 @@
package org.apache.streampipes.manager.verification;
-import
org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import
org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
import org.apache.streampipes.resource.management.SpResourceManager;
import
org.apache.streampipes.storage.api.pipeline.IPipelineElementDescriptionStorage;
-import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery;
import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
import java.io.IOException;
@@ -37,7 +36,6 @@ public class TypedElementVerifier<T extends
NamedStreamPipesEntity> extends Elem
private final Consumer<T> updateOperation;
private final SpServiceUrlProvider serviceUrlProvider;
private final ExtensionServiceRequestManager requestManager;
- private final ISpServiceDiscovery serviceDiscovery;
public TypedElementVerifier(
String graphData,
@@ -56,7 +54,6 @@ public class TypedElementVerifier<T extends
NamedStreamPipesEntity> extends Elem
this.updateOperation = updateOperation;
this.serviceUrlProvider = serviceUrlProvider;
this.requestManager = requestManager;
- this.serviceDiscovery = resourceManager.getServiceDiscovery();
}
public TypedElementVerifier(
@@ -75,7 +72,6 @@ public class TypedElementVerifier<T extends
NamedStreamPipesEntity> extends Elem
this.updateOperation = updateOperation;
this.serviceUrlProvider = serviceUrlProvider;
this.requestManager = requestManager;
- this.serviceDiscovery = resourceManager.getServiceDiscovery();
}
@Override
@@ -93,9 +89,12 @@ public class TypedElementVerifier<T extends
NamedStreamPipesEntity> extends Elem
}
@Override
- protected void storeAssets() throws IOException,
NoServiceEndpointsAvailableException {
+ protected void storeAssets(SpServiceRegistration service) throws IOException
{
if (elementDescription.isIncludesAssets()) {
- assetManager.storeAsset(serviceUrlProvider,
elementDescription.getAppId(), requestManager, serviceDiscovery);
+ if (service == null) {
+ throw new IOException("No source service provided for assets of " +
elementDescription.getAppId());
+ }
+ assetManager.storeAsset(serviceUrlProvider,
elementDescription.getAppId(), requestManager, service);
}
}
}
diff --git
a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/migration/MigrationAssetUpdateTest.java
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/migration/MigrationAssetUpdateTest.java
new file mode 100644
index 0000000000..fe7fb4ca13
--- /dev/null
+++
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/migration/MigrationAssetUpdateTest.java
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.streampipes.manager.migration;
+
+import
org.apache.streampipes.manager.api.extensions.ExtensionServiceOperationResult;
+import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
+import org.apache.streampipes.manager.util.AuthTokenProvider;
+import org.apache.streampipes.model.client.user.Principal;
+import org.apache.streampipes.model.configuration.SpCoreConfiguration;
+import
org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix;
+import
org.apache.streampipes.model.extensions.transport.ExtensionServiceBrokerOperations;
+import org.apache.streampipes.model.graph.DataSinkDescription;
+import org.apache.streampipes.model.migration.ModelMigratorConfig;
+import org.apache.streampipes.resource.management.SpResourceManager;
+import org.apache.streampipes.resource.management.UserResourceManager;
+import
org.apache.streampipes.storage.api.pipeline.IPipelineElementDescriptionStorage;
+import org.apache.streampipes.storage.api.system.ISpCoreConfigurationStorage;
+import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipOutputStream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+class MigrationAssetUpdateTest {
+
+ private static final String APP_ID =
"org.apache.streampipes.sinks.internal.jvm.dataset";
+
+ @TempDir
+ Path directory;
+
+ private SpResourceManager resources;
+ private IPipelineElementDescriptionStorage descriptions;
+ private ISpServiceDiscovery discovery;
+ private SpServiceRegistration service;
+
+ @BeforeEach
+ void setUp() {
+ resources = mock(SpResourceManager.class);
+ descriptions = mock(IPipelineElementDescriptionStorage.class);
+ discovery = mock(ISpServiceDiscovery.class);
+ when(resources.getServiceDiscovery()).thenReturn(discovery);
+ var configurationStorage = mock(ISpCoreConfigurationStorage.class);
+ var configuration = new SpCoreConfiguration();
+ configuration.setAssetDir(directory.toString());
+ when(configurationStorage.get()).thenReturn(configuration);
+
when(resources.getCoreConfigurationStorage()).thenReturn(configurationStorage);
+ service = new SpServiceRegistration();
+ service.setSvcId("streampipes-current-instance");
+ service.setHost("localhost");
+ service.setPort(8090);
+ }
+
+ @Test
+ void migrationFetchesAssetsFromTheServiceProvidingTheDescription() throws
Exception {
+ var archive = assetArchive();
+ ExtensionServiceRequestManager requests = request -> {
+ assertEquals(service.getSvcId(), request.target().serviceId());
+ if
(request.target().operation().equals(ExtensionServiceBrokerOperations.PIPELINE_ELEMENT_ASSETS.operationId()))
{
+ return new ExtensionServiceOperationResult(200, archive);
+ }
+ return new ExtensionServiceOperationResult(200,
description().getBytes(StandardCharsets.UTF_8));
+ };
+ var manager = new AbstractMigrationManager(requests, resources,
descriptions) {
+ @Override
+ protected boolean isInstalled(SpServiceTagPrefix modelType, String
appId) {
+ return true;
+ }
+ };
+
+ var users = mock(UserResourceManager.class);
+ when(resources.manageUsers()).thenReturn(users);
+ when(users.getServiceAdmin()).thenReturn(mock(Principal.class));
+ try (var tokens = mockConstruction(AuthTokenProvider.class)) {
+ manager.updateDescriptions(List.of(new ModelMigratorConfig(APP_ID,
SpServiceTagPrefix.DATA_SINK, 1, 2)), service);
+ }
+
+ assertEquals("new icon",
Files.readString(directory.resolve(APP_ID).resolve("icon.png")));
+ verify(descriptions).update(any(DataSinkDescription.class));
+ verifyNoInteractions(discovery);
+ }
+
+ private String description() {
+ return "{\"@class\":\"" + DataSinkDescription.class.getCanonicalName()
+ + "\",\"appId\":\"" + APP_ID +
"\",\"elementId\":\"test-element\",\"includesAssets\":true}";
+ }
+
+ private byte[] assetArchive() throws IOException {
+ var bytes = new ByteArrayOutputStream();
+ try (var zip = new ZipOutputStream(bytes)) {
+ zip.putNextEntry(new ZipEntry("icon.png"));
+ zip.write("new icon".getBytes(StandardCharsets.UTF_8));
+ zip.closeEntry();
+ zip.putNextEntry(new ZipEntry("documentation.md"));
+ zip.write("documentation".getBytes(StandardCharsets.UTF_8));
+ zip.closeEntry();
+ }
+ return bytes.toByteArray();
+ }
+}
diff --git
a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/verification/TypeExtractorTest.java
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/verification/TypeExtractorTest.java
index 86d345ce87..4b13bc0774 100644
---
a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/verification/TypeExtractorTest.java
+++
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/verification/TypeExtractorTest.java
@@ -23,6 +23,7 @@ import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestMana
import org.apache.streampipes.manager.verification.extractor.TypeExtractor;
import org.apache.streampipes.model.SpDataStream;
import org.apache.streampipes.model.connect.adapter.AdapterDescription;
+import
org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
import org.apache.streampipes.model.graph.DataProcessorDescription;
import org.apache.streampipes.model.graph.DataSinkDescription;
import org.apache.streampipes.model.message.NotificationType;
@@ -35,15 +36,20 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
public class TypeExtractorTest {
+ private final SpServiceRegistration service = new SpServiceRegistration();
+
private IPipelineElementDescriptionStorage storageApi;
private SpResourceManager resourceManager;
private final ExtensionServiceRequestManager requestManager =
mock(ExtensionServiceRequestManager.class);
@@ -60,7 +66,7 @@ public class TypeExtractorTest {
public void verifyAndUpdateDataStream() throws SepaParseException {
var message = new TypeExtractor(payload(SpDataStream.class), storageApi,
requestManager, resourceManager)
.getTypeVerifier()
- .verifyAndUpdate();
+ .verifyAndUpdate(service);
assertTrue(message.isSuccess());
verify(storageApi).update(any(SpDataStream.class));
@@ -70,7 +76,7 @@ public class TypeExtractorTest {
public void verifyAndUpdateDataProcessor() throws SepaParseException {
var message = new TypeExtractor(payload(DataProcessorDescription.class),
storageApi, requestManager, resourceManager)
.getTypeVerifier()
- .verifyAndUpdate();
+ .verifyAndUpdate(service);
assertTrue(message.isSuccess());
verify(storageApi).update(any(DataProcessorDescription.class));
@@ -80,7 +86,7 @@ public class TypeExtractorTest {
public void verifyAndUpdateDataSink() throws SepaParseException {
var message = new TypeExtractor(payload(DataSinkDescription.class),
storageApi, requestManager, resourceManager)
.getTypeVerifier()
- .verifyAndUpdate();
+ .verifyAndUpdate(service);
assertTrue(message.isSuccess());
verify(storageApi).update(any(DataSinkDescription.class));
@@ -90,7 +96,7 @@ public class TypeExtractorTest {
public void verifyAndUpdateAdapter() throws SepaParseException {
var message = new TypeExtractor(payload(AdapterDescription.class),
storageApi, requestManager, resourceManager)
.getTypeVerifier()
- .verifyAndUpdate();
+ .verifyAndUpdate(service);
assertTrue(message.isSuccess());
verify(storageApi).update(any(AdapterDescription.class));
@@ -100,12 +106,34 @@ public class TypeExtractorTest {
public void
verifyAndUpdateWithoutNameOrIcon_onlyContainsStorageSuccessNotification()
throws SepaParseException {
var message = new TypeExtractor(payload(DataProcessorDescription.class),
storageApi, requestManager, resourceManager)
.getTypeVerifier()
- .verifyAndUpdate();
+ .verifyAndUpdate(service);
assertEquals(1, message.getNotifications().size());
assertEquals(NotificationType.STORAGE_SUCCESS.title(),
message.getNotifications().get(0).getTitle());
}
+ @Test
+ public void updateWithoutAssetsDoesNotRequireAService() throws
SepaParseException {
+ var message = new TypeExtractor(payload(SpDataStream.class), storageApi,
requestManager, resourceManager)
+ .getTypeVerifier().verifyAndUpdate(null);
+
+ assertTrue(message.isSuccess());
+ verifyNoInteractions(requestManager);
+ verify(resourceManager, never()).getServiceDiscovery();
+ }
+
+ @Test
+ public void updateWithAssetsReportsMissingServiceWithoutDiscovery() throws
SepaParseException {
+ var description =
payload(DataSinkDescription.class).replace("\"elementId\"",
"\"includesAssets\":true,\"elementId\"");
+ var message = new TypeExtractor(description, storageApi, requestManager,
resourceManager)
+ .getTypeVerifier().verifyAndUpdate(null);
+
+ assertFalse(message.isSuccess());
+ assertTrue(message.getNotifications().stream().anyMatch(n ->
n.getTitle().equals("Asset update failed")));
+ verifyNoInteractions(requestManager);
+ verify(resourceManager, never()).getServiceDiscovery();
+ }
+
@Test
public void missingClassProperty_throwsSepaParseException() {
assertThrows(
