diff --git a/test/src/test/groovy/org/openremote/test/gateway/GatewayTest.groovy b/test/src/test/groovy/org/openremote/test/gateway/GatewayTest.groovy index b007b01be0..c3d536e334 100644 --- a/test/src/test/groovy/org/openremote/test/gateway/GatewayTest.groovy +++ b/test/src/test/groovy/org/openremote/test/gateway/GatewayTest.groovy @@ -61,12 +61,14 @@ import org.openremote.model.value.ValueFormat import org.openremote.setup.integration.ManagerTestSetup import org.openremote.test.ManagerContainerTrait import spock.lang.Ignore +import spock.lang.Shared import spock.lang.Specification import spock.util.concurrent.PollingConditions import javax.net.ssl.SSLSession import java.nio.file.Paths import java.time.Duration +import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.CopyOnWriteArrayList import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicBoolean @@ -86,6 +88,9 @@ import static org.openremote.model.value.ValueType.* class GatewayTest extends Specification implements ManagerContainerTrait { + @Shared + Map gatewayDisconnectReasons = new ConcurrentHashMap<>() + def setupSpec() { // Mock the GatewayClientService so we can use a different port for authentication on the GatewayClientConnector def services = defaultServices().toList() @@ -102,7 +107,17 @@ class GatewayTest extends Specification implements ManagerContainerTrait { callRealMethod() } - // Start the container with the spy service + // Observe gateway rejections before the asynchronous disconnect message races with socket closure. + def clientEventServiceIndex = services.findIndexOf {it instanceof ClientEventService} + def clientEventService = services.removeAt(clientEventServiceIndex) as ClientEventService + def spyClientEventService = Spy(clientEventService) + services.add(clientEventServiceIndex, spyClientEventService) + spyClientEventService.sendToWebsocketSession(_ as String, _ as GatewayDisconnectEvent) >> { String sessionId, GatewayDisconnectEvent event -> + callRealMethod() + gatewayDisconnectReasons.put(sessionId, event.reason) + } + + // Start the container with the spy services startContainer(defaultConfig(), services) } @@ -115,6 +130,7 @@ class GatewayTest extends Specification implements ManagerContainerTrait { def assetStorageService = container.getService(AssetStorageService.class) def agentService = container.getService(AgentService.class) def gatewayService = container.getService(GatewayService.class) + def clientEventService = container.getService(ClientEventService.class) def managerTestSetup = container.getService(SetupService.class).getTaskOfType(ManagerTestSetup.class) def identityProvider = container.getService(ManagerIdentityService.class).identityProvider as ManagerKeycloakIdentityProvider @@ -684,30 +700,47 @@ class GatewayTest extends Specification implements ManagerContainerTrait { [new AbstractNettyIOClient.MessageToMessageDecoder(String.class, rogueGatewayClient)].toArray(new ChannelHandler[0]) }) - and: "we add callback consumers to the rogue client" - List rogueConnectionStatuses = new CopyOnWriteArrayList<>() + and: "we observe the manager's rejection of the rogue session" + gatewayDisconnectReasons.clear() + def registeredGatewayConnector = gatewayService.gatewayConnectorMap.get(gateway.getId().toLowerCase(Locale.ROOT)) + def legitimateSessionId = registeredGatewayConnector.getSessionId() + def legitimateDisconnect = registeredGatewayConnector.requestDisconnect + assert legitimateSessionId != null + + and: "we record messages received by the rogue client" List rogueClientReceivedMessages = new CopyOnWriteArrayList<>() rogueGatewayClient.addMessageConsumer({message -> getLOG().info("Message received from central manager: $message") rogueClientReceivedMessages.add(message as String) }) - rogueGatewayClient.addConnectionStatusConsumer({status -> - getLOG().info("Rouge gateway client connection staus changed: $status") - rogueConnectionStatuses.add(status) - }) and: "the rogue client connects" rogueGatewayClient.connect() - then: "the rogue netty client status should continually connect as legitimate gateway client ALREADY_CONNECTED" + then: "the manager rejects the rogue session and finishes processing its closure" + // Removal from sessionStates follows both the session open and close handlers. conditions.eventually { - assert rogueConnectionStatuses.count(ConnectionStatus.CONNECTING) > 5 - assert rogueConnectionStatuses.count(ConnectionStatus.CONNECTED) > 5 + assert gatewayDisconnectReasons.any { sessionId, reason -> + reason == GatewayDisconnectEvent.Reason.UNRECOGNISED && !clientEventService.sessionStates.containsKey(sessionId) + } } - then: "the rogue client is disconnected" + and: "the legitimate gateway session is preserved and the rogue client cannot start synchronization" + assert registeredGatewayConnector.getSessionId() == legitimateSessionId + assert registeredGatewayConnector.requestDisconnect.is(legitimateDisconnect) + assert gatewayClient.connectionStatus == ConnectionStatus.CONNECTED + assert assetStorageService.find(gateway.id).getGatewayStatus().orElse(null) == ConnectionStatus.CONNECTED + assert !clientReceivedMessages.any {it.contains(GatewayDisconnectEvent.TYPE)} + assert !rogueClientReceivedMessages.any {it.contains(GatewayInitStartEvent.TYPE)} + + when: "the rogue client is disconnected" rogueGatewayClient.disconnect() + then: "all its session events are processed before testing another connection attempt" + conditions.eventually { + assert clientEventService.sessionStates.keySet() == [legitimateSessionId] as Set + } + when: "time advances" advancePseudoClock(1, TimeUnit.SECONDS, container) @@ -765,18 +798,24 @@ class GatewayTest extends Specification implements ManagerContainerTrait { } when: "the rogue client connects now the legitimate gateway client is disconnected" - rogueConnectionStatuses.clear() + gatewayDisconnectReasons.clear() + rogueClientReceivedMessages.clear() rogueGatewayClient.connect() - then: "the gateway connector disconnect runnable should not be set" - assert gatewayService.gatewayConnectorMap.get(gateway.getId().toLowerCase(Locale.ROOT)).requestDisconnect == null - - and: "the gateway netty client status should still continually connect as it's rogue" + then: "the manager rejects the new rogue session and finishes processing its closure" conditions.eventually { - assert rogueConnectionStatuses.count(ConnectionStatus.CONNECTING) > 5 - assert rogueConnectionStatuses.count(ConnectionStatus.CONNECTED) > 5 + assert gatewayDisconnectReasons.any { sessionId, reason -> + reason == GatewayDisconnectEvent.Reason.UNRECOGNISED && !clientEventService.sessionStates.containsKey(sessionId) + } } + and: "the rogue client cannot establish a gateway session or start synchronization" + assert registeredGatewayConnector.getSessionId() == null + assert !registeredGatewayConnector.connected + assert registeredGatewayConnector.requestDisconnect == null + assert assetStorageService.find(gateway.id).getGatewayStatus().orElse(null) == ConnectionStatus.DISCONNECTED + assert !rogueClientReceivedMessages.any {it.contains(GatewayInitStartEvent.TYPE)} + when: "the rogue client is disconnected" rogueGatewayClient.disconnect()