From 8c4edc8807ae53b7cc8d01ace0b895886bed2560 Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 3 Jun 2026 13:38:47 +0200 Subject: [PATCH 01/11] Add GOPACSHandler test scaffold and test-support constructor --- .../ems/manager/gopacs/GOPACSHandler.java | 46 ++++++++ .../manager/gopacs/GOPACSHandlerTest.groovy | 108 ++++++++++++++++++ 2 files changed, 154 insertions(+) create mode 100644 ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java index 0871d66..ba098b5 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java @@ -211,6 +211,52 @@ protected GOPACSHandler(String contractedEAN, String realm, String electricitySu deploy(container); } + /** + * Test-support constructor. Wires the message-processing collaborators directly and skips + * remote configuration (OAuth client, private-key file) and JAX-RS deployment, so the + * day-ahead UFTP message flow can be exercised in isolation. Not used in production wiring. + */ + protected GOPACSHandler(String contractedEAN, + String realm, + String electricitySupplierAssetId, + AssetProcessingService assetProcessingService, + AssetPredictedDatapointService assetPredictedDatapointService, + TimerService timerService, + ScheduledExecutorService scheduledExecutorService, + String privateKey) { + this.devMode = false; + this.contractedEAN = contractedEAN; + this.realm = realm; + this.electricitySupplierAssetId = electricitySupplierAssetId; + this.participants = new HashMap<>(); + + this.assetProcessingService = assetProcessingService; + this.assetPredictedDatapointService = assetPredictedDatapointService; + this.timerService = timerService; + this.scheduledExecutorService = scheduledExecutorService; + this.webService = null; + + this.gopacsBrokerUrl = ""; + this.responseDelaySeconds = 0; + this.flexOfferDelaySeconds = 0; + this.clientId = null; + this.clientSecret = null; + this.privateKey = privateKey; + + this.client = null; + this.gopacsAddressBookResource = null; + this.gopacsAuthResource = null; + this.gopacsServerResource = null; + + this.participantResolutionService = new ParticipantResolutionService(this); + this.cryptoService = new UftpCryptoService(participantResolutionService, new LazySodiumFactory(), new LazySodiumBase64Pool()); + this.uftpValidationService = new UftpValidationService(new ArrayList<>()); + this.uftpReceivedMessageService = new UftpReceivedMessageService(uftpValidationService, this); + this.uftpSendMessageService = new UftpSendMessageService(serializer, cryptoService, participantResolutionService, this, uftpValidationService); + + this.objectMapper = new ObjectMapper(); + } + protected static String getDeploymentName(String contractedEAN) { return "GOPACS: " + contractedEAN; } diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy new file mode 100644 index 0000000..efc1470 --- /dev/null +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy @@ -0,0 +1,108 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * See the CONTRIBUTORS.txt file in the distribution for a + * full listing of individual contributors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +package org.openremote.extension.ems.manager.gopacs + +import com.goterl.lazysodium.LazySodiumJava +import com.goterl.lazysodium.SodiumJava +import com.goterl.lazysodium.utils.KeyPair +import org.lfenergy.shapeshifter.api.* +import org.lfenergy.shapeshifter.api.model.UftpParticipantInformation +import org.lfenergy.shapeshifter.core.model.OutgoingUftpMessage +import org.lfenergy.shapeshifter.core.model.UftpParticipant +import org.openremote.manager.asset.AssetProcessingService +import org.openremote.manager.datapoint.AssetPredictedDatapointService +import org.openremote.container.timer.TimerService +import spock.lang.Specification + +import java.time.Duration +import java.time.LocalDate +import java.time.OffsetDateTime +import java.time.ZoneOffset +import java.util.concurrent.ScheduledExecutorService +import java.util.concurrent.ScheduledFuture +import java.util.concurrent.TimeUnit + +class GOPACSHandlerTest extends Specification { + + static final String CONTRACTED_EAN = "871234567890123456" + static final String ASSET_ID = "0abcDEFghiJKLmnoPQRstu" + static final String AGR_DOMAIN = "agr.example.com" + static final String DSO_DOMAIN = "dso.example.com" + static final String TIME_ZONE = "Europe/Amsterdam" + static final LocalDate PERIOD = LocalDate.of(2026, 6, 4) + + AssetProcessingService assetProcessingService + AssetPredictedDatapointService assetPredictedDatapointService + TimerService timerService + ScheduledExecutorService executor + RecordingGOPACSHandler handler + String pubB64 + + def setup() { + assetProcessingService = Mock(AssetProcessingService) + assetPredictedDatapointService = Mock(AssetPredictedDatapointService) + timerService = Stub(TimerService) { + getCurrentTimeMillis() >> PERIOD.atStartOfDay(ZoneOffset.UTC).toInstant().toEpochMilli() + } + // Run every scheduled task inline so processRawMessage is fully synchronous. + executor = Stub(ScheduledExecutorService) { + schedule(_ as Runnable, _ as Long, _ as TimeUnit) >> { Runnable r, long d, TimeUnit u -> + r.run() + Stub(ScheduledFuture) + } + } + + // Generate an ed25519 keypair; hex secret signs, base64 public verifies. + def lazySodium = new LazySodiumJava(new SodiumJava()) + KeyPair kp = lazySodium.cryptoSignKeypair() + String privHex = kp.secretKey.asHexString + pubB64 = Base64.encoder.encodeToString(kp.publicKey.asBytes) + + handler = new RecordingGOPACSHandler(CONTRACTED_EAN, "master", ASSET_ID, + assetProcessingService, assetPredictedDatapointService, timerService, executor, privHex) + + // Pre-seed the DSO participant so signature verification never makes an HTTP call. + handler.participants.put(DSO_DOMAIN, + new UftpParticipantInformation(DSO_DOMAIN, pubB64, "https://dso.example.com/endpoint", true)) + } + + def "handler is constructed via the test-support constructor and is in scope for its contracted EAN"() { + expect: + handler != null + handler.isWithinContractedScope("FlexRequest", "conv-1", CONTRACTED_EAN) + !handler.isWithinContractedScope("FlexRequest", "conv-1", "999999999999999999") + } + + // ---- Test subclass: records outbound messages instead of signing/sending them ---- + static class RecordingGOPACSHandler extends GOPACSHandler { + final List sent = new ArrayList<>() + + RecordingGOPACSHandler(String ean, String realm, String assetId, + AssetProcessingService aps, AssetPredictedDatapointService apds, + TimerService ts, ScheduledExecutorService exec, String privateKey) { + super(ean, realm, assetId, aps, apds, ts, exec, privateKey) + } + + @Override + void notifyNewOutgoingMessage(OutgoingUftpMessage message) { + sent.add(message.payloadMessage()) + } + } +} From ae4f0f122dec2494d62d2205ef2decad07f41460 Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 3 Jun 2026 13:42:58 +0200 Subject: [PATCH 02/11] Add GOPACSHandler FlexRequest flow test --- .../manager/gopacs/GOPACSHandlerTest.groovy | 73 ++++++++++++++++++- 1 file changed, 69 insertions(+), 4 deletions(-) diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy index efc1470..fac0dea 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy @@ -41,7 +41,7 @@ import java.util.concurrent.TimeUnit class GOPACSHandlerTest extends Specification { - static final String CONTRACTED_EAN = "871234567890123456" + static final String CONTRACTED_EAN = "ean.871234567890123456" static final String ASSET_ID = "0abcDEFghiJKLmnoPQRstu" static final String AGR_DOMAIN = "agr.example.com" static final String DSO_DOMAIN = "dso.example.com" @@ -69,14 +69,15 @@ class GOPACSHandlerTest extends Specification { } } - // Generate an ed25519 keypair; hex secret signs, base64 public verifies. + // Generate an ed25519 keypair. The handler's crypto pool (LazySodiumBase64Pool) decodes + // both the signing secret key and the verifying public key as base64. def lazySodium = new LazySodiumJava(new SodiumJava()) KeyPair kp = lazySodium.cryptoSignKeypair() - String privHex = kp.secretKey.asHexString + String privKeyB64 = Base64.encoder.encodeToString(kp.secretKey.asBytes) pubB64 = Base64.encoder.encodeToString(kp.publicKey.asBytes) handler = new RecordingGOPACSHandler(CONTRACTED_EAN, "master", ASSET_ID, - assetProcessingService, assetPredictedDatapointService, timerService, executor, privHex) + assetProcessingService, assetPredictedDatapointService, timerService, executor, privKeyB64) // Pre-seed the DSO participant so signature verification never makes an HTTP call. handler.participants.put(DSO_DOMAIN, @@ -90,6 +91,70 @@ class GOPACSHandlerTest extends Specification { !handler.isWithinContractedScope("FlexRequest", "conv-1", "999999999999999999") } + // Signs the payload as the DSO and feeds the transport XML through the real entry point. + private void signAndProcess(PayloadMessageType payload) { + def sender = new UftpParticipant(DSO_DOMAIN, USEFRoleType.DSO) + String payloadXml = GOPACSHandler.serializer.toXml(payload) + SignedMessage signed = handler.cryptoService.signMessage(payloadXml, sender, handler.privateKey) + String transportXml = GOPACSHandler.serializer.toXml(signed) + handler.processRawMessage(transportXml) + } + + private static void applyHeader(PayloadMessageType m) { + m.setVersion("3.0.0") + m.setSenderDomain(DSO_DOMAIN) + m.setRecipientDomain(AGR_DOMAIN) + m.setTimeStamp(OffsetDateTime.now(ZoneOffset.UTC)) + m.setMessageID(UUID.randomUUID().toString()) + m.setConversationID(UUID.randomUUID().toString()) + } + + private static FlexRequestISPType reqIsp(long start, long maxPower, long minPower) { + def isp = new FlexRequestISPType() + isp.setDisposition(AvailableRequestedType.REQUESTED) + isp.setStart(start) + isp.setDuration(1L) + isp.setMaxPower(maxPower) + isp.setMinPower(minPower) + return isp + } + + private static FlexRequest buildFlexRequest(String congestionPoint) { + def fr = new FlexRequest() + applyHeader(fr) + fr.setISPDuration(Duration.ofMinutes(15)) + fr.setTimeZone(TIME_ZONE) + fr.setPeriod(PERIOD) + fr.setCongestionPoint(congestionPoint) + fr.setExpirationDateTime(OffsetDateTime.now(ZoneOffset.UTC).plusHours(6)) + fr.setRevision(1L) + fr.setContractID("contract-1") + fr.getISPS().add(reqIsp(1L, 5000L, -3000L)) // importMax 5.0, exportMax 3.0 + fr.getISPS().add(reqIsp(2L, 6000L, -4000L)) // importMax 6.0, exportMax 4.0 + return fr + } + + def "FlexRequest updates the asset from request ISPs and replies with FlexRequestResponse then FlexOffer"() { + when: "a signed in-scope FlexRequest is processed" + signAndProcess(buildFlexRequest(CONTRACTED_EAN)) + + then: "predicted datapoints are written: max=importMax, min=exportMax" + 1 * assetPredictedDatapointService.updateValues(ASSET_ID, "powerMaximumFlexRequest", { List dps -> + dps.size() == 2 && dps.collect { it.value as double } == [5.0d, 6.0d] + }) + 1 * assetPredictedDatapointService.updateValues(ASSET_ID, "powerMinimumFlexRequest", { List dps -> + dps.size() == 2 && dps.collect { it.value as double } == [3.0d, 4.0d] + }) + + and: "a FlexRequestResponse (Accepted) is sent, followed by a FlexOffer for the same congestion point" + handler.sent.size() == 2 + handler.sent[0] instanceof FlexRequestResponse + ((FlexRequestResponse) handler.sent[0]).result == AcceptedRejectedType.ACCEPTED + handler.sent[1] instanceof FlexOffer + ((FlexOffer) handler.sent[1]).congestionPoint == CONTRACTED_EAN + !((FlexOffer) handler.sent[1]).offerOptions.isEmpty() + } + // ---- Test subclass: records outbound messages instead of signing/sending them ---- static class RecordingGOPACSHandler extends GOPACSHandler { final List sent = new ArrayList<>() From ee8fd1745c46afbd4a270401736d5fde7762113f Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 3 Jun 2026 13:43:46 +0200 Subject: [PATCH 03/11] Add GOPACSHandler FlexOfferResponse no-reply test --- .../manager/gopacs/GOPACSHandlerTest.groovy | 24 +++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy index fac0dea..1853378 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy @@ -155,6 +155,30 @@ class GOPACSHandlerTest extends Specification { !((FlexOffer) handler.sent[1]).offerOptions.isEmpty() } + private static FlexOfferResponse buildFlexOfferResponse(AcceptedRejectedType result) { + def r = new FlexOfferResponse() + applyHeader(r) + r.setFlexOfferMessageID(UUID.randomUUID().toString()) + r.setResult(result) + if (result == AcceptedRejectedType.REJECTED) { + r.setRejectionReason("insufficient flexibility") + } + return r + } + + def "FlexOfferResponse (#result) is handled without mutating the asset or sending a reply"() { + when: "a signed FlexOfferResponse is processed" + signAndProcess(buildFlexOfferResponse(result)) + + then: "no asset mutation and no outbound message" + 0 * assetPredictedDatapointService.updateValues(_, _, _) + 0 * assetProcessingService.sendAttributeEvent(_, _) + handler.sent.isEmpty() + + where: + result << [AcceptedRejectedType.ACCEPTED, AcceptedRejectedType.REJECTED] + } + // ---- Test subclass: records outbound messages instead of signing/sending them ---- static class RecordingGOPACSHandler extends GOPACSHandler { final List sent = new ArrayList<>() From 7300781a4543e5c1ac996515e77f80c3359b8d0e Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 3 Jun 2026 13:44:38 +0200 Subject: [PATCH 04/11] Add GOPACSHandler FlexOrder offtake and feed-in flow tests --- .../manager/gopacs/GOPACSHandlerTest.groovy | 63 +++++++++++++++++++ 1 file changed, 63 insertions(+) diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy index 1853378..fbb6507 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy @@ -179,6 +179,69 @@ class GOPACSHandlerTest extends Specification { result << [AcceptedRejectedType.ACCEPTED, AcceptedRejectedType.REJECTED] } + private static FlexOrderISPType orderIsp(long start, long power) { + def isp = new FlexOrderISPType() + isp.setStart(start) + isp.setDuration(1L) + isp.setPower(power) + return isp + } + + private static FlexOrder buildFlexOrder(String congestionPoint, List powers) { + def fo = new FlexOrder() + applyHeader(fo) + fo.setISPDuration(Duration.ofMinutes(15)) + fo.setTimeZone(TIME_ZONE) + fo.setPeriod(PERIOD) + fo.setCongestionPoint(congestionPoint) + fo.setFlexOfferMessageID(UUID.randomUUID().toString()) + fo.setOrderReference(UUID.randomUUID().toString()) + fo.setContractID("contract-1") + fo.setCurrency("EUR") + fo.setPrice(new BigDecimal("0.00")) + long start = 1L + powers.each { p -> fo.getISPS().add(orderIsp(start++, p)) } + return fo + } + + def "FlexOrder with offtake power updates currentPower and the max-profile and replies with FlexOrderResponse"() { + when: "a signed in-scope FlexOrder with positive (offtake) power is processed" + signAndProcess(buildFlexOrder(CONTRACTED_EAN, [4000L, 8000L])) // 4.0, 8.0 kW + + then: "current power and the offtake (max) profile are written; the feed-in (min) profile is not" + 1 * assetPredictedDatapointService.updateValues(ASSET_ID, "currentPowerFlexRequest", { List dps -> + dps.size() == 2 && dps.collect { it.value as double } == [4.0d, 8.0d] + }) + 1 * assetPredictedDatapointService.updateValues(ASSET_ID, "powerLimitMaximumProfileFlexOrder", { List dps -> + dps.size() == 2 && dps.collect { it.value as double } == [4.0d, 8.0d] + }) + 0 * assetPredictedDatapointService.updateValues(ASSET_ID, "powerLimitMinimumProfileFlexOrder", _) + + and: "a FlexOrderResponse (Accepted) is sent back" + handler.sent.size() == 1 + handler.sent[0] instanceof FlexOrderResponse + ((FlexOrderResponse) handler.sent[0]).result == AcceptedRejectedType.ACCEPTED + } + + def "FlexOrder with feed-in power updates currentPower and the min-profile"() { + when: "a signed in-scope FlexOrder with negative (feed-in) power is processed" + signAndProcess(buildFlexOrder(CONTRACTED_EAN, [-2000L, -5000L])) // -2.0, -5.0 kW + + then: "current power and the feed-in (min) profile are written; the offtake (max) profile is not" + 1 * assetPredictedDatapointService.updateValues(ASSET_ID, "currentPowerFlexRequest", { List dps -> + dps.size() == 2 && dps.collect { it.value as double } == [-2.0d, -5.0d] + }) + 1 * assetPredictedDatapointService.updateValues(ASSET_ID, "powerLimitMinimumProfileFlexOrder", { List dps -> + dps.size() == 2 && dps.collect { it.value as double } == [-2.0d, -5.0d] + }) + 0 * assetPredictedDatapointService.updateValues(ASSET_ID, "powerLimitMaximumProfileFlexOrder", _) + + and: "a FlexOrderResponse (Accepted) is sent back" + handler.sent.size() == 1 + handler.sent[0] instanceof FlexOrderResponse + ((FlexOrderResponse) handler.sent[0]).result == AcceptedRejectedType.ACCEPTED + } + // ---- Test subclass: records outbound messages instead of signing/sending them ---- static class RecordingGOPACSHandler extends GOPACSHandler { final List sent = new ArrayList<>() From 3f88e1538770f00814e1e925fa334cee390c9a46 Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 3 Jun 2026 13:45:45 +0200 Subject: [PATCH 05/11] Add GOPACSHandler out-of-scope congestion point drop test --- .../ems/manager/gopacs/GOPACSHandlerTest.groovy | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy index fbb6507..dfcfa67 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy @@ -242,6 +242,19 @@ class GOPACSHandlerTest extends Specification { ((FlexOrderResponse) handler.sent[0]).result == AcceptedRejectedType.ACCEPTED } + def "a validly-signed FlexRequest for a different congestion point is dropped with no mutation and no reply"() { + given: "a FlexRequest whose congestion point is not the contracted EAN" + def otherEan = "ean.999999999999999999" + + when: "the signed out-of-scope FlexRequest is processed" + signAndProcess(buildFlexRequest(otherEan)) + + then: "the asset is not mutated and nothing is sent back" + 0 * assetPredictedDatapointService.updateValues(_, _, _) + 0 * assetProcessingService.sendAttributeEvent(_, _) + handler.sent.isEmpty() + } + // ---- Test subclass: records outbound messages instead of signing/sending them ---- static class RecordingGOPACSHandler extends GOPACSHandler { final List sent = new ArrayList<>() From 610007ef16c106faf30741d2a6e653f2374f1976 Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 3 Jun 2026 15:36:27 +0200 Subject: [PATCH 06/11] Restyle GOPACS handler tests to embedded UFTP XML fixtures matching the example message format --- .../manager/gopacs/GOPACSHandlerTest.groovy | 192 ++++++++---------- 1 file changed, 82 insertions(+), 110 deletions(-) diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy index dfcfa67..98bdfb0 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy @@ -26,34 +26,43 @@ import org.lfenergy.shapeshifter.api.* import org.lfenergy.shapeshifter.api.model.UftpParticipantInformation import org.lfenergy.shapeshifter.core.model.OutgoingUftpMessage import org.lfenergy.shapeshifter.core.model.UftpParticipant +import org.openremote.container.timer.TimerService import org.openremote.manager.asset.AssetProcessingService import org.openremote.manager.datapoint.AssetPredictedDatapointService -import org.openremote.container.timer.TimerService import spock.lang.Specification -import java.time.Duration import java.time.LocalDate -import java.time.OffsetDateTime import java.time.ZoneOffset import java.util.concurrent.ScheduledExecutorService import java.util.concurrent.ScheduledFuture import java.util.concurrent.TimeUnit +/** + * Exercises the day-ahead UFTP message flow handled by {@link GOPACSHandler} on the AGR (EMS) side. + * + * Real signed messages are driven through the handler's actual entry point ({@code processRawMessage}) + * using libsodium crypto -- no database, no container. The embedded payload XML matches the wire format + * of the example messages (an outbound FlexRequestResponse + its SignedMessage envelope), with the + * inbound direction being DSO (nilsgrid.net) -> AGR (openremote.io). Outbound messages -- both the + * library-generated responses and the handler-built FlexOffer -- are captured by overriding + * {@code notifyNewOutgoingMessage}. + */ class GOPACSHandlerTest extends Specification { static final String CONTRACTED_EAN = "ean.871234567890123456" static final String ASSET_ID = "0abcDEFghiJKLmnoPQRstu" - static final String AGR_DOMAIN = "agr.example.com" - static final String DSO_DOMAIN = "dso.example.com" - static final String TIME_ZONE = "Europe/Amsterdam" + static final String AGR_DOMAIN = "openremote.io" // our AGR: recipient of inbound, sender of replies + static final String DSO_DOMAIN = "nilsgrid.net" // the DSO: sender of inbound, recipient of replies static final LocalDate PERIOD = LocalDate.of(2026, 6, 4) + // IDs taken from the provided example payload.xml so that file is literally the expected FlexRequest reply. + static final String FLEX_REQUEST_MESSAGE_ID = "b3030b8b-2f45-43f4-8bf3-f00ef6fd74fc" + static final String CONVERSATION_ID = "5f7c9c4d-5988-4a17-a479-2f716051fd6d" AssetProcessingService assetProcessingService AssetPredictedDatapointService assetPredictedDatapointService TimerService timerService ScheduledExecutorService executor RecordingGOPACSHandler handler - String pubB64 def setup() { assetProcessingService = Mock(AssetProcessingService) @@ -69,74 +78,27 @@ class GOPACSHandlerTest extends Specification { } } - // Generate an ed25519 keypair. The handler's crypto pool (LazySodiumBase64Pool) decodes - // both the signing secret key and the verifying public key as base64. + // Generate an ed25519 keypair. The handler's crypto pool (LazySodiumBase64Pool) decodes both the + // signing secret key and the verifying public key as base64. def lazySodium = new LazySodiumJava(new SodiumJava()) KeyPair kp = lazySodium.cryptoSignKeypair() String privKeyB64 = Base64.encoder.encodeToString(kp.secretKey.asBytes) - pubB64 = Base64.encoder.encodeToString(kp.publicKey.asBytes) + String pubB64 = Base64.encoder.encodeToString(kp.publicKey.asBytes) handler = new RecordingGOPACSHandler(CONTRACTED_EAN, "master", ASSET_ID, assetProcessingService, assetPredictedDatapointService, timerService, executor, privKeyB64) - // Pre-seed the DSO participant so signature verification never makes an HTTP call. + // Pre-seed the DSO participant so inbound signature verification never makes an HTTP call. handler.participants.put(DSO_DOMAIN, - new UftpParticipantInformation(DSO_DOMAIN, pubB64, "https://dso.example.com/endpoint", true)) - } - - def "handler is constructed via the test-support constructor and is in scope for its contracted EAN"() { - expect: - handler != null - handler.isWithinContractedScope("FlexRequest", "conv-1", CONTRACTED_EAN) - !handler.isWithinContractedScope("FlexRequest", "conv-1", "999999999999999999") - } - - // Signs the payload as the DSO and feeds the transport XML through the real entry point. - private void signAndProcess(PayloadMessageType payload) { - def sender = new UftpParticipant(DSO_DOMAIN, USEFRoleType.DSO) - String payloadXml = GOPACSHandler.serializer.toXml(payload) - SignedMessage signed = handler.cryptoService.signMessage(payloadXml, sender, handler.privateKey) - String transportXml = GOPACSHandler.serializer.toXml(signed) - handler.processRawMessage(transportXml) - } - - private static void applyHeader(PayloadMessageType m) { - m.setVersion("3.0.0") - m.setSenderDomain(DSO_DOMAIN) - m.setRecipientDomain(AGR_DOMAIN) - m.setTimeStamp(OffsetDateTime.now(ZoneOffset.UTC)) - m.setMessageID(UUID.randomUUID().toString()) - m.setConversationID(UUID.randomUUID().toString()) - } - - private static FlexRequestISPType reqIsp(long start, long maxPower, long minPower) { - def isp = new FlexRequestISPType() - isp.setDisposition(AvailableRequestedType.REQUESTED) - isp.setStart(start) - isp.setDuration(1L) - isp.setMaxPower(maxPower) - isp.setMinPower(minPower) - return isp - } - - private static FlexRequest buildFlexRequest(String congestionPoint) { - def fr = new FlexRequest() - applyHeader(fr) - fr.setISPDuration(Duration.ofMinutes(15)) - fr.setTimeZone(TIME_ZONE) - fr.setPeriod(PERIOD) - fr.setCongestionPoint(congestionPoint) - fr.setExpirationDateTime(OffsetDateTime.now(ZoneOffset.UTC).plusHours(6)) - fr.setRevision(1L) - fr.setContractID("contract-1") - fr.getISPS().add(reqIsp(1L, 5000L, -3000L)) // importMax 5.0, exportMax 3.0 - fr.getISPS().add(reqIsp(2L, 6000L, -4000L)) // importMax 6.0, exportMax 4.0 - return fr + new UftpParticipantInformation(DSO_DOMAIN, pubB64, "https://nilsgrid.net/shapeshifter/api/v3/message", true)) } def "FlexRequest updates the asset from request ISPs and replies with FlexRequestResponse then FlexOffer"() { - when: "a signed in-scope FlexRequest is processed" - signAndProcess(buildFlexRequest(CONTRACTED_EAN)) + given: "a signed in-scope FlexRequest in the example wire format" + def xml = flexRequestXml(CONTRACTED_EAN) + + when: "it is processed" + signAndProcess(xml) then: "predicted datapoints are written: max=importMax, min=exportMax" 1 * assetPredictedDatapointService.updateValues(ASSET_ID, "powerMaximumFlexRequest", { List dps -> @@ -146,29 +108,26 @@ class GOPACSHandlerTest extends Specification { dps.size() == 2 && dps.collect { it.value as double } == [3.0d, 4.0d] }) - and: "a FlexRequestResponse (Accepted) is sent, followed by a FlexOffer for the same congestion point" + and: "a FlexRequestResponse is sent, followed by a FlexOffer for the same congestion point" handler.sent.size() == 2 handler.sent[0] instanceof FlexRequestResponse - ((FlexRequestResponse) handler.sent[0]).result == AcceptedRejectedType.ACCEPTED handler.sent[1] instanceof FlexOffer ((FlexOffer) handler.sent[1]).congestionPoint == CONTRACTED_EAN !((FlexOffer) handler.sent[1]).offerOptions.isEmpty() - } - private static FlexOfferResponse buildFlexOfferResponse(AcceptedRejectedType result) { - def r = new FlexOfferResponse() - applyHeader(r) - r.setFlexOfferMessageID(UUID.randomUUID().toString()) - r.setResult(result) - if (result == AcceptedRejectedType.REJECTED) { - r.setRejectionReason("insufficient flexibility") - } - return r + and: "the emitted FlexRequestResponse matches the example payload.xml shape" + def response = (FlexRequestResponse) handler.sent[0] + response.result == AcceptedRejectedType.ACCEPTED + response.flexRequestMessageID == FLEX_REQUEST_MESSAGE_ID + response.senderDomain == AGR_DOMAIN + response.recipientDomain == DSO_DOMAIN + response.version == "3.0.0" + response.conversationID == CONVERSATION_ID } def "FlexOfferResponse (#result) is handled without mutating the asset or sending a reply"() { when: "a signed FlexOfferResponse is processed" - signAndProcess(buildFlexOfferResponse(result)) + signAndProcess(flexOfferResponseXml(result)) then: "no asset mutation and no outbound message" 0 * assetPredictedDatapointService.updateValues(_, _, _) @@ -179,34 +138,9 @@ class GOPACSHandlerTest extends Specification { result << [AcceptedRejectedType.ACCEPTED, AcceptedRejectedType.REJECTED] } - private static FlexOrderISPType orderIsp(long start, long power) { - def isp = new FlexOrderISPType() - isp.setStart(start) - isp.setDuration(1L) - isp.setPower(power) - return isp - } - - private static FlexOrder buildFlexOrder(String congestionPoint, List powers) { - def fo = new FlexOrder() - applyHeader(fo) - fo.setISPDuration(Duration.ofMinutes(15)) - fo.setTimeZone(TIME_ZONE) - fo.setPeriod(PERIOD) - fo.setCongestionPoint(congestionPoint) - fo.setFlexOfferMessageID(UUID.randomUUID().toString()) - fo.setOrderReference(UUID.randomUUID().toString()) - fo.setContractID("contract-1") - fo.setCurrency("EUR") - fo.setPrice(new BigDecimal("0.00")) - long start = 1L - powers.each { p -> fo.getISPS().add(orderIsp(start++, p)) } - return fo - } - def "FlexOrder with offtake power updates currentPower and the max-profile and replies with FlexOrderResponse"() { when: "a signed in-scope FlexOrder with positive (offtake) power is processed" - signAndProcess(buildFlexOrder(CONTRACTED_EAN, [4000L, 8000L])) // 4.0, 8.0 kW + signAndProcess(flexOrderXml(CONTRACTED_EAN, [4000, 8000])) // 4.0, 8.0 kW then: "current power and the offtake (max) profile are written; the feed-in (min) profile is not" 1 * assetPredictedDatapointService.updateValues(ASSET_ID, "currentPowerFlexRequest", { List dps -> @@ -225,7 +159,7 @@ class GOPACSHandlerTest extends Specification { def "FlexOrder with feed-in power updates currentPower and the min-profile"() { when: "a signed in-scope FlexOrder with negative (feed-in) power is processed" - signAndProcess(buildFlexOrder(CONTRACTED_EAN, [-2000L, -5000L])) // -2.0, -5.0 kW + signAndProcess(flexOrderXml(CONTRACTED_EAN, [-2000, -5000])) // -2.0, -5.0 kW then: "current power and the feed-in (min) profile are written; the offtake (max) profile is not" 1 * assetPredictedDatapointService.updateValues(ASSET_ID, "currentPowerFlexRequest", { List dps -> @@ -243,11 +177,8 @@ class GOPACSHandlerTest extends Specification { } def "a validly-signed FlexRequest for a different congestion point is dropped with no mutation and no reply"() { - given: "a FlexRequest whose congestion point is not the contracted EAN" - def otherEan = "ean.999999999999999999" - - when: "the signed out-of-scope FlexRequest is processed" - signAndProcess(buildFlexRequest(otherEan)) + when: "a signed out-of-scope FlexRequest (different EAN) is processed" + signAndProcess(flexRequestXml("ean.999999999999999999")) then: "the asset is not mutated and nothing is sent back" 0 * assetPredictedDatapointService.updateValues(_, _, _) @@ -255,7 +186,48 @@ class GOPACSHandlerTest extends Specification { handler.sent.isEmpty() } - // ---- Test subclass: records outbound messages instead of signing/sending them ---- + // ---- Embedded UFTP payload fixtures (attribute-style XML, matching the example message format) ---- + + private static String flexRequestXml(String congestionPoint) { + """ + + + +""" + } + + private static String flexOrderXml(String congestionPoint, List powers) { + def isps = (0.. + """ """ + }.join("\n") + """ + +${isps} +""" + } + + private static String flexOfferResponseXml(AcceptedRejectedType result) { + def reason = result == AcceptedRejectedType.REJECTED ? ' RejectionReason="insufficient flexibility"' : '' + """ +""" + } + + // Signs the payload XML as the DSO and feeds the transport XML through the real entry point. + private void signAndProcess(String payloadXml) { + def sender = new UftpParticipant(DSO_DOMAIN, USEFRoleType.DSO) + SignedMessage signed = handler.cryptoService.signMessage(payloadXml, sender, handler.privateKey) + handler.processRawMessage(GOPACSHandler.serializer.toXml(signed)) + } + + // Records outbound messages (library-generated responses + handler-built FlexOffer) instead of sending them. static class RecordingGOPACSHandler extends GOPACSHandler { final List sent = new ArrayList<>() From 086187d5fce7c4ef8f91a5665a98fa81f79a5b63 Mon Sep 17 00:00:00 2001 From: Michael Date: Thu, 4 Jun 2026 14:39:24 +0200 Subject: [PATCH 07/11] Add WireMock test dependency for GOPACS HTTP-level tests The GOPACS broker delivery goes through java.net.http.HttpClient (via shapeshifter's UftpSendMessageService), which a JAX-RS ClientRequestFilter cannot intercept. WireMock (standalone, shaded to avoid clashing with the manager's web stack) provides a local HTTP server to stub and verify the broker, OAuth2 token and address-book endpoints on the wire -- the same approach shapeshifter uses to test UftpSendMessageService. --- ems/build.gradle | 4 ++++ gradle.properties | 1 + 2 files changed, 5 insertions(+) diff --git a/ems/build.gradle b/ems/build.gradle index 5557af3..1a99709 100644 --- a/ems/build.gradle +++ b/ems/build.gradle @@ -14,6 +14,10 @@ dependencies { api "org.jboss.resteasy:resteasy-client-api:$resteasyVersion" implementation "org.lfenergy.shapeshifter:shapeshifter-core:$shapeshifterVersion" testImplementation "io.openremote:openremote-test:$openremoteVersion" + // Standalone (shaded) WireMock so its embedded Jetty/Jackson don't clash with the manager's web stack; + // used to mock the GOPACS broker / OAuth2 / address-book HTTP endpoints (the broker send goes through + // java.net.http.HttpClient, which a JAX-RS ClientRequestFilter cannot intercept). + testImplementation "org.wiremock:wiremock-standalone:$wiremockVersion" } jar { diff --git a/gradle.properties b/gradle.properties index 75531c3..aa28d06 100644 --- a/gradle.properties +++ b/gradle.properties @@ -20,3 +20,4 @@ jacksonVersion = 2.21.3 resteasyVersion = 6.2.16.Final shapeshifterVersion = 3.5.0 testLoggerVersion = 4.0.0 +wiremockVersion = 3.13.1 From f69c2c92fcbae1349798c30331a3a96d1424c694 Mon Sep 17 00:00:00 2001 From: Michael Date: Thu, 4 Jun 2026 14:39:24 +0200 Subject: [PATCH 08/11] Add GOPACSHandler HTTP-contract integration test Complements the in-process GOPACSHandlerTest (which calls processRawMessage directly and records outgoing payload objects) by exercising the real transport chain end to end, addressing the HTTP coverage gap raised in review: - Inbound: real POST to the deployed /gopacs/message JAX-RS resource, asserting accepted media type, JAX-RS delegation and the WebApplicationException status mapping (success vs 400). - Outbound: real broker delivery via UftpSendMessageService (over java.net.http.HttpClient), plus the OAuth2 token and address-book lookups via the handler's RESTEasy client. All three are pointed at one WireMock server, so the signed XML, synthesised broker endpoint and 'Authorization: Bearer' header are verified on the wire. Covers: in-scope FlexRequest -> FlexRequestResponse + FlexOffer broker delivery; FlexOrder -> FlexOrderResponse; out-of-scope drop (accepted, no delivery); malformed XML and signature-mismatch -> 400; OAuth2 500 / invalid token JSON and address-book 404 -> 400; and participant-lookup caching. Like the other manager-container integration tests in this repo (e.g. EntsoeProtocolTest) it requires a running OpenRemote dev stack and is skipped on CI via @IgnoreIf(GITHUB_ACTIONS). --- .../gopacs/GOPACSHandlerHttpTest.groovy | 391 ++++++++++++++++++ 1 file changed, 391 insertions(+) create mode 100644 ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerHttpTest.groovy diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerHttpTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerHttpTest.groovy new file mode 100644 index 0000000..50d9df1 --- /dev/null +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerHttpTest.groovy @@ -0,0 +1,391 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * See the CONTRIBUTORS.txt file in the distribution for a + * full listing of individual contributors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +package org.openremote.extension.ems.manager.gopacs + +import com.github.tomakehurst.wiremock.WireMockServer +import com.goterl.lazysodium.LazySodiumJava +import com.goterl.lazysodium.SodiumJava +import com.goterl.lazysodium.utils.KeyPair +import jakarta.ws.rs.client.Entity +import jakarta.ws.rs.core.Response +import org.lfenergy.shapeshifter.api.USEFRoleType +import org.lfenergy.shapeshifter.api.SignedMessage +import org.lfenergy.shapeshifter.core.model.UftpParticipant +import org.openremote.container.web.WebService +import org.openremote.extension.ems.manager.EmsOptimisationService +import org.openremote.extension.ems.manager.EmsOptimisationSetupService +import org.openremote.model.ContainerService +import org.openremote.test.ManagerContainerTrait +import spock.lang.IgnoreIf +import spock.lang.Shared +import spock.lang.Specification +import spock.util.concurrent.PollingConditions + +import java.time.LocalDate + +import static com.github.tomakehurst.wiremock.client.WireMock.aResponse +import static com.github.tomakehurst.wiremock.client.WireMock.containing +import static com.github.tomakehurst.wiremock.client.WireMock.equalTo +import static com.github.tomakehurst.wiremock.client.WireMock.get +import static com.github.tomakehurst.wiremock.client.WireMock.getRequestedFor +import static com.github.tomakehurst.wiremock.client.WireMock.okJson +import static com.github.tomakehurst.wiremock.client.WireMock.post +import static com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor +import static com.github.tomakehurst.wiremock.client.WireMock.urlPathEqualTo +import static org.openremote.model.Constants.MASTER_REALM + +/** + * HTTP-contract integration test for {@link GOPACSHandler}, complementing the in-process message-flow + * unit test in {@code GOPACSHandlerTest}. Where that test calls {@code processRawMessage} directly and + * records outgoing payload objects, this test exercises the real transport chain: + * + *
    + *
  • Inbound: a real {@code POST} to the deployed JAX-RS resource ({@code /gopacs/message}), + * asserting the accepted media type, JAX-RS delegation and {@code WebApplicationException} status + * mapping in {@link GOPACSHandler#processRawMessage}.
    • + *
  • Outbound: the real broker delivery via {@code UftpSendMessageService} (which posts over + * {@code java.net.http.HttpClient}), plus the OAuth2 token and address-book lookups via the + * handler's RESTEasy client. All three are pointed at a single WireMock server, so signed XML, + * the synthesised broker endpoint and the {@code Authorization: Bearer} header are asserted on the + * wire.
    • + *
+ * + * The broker send uses the JDK {@code HttpClient}, which a JAX-RS {@code ClientRequestFilter} (the style + * used by {@code EntsoeProtocolTest}) cannot intercept; hence the WireMock server, matching the approach + * shapeshifter itself uses to test {@code UftpSendMessageService}. + * + * Like the other manager-container integration tests in this repository (e.g. {@code EntsoeProtocolTest}), + * this requires a running OpenRemote dev stack and is skipped on CI. + * + * Asset mutation (predicted datapoints / attribute events) is intentionally not re-asserted here -- those + * writes are scheduled asynchronously and are already covered precisely by {@code GOPACSHandlerTest}. This + * test focuses on the HTTP contract, so a placeholder asset id is sufficient. + */ +@IgnoreIf({ System.getenv("GITHUB_ACTIONS") == "true" }) +class GOPACSHandlerHttpTest extends Specification implements ManagerContainerTrait { + + static final String CONTRACTED_EAN = "ean.871234567890123456" + static final String ASSET_ID = "0abcDEFghiJKLmnoPQRstu" + static final String AGR_DOMAIN = "openremote.io" // our AGR: recipient of inbound, sender of replies + static final String DSO_DOMAIN = "nilsgrid.net" // the DSO: sender of inbound, recipient of replies + static final LocalDate PERIOD = LocalDate.of(2026, 6, 4) + static final String FLEX_REQUEST_MESSAGE_ID = "b3030b8b-2f45-43f4-8bf3-f00ef6fd74fc" + static final String CONVERSATION_ID = "5f7c9c4d-5988-4a17-a479-2f716051fd6d" + static final String ACCESS_TOKEN = "test-access-token" + + static final String OAUTH_PATH = "/oauth/token" + static final String ADDRESS_BOOK_PATH = "/uftp-participants/v3/participants/" + DSO_DOMAIN + static final String BROKER_PATH = "/shapeshifter/api/v3/message" + + @Shared + WireMockServer wireMock + @Shared + String dsoPrivateKeyB64 + @Shared + String dsoPublicKeyB64 + @Shared + File privateKeyFile + @Shared + GOPACSHandler handler + @Shared + int serverPort + + def setupSpec() { + // Two ed25519 keypairs: the DSO signs the inbound messages (its public key is served by the + // address-book stub so the handler can verify them), and the AGR signs its outbound replies. + def lazySodium = new LazySodiumJava(new SodiumJava()) + KeyPair dso = lazySodium.cryptoSignKeypair() + dsoPrivateKeyB64 = Base64.encoder.encodeToString(dso.secretKey.asBytes) + dsoPublicKeyB64 = Base64.encoder.encodeToString(dso.publicKey.asBytes) + + KeyPair agr = lazySodium.cryptoSignKeypair() + privateKeyFile = File.createTempFile("gopacs-agr-private-key", ".txt") + privateKeyFile.deleteOnExit() + privateKeyFile.text = Base64.encoder.encodeToString(agr.secretKey.asBytes) + + wireMock = new WireMockServer(0) + wireMock.start() + } + + def cleanupSpec() { + handler?.undeploy() + wireMock?.stop() + } + + def setup() { + // Must be a plain String (not a Groovy GString): OpenRemote's Config.init casts every config value to String. + String base = "http://localhost:${wireMock.port()}".toString() + def config = defaultConfig() << [ + (GOPACSHandler.GOPACS_CLIENT_ID) : "test-client", + (GOPACSHandler.GOPACS_CLIENT_SECRET) : "test-secret", + (GOPACSHandler.GOPACS_PRIVATE_KEY_FILE) : privateKeyFile.absolutePath, + (GOPACSHandler.GOPACS_OAUTH2_URL) : base + OAUTH_PATH, + (GOPACSHandler.GOPACS_PARTICIPANT_URL) : base, + (GOPACSHandler.GOPACS_BROKER_URL) : base, + // Send replies immediately so assertions don't wait on the production response/offer delays. + (GOPACSHandler.GOPACS_RESPONSE_DELAY_SECONDS) : "0", + (GOPACSHandler.GOPACS_FLEX_OFFER_DELAY_SECONDS): "0" + ] + + // startContainer reuses the already-running container when config (ignoring the web-server port) + // and services match, so this is cheap after the first feature. + def runningContainer = startContainer(config, gopacsServices()) + + if (handler == null) { + serverPort = runningContainer.getConfig().get(WebService.OR_WEBSERVER_LISTEN_PORT) as int + + // Construct the production handler directly (real RESTEasy client, real send service, real + // JAX-RS deployment). EmsOptimisationService is excluded from the service list above so it + // cannot spin up a second handler on the same /gopacs deployment path. + handler = new GOPACSHandler(CONTRACTED_EAN, MASTER_REALM, ASSET_ID, runningContainer) + } + + // Reset cross-test state: the participant cache leaks across features otherwise, and the request + // journal / stubs must start clean for the per-feature verifications. + handler.participants.clear() + wireMock.resetAll() + stubOAuthToken(ACCESS_TOKEN) + stubAddressBook(dsoPublicKeyB64) + stubBrokerAccepts() + } + + // The manager services minus the EMS optimisation services, which would otherwise deploy their own + // GOPACS handler (on the same /gopacs path) from any persisted EmsGOPACSAsset. + private Iterable gopacsServices() { + defaultServices().findAll { + !(it instanceof EmsOptimisationService) && !(it instanceof EmsOptimisationSetupService) + } + } + + def "a signed in-scope FlexRequest POSTed to /gopacs/message is accepted and the reply is delivered to the broker with a bearer token"() { + given: "polling for the asynchronous outbound deliveries" + def conditions = new PollingConditions(timeout: 10, delay: 0.2) + + when: "a signed FlexRequest is POSTed to the deployed endpoint" + Response response = postSignedAsDso(flexRequestXml(CONTRACTED_EAN)) + + then: "the transport call succeeds" + response.statusInfo.family == Response.Status.Family.SUCCESSFUL + response.close() + + and: "verifying the signature fetched an OAuth2 token and resolved the DSO via the address book" + conditions.eventually { + wireMock.verify(postRequestedFor(urlPathEqualTo(OAUTH_PATH)) + .withRequestBody(containing("grant_type=client_credentials"))) + wireMock.verify(getRequestedFor(urlPathEqualTo(ADDRESS_BOOK_PATH)) + .withHeader("Authorization", equalTo("Bearer " + ACCESS_TOKEN))) + } + + and: "the FlexRequestResponse and FlexOffer are actually POSTed to the broker as signed XML with a bearer token" + conditions.eventually { + wireMock.verify(postRequestedFor(urlPathEqualTo(BROKER_PATH)) + .withHeader("Authorization", equalTo("Bearer " + ACCESS_TOKEN)) + .withRequestBody(containing("SignedMessage"))) + def payloads = decodedBrokerPayloads() + assert payloads.any { it.contains("FlexRequestResponse") } + assert payloads.any { it.contains("FlexOffer") } + } + } + + def "a signed in-scope FlexOrder POSTed to /gopacs/message is accepted and a FlexOrderResponse is delivered to the broker"() { + given: + def conditions = new PollingConditions(timeout: 10, delay: 0.2) + + when: "a signed FlexOrder is POSTed to the deployed endpoint" + Response response = postSignedAsDso(flexOrderXml(CONTRACTED_EAN, [4000, 8000])) + + then: "the transport call succeeds" + response.statusInfo.family == Response.Status.Family.SUCCESSFUL + response.close() + + and: "a FlexOrderResponse is POSTed to the broker as signed XML with a bearer token" + conditions.eventually { + wireMock.verify(postRequestedFor(urlPathEqualTo(BROKER_PATH)) + .withHeader("Authorization", equalTo("Bearer " + ACCESS_TOKEN)) + .withRequestBody(containing("SignedMessage"))) + assert decodedBrokerPayloads().any { it.contains("FlexOrderResponse") } + } + } + + def "a validly-signed FlexRequest for a different congestion point is accepted but produces no broker delivery"() { + when: "a signed out-of-scope FlexRequest (different EAN) is POSTed" + Response response = postSignedAsDso(flexRequestXml("ean.999999999999999999")) + + then: "the transport call still succeeds (the signed envelope is accepted)" + response.statusInfo.family == Response.Status.Family.SUCCESSFUL + response.close() + + and: "no message is ever delivered to the broker" + // The drop happens before any reply is scheduled; allow the (zero-delay) scheduler to run first. + Thread.sleep(1000) + wireMock.verify(0, postRequestedFor(urlPathEqualTo(BROKER_PATH))) + } + + def "malformed transport XML is rejected with 400"() { + when: "a body that is not a SignedMessage envelope is POSTed" + Response response = postXml("") + + then: "the endpoint maps the failure to 400 Bad Request" + response.status == 400 + response.close() + + and: "nothing is delivered to the broker" + wireMock.verify(0, postRequestedFor(urlPathEqualTo(BROKER_PATH))) + } + + def "a correctly-formed message whose signature does not match the resolved public key is rejected with 400"() { + given: "the address book returns a public key that does not match the DSO signing key" + def lazySodium = new LazySodiumJava(new SodiumJava()) + def wrongPublicKeyB64 = Base64.encoder.encodeToString(lazySodium.cryptoSignKeypair().publicKey.asBytes) + stubAddressBook(wrongPublicKeyB64) + + when: "a validly-signed FlexRequest is POSTed" + Response response = postSignedAsDso(flexRequestXml(CONTRACTED_EAN)) + + then: "signature verification fails and the endpoint returns 400 (the UftpConnectorException branch)" + response.status == 400 + response.close() + + and: "nothing is delivered to the broker" + wireMock.verify(0, postRequestedFor(urlPathEqualTo(BROKER_PATH))) + } + + def "an OAuth2 token failure (#scenario) prevents participant resolution so the inbound message is rejected with 400"() { + given: "the token endpoint is unhealthy" + wireMock.stubFor(post(urlPathEqualTo(OAUTH_PATH)).willReturn(tokenResponse)) + + when: "a validly-signed FlexRequest is POSTed" + Response response = postSignedAsDso(flexRequestXml(CONTRACTED_EAN)) + + then: "without a bearer token the DSO cannot be resolved, verification fails and the endpoint returns 400" + response.status == 400 + response.close() + + and: "nothing is delivered to the broker" + wireMock.verify(0, postRequestedFor(urlPathEqualTo(BROKER_PATH))) + + where: + scenario | tokenResponse + "500 error" | aResponse().withStatus(500) + "invalid token JSON" | okJson("not-json") + } + + def "an address-book 404 for the sender domain is rejected with 400"() { + given: "the address book does not know the DSO" + wireMock.stubFor(get(urlPathEqualTo(ADDRESS_BOOK_PATH)).willReturn(aResponse().withStatus(404))) + + when: "a validly-signed FlexRequest is POSTed" + Response response = postSignedAsDso(flexRequestXml(CONTRACTED_EAN)) + + then: "the sender cannot be resolved, verification fails and the endpoint returns 400" + response.status == 400 + response.close() + + and: "nothing is delivered to the broker" + wireMock.verify(0, postRequestedFor(urlPathEqualTo(BROKER_PATH))) + } + + def "the resolved participant is cached so a second inbound message does not trigger a second address-book lookup"() { + given: + def conditions = new PollingConditions(timeout: 10, delay: 0.2) + + when: "two signed in-scope FlexRequests are POSTed" + postSignedAsDso(flexRequestXml(CONTRACTED_EAN)).close() + conditions.eventually { + wireMock.verify(getRequestedFor(urlPathEqualTo(ADDRESS_BOOK_PATH))) + } + postSignedAsDso(flexRequestXml(CONTRACTED_EAN)).close() + // Let any (zero-delay) second lookup happen before asserting it did not. + Thread.sleep(1000) + + then: "the address book is queried exactly once -- the second message uses the cached participant" + wireMock.verify(1, getRequestedFor(urlPathEqualTo(ADDRESS_BOOK_PATH))) + } + + // ---- WireMock stubs ---- + + private void stubOAuthToken(String token) { + wireMock.stubFor(post(urlPathEqualTo(OAUTH_PATH)).willReturn( + okJson("""{"access_token":"${token}","token_type":"Bearer","expires_in":3600,"scope":"uftp"}"""))) + } + + private void stubAddressBook(String publicKeyB64) { + wireMock.stubFor(get(urlPathEqualTo(ADDRESS_BOOK_PATH)).willReturn( + okJson("""{"domain":"${DSO_DOMAIN}","publicKey":"${publicKeyB64}"}"""))) + } + + private void stubBrokerAccepts() { + wireMock.stubFor(post(urlPathEqualTo(BROKER_PATH)).willReturn(aResponse().withStatus(200))) + } + + // ---- HTTP helpers ---- + + // Signs the payload XML as the DSO and POSTs the resulting transport XML to the deployed endpoint. + private Response postSignedAsDso(String payloadXml) { + def sender = new UftpParticipant(DSO_DOMAIN, USEFRoleType.DSO) + SignedMessage signed = handler.cryptoService.signMessage(payloadXml, sender, dsoPrivateKeyB64) + return postXml(GOPACSHandler.serializer.toXml(signed)) + } + + // The broker receives SignedMessage envelopes carrying the payload in the base64 "Body" attribute. The + // decoded bytes are the libsodium-signed payload (a 64-byte signature prefixed to the payload XML), so the + // outgoing message type is still a substring of the decoded text and can be asserted on the wire. + private List decodedBrokerPayloads() { + wireMock.findAll(postRequestedFor(urlPathEqualTo(BROKER_PATH))).collect { req -> + def matcher = (req.bodyAsString =~ /Body="([^"]+)"/) + matcher ? new String(Base64.decoder.decode(matcher[0][1] as String)) : "" + } + } + + private Response postXml(String transportXml) { + // The deployment registers a realm-path-extractor filter (realmIndex 0), so the realm is the first + // segment after the /gopacs context path: /gopacs/{realm}/message -> resource @Path("message"). + createClient(null) + .target(serverUri(serverPort).path("gopacs").path(MASTER_REALM).path("message")) + .request() + .post(Entity.entity(transportXml, "text/xml")) + } + + // ---- Embedded UFTP payload fixtures (attribute-style XML, matching the example message format) ---- + + private static String flexRequestXml(String congestionPoint) { + """ + + + +""" + } + + private static String flexOrderXml(String congestionPoint, List powers) { + def isps = (0.. + """ """ + }.join("\n") + """ + +${isps} +""" + } +} From c921483ae4242227840bf4e2a9dafb1acfc13a2e Mon Sep 17 00:00:00 2001 From: Michael Date: Mon, 29 Jun 2026 10:58:37 +0200 Subject: [PATCH 09/11] Fix GOPACS congestion-point scope check to ignore "ean." prefix GOPACS flex messages carry the congestion point with an "ean." scheme prefix (e.g. "ean.265987182507322951"), while the contracted EAN is configured on the asset as the bare code. The exact equals() comparison in isWithinContractedScope therefore never matched, so in-scope flex messages were rejected as out-of-scope and silently dropped. Normalise both sides by stripping the optional "ean." prefix before comparing, making the scope check agnostic to whether either value includes it. --- .../ems/manager/gopacs/GOPACSHandler.java | 21 ++++++++++++++++++- 1 file changed, 20 insertions(+), 1 deletion(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java index 15cfc2d..f0c5580 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java @@ -100,6 +100,8 @@ public class GOPACSHandler implements UftpPayloadHandler, UftpParticipantService public static final String GOPACS_FLEX_OFFER_DELAY_SECONDS = "GOPACS_FLEX_OFFER_DELAY_SECONDS"; public static final String DEFAULT_GOPACS_FLEX_OFFER_DELAY_SECONDS = "30"; public static final String DEPLOYMENT_PATH = "/gopacs"; + /** Scheme prefix GOPACS uses on congestion-point identifiers, e.g. "ean.265987182507322951". */ + public static final String EAN_PREFIX = "ean."; protected static final UftpSerializer serializer = new UftpSerializer(new XmlSerializer(), new XsdValidator(new XsdSchemaProvider(new XsdFactory(new XsdSchemaFactoryPool())))); @@ -507,7 +509,7 @@ protected long calculatePower(FlexRequestISPType requestIsp) { * or outbound response. See issue #28 for full per-contract/role scoping via the V3 contracts endpoint. */ protected boolean isWithinContractedScope(String messageType, String conversationId, String congestionPoint) { - if (contractedEAN.equals(congestionPoint)) { + if (Objects.equals(normaliseEan(contractedEAN), normaliseEan(congestionPoint))) { return true; } LOG.warning("Rejecting " + messageType + " " + conversationId + " for out-of-scope congestion point " @@ -515,6 +517,23 @@ protected boolean isWithinContractedScope(String messageType, String conversatio return false; } + /** + * Normalises an EAN / congestion-point identifier for comparison by stripping the optional "ean." + * scheme prefix. GOPACS flex messages carry the congestion point as "ean.<code>" (for example + * "ean.265987182507322951"), whereas the contracted EAN may be configured with or without the prefix. + * Comparing the normalised forms keeps the scope check agnostic to whether either side includes it. + */ + protected static String normaliseEan(String ean) { + if (ean == null) { + return null; + } + String trimmed = ean.trim(); + if (trimmed.regionMatches(true, 0, EAN_PREFIX, 0, EAN_PREFIX.length())) { + return trimmed.substring(EAN_PREFIX.length()); + } + return trimmed; + } + protected void processRawMessage(String transportXml) { try { SignedMessage signedMessage = serializer.fromSignedXml(transportXml); From 1b655bd935035cfa0b55ffe7523ac14831bc6979 Mon Sep 17 00:00:00 2001 From: Michael Date: Mon, 29 Jun 2026 11:00:16 +0200 Subject: [PATCH 10/11] Fix GOPACS congestion-point scope check to ignore "ean." prefix GOPACS flex messages carry the congestion point in the canonical format ean.[0-9]{18} (e.g. "ean.265987182507322951"), while the contracted EAN is configured on the asset as the bare code. The exact equals() comparison in isWithinContractedScope therefore never matched, so in-scope flex messages were rejected as out-of-scope and silently dropped. Canonicalise both sides to the documented ean. congestion-point format before comparing, so the scope check is correct whether or not the contracted EAN is configured with the prefix. --- .../ems/manager/gopacs/GOPACSHandler.java | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java index f0c5580..2a6c875 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java @@ -509,7 +509,7 @@ protected long calculatePower(FlexRequestISPType requestIsp) { * or outbound response. See issue #28 for full per-contract/role scoping via the V3 contracts endpoint. */ protected boolean isWithinContractedScope(String messageType, String conversationId, String congestionPoint) { - if (Objects.equals(normaliseEan(contractedEAN), normaliseEan(congestionPoint))) { + if (Objects.equals(toCongestionPoint(contractedEAN), toCongestionPoint(congestionPoint))) { return true; } LOG.warning("Rejecting " + messageType + " " + conversationId + " for out-of-scope congestion point " @@ -518,20 +518,21 @@ protected boolean isWithinContractedScope(String messageType, String conversatio } /** - * Normalises an EAN / congestion-point identifier for comparison by stripping the optional "ean." - * scheme prefix. GOPACS flex messages carry the congestion point as "ean.<code>" (for example - * "ean.265987182507322951"), whereas the contracted EAN may be configured with or without the prefix. - * Comparing the normalised forms keeps the scope check agnostic to whether either side includes it. + * Converts an EAN / congestion-point identifier to the canonical GOPACS congestion-point format + * "ean.<code>" (for example "ean.265987182507322951"). GOPACS flex messages always carry the + * congestion point with the "ean." prefix, whereas the contracted EAN may be configured with or + * without it; canonicalising both sides keeps the scope check correct either way. */ - protected static String normaliseEan(String ean) { + protected static String toCongestionPoint(String ean) { if (ean == null) { return null; } String trimmed = ean.trim(); if (trimmed.regionMatches(true, 0, EAN_PREFIX, 0, EAN_PREFIX.length())) { - return trimmed.substring(EAN_PREFIX.length()); + // Already prefixed (any case) — normalise the prefix to its canonical lower-case form. + return EAN_PREFIX + trimmed.substring(EAN_PREFIX.length()); } - return trimmed; + return EAN_PREFIX + trimmed; } protected void processRawMessage(String transportXml) { From 5285712d7d991bb880e6e705c52b48fa11a653cd Mon Sep 17 00:00:00 2001 From: Michael Date: Mon, 29 Jun 2026 11:07:46 +0200 Subject: [PATCH 11/11] Add unit tests for GOPACS EAN congestion-point canonicalisation Covers toCongestionPoint directly (prefix present/absent, mixed case, surrounding whitespace, empty and null), and adds an end-to-end scope test proving a FlexRequest is accepted when the contracted EAN is configured without the "ean." prefix while the message carries the canonical ean. congestion point. Extracts a newHandler() helper so a second handler with a different contracted EAN can be built from the shared keypair. --- .../manager/gopacs/GOPACSHandlerTest.groovy | 53 ++++++++++++++++--- 1 file changed, 47 insertions(+), 6 deletions(-) diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy index 98bdfb0..bc145d4 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy @@ -63,6 +63,8 @@ class GOPACSHandlerTest extends Specification { TimerService timerService ScheduledExecutorService executor RecordingGOPACSHandler handler + String privKeyB64 + String pubB64 def setup() { assetProcessingService = Mock(AssetProcessingService) @@ -82,15 +84,20 @@ class GOPACSHandlerTest extends Specification { // signing secret key and the verifying public key as base64. def lazySodium = new LazySodiumJava(new SodiumJava()) KeyPair kp = lazySodium.cryptoSignKeypair() - String privKeyB64 = Base64.encoder.encodeToString(kp.secretKey.asBytes) - String pubB64 = Base64.encoder.encodeToString(kp.publicKey.asBytes) + privKeyB64 = Base64.encoder.encodeToString(kp.secretKey.asBytes) + pubB64 = Base64.encoder.encodeToString(kp.publicKey.asBytes) - handler = new RecordingGOPACSHandler(CONTRACTED_EAN, "master", ASSET_ID, - assetProcessingService, assetPredictedDatapointService, timerService, executor, privKeyB64) + handler = newHandler(CONTRACTED_EAN) + } - // Pre-seed the DSO participant so inbound signature verification never makes an HTTP call. - handler.participants.put(DSO_DOMAIN, + // Builds a handler for the given contracted EAN and pre-seeds the DSO participant so inbound + // signature verification never makes an HTTP call. + private RecordingGOPACSHandler newHandler(String contractedEan) { + def h = new RecordingGOPACSHandler(contractedEan, "master", ASSET_ID, + assetProcessingService, assetPredictedDatapointService, timerService, executor, privKeyB64) + h.participants.put(DSO_DOMAIN, new UftpParticipantInformation(DSO_DOMAIN, pubB64, "https://nilsgrid.net/shapeshifter/api/v3/message", true)) + return h } def "FlexRequest updates the asset from request ISPs and replies with FlexRequestResponse then FlexOffer"() { @@ -186,6 +193,40 @@ class GOPACSHandlerTest extends Specification { handler.sent.isEmpty() } + def "toCongestionPoint canonicalises an EAN to the GOPACS ean. format (#input -> #expected)"() { + expect: "the optional, case-insensitive ean. prefix is normalised to lower-case and added when missing" + GOPACSHandler.toCongestionPoint(input) == expected + + where: + input || expected + "ean.265987182507322951" || "ean.265987182507322951" + "265987182507322951" || "ean.265987182507322951" + "EAN.265987182507322951" || "ean.265987182507322951" + "Ean.265987182507322951" || "ean.265987182507322951" + " ean.265987182507322951 " || "ean.265987182507322951" + " 265987182507322951 " || "ean.265987182507322951" + "" || "ean." + null || null + } + + def "FlexRequest is in scope when the contracted EAN is configured without the ean. prefix"() { + given: "a handler whose contracted EAN omits the ean. prefix, and a standard prefixed FlexRequest" + def bareEan = CONTRACTED_EAN.substring(GOPACSHandler.EAN_PREFIX.length()) + def bareHandler = newHandler(bareEan) + def sender = new UftpParticipant(DSO_DOMAIN, USEFRoleType.DSO) + def signed = bareHandler.cryptoService.signMessage(flexRequestXml(CONTRACTED_EAN), sender, bareHandler.privateKey) + + when: "the signed FlexRequest is processed" + bareHandler.processRawMessage(GOPACSHandler.serializer.toXml(signed)) + + then: "it is treated as in scope: the asset is updated and both replies are sent" + 1 * assetPredictedDatapointService.updateValues(ASSET_ID, "powerMaximumFlexRequest", _) + 1 * assetPredictedDatapointService.updateValues(ASSET_ID, "powerMinimumFlexRequest", _) + bareHandler.sent.size() == 2 + bareHandler.sent[0] instanceof FlexRequestResponse + bareHandler.sent[1] instanceof FlexOffer + } + // ---- Embedded UFTP payload fixtures (attribute-style XML, matching the example message format) ---- private static String flexRequestXml(String congestionPoint) {