Fix node integration tests. (corda#2233)

* Add tests for node registration and communication

docs/source/example-code/src/main/kotlin/net/corda/docs/tutorial/twoparty/flow.kt

@@ -10,7 +10,6 @@ import net.corda.core.messaging.CordaRPCOps
 import net.corda.core.serialization.SerializationWhitelist
 import net.corda.core.transactions.TransactionBuilder
 import net.corda.core.utilities.ProgressTracker
-import net.corda.webserver.services.WebServerPluginRegistry
 import java.util.function.Function
 import javax.ws.rs.GET
 import javax.ws.rs.Path

node/src/integration-test/kotlin/net/corda/node/services/network/NetworkMapTest.kt

@@ -10,6 +10,7 @@ import net.corda.core.utilities.seconds
 import net.corda.nodeapi.internal.network.NETWORK_PARAMS_FILE_NAME
 import net.corda.nodeapi.internal.network.NetworkParameters
 import net.corda.testing.ALICE_NAME
+import net.corda.testing.ROOT_CA
 import net.corda.testing.BOB_NAME
 import net.corda.testing.SerializationEnvironmentRule
 import net.corda.testing.driver.NodeHandle
@@ -49,20 +50,22 @@ class NetworkMapTest {
 
     @Test
     fun `node correctly downloads and saves network parameters file on startup`() {
-        internalDriver(portAllocation = portAllocation, compatibilityZone = compatibilityZone, initialiseSerialization = false) {
+        internalDriver(portAllocation = portAllocation, compatibilityZone = compatibilityZone,
+                initialiseSerialization = false, notarySpecs = emptyList()) {
             val alice = startNode(providedName = ALICE_NAME).getOrThrow()
             val networkParameters = alice.configuration.baseDirectory
                     .list { paths -> paths.filter { it.fileName.toString() == NETWORK_PARAMS_FILE_NAME }.findFirst().get() }
                     .readAll()
                     .deserialize<SignedData<NetworkParameters>>()
                     .verified()
-            assertEquals(NetworkMapServer.stubNetworkParameter, networkParameters)
+            assertEquals(networkMapServer.networkParameters, networkParameters)
         }
     }
 
     @Test
     fun `nodes can see each other using the http network map`() {
-        internalDriver(portAllocation = portAllocation, compatibilityZone = compatibilityZone, initialiseSerialization = false) {
+        internalDriver(portAllocation = portAllocation, compatibilityZone = compatibilityZone,
+                initialiseSerialization = false, onNetworkParametersGeneration = { networkMapServer.networkParameters = it }) {
             val alice = startNode(providedName = ALICE_NAME)
             val bob = startNode(providedName = BOB_NAME)
             val notaryNode = defaultNotaryNode.get()
@@ -77,7 +80,8 @@ class NetworkMapTest {
 
     @Test
     fun `nodes process network map add updates correctly when adding new node to network map`() {
-        internalDriver(portAllocation = portAllocation, compatibilityZone = compatibilityZone, initialiseSerialization = false) {
+        internalDriver(portAllocation = portAllocation, compatibilityZone = compatibilityZone,
+                initialiseSerialization = false, onNetworkParametersGeneration = { networkMapServer.networkParameters = it }) {
             val alice = startNode(providedName = ALICE_NAME)
             val notaryNode = defaultNotaryNode.get()
             val aliceNode = alice.get()
@@ -98,7 +102,8 @@ class NetworkMapTest {
 
     @Test
     fun `nodes process network map remove updates correctly`() {
-        internalDriver(portAllocation = portAllocation, compatibilityZone = compatibilityZone, initialiseSerialization = false) {
+        internalDriver(portAllocation = portAllocation, compatibilityZone = compatibilityZone,
+                initialiseSerialization = false, onNetworkParametersGeneration = { networkMapServer.networkParameters = it }) {
             val alice = startNode(providedName = ALICE_NAME)
             val bob = startNode(providedName = BOB_NAME)
             val notaryNode = defaultNotaryNode.get()

node/src/integration-test/kotlin/net/corda/node/utilities/registration/NodeRegistrationTest.kt

@@ -3,22 +3,29 @@ package net.corda.node.utilities.registration
 import net.corda.core.crypto.Crypto
 import net.corda.core.identity.CordaX500Name
 import net.corda.core.internal.cert
+import net.corda.core.internal.concurrent.transpose
 import net.corda.core.internal.toX509CertHolder
-import net.corda.core.utilities.NetworkHostAndPort
-import net.corda.core.utilities.getOrThrow
-import net.corda.core.utilities.minutes
+import net.corda.core.messaging.startFlow
+import net.corda.core.utilities.*
+import net.corda.finance.DOLLARS
+import net.corda.finance.flows.CashIssueAndPaymentFlow
 import net.corda.nodeapi.internal.crypto.CertificateAndKeyPair
 import net.corda.nodeapi.internal.crypto.CertificateType
 import net.corda.nodeapi.internal.crypto.X509CertificateFactory
 import net.corda.nodeapi.internal.crypto.X509Utilities
 import net.corda.nodeapi.internal.crypto.X509Utilities.CORDA_CLIENT_CA
 import net.corda.nodeapi.internal.crypto.X509Utilities.CORDA_INTERMEDIATE_CA
 import net.corda.nodeapi.internal.crypto.X509Utilities.CORDA_ROOT_CA
+import net.corda.nodeapi.internal.network.NotaryInfo
+import net.corda.testing.ROOT_CA
 import net.corda.testing.SerializationEnvironmentRule
-import net.corda.testing.node.internal.CompatibilityZoneParams
 import net.corda.testing.driver.PortAllocation
+import net.corda.testing.node.NotarySpec
+import net.corda.testing.node.internal.CompatibilityZoneParams
 import net.corda.testing.node.internal.internalDriver
 import net.corda.testing.node.internal.network.NetworkMapServer
+import net.corda.testing.singleIdentity
+import net.corda.testing.singleIdentityAndCert
 import org.assertj.core.api.Assertions.assertThat
 import org.assertj.core.api.Assertions.assertThatThrownBy
 import org.bouncycastle.pkcs.PKCS10CertificationRequest
@@ -44,15 +51,13 @@ class NodeRegistrationTest {
     @JvmField
     val testSerialization = SerializationEnvironmentRule(true)
     private val portAllocation = PortAllocation.Incremental(13000)
-    private val rootCertAndKeyPair = createSelfKeyAndSelfSignedCertificate()
-    private val registrationHandler = RegistrationHandler(rootCertAndKeyPair)
-
+    private val registrationHandler = RegistrationHandler(ROOT_CA)
     private lateinit var server: NetworkMapServer
     private lateinit var serverHostAndPort: NetworkHostAndPort
 
     @Before
     fun startServer() {
-        server = NetworkMapServer(1.minutes, portAllocation.nextHostAndPort(), rootCertAndKeyPair, registrationHandler)
+        server = NetworkMapServer(1.minutes, portAllocation.nextHostAndPort(), ROOT_CA, "localhost", registrationHandler)
         serverHostAndPort = server.start()
     }
 
@@ -61,19 +66,47 @@ class NodeRegistrationTest {
         server.close()
     }
 
-    // TODO Ideally this test should be checking that two nodes that register are able to transact with each other. However
-    // starting a second node hangs so that needs to be fixed.
     @Test
     fun `node registration correct root cert`() {
-        val compatibilityZone = CompatibilityZoneParams(URL("http://$serverHostAndPort"), rootCert = rootCertAndKeyPair.certificate.cert)
+        val compatibilityZone = CompatibilityZoneParams(URL("http://$serverHostAndPort"), rootCert = ROOT_CA.certificate.cert)
         internalDriver(
                 portAllocation = portAllocation,
-                notarySpecs = emptyList(),
                 compatibilityZone = compatibilityZone,
-                initialiseSerialization = false
+                initialiseSerialization = false,
+                notarySpecs = listOf(NotarySpec(CordaX500Name(organisation = "NotaryService", locality = "Zurich", country = "CH"), validating = false)),
+                extraCordappPackagesToScan = listOf("net.corda.finance"),
+                onNetworkParametersGeneration = { server.networkParameters = it }
         ) {
-            startNode(providedName = CordaX500Name("Alice", "London", "GB")).getOrThrow()
-            assertThat(registrationHandler.idsPolled).contains("Alice")
+            val notary = defaultNotaryNode.get()
+
+            val ALICE_NAME = "Alice"
+            val GENEVIEVE_NAME = "Genevieve"
+            val nodesFutures = listOf(startNode(providedName = CordaX500Name(ALICE_NAME, "London", "GB")),
+                    startNode(providedName = CordaX500Name(GENEVIEVE_NAME, "London", "GB")))
+
+            val (alice, genevieve) = nodesFutures.transpose().get()
+            val nodes = listOf(alice, genevieve, notary)
+
+            assertThat(registrationHandler.idsPolled).contains(ALICE_NAME, GENEVIEVE_NAME)
+            // Notary identities are generated beforehand hence notary nodes don't go through registration.
+            // This test isn't specifically testing this, or relying on this behavior, though if this check fail,
+            // this will probably lead to the rest of the test to fail.
+            assertThat(registrationHandler.idsPolled).doesNotContain("NotaryService")
+
+            // Check each node has each other identity in their network map cache.
+            val nodeIdentities = nodes.map { it.nodeInfo.singleIdentity() }
+            for (node in nodes) {
+                assertThat(node.rpc.networkMapSnapshot().map { it.singleIdentity() }).containsAll(nodeIdentities)
+            }
+
+            // Check we nodes communicate among themselves (and the notary).
+            val anonymous = false
+            genevieve.rpc.startFlow(::CashIssueAndPaymentFlow, 1000.DOLLARS, OpaqueBytes.of(12),
+                    alice.nodeInfo.singleIdentity(),
+                    anonymous,
+                    notary.nodeInfo.singleIdentity())
+                    .returnValue
+                    .getOrThrow()
         }
     }
 
@@ -85,6 +118,7 @@ class NodeRegistrationTest {
                 portAllocation = portAllocation,
                 notarySpecs = emptyList(),
                 compatibilityZone = compatibilityZone,
+                initialiseSerialization = false,
                 // Changing the content of the truststore makes the node fail in a number of ways if started out process.
                 startNodesInProcess = true
         ) {

node/src/test/kotlin/net/corda/node/services/network/NetworkMapClientTest.kt

@@ -5,10 +5,7 @@ import net.corda.core.crypto.sha256
 import net.corda.core.internal.cert
 import net.corda.core.serialization.serialize
 import net.corda.core.utilities.seconds
-import net.corda.testing.ALICE_NAME
-import net.corda.testing.BOB_NAME
-import net.corda.testing.DEV_TRUST_ROOT
-import net.corda.testing.SerializationEnvironmentRule
+import net.corda.testing.*
 import net.corda.testing.driver.PortAllocation
 import net.corda.testing.internal.createNodeInfoAndSigned
 import net.corda.testing.node.internal.network.NetworkMapServer
@@ -33,7 +30,7 @@ class NetworkMapClientTest {
 
     @Before
     fun setUp() {
-        server = NetworkMapServer(cacheTimeout, PortAllocation.Incremental(10000).nextHostAndPort())
+        server = NetworkMapServer(cacheTimeout, PortAllocation.Incremental(10000).nextHostAndPort(), root_ca = ROOT_CA)
         val hostAndPort = server.start()
         networkMapClient = NetworkMapClient(URL("http://${hostAndPort.host}:${hostAndPort.port}"), DEV_TRUST_ROOT.cert)
     }
@@ -70,7 +67,7 @@ class NetworkMapClientTest {
         // The test server returns same network parameter for any hash.
         val networkParameter = networkMapClient.getNetworkParameter(SecureHash.randomSHA256())?.verified()
         assertNotNull(networkParameter)
-        assertEquals(NetworkMapServer.stubNetworkParameter, networkParameter)
+        assertEquals(server.networkParameters, networkParameter)
     }
 
     @Test

testing/node-driver/src/main/kotlin/net/corda/testing/driver/Driver.kt

@@ -193,7 +193,8 @@ fun <A> driver(
                     notarySpecs = notarySpecs,
                     extraCordappPackagesToScan = extraCordappPackagesToScan,
                     jmxPolicy = jmxPolicy,
-                    compatibilityZone = null
+                    compatibilityZone = null,
+                    onNetworkParametersGeneration = { }
             ),
             coerce = { it },
             dsl = dsl,

testing/node-driver/src/main/kotlin/net/corda/testing/node/internal/DriverDSLImpl.kt

@@ -10,13 +10,12 @@ import net.corda.cordform.CordformContext
 import net.corda.cordform.CordformNode
 import net.corda.core.concurrent.CordaFuture
 import net.corda.core.concurrent.firstOf
+import net.corda.core.crypto.generateKeyPair
 import net.corda.core.crypto.random63BitValue
 import net.corda.core.identity.CordaX500Name
-import net.corda.core.internal.ThreadBox
+import net.corda.core.identity.Party
+import net.corda.core.internal.*
 import net.corda.core.internal.concurrent.*
-import net.corda.core.internal.copyTo
-import net.corda.core.internal.createDirectories
-import net.corda.core.internal.div
 import net.corda.core.messaging.CordaRPCOps
 import net.corda.core.node.services.NetworkMapCache
 import net.corda.core.toFuture
@@ -32,6 +31,7 @@ import net.corda.node.services.config.*
 import net.corda.node.utilities.registration.HTTPNetworkRegistrationService
 import net.corda.node.utilities.registration.NetworkRegistrationHelper
 import net.corda.nodeapi.internal.IdentityGenerator
+import net.corda.nodeapi.internal.IdentityGenerator.NODE_IDENTITY_ALIAS_PREFIX
 import net.corda.nodeapi.internal.addShutdownHook
 import net.corda.nodeapi.internal.config.User
 import net.corda.nodeapi.internal.config.parseAs
@@ -46,6 +46,8 @@ import net.corda.nodeapi.internal.network.NotaryInfo
 import net.corda.testing.ALICE_NAME
 import net.corda.testing.BOB_NAME
 import net.corda.testing.DUMMY_BANK_A_NAME
+import net.corda.nodeapi.internal.crypto.*
+import net.corda.nodeapi.internal.network.NetworkParameters
 import net.corda.testing.common.internal.testNetworkParameters
 import net.corda.testing.driver.*
 import net.corda.testing.node.ClusterSpec
@@ -89,7 +91,8 @@ class DriverDSLImpl(
         extraCordappPackagesToScan: List<String>,
         val jmxPolicy: JmxPolicy,
         val notarySpecs: List<NotarySpec>,
-        val compatibilityZone: CompatibilityZoneParams?
+        val compatibilityZone: CompatibilityZoneParams?,
+        val onNetworkParametersGeneration: (NetworkParameters) -> Unit
 ) : InternalDriverDSL {
     private var _executorService: ScheduledExecutorService? = null
     val executorService get() = _executorService!!
@@ -104,7 +107,8 @@ class DriverDSLImpl(
     private val countObservables = mutableMapOf<CordaX500Name, Observable<Int>>()
     private lateinit var _notaries: List<NotaryHandle>
     override val notaryHandles: List<NotaryHandle> get() = _notaries
-    private var networkParameters: NetworkParametersCopier? = null
+    private var networkParametersCopier: NetworkParametersCopier? = null
+    lateinit var networkParameters: NetworkParameters
 
     class State {
         val processes = ArrayList<Process>()
@@ -179,8 +183,8 @@ class DriverDSLImpl(
         val p2pAddress = portAllocation.nextHostAndPort()
         // TODO: Derive name from the full picked name, don't just wrap the common name
         val name = providedName ?: CordaX500Name(organisation = "${oneOf(names).organisation}-${p2pAddress.port}", locality = "London", country = "GB")
-
-        val registrationFuture = if (compatibilityZone?.rootCert != null) {
+        val isNotary =  name in notarySpecs.map { it.name }
+        val registrationFuture = if (compatibilityZone?.rootCert != null && !isNotary) {
             nodeRegistration(name, compatibilityZone.rootCert, compatibilityZone.url)
         } else {
             doneFuture(Unit)
@@ -279,7 +283,9 @@ class DriverDSLImpl(
             notaryInfos += NotaryInfo(identity, type.validating)
         }
 
-        networkParameters = NetworkParametersCopier(testNetworkParameters(notaryInfos))
+        networkParameters = testNetworkParameters(notaryInfos)
+        onNetworkParametersGeneration(networkParameters)
+        networkParametersCopier = NetworkParametersCopier(networkParameters)
 
         return cordforms.map {
             val startedNode = startCordformNode(it)
@@ -347,8 +353,10 @@ class DriverDSLImpl(
             }
         }
         val notaryInfos = generateNotaryIdentities()
+        networkParameters = testNetworkParameters(notaryInfos)
+        onNetworkParametersGeneration(networkParameters)
         // The network parameters must be serialised before starting any of the nodes
-        if (compatibilityZone == null) networkParameters = NetworkParametersCopier(testNetworkParameters(notaryInfos))
+        networkParametersCopier = NetworkParametersCopier(networkParameters)
         val nodeHandles = startNotaries()
         _notaries = notaryInfos.zip(nodeHandles) { (identity, validating), nodes -> NotaryHandle(identity, validating, nodes) }
     }
@@ -504,7 +512,11 @@ class DriverDSLImpl(
         val configuration = config.parseAsNodeConfiguration()
         val baseDirectory = configuration.baseDirectory.createDirectories()
         nodeInfoFilesCopier?.addConfig(baseDirectory)
-        networkParameters?.install(baseDirectory)
+        if (configuration.myLegalName in networkParameters.notaries.map { it.identity.name } || compatibilityZone == null) {
+            // If a notary is being started, then its identity appears in networkParameters (generated by Driver),
+            // and Driver itself must pass the network parameters to the notary.
+            networkParametersCopier?.install(baseDirectory)
+        }
         val onNodeExit: () -> Unit = {
             nodeInfoFilesCopier?.removeConfig(baseDirectory)
             countObservables.remove(configuration.myLegalName)
@@ -817,7 +829,8 @@ fun <DI : DriverDSL, D : InternalDriverDSL, A> genericDriver(
                     extraCordappPackagesToScan = extraCordappPackagesToScan,
                     jmxPolicy = jmxPolicy,
                     notarySpecs = notarySpecs,
-                    compatibilityZone = null
+                    compatibilityZone = null,
+                    onNetworkParametersGeneration = {}
             )
     )
     val shutdownHook = addShutdownHook(driverDsl::shutdown)
@@ -855,6 +868,7 @@ fun <A> internalDriver(
         extraCordappPackagesToScan: List<String> = DriverParameters().extraCordappPackagesToScan,
         jmxPolicy: JmxPolicy = DriverParameters().jmxPolicy,
         compatibilityZone: CompatibilityZoneParams? = null,
+        onNetworkParametersGeneration: (NetworkParameters) -> Unit = {},
         dsl: DriverDSLImpl.() -> A
 ): A {
     return genericDriver(
@@ -870,7 +884,8 @@ fun <A> internalDriver(
                     notarySpecs = notarySpecs,
                     extraCordappPackagesToScan = extraCordappPackagesToScan,
                     jmxPolicy = jmxPolicy,
-                    compatibilityZone = compatibilityZone
+                    compatibilityZone = compatibilityZone,
+                    onNetworkParametersGeneration = onNetworkParametersGeneration
             ),
             coerce = { it },
             dsl = dsl,

testing/node-driver/src/main/kotlin/net/corda/testing/node/internal/RPCDriver.kt

@@ -121,7 +121,8 @@ fun <A> rpcDriver(
                             extraCordappPackagesToScan = extraCordappPackagesToScan,
                             notarySpecs = notarySpecs,
                             jmxPolicy = jmxPolicy,
-                            compatibilityZone = null
+                            compatibilityZone = null,
+                            onNetworkParametersGeneration = {}
                     ), externalTrace
             ),
             coerce = { it },