atiaomar1978-hub commented on code in PR #26010: URL: https://github.com/apache/camel/pull/26010#discussion_r3910978231
########## components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/EventSourceCache.java: ########## @@ -0,0 +1,362 @@ +/* + * 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.camel.component.alibaba.eventbridge; + +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +import com.aliyun.eventbridge.EventBridgeClient; +import com.aliyun.eventbridge.models.EventBusEntry; +import com.aliyun.eventbridge.models.EventRuleDTO; +import com.aliyun.eventbridge.models.ListEventBusesRequest; +import com.aliyun.eventbridge.models.ListEventBusesResponse; +import com.aliyun.eventbridge.models.ListRulesRequest; +import com.aliyun.eventbridge.models.ListRulesResponse; +import com.google.gson.Gson; +import com.google.gson.JsonArray; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import org.apache.camel.component.alibaba.eventbridge.models.AllowedEventBus; +import org.apache.camel.component.alibaba.eventbridge.models.AllowedEventSource; +import org.apache.camel.util.ObjectHelper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * TTL-based in-memory cache for verified Alibaba Cloud EventBridge bus, event source, and source-scoped event type + * definitions. + * <p> + * Implements a two-phase validated cache workflow: + * <ol> + * <li>Fetch and validate configured event sources and event types against Alibaba Cloud API ({@code listEventBuses} and + * {@code listRules}).</li> + * <li><b>Only after validation passes</b>, populate the cache with the verified {@link BusMetadata} for fast runtime + * comparison.</li> + * </ol> + */ +final class EventSourceCache { + + private static final Logger LOG = LoggerFactory.getLogger(EventSourceCache.class); + private static final Gson GSON = new Gson(); + private static final int PAGE_LIMIT = 100; + + /** + * Java 16 record holding cached bus metadata including mapped event sources and their permitted event types. + */ + public record BusMetadata(boolean exists, Map<String, Set<String>> sourceToTypesMap) { + public BusMetadata { + if (sourceToTypesMap == null) { + sourceToTypesMap = Collections.emptyMap(); + } else { + Map<String, Set<String>> unmodifiable = new HashMap<>(); + for (Map.Entry<String, Set<String>> entry : sourceToTypesMap.entrySet()) { + unmodifiable.put(entry.getKey(), Collections.unmodifiableSet(new HashSet<>(entry.getValue()))); + } + sourceToTypesMap = Collections.unmodifiableMap(unmodifiable); + } + } Review Comment: When rule metadata is unavailable, isKnownSource/isKnownType return true. Document this explicitly in the component guide, or tighten behaviour when validateEventSource/validateEventType is enabled. ########## components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/AlibabaEventBridgeUtils.java: ########## @@ -62,117 +74,346 @@ public static EventBridgeClient createClient(AlibabaEventBridgeEndpoint endpoint } public static ClientConfigurations createClientConfigurations(AlibabaEventBridgeEndpoint endpoint, Exchange exchange) { - ClientConfigurations configuration = new ClientConfigurations(); - configuration.setOperation( - OpenApiClientSupport.resolveString(exchange, AlibabaEventBridgeProperties.OPERATION, endpoint.getOperation())); - configuration.setEventBusName( - OpenApiClientSupport.resolveString(exchange, AlibabaEventBridgeProperties.EVENT_BUS_NAME, - endpoint.getEventBusName())); - configuration.setEventSource( + String defaultBusName = OpenApiClientSupport.resolveString( + exchange, AlibabaEventBridgeProperties.EVENT_BUS_NAME, endpoint.getEventBusName()); + + return new ClientConfigurations( + OpenApiClientSupport.resolveString(exchange, AlibabaEventBridgeProperties.OPERATION, endpoint.getOperation()), + defaultBusName, OpenApiClientSupport.resolveString(exchange, AlibabaEventBridgeProperties.EVENT_SOURCE, - endpoint.getEventSource())); - configuration.setEventType( - OpenApiClientSupport.resolveString(exchange, AlibabaEventBridgeProperties.EVENT_TYPE, endpoint.getEventType())); - configuration.setEventSubject( + endpoint.getEventSource()), + OpenApiClientSupport.resolveString(exchange, AlibabaEventBridgeProperties.EVENT_TYPE, endpoint.getEventType()), OpenApiClientSupport.resolveString(exchange, AlibabaEventBridgeProperties.EVENT_SUBJECT, - endpoint.getEventSubject())); - return configuration; + endpoint.getEventSubject()), + OpenApiClientSupport.resolveBoolean(exchange, AlibabaEventBridgeProperties.VALIDATE_EVENT_SOURCE, + endpoint.isValidateEventSource()), + OpenApiClientSupport.resolveBoolean(exchange, AlibabaEventBridgeProperties.VALIDATE_EVENT_TYPE, + endpoint.isValidateEventType()), + OpenApiClientSupport.resolveBoolean(exchange, AlibabaEventBridgeProperties.VALIDATE_EVENT_SPEC, + endpoint.isValidateEventSpec()), + resolveAllowedEventBuses(exchange, AlibabaEventBridgeProperties.ALLOWED_EVENT_SOURCES, + endpoint.getAllowedEventSources(), defaultBusName), + OpenApiClientSupport.resolveLong(exchange, AlibabaEventBridgeProperties.EVENT_SOURCE_CACHE_TTL, + endpoint.getEventSourceCacheTtl())); } public static List<CloudEvent> resolveCloudEvents(Exchange exchange, ClientConfigurations configuration) { + return resolveCloudEvents(exchange, configuration, null, null); + } + + public static List<CloudEvent> resolveCloudEvents( + Exchange exchange, ClientConfigurations configuration, + EventSourceCache eventSourceCache, EventBridgeClient client) { Object body = exchange.getMessage().getBody(); List<CloudEvent> events = new ArrayList<>(); + MapCloudEventValidator mapValidator = new MapCloudEventValidator(eventSourceCache); if (body instanceof List<?> listBody) { for (Object item : listBody) { - events.add(toCloudEvent(item, configuration)); + events.add(toCloudEvent(item, configuration, mapValidator, client)); } return events; } - events.add(toCloudEvent(body, configuration)); + events.add(toCloudEvent(body, configuration, mapValidator, client)); return events; } - private static CloudEvent toCloudEvent(Object body, ClientConfigurations configuration) { + private static CloudEvent toCloudEvent( + Object body, ClientConfigurations configuration, + MapCloudEventValidator mapValidator, EventBridgeClient client) { if (body instanceof CloudEvent cloudEvent) { + mapValidator.validateCloudEvent(cloudEvent, configuration, client); return cloudEvent; } if (body instanceof Map<?, ?> mapBody) { - String eventBusName - = stringValue(mapBody.get(AlibabaEventBridgeConstants.EVENT_BUS_NAME), configuration.getEventBusName()); - String source = stringValue(mapBody.get(AlibabaEventBridgeConstants.EVENT_SOURCE), configuration.getEventSource()); - String type = stringValue(mapBody.get(AlibabaEventBridgeConstants.EVENT_TYPE), configuration.getEventType()); - String subject - = stringValue(mapBody.get(AlibabaEventBridgeConstants.EVENT_SUBJECT), configuration.getEventSubject()); - String data = jsonDataValue(mapBody.get(AlibabaEventBridgeConstants.EVENT_DATA)); - - if (ObjectHelper.isEmpty(source) || ObjectHelper.isEmpty(type) || ObjectHelper.isEmpty(eventBusName)) { - throw new IllegalArgumentException("Event source, type and event bus name are required"); - } - - EventBuilder builder = EventBuilder.builder() - .withSource(URI.create(source)) - .withType(type) - .withAliyunEventBus(eventBusName); - - if (ObjectHelper.isNotEmpty(subject)) { - builder.withSubject(subject); - } - if (data != null) { - builder.withJsonStringData(data); - } - return builder.build(); + return mapValidator.validateAndBuild(mapBody, configuration, client); } if (body instanceof String stringBody) { - if (ObjectHelper.isEmpty(configuration.getEventSource()) - || ObjectHelper.isEmpty(configuration.getEventType()) - || ObjectHelper.isEmpty(configuration.getEventBusName())) { + if (ObjectHelper.isEmpty(configuration.eventSource()) + || ObjectHelper.isEmpty(configuration.eventType()) + || ObjectHelper.isEmpty(configuration.eventBusName())) { throw new IllegalArgumentException("Event source, type and event bus name are required when body is a string"); } + mapValidator.validateBusSourceAndType( + configuration.eventBusName(), configuration.eventSource(), configuration.eventType(), + configuration, client); + EventBuilder builder = EventBuilder.builder() Review Comment: Security note: ALLOWED_EVENT_SOURCES can be overridden via header/property. If routes accept untrusted ingress, this can replace the endpoint whitelist per message. Add untrusted-ingress guidance to strip CamelAlibabaEventBridge* headers before the producer when URI options are policy. ########## components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/AlibabaEventBridgeEndpoint.java: ########## @@ -74,12 +74,38 @@ public class AlibabaEventBridgeEndpoint extends DefaultEndpoint { @UriParam(description = "Default event subject", displayName = "Event Subject") Review Comment: Missing upgrade guide: validateEventSpec defaults to true for map payloads. Routes with non-1.0 specversion or blank id will start failing. Add a camel-alibaba-eventbridge section to the 4.23 upgrade guide. ########## components/camel-alibaba/camel-alibaba-eventbridge/src/main/docs/alibaba-eventbridge-component.adoc: ########## @@ -44,56 +44,243 @@ include::partial$component-endpoint-headers.adoc[] == Usage -=== Message headers evaluated by the EventBridge producer +=== Message Headers and Properties -[width="100%",cols="10%,10%,80%",options="header",] +The component defines constants in package `org.apache.camel.component.alibaba.eventbridge.constants`: + +* `AlibabaEventBridgeHeaders` defines all message header constants evaluated or set by the EventBridge producer. +* `AlibabaEventBridgeProperties` extends `AlibabaEventBridgeHeaders` and provides constants for Exchange properties (such as `OPERATION`), while inheriting all header constants. + +==== Message headers evaluated by the EventBridge producer + +[width="100%",cols="25%,25%,10%,40%",options="header",] |======================================================================= -|Header |Type |Description +|Constant (`AlibabaEventBridgeHeaders`) |Header String |Type |Description + +|`EVENT_BUS_NAME` |`CamelAlibabaEventBridgeEventBusName` |`String` | Event bus name to publish events to (overrides the endpoint option). + +|`EVENT_SOURCE` |`CamelAlibabaEventBridgeEventSource` |`String` | Event source URI (overrides the endpoint option). -|`CamelAlibabaEventBridgeOperation` |`String` | Name of operation to invoke +|`EVENT_TYPE` |`CamelAlibabaEventBridgeEventType` |`String` | Event type (overrides the endpoint option). -|`CamelAlibabaEventBridgeEventBusName` |`String` | Event bus name to publish events to (overrides the endpoint option) +|`EVENT_SUBJECT` |`CamelAlibabaEventBridgeEventSubject` |`String` | Event subject (overrides the endpoint option). -|`CamelAlibabaEventBridgeEventSource` |`String` | Event source URI (overrides the endpoint option) +|`VALIDATE_EVENT_SOURCE` |`CamelAlibabaEventBridgeValidateEventSource` |`boolean` | Whether to validate the event bus and event source against Alibaba Cloud. -|`CamelAlibabaEventBridgeEventType` |`String` | Event type (overrides the endpoint option) +|`VALIDATE_EVENT_TYPE` |`CamelAlibabaEventBridgeValidateEventType` |`boolean` | Whether to validate the event type against Alibaba Cloud rule filter patterns for that source. -|`CamelAlibabaEventBridgeEventSubject` |`String` | Event subject (overrides the endpoint option) +|`VALIDATE_EVENT_SPEC` |`CamelAlibabaEventBridgeValidateEventSpec` |`boolean` | Whether to enforce CloudEvents 1.0 specification compliance checks (defaults to `true`). + +|`ALLOWED_EVENT_SOURCES` |`CamelAlibabaEventBridgeAllowedEventSources` |`Object` | Allowed event sources and source-scoped event types per bus (DSL string, JSON string, Map, or List). + +|`EVENT_SOURCE_CACHE_TTL` |`CamelAlibabaEventBridgeEventSourceCacheTtl` |`long` | Cache time-to-live in milliseconds for verified Alibaba Cloud event bus and type metadata (defaults to `300000` / 5 minutes). |======================================================================= -If any of the above headers are set, they will override their corresponding query parameter value. +If any of the above headers are set, they will override their corresponding exchange property or query parameter value. -=== Message headers set by the EventBridge producer +==== Message headers set by the EventBridge producer -[width="100%",cols="10%,10%,80%",options="header",] +[width="100%",cols="25%,25%,10%,40%",options="header",] +|======================================================================= +|Constant (`AlibabaEventBridgeHeaders`) |Header String |Type |Description + +|`REQUEST_ID` |`CamelAlibabaEventBridgeRequestId` |`String` | Alibaba Cloud request ID returned by EventBridge. + +|======================================================================= Review Comment: Docs CI failure: website build reports AsciiDoc list item index warnings at lines 140-145 (nested ordered list under item 1). Use list continuations (+) or restructure the numbered sub-list so the site build passes. ########## components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/EventSourceCache.java: ########## @@ -0,0 +1,362 @@ +/* + * 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.camel.component.alibaba.eventbridge; + +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +import com.aliyun.eventbridge.EventBridgeClient; +import com.aliyun.eventbridge.models.EventBusEntry; +import com.aliyun.eventbridge.models.EventRuleDTO; +import com.aliyun.eventbridge.models.ListEventBusesRequest; +import com.aliyun.eventbridge.models.ListEventBusesResponse; +import com.aliyun.eventbridge.models.ListRulesRequest; +import com.aliyun.eventbridge.models.ListRulesResponse; +import com.google.gson.Gson; +import com.google.gson.JsonArray; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import org.apache.camel.component.alibaba.eventbridge.models.AllowedEventBus; +import org.apache.camel.component.alibaba.eventbridge.models.AllowedEventSource; +import org.apache.camel.util.ObjectHelper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * TTL-based in-memory cache for verified Alibaba Cloud EventBridge bus, event source, and source-scoped event type + * definitions. + * <p> + * Implements a two-phase validated cache workflow: + * <ol> + * <li>Fetch and validate configured event sources and event types against Alibaba Cloud API ({@code listEventBuses} and + * {@code listRules}).</li> + * <li><b>Only after validation passes</b>, populate the cache with the verified {@link BusMetadata} for fast runtime + * comparison.</li> + * </ol> + */ +final class EventSourceCache { + + private static final Logger LOG = LoggerFactory.getLogger(EventSourceCache.class); + private static final Gson GSON = new Gson(); + private static final int PAGE_LIMIT = 100; + + /** + * Java 16 record holding cached bus metadata including mapped event sources and their permitted event types. + */ + public record BusMetadata(boolean exists, Map<String, Set<String>> sourceToTypesMap) { + public BusMetadata { + if (sourceToTypesMap == null) { + sourceToTypesMap = Collections.emptyMap(); + } else { + Map<String, Set<String>> unmodifiable = new HashMap<>(); + for (Map.Entry<String, Set<String>> entry : sourceToTypesMap.entrySet()) { + unmodifiable.put(entry.getKey(), Collections.unmodifiableSet(new HashSet<>(entry.getValue()))); + } + sourceToTypesMap = Collections.unmodifiableMap(unmodifiable); + } + } + + public boolean isKnownSource(String source) { + if (source == null || sourceToTypesMap.isEmpty()) { + return true; + } + return sourceToTypesMap.containsKey(source.trim()) || sourceToTypesMap.containsKey("*"); + } + + public boolean isKnownType(String source, String eventType) { + if (eventType == null || sourceToTypesMap.isEmpty()) { + return true; + } + Set<String> types = sourceToTypesMap.get(source != null ? source.trim() : null); + if (types == null || types.isEmpty()) { + types = sourceToTypesMap.get("*"); + } + if (types == null || types.isEmpty()) { + return true; + } + return types.contains(eventType.trim()); + } + } + + /** + * Java 16 record representing a TTL-aware cache entry. + */ + public record CacheEntry<T>(T value, long expiryTime) { + public boolean isExpired(long now) { + return now >= expiryTime; + } + } + + private final Map<String, CacheEntry<BusMetadata>> cache = new ConcurrentHashMap<>(); + private final long ttlMillis; + + EventSourceCache(long ttlMillis) { + this.ttlMillis = ttlMillis; + } + + /** + * Validates the given {@code eventBusName} and any configured {@link AllowedEventBus} definitions against Alibaba + * Cloud. Upon successful validation, the verified metadata is stored in the cache. + * + * @param eventBusName the target event bus name + * @param allowedBus the configured whitelist rules for this bus, or {@code null} + * @param validateSource whether to validate event source existence against Alibaba Cloud + * @param validateType whether to validate event types against Alibaba Cloud rule filter patterns + * @param client the EventBridge client instance; if {@code null}, validation is bypassed + * @return the verified {@link BusMetadata} + */ + BusMetadata validateAndUpdateCache( + String eventBusName, AllowedEventBus allowedBus, + boolean validateSource, boolean validateType, EventBridgeClient client) { + if (client == null || ObjectHelper.isEmpty(eventBusName)) { + return new BusMetadata(true, Collections.emptyMap()); + } + + long now = System.currentTimeMillis(); + CacheEntry<BusMetadata> entry = cache.get(eventBusName); + if (entry != null && !entry.isExpired(now)) { + return entry.value(); + } + + boolean busExists = fetchEventBusExists(eventBusName, client); + if (!busExists) { + throw new IllegalArgumentException( + String.format("Event bus '%s' does not exist in Alibaba Cloud EventBridge", eventBusName)); + } + + Map<String, Set<String>> cloudSourceToTypes = fetchCloudSourceToTypes(eventBusName, client); + + if (validateSource && allowedBus != null && !allowedBus.allowedSources().isEmpty() && !cloudSourceToTypes.isEmpty()) { + for (String source : allowedBus.allowedSources().keySet()) { + if (!cloudSourceToTypes.containsKey(source) && !cloudSourceToTypes.containsKey("*")) { + throw new IllegalArgumentException( + String.format( + "Event source '%s' is not registered in Alibaba Cloud rules for event bus '%s'. Known sources: %s", + source, eventBusName, cloudSourceToTypes.keySet())); + } + } + } + + if (validateType && allowedBus != null && !allowedBus.allowedSources().isEmpty() && !cloudSourceToTypes.isEmpty()) { + for (Map.Entry<String, AllowedEventSource> sourceEntry : allowedBus.allowedSources().entrySet()) { + String source = sourceEntry.getKey(); + Set<String> allowedTypes = sourceEntry.getValue().allowedEventTypes(); + if (allowedTypes != null && !allowedTypes.isEmpty()) { + Set<String> cloudTypes = cloudSourceToTypes.get(source); + if (cloudTypes == null || cloudTypes.isEmpty()) { + cloudTypes = cloudSourceToTypes.get("*"); + } + if (cloudTypes != null && !cloudTypes.isEmpty()) { + for (String type : allowedTypes) { + if (!cloudTypes.contains(type)) { + throw new IllegalArgumentException( + String.format( + "Event type '%s' is not registered in Alibaba Cloud rules for source '%s' on bus '%s'. Allowed in Cloud: %s", + type, source, eventBusName, cloudTypes)); + } + } + } + } + } + } + + BusMetadata metadata = new BusMetadata(true, cloudSourceToTypes); + cache.put(eventBusName, new CacheEntry<>(metadata, now + ttlMillis)); + return metadata; + } + + /** + * Checks if the event bus is known to exist. + */ + boolean isKnownEventBus(String eventBusName, EventBridgeClient client) { + if (client == null || ObjectHelper.isEmpty(eventBusName)) { + return true; + } + long now = System.currentTimeMillis(); + CacheEntry<BusMetadata> entry = cache.get(eventBusName); + if (entry != null && !entry.isExpired(now)) { + return entry.value().exists(); + } + return fetchEventBusExists(eventBusName, client); + } + + /** + * Checks if the event source is registered for the given bus. + */ + boolean isKnownEventSource(String eventBusName, String eventSource, EventBridgeClient client) { + if (client == null || ObjectHelper.isEmpty(eventBusName) || ObjectHelper.isEmpty(eventSource)) { + return true; + } + BusMetadata metadata = validateAndUpdateCache(eventBusName, null, false, false, client); + return metadata.isKnownSource(eventSource); + } + + /** + * Checks if the event type is valid for the given event source on the bus. + */ + boolean isKnownEventType(String eventBusName, String eventSource, String eventType, EventBridgeClient client) { + if (client == null || ObjectHelper.isEmpty(eventBusName) || ObjectHelper.isEmpty(eventType)) { + return true; + } + BusMetadata metadata = validateAndUpdateCache(eventBusName, null, false, false, client); + return metadata.isKnownType(eventSource, eventType); + } + + /** + * Fetches whether the event bus exists via {@code listEventBuses}. + */ + boolean fetchEventBusExists(String eventBusName, EventBridgeClient client) { + try { Review Comment: Fail-open validation (medium): On listEventBuses API failure this returns true (see also line 230), so validateEventSource can be bypassed during cloud/API errors. Same pattern in fetchCloudSourceToTypes with empty metadata. ########## components/camel-alibaba/camel-alibaba-eventbridge/src/test/java/org/apache/camel/component/alibaba/eventbridge/MapCloudEventValidationTest.java: ########## @@ -0,0 +1,332 @@ +/* + * 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.camel.component.alibaba.eventbridge; + +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import com.aliyun.eventbridge.EventBridgeClient; +import com.aliyun.eventbridge.models.CloudEvent; +import com.aliyun.eventbridge.models.EventBusEntry; +import com.aliyun.eventbridge.models.EventRuleDTO; +import com.aliyun.eventbridge.models.ListEventBusesRequest; +import com.aliyun.eventbridge.models.ListEventBusesResponse; +import com.aliyun.eventbridge.models.ListRulesRequest; +import com.aliyun.eventbridge.models.ListRulesResponse; +import org.apache.camel.Exchange; +import org.apache.camel.component.alibaba.eventbridge.models.AllowedEventBus; +import org.apache.camel.component.alibaba.eventbridge.models.AllowedEventSource; +import org.apache.camel.component.alibaba.eventbridge.models.ClientConfigurations; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class MapCloudEventValidationTest extends CamelTestSupport { + + private EventBridgeClient eventBridgeClient; + private EventSourceCache eventSourceCache; + private MapCloudEventValidator validator; + + @BeforeEach + void initTest() { + eventBridgeClient = mock(EventBridgeClient.class); + eventSourceCache = new EventSourceCache(300000L); + validator = new MapCloudEventValidator(eventSourceCache); + } + + @Test + void testValidateAndBuildSuccessWithFullMap() { + ClientConfigurations config + = new ClientConfigurations(null, "default-bus", null, null, null, false, false, true, Map.of(), 300000L); + + Map<String, Object> map = new HashMap<>(); + map.put("eventBusName", "my-bus"); + map.put("source", "acs:oss:cn-hangzhou:12345:my-bucket"); + map.put("type", "oss:ObjectCreated:PutObject"); + map.put("id", "event-id-123"); + map.put("specversion", "1.0"); + map.put("subject", "my-object.jpg"); + map.put("time", "2026-08-23T10:15:30Z"); + map.put("datacontenttype", "application/json"); + map.put("dataschema", "http://example.com/schema.json";); + map.put("data", Map.of("fileSize", 1024, "bucket", "my-bucket")); + + CloudEvent event = validator.validateAndBuild(map, config, eventBridgeClient); + + assertThat(event).isNotNull(); + assertThat(event.getSource().toString()).isEqualTo("acs:oss:cn-hangzhou:12345:my-bucket"); + assertThat(event.getType()).isEqualTo("oss:ObjectCreated:PutObject"); + assertThat(event.getId()).isEqualTo("event-id-123"); + assertThat(event.getSubject()).isEqualTo("my-object.jpg"); + assertThat(event.getSpecversion()).isEqualTo("1.0"); + assertThat(event.getDatacontenttype()).isEqualTo("application/json"); + assertThat(event.getDataschema().toString()).isEqualTo("http://example.com/schema.json";); + assertThat(new String(event.getData(), StandardCharsets.UTF_8)).contains("\"fileSize\":1024"); + } + + @Test + void testValidateAndBuildWithFallbackConfig() { + ClientConfigurations config + = new ClientConfigurations(null, "default-bus", "my.custom.app", "order.created", "order-999"); + + Map<String, Object> map = new HashMap<>(); + map.put("data", "{\"orderId\":\"999\"}"); + + CloudEvent event = validator.validateAndBuild(map, config, eventBridgeClient); + + assertThat(event).isNotNull(); + assertThat(event.getSource().toString()).isEqualTo("my.custom.app"); + assertThat(event.getType()).isEqualTo("order.created"); + assertThat(event.getSubject()).isEqualTo("order-999"); + assertThat(new String(event.getData(), StandardCharsets.UTF_8)).isEqualTo("{\"orderId\":\"999\"}"); + } + + @Test + void testValidateFailsWhenBusNameMissing() { + ClientConfigurations config = new ClientConfigurations(); + Map<String, Object> map = Map.of("source", "my.source", "type", "my.type"); + + assertThatThrownBy(() -> validator.validateAndBuild(map, config, eventBridgeClient)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Event bus name is required"); + } + + @Test + void testValidateFailsWhenSourceMissing() { + ClientConfigurations config = new ClientConfigurations(null, "test-bus", null, null, null); + Map<String, Object> map = Map.of("type", "my.type"); + + assertThatThrownBy(() -> validator.validateAndBuild(map, config, eventBridgeClient)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Event 'source' cannot be empty"); + } + + @Test + void testValidateFailsWhenTypeMissing() { + ClientConfigurations config = new ClientConfigurations(null, "test-bus", "my.source", null, null); + Map<String, Object> map = Map.of("source", "my.source"); + + assertThatThrownBy(() -> validator.validateAndBuild(map, config, eventBridgeClient)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Event 'type' cannot be empty"); + } + + @Test + void testValidateFailsWhenInvalidSpecversion() { + ClientConfigurations config = new ClientConfigurations(null, "test-bus", null, null, null, false, true); + + Map<String, Object> map = Map.of( + "source", "my.source", + "type", "my.type", + "specversion", "0.3"); + + assertThatThrownBy(() -> validator.validateAndBuild(map, config, eventBridgeClient)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Invalid CloudEvent specversion: '0.3'"); + } + + @Test + void testSingleBusDslWithColonsInSourceAndTypes() { + String dsl + = "acs:oss:cn-hangzhou:12345:my-bucket -> oss:ObjectCreated:PutObject, oss:ObjectCreated:PostObject ; app.orders -> order:created:v1"; + Map<String, AllowedEventBus> buses = AlibabaEventBridgeUtils.parseAllowedBusesFromString(dsl, "order-bus"); + + assertThat(buses).containsKey("order-bus"); + AllowedEventBus bus = buses.get("order-bus"); + assertThat(bus.allowedSources()).containsKeys("acs:oss:cn-hangzhou:12345:my-bucket", "app.orders"); + + AllowedEventSource ossSource = bus.allowedSources().get("acs:oss:cn-hangzhou:12345:my-bucket"); + assertThat(ossSource.allowedEventTypes()).containsExactlyInAnyOrder( + "oss:ObjectCreated:PutObject", "oss:ObjectCreated:PostObject"); + + ClientConfigurations config = new ClientConfigurations( + null, "order-bus", null, null, null, false, false, true, buses, 300000L); + + Map<String, Object> validEvent = Map.of( + "source", "acs:oss:cn-hangzhou:12345:my-bucket", + "type", "oss:ObjectCreated:PutObject"); + + CloudEvent event = validator.validateAndBuild(validEvent, config, eventBridgeClient); + assertThat(event).isNotNull(); + + Map<String, Object> invalidTypeEvent = Map.of( + "source", "acs:oss:cn-hangzhou:12345:my-bucket", + "type", "oss:ObjectDeleted:DeleteObject"); + + assertThatThrownBy(() -> validator.validateAndBuild(invalidTypeEvent, config, eventBridgeClient)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Event type 'oss:ObjectDeleted:DeleteObject' is not allowed for event source"); + } + + @Test + void testMultiBusDslWithColonsInSourceAndTypes() { + String dsl + = "orders-bus[ acs:oss:cn-hangzhou:12345:orders -> oss:ObjectCreated:PutObject ; app.orders -> order:created:v1 ]" + + " | payments-bus[ app.payments -> payment:authorized:v1, payment:captured:v1 ]"; + + Map<String, AllowedEventBus> buses = AlibabaEventBridgeUtils.parseAllowedBusesFromString(dsl, null); + assertThat(buses).containsKeys("orders-bus", "payments-bus"); + + ClientConfigurations config = new ClientConfigurations( + null, "orders-bus", null, null, null, false, false, true, buses, 300000L); + + Map<String, Object> event1 = Map.of( + "eventBusName", "orders-bus", + "source", "app.orders", + "type", "order:created:v1"); + assertThat(validator.validateAndBuild(event1, config, eventBridgeClient)).isNotNull(); + + Map<String, Object> event2 = Map.of( + "eventBusName", "payments-bus", + "source", "app.payments", + "type", "payment:captured:v1"); + assertThat(validator.validateAndBuild(event2, config, eventBridgeClient)).isNotNull(); + + Map<String, Object> invalidEvent1 = Map.of( + "eventBusName", "orders-bus", + "source", "app.payments", + "type", "payment:captured:v1"); + assertThatThrownBy(() -> validator.validateAndBuild(invalidEvent1, config, eventBridgeClient)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Event source 'app.payments' is not in the allowed sources list for bus 'orders-bus'"); + + Map<String, Object> invalidEvent2 = Map.of( + "eventBusName", "unknown-bus", + "source", "app.orders", + "type", "order:created:v1"); + assertThatThrownBy(() -> validator.validateAndBuild(invalidEvent2, config, eventBridgeClient)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Event bus 'unknown-bus' is not in the allowed event buses list"); + } + + @Test + void testJsonConfigurationParsing() { + String json = """ + { + "orders-bus": { + "acs:oss:cn-hangzhou:12345:orders": ["oss:ObjectCreated:PutObject", "oss:ObjectCreated:PostObject"], + "app.orders": ["order:created:v1"] + }, + "payments-bus": { + "app.payments": ["payment:authorized:v1"] + } + } + """; + + Map<String, AllowedEventBus> buses = AlibabaEventBridgeUtils.parseAllowedBusesFromString(json, null); + assertThat(buses).containsKeys("orders-bus", "payments-bus"); + + AllowedEventBus ordersBus = buses.get("orders-bus"); + assertThat(ordersBus.allowedSources().get("acs:oss:cn-hangzhou:12345:orders").allowedEventTypes()) + .contains("oss:ObjectCreated:PutObject", "oss:ObjectCreated:PostObject"); + } + + @Test + void testValidatedCacheWorkflowSuccess() { Review Comment: Test coverage gaps: add validateCloudEvent for CloudEvent passthrough; API failure behaviour when validateEventSource=true; rejection when map body overrides bus/source outside whitelist; validateEventSpec=false disables spec checks. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
