Merged in aslemmer-fix-wallet-monitor-service-tests (pull request #319)

Fix wallet monitor service tests

client/build.gradle

@@ -48,6 +48,8 @@ dependencies {
     // Unit testing helpers.
     testCompile 'junit:junit:4.12'
     testCompile "org.assertj:assertj-core:${assertj_version}"
+
+    testCompile project(':test-utils')
 }
 
 quasarScan.dependsOn('classes', ':core:classes', ':contracts:classes')

client/src/test/kotlin/com/r3corda/client/WalletMonitorClientTests.kt

@@ -1,7 +1,6 @@
 package com.r3corda.client
 
 import co.paralleluniverse.strands.SettableFuture
-import com.r3corda.client.testing.*
 import com.r3corda.core.contracts.*
 import com.r3corda.core.serialization.OpaqueBytes
 import com.r3corda.node.driver.driver
@@ -10,6 +9,7 @@ import com.r3corda.node.services.monitor.ServiceToClientEvent
 import com.r3corda.node.services.monitor.TransactionBuildResult
 import com.r3corda.node.services.transactions.SimpleNotaryService
 import com.r3corda.node.utilities.AddOrRemove
+import com.r3corda.testing.*
 import org.junit.Test
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
@@ -163,7 +163,7 @@ class WalletMonitorServiceTests {
                                                 is TransactionBuildResult.Failed -> fail(state.message)
                                             }
                                         },
-                                        repeat(7) {
+                                        replicate(7) {
                                             expect { build: ServiceToClientEvent.Progress -> }
                                         }
                                 ),

node/src/test/kotlin/com/r3corda/node/services/WalletMonitorServiceTests.kt

@@ -15,11 +15,16 @@ import com.r3corda.core.serialization.OpaqueBytes
 import com.r3corda.core.serialization.deserialize
 import com.r3corda.core.serialization.serialize
 import com.r3corda.core.utilities.DUMMY_NOTARY
+import com.r3corda.core.utilities.DUMMY_NOTARY_KEY
 import com.r3corda.core.utilities.DUMMY_PUBKEY_1
 import com.r3corda.testing.node.MockNetwork
 import com.r3corda.node.services.monitor.*
+import com.r3corda.node.utilities.AddOrRemove
+import com.r3corda.testing.*
 import org.junit.Before
 import org.junit.Test
+import rx.subjects.PublishSubject
+import rx.subjects.ReplaySubject
 import java.util.*
 import java.util.concurrent.TimeUnit
 import kotlin.test.*
@@ -117,11 +122,11 @@ class WalletMonitorServiceTests {
         val (monitorServiceNode, registerNode) = network.createTwoNodes()
         val sessionID = authenticate(monitorServiceNode, registerNode)
         val quantity = 1000L
-        val events = Collections.synchronizedList(ArrayList<ServiceToClientEvent>())
+        val events = ReplaySubject.create<ServiceToClientEvent>()
         val ref = OpaqueBytes(ByteArray(1) {1})
 
-        registerNode.net.addMessageHandler(WalletMonitorService.IN_EVENT_TOPIC + ".0") { msg, reg ->
+        registerNode.net.addMessageHandler(WalletMonitorService.IN_EVENT_TOPIC, sessionID) { msg, reg ->
-            events.add(msg.data.deserialize<ServiceToClientEvent>())
+            events.onNext(msg.data.deserialize<ServiceToClientEvent>())
         }
 
         // Check the monitoring service wallet is empty
@@ -140,21 +145,29 @@ class WalletMonitorServiceTests {
                 recipient.owningKey)
 
         // Check we've received a response
-        events.forEach { event ->
+        events.expectEvents {
-            when (event) {
+            parallel(
-                is ServiceToClientEvent.TransactionBuild -> {
+                    sequence(
-                    // Check the returned event is correct
+                            expect { event: ServiceToClientEvent.StateMachine ->
-                    val tx = (event.state as TransactionBuildResult.ProtocolStarted).transaction
+                                require(event.addOrRemove == AddOrRemove.ADD)
-                    assertNotNull(tx)
+                            },
-                    assertEquals(expectedState, tx!!.tx.outputs.single().data)
+                            expect { event: ServiceToClientEvent.StateMachine ->
-                }
+                                require(event.addOrRemove == AddOrRemove.REMOVE)
-                is ServiceToClientEvent.OutputState -> {
+                            }
-                    // Check the generated state is correct
+                    ),
-                    val actual = event.produced.single().state.data
+                    expect { event: ServiceToClientEvent.Transaction -> },
-                    assertEquals(expectedState, actual)
+                    expect { event: ServiceToClientEvent.TransactionBuild ->
-                }
+                        // Check the returned event is correct
-                else -> fail("Unexpected in event ${event}")
+                        val tx = (event.state as TransactionBuildResult.ProtocolStarted).transaction
-            }
+                        assertNotNull(tx)
+                        assertEquals(expectedState, tx!!.tx.outputs.single().data)
+                    },
+                    expect { event: ServiceToClientEvent.OutputState ->
+                        // Check the generated state is correct
+                        val actual = event.produced.single().state.data
+                        assertEquals(expectedState, actual)
+                    }
+            )
         }
     }
 
@@ -163,46 +176,68 @@ class WalletMonitorServiceTests {
         val (monitorServiceNode, registerNode) = network.createTwoNodes()
         val sessionID = authenticate(monitorServiceNode, registerNode)
         val quantity = 1000L
-        val events = Collections.synchronizedList(ArrayList<ServiceToClientEvent>())
+        val events = ReplaySubject.create<ServiceToClientEvent>()
-        val ref = OpaqueBytes(ByteArray(1) {1})
-        var handlerReg: MessageHandlerRegistration? = null
 
-        registerNode.net.addMessageHandler(WalletMonitorService.IN_EVENT_TOPIC + ".0") { msg, reg ->
+        registerNode.net.addMessageHandler(WalletMonitorService.IN_EVENT_TOPIC, sessionID) { msg, reg ->
-            events.add(msg.data.deserialize<ServiceToClientEvent>())
+            events.onNext(msg.data.deserialize<ServiceToClientEvent>())
-            handlerReg = reg
         }
 
-        // Check the monitoring service wallet is empty
-        assertFalse(monitorServiceNode.services.walletService.currentWallet.states.iterator().hasNext())
-
-        // Tell the monitoring service node to issue some cash
         val recipient = monitorServiceNode.services.storageService.myLegalIdentity
-        val outEvent = ClientToServiceCommand.IssueCash(Amount(quantity, GBP), ref, recipient, DUMMY_NOTARY)
+
-        val message = registerNode.net.createMessage(WalletMonitorService.OUT_EVENT_TOPIC, DEFAULT_SESSION_ID,
+        // Tell the monitoring service node to issue some cash so we can spend it later
-                ClientToServiceCommandMessage(sessionID, registerNode.net.myAddress, outEvent).serialize().bits)
+        val issueCommand = ClientToServiceCommand.IssueCash(Amount(quantity, GBP), OpaqueBytes.of(0), recipient, recipient)
-        registerNode.net.send(message, monitorServiceNode.net.myAddress)
+        val issueMessage = registerNode.net.createMessage(WalletMonitorService.OUT_EVENT_TOPIC, DEFAULT_SESSION_ID,
+                ClientToServiceCommandMessage(sessionID, registerNode.net.myAddress, issueCommand).serialize().bits)
+        registerNode.net.send(issueMessage, monitorServiceNode.net.myAddress)
+        val payCommand = ClientToServiceCommand.PayCash(Amount(quantity, Issued(recipient.ref(0), GBP)), recipient)
+        val payMessage = registerNode.net.createMessage(WalletMonitorService.OUT_EVENT_TOPIC, DEFAULT_SESSION_ID,
+                ClientToServiceCommandMessage(sessionID, registerNode.net.myAddress, payCommand).serialize().bits)
+        registerNode.net.send(payMessage, monitorServiceNode.net.myAddress)
         network.runNetwork()
 
-        val expectedState = Cash.State(Amount(quantity,
+        events.expectEvents(isStrict = false) {
-                Issued(monitorServiceNode.services.storageService.myLegalIdentity.ref(ref), GBP)),
+            sequence(
-                recipient.owningKey)
+                    // ISSUE
-
+                    parallel(
-        // Check we've received a response
+                            sequence(
-        events.forEach { event ->
+                                    expect { event: ServiceToClientEvent.StateMachine ->
-            when (event) {
+                                        require(event.addOrRemove == AddOrRemove.ADD)
-                is ServiceToClientEvent.TransactionBuild -> {
+                                    },
-                    // Check the returned event is correct
+                                    expect { event: ServiceToClientEvent.StateMachine ->
-                    val tx = (event.state as TransactionBuildResult.ProtocolStarted).transaction
+                                        require(event.addOrRemove == AddOrRemove.REMOVE)
-                    assertNotNull(tx)
+                                    }
-                    assertEquals(expectedState, tx!!.tx.outputs.single().data)
+                            ),
-                }
+                            expect { event: ServiceToClientEvent.Transaction -> },
-                is ServiceToClientEvent.OutputState -> {
+                            expect { event: ServiceToClientEvent.TransactionBuild -> },
-                    // Check the generated state is correct
+                            expect { event: ServiceToClientEvent.OutputState -> }
-                    val actual = event.produced.single().state.data
+                    ),
-                    assertEquals(expectedState, actual)
+                    // MOVE
-                }
+                    parallel(
-                else -> fail("Unexpected in event ${event}")
+                            sequence(
-            }
+                                    expect { event: ServiceToClientEvent.StateMachine ->
+                                        require(event.addOrRemove == AddOrRemove.ADD)
+                                    },
+                                    expect { event: ServiceToClientEvent.StateMachine ->
+                                        require(event.addOrRemove == AddOrRemove.REMOVE)
+                                    }
+                            ),
+                            expect { event: ServiceToClientEvent.Transaction ->
+                                require(event.transaction.sigs.size == 1)
+                                event.transaction.sigs.map { it.by }.toSet().containsAll(
+                                        listOf(
+                                                monitorServiceNode.services.storageService.myLegalIdentity.owningKey
+                                        )
+                                )
+                            },
+                            expect { event: ServiceToClientEvent.TransactionBuild ->
+                                require(event.state is TransactionBuildResult.ProtocolStarted)
+                            },
+                            expect { event: ServiceToClientEvent.OutputState ->
+                                require(event.consumed.size == 1)
+                                require(event.produced.size == 1)
+                            }
+                    )
+            )
         }
     }
 }

test-utils/src/main/kotlin/com/r3corda/testing/Expect.kt

@@ -1,4 +1,4 @@
-package com.r3corda.client.testing
+package com.r3corda.testing
 
 import co.paralleluniverse.strands.SettableFuture
 import org.slf4j.Logger
@@ -61,7 +61,7 @@ fun <E> parallel(vararg expectations: ExpectCompose<E>): ExpectCompose<E> = Expe
  * @param number The number of events expected.
  * @param expectation The piece of DSL to run on each event, with the index of the event passed in.
  */
-inline fun <E> repeat(number: Int, expectation: (Int) -> ExpectCompose<E>) = sequence(*Array(number) { expectation(it) })
+inline fun <E> replicate(number: Int, expectation: (Int) -> ExpectCompose<E>) = sequence(*Array(number) { expectation(it) })
 
 /**
  * Run the specified DSL against the event stream.