@Test
public void inPortHasSignal() {

    try (ModuleInterpretation moduleInterpretation = testCaseBuilder("inPortHasSignal", PROXY_CONTEXT).build()) {
        MapStagingArea stagingArea = moduleInterpretation.getStagingArea();
        Map<ExecutionTrace, ObjectNode> stagingAreaMap = stagingArea.toUnmodifiableMap();
        put(stagingArea, ":in:x", ":in:y");
        int moduleId = 3;
        Props props = moduleInterpretation.toCompositeModuleInterpreterProps(moduleId, bitSetOf(0), bitSetOf(0, 1));
        ActorSystem actorSystem = moduleInterpretation.getActorSystem();
        TestActorRef<CompositeModuleInterpreterActor> actorRef
            = TestActorRef.create(actorSystem, props, moduleInterpretation.getSupervisor());
        TestProbe akkaEventProbe = new TestProbe(actorSystem);
        actorSystem.eventStream().subscribe(akkaEventProbe.ref(), Object.class);
        // Verify that we see a warning if the InPortHasSignal message is sent twice
        actorRef.receive(new InPortHasSignal(0));
        actorRef.receive(new InPortHasSignal(0));
        LogEvent redundantMsgLogEvent = akkaEventProbe.expectMsgClass(LogEvent.class);
        Assert.assertEquals(redundantMsgLogEvent.logSource(), actorRef.path().toString());
        String redundantMsg = (String) redundantMsgLogEvent.message();
        Assert.assertTrue(redundantMsg.contains("redundant"));
        Assert.assertTrue(redundantMsg.contains(PROXY_CONTEXT.getModule().getInPorts().get(0).toString()));
        // Empty the executor queue, and subsequently verify that the in-port values have been copied and that the
        // parent was informed about the new out-port value.
        CallingThreadExecutor asyncTaskExecutor = moduleInterpretation.getAsyncTaskExecutor();
        asyncTaskExecutor.executeAll();
        TestProbe parentProbe = moduleInterpretation.getParentProbe();
        Assert.assertEquals(
            parentProbe.expectMsgClass(Object.class),
            new SubmoduleOutPortHasSignal(moduleId, 1)
        );
        Assert.assertFalse(parentProbe.msgAvailable());
        Assert.assertFalse(akkaEventProbe.msgAvailable());
        Assert.assertEquals(
            stagingAreaMap.keySet(),
            tracesSet(":in:x", ":in:y", "/a:in:ax", "/a:in:ay", "/b:in:bx", ":out:q")
        );
        // Verify that messages have been sent to the child interpreters
        Assert.assertEquals(
            IntStream.range(0, 2)
                .mapToObj(
                    id -> moduleInterpretation.getTestProbeForChild("a").expectMsgClass(InPortHasSignal.class)
                )
                .mapToInt(InPortHasSignal::getInPortId)
                .collect(BitSet::new, BitSet::set, BitSet::or),
            bitSetOf(0, 1)
        );
        Assert.assertFalse(moduleInterpretation.getTestProbeForChild("a").msgAvailable());
        moduleInterpretation.getTestProbeForChild("b").expectMsg(new InPortHasSignal(0));
        Assert.assertFalse(moduleInterpretation.getTestProbeForChild("b").msgAvailable());
        Assert.assertFalse(moduleInterpretation.getTestProbeForChild("c").msgAvailable());
        // Verify that we see a warning if the InPortHasSignal message message is sent about an unexpected port
        // that was not included in the set of requested in-ports
        actorRef.receive(new InPortHasSignal(1));
        LogEvent unexpectedMsgLogEvent = akkaEventProbe.expectMsgClass(LogEvent.class);
        Assert.assertEquals(unexpectedMsgLogEvent.logSource(), actorRef.path().toString());
        String unexpectedMsg = (String) unexpectedMsgLogEvent.message();
        Assert.assertTrue(unexpectedMsg.contains("unexpected"));
        Assert.assertTrue(unexpectedMsg.contains(PROXY_CONTEXT.getModule().getInPorts().get(1).toString()));
    }
}