ms/blueprintsprocessor/functions/netconf-executor/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfDeviceCommunicatorTest.kt

   1 /*
   2  * Copyright © 2019 Bell Canada
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 package org.onap.ccsdk.cds.blueprintsprocessor.functions.netconf.executor.core
  18
  19 import io.mockk.CapturingSlot
  20 import io.mockk.Runs
  21 import io.mockk.every
  22 import io.mockk.just
  23 import io.mockk.mockk
  24 import io.mockk.spyk
  25 import io.mockk.verify
  26 import org.junit.Before
  27 import org.junit.Test
  28 import org.onap.ccsdk.cds.blueprintsprocessor.functions.netconf.executor.api.DeviceInfo
  29 import org.onap.ccsdk.cds.blueprintsprocessor.functions.netconf.executor.api.NetconfReceivedEvent
  30 import org.onap.ccsdk.cds.blueprintsprocessor.functions.netconf.executor.api.NetconfSession
  31 import org.onap.ccsdk.cds.blueprintsprocessor.functions.netconf.executor.api.NetconfSessionListener
  32 import org.onap.ccsdk.cds.blueprintsprocessor.functions.netconf.executor.utils.RpcMessageUtils
  33 import java.io.IOException
  34 import java.io.InputStream
  35 import java.io.OutputStream
  36 import java.nio.charset.StandardCharsets
  37 import java.util.concurrent.CompletableFuture
  38 import java.util.concurrent.ConcurrentHashMap
  39 import java.util.regex.Pattern
  40 import kotlin.test.assertEquals
  41 import kotlin.test.assertFalse
  42 import kotlin.test.assertTrue
  43
  44 class NetconfDeviceCommunicatorTest {
  45     private lateinit var netconfSession: NetconfSession
  46     private lateinit var netconfSessionListener: NetconfSessionListener
  47     private lateinit var mockInputStream: InputStream
  48     private lateinit var mockOutputStream: OutputStream
  49     private lateinit var stubInputStream: InputStream
  50     private lateinit var replies: MutableMap<String, CompletableFuture<String>>
  51     private val endPatternCharArray: List<Int> = stringToCharArray(RpcMessageUtils.END_PATTERN)
  52
  53     companion object {
  54         private val chunkedEnding = "\n##\n"
  55         // using example from section 4.2 of RFC6242 (https://tools.ietf.org/html/rfc6242#section-4.2)
  56         private val validChunkedEncodedMsg = """
  57             |
  58             |#4
  59             |<rpc
  60             |#18
  61             | message-id="102"
  62             |
  63             |#79
  64             |     xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  65             |  <close-session/>
  66             |</rpc>
  67             |##
  68             |""".trimMargin()
  69     }
  70
  71     private fun stringToCharArray(str: String): List<Int> {
  72         return str.toCharArray().map(Char::toInt)
  73     }
  74
  75     @Before
  76     fun setup() {
  77         netconfSession = mockk()
  78         netconfSessionListener = mockk()
  79         mockInputStream = mockk()
  80         mockOutputStream = mockk()
  81         replies = ConcurrentHashMap()
  82     }
  83
  84     @Test
  85     fun `NetconfDeviceCommunicator should read from supplied reader`() {
  86         every { mockInputStream.read() } returns -1
  87         every { mockInputStream.read(any(), any(), any()) } returns -1
  88         val communicator: NetconfDeviceCommunicator =
  89             NetconfDeviceCommunicator(mockInputStream, mockOutputStream, genDeviceInfo(), netconfSessionListener, replies)
  90         communicator.join()
  91         // verify
  92         verify { mockInputStream.read(any(), any(), any()) }
  93     }
  94
  95     @Test
  96     fun `NetconfDeviceCommunicator unregisters device on END_PATTERN`() {
  97         // The reader will generate RpcMessageUtils.END_PATTERN "]]>]]>" which tells Netconf
  98         // to unregister the device.
  99         // we want to capture the slot to return the value as inputStreamReader will pass a char array
 100         // create a slot where NetconfReceivedEvent will be placed to further verify Type.DEVICE_UNREGISTERED
 101         val eventSlot = CapturingSlot<NetconfReceivedEvent>()
 102         every { netconfSessionListener.accept(event = capture(eventSlot)) } just Runs
 103         stubInputStream = RpcMessageUtils.END_PATTERN.byteInputStream(StandardCharsets.UTF_8)
 104         val inputStreamSpy = spyk(stubInputStream)
 105         // RUN the test
 106         val communicator = NetconfDeviceCommunicator(
 107             inputStreamSpy, mockOutputStream,
 108             genDeviceInfo(), netconfSessionListener, replies
 109         )
 110         communicator.join()
 111         // Verify
 112         verify { inputStreamSpy.close() }
 113         assertTrue { eventSlot.isCaptured }
 114         assertEquals(NetconfReceivedEvent.Type.DEVICE_UNREGISTERED, eventSlot.captured.type)
 115         assertEquals(genDeviceInfo(), eventSlot.captured.deviceInfo)
 116     }
 117
 118     @Test
 119     fun `NetconfDeviceCommunicator on IOException generated DEVICE_ERROR event`() {
 120         val eventSlot = CapturingSlot<NetconfReceivedEvent>()
 121         every { netconfSessionListener.accept(event = capture(eventSlot)) } just Runs
 122         stubInputStream = "".byteInputStream(StandardCharsets.UTF_8)
 123         val inputStreamSpy = spyk(stubInputStream)
 124         every { inputStreamSpy.read(any(), any(), any()) } returns 1 andThenThrows IOException("Fake IO Exception")
 125         // RUN THE TEST
 126         val communicator = NetconfDeviceCommunicator(
 127             inputStreamSpy, mockOutputStream,
 128             genDeviceInfo(), netconfSessionListener, replies
 129         )
 130         communicator.join()
 131         // Verify
 132         assertTrue { eventSlot.isCaptured }
 133         assertEquals(genDeviceInfo(), eventSlot.captured.deviceInfo)
 134         assertEquals(NetconfReceivedEvent.Type.DEVICE_ERROR, eventSlot.captured.type)
 135     }
 136
 137     @Test
 138     fun `NetconfDeviceCommunicator in END_PATTERN state but fails RpcMessageUtils end pattern validation`() {
 139         val eventSlot = CapturingSlot<NetconfReceivedEvent>()
 140         val payload = "<rpc-reply>blah</rpc-reply>"
 141         stubInputStream = "$payload${RpcMessageUtils.END_PATTERN}".byteInputStream(StandardCharsets.UTF_8)
 142         every { netconfSessionListener.accept(event = capture(eventSlot)) } just Runs
 143         // RUN the test
 144         val communicator = NetconfDeviceCommunicator(
 145             stubInputStream, mockOutputStream,
 146             genDeviceInfo(), netconfSessionListener, replies
 147         )
 148         communicator.join()
 149         // Verify
 150         verify(exactly = 0) { mockInputStream.close() } // make sure the reader is not closed as this could cause problems
 151         assertTrue { eventSlot.isCaptured }
 152         // eventually, sessionListener is called with message type DEVICE_REPLY
 153         assertEquals(NetconfReceivedEvent.Type.DEVICE_REPLY, eventSlot.captured.type)
 154         assertEquals(payload, eventSlot.captured.messagePayload)
 155     }
 156
 157     @Test
 158     fun `NetconfDeviceCommunicator in END_CHUNKED_PATTERN but validation failing produces DEVICE_ERROR`() {
 159         val eventSlot = CapturingSlot<NetconfReceivedEvent>()
 160         val payload = "<rpc-reply>blah</rpc-reply>"
 161         val payloadWithChunkedEnding = "$payload$chunkedEnding"
 162         every { netconfSessionListener.accept(event = capture(eventSlot)) } just Runs
 163
 164         stubInputStream = payloadWithChunkedEnding.byteInputStream(StandardCharsets.UTF_8)
 165         // we have to ensure that the input stream is processed, so need to create a spy object.
 166         val inputStreamSpy = spyk(stubInputStream)
 167         // RUN the test
 168         val communicator = NetconfDeviceCommunicator(
 169             inputStreamSpy, mockOutputStream, genDeviceInfo(),
 170             netconfSessionListener, replies
 171         )
 172         communicator.join()
 173         // Verify
 174         verify(exactly = 0) { inputStreamSpy.close() } // make sure the reader is not closed as this could cause problems
 175         assertTrue { eventSlot.isCaptured }
 176         // eventually, sessionListener is called with message type DEVICE_REPLY
 177         assertEquals(NetconfReceivedEvent.Type.DEVICE_ERROR, eventSlot.captured.type)
 178         assertEquals("", eventSlot.captured.messagePayload)
 179     }
 180
 181     @Test
 182     fun `NetconfDeviceCommunicator in END_CHUNKED_PATTERN passing validation generates DEVICE_REPLY`() {
 183         val eventSlot = CapturingSlot<NetconfReceivedEvent>()
 184         stubInputStream = validChunkedEncodedMsg.byteInputStream(StandardCharsets.UTF_8)
 185         val inputStreamSpy = spyk(stubInputStream)
 186         every { netconfSessionListener.accept(event = capture(eventSlot)) } just Runs
 187         // RUN the test
 188         NetconfDeviceCommunicator(inputStreamSpy, mockOutputStream, genDeviceInfo(), netconfSessionListener, replies).join()
 189         // Verify
 190         verify(exactly = 0) { inputStreamSpy.close() } // make sure the reader is not closed as this could cause problems
 191         assertTrue { eventSlot.isCaptured }
 192         // eventually, sessionListener is called with message type DEVICE_REPLY
 193         assertEquals(NetconfReceivedEvent.Type.DEVICE_REPLY, eventSlot.captured.type)
 194         assertEquals(
 195             """
 196         <rpc message-id="102"
 197              xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
 198           <close-session/>
 199         </rpc>
 200         """.trimIndent(), eventSlot.captured.messagePayload
 201         )
 202     }
 203
 204     @Test
 205     // test to ensure that we have a valid test message to be then used in the case of chunked message
 206     // validation code path
 207     fun `chunked sample is validated by the chunked response regex`() {
 208         val test1 = "\n#10\nblah\n##\n"
 209         val chunkedFramingPattern = Pattern.compile("(\\n#([1-9][0-9]*)\\n(.+))+\\n##\\n", Pattern.DOTALL)
 210         val matcher = chunkedFramingPattern.matcher(test1)
 211         assertTrue { matcher.matches() }
 212     }
 213
 214     @Test
 215     // Verify that our test sample passes the second pattern for chunked size
 216     fun `chunkSizeMatcher pattern finds matches in chunkedMessageSample`() {
 217         val sizePattern = Pattern.compile("\\n#([1-9][0-9]*)\\n")
 218         val matcher = sizePattern.matcher(validChunkedEncodedMsg)
 219         assertTrue { matcher.find() }
 220     }
 221
 222     @Test
 223     fun `sendMessage writes the request to NetconfDeviceCommunicator Writer`() {
 224         val msgPayload = "some text"
 225         val msgId = "100"
 226         stubInputStream = "".byteInputStream(StandardCharsets.UTF_8) // no data available in the stream...
 227         every { mockOutputStream.write(any(), any(), any()) } just Runs
 228         every { mockOutputStream.write(msgPayload.toByteArray(Charsets.UTF_8)) } just Runs
 229         every { mockOutputStream.flush() } just Runs
 230         // Run the command
 231         val communicator = NetconfDeviceCommunicator(
 232             stubInputStream, mockOutputStream,
 233             genDeviceInfo(), netconfSessionListener, replies
 234         )
 235         val completableFuture = communicator.sendMessage(msgPayload, msgId)
 236         communicator.join()
 237         // verify
 238         verify { mockOutputStream.write(any(), any(), any()) }
 239         verify { mockOutputStream.flush() }
 240         assertFalse { completableFuture.isCompletedExceptionally }
 241     }
 242
 243     @Test
 244     fun `sendMessage on IOError returns completed exceptionally future`() {
 245         val msgPayload = "some text"
 246         val msgId = "100"
 247         every { mockOutputStream.write(any(), any(), any()) } throws IOException("Some IO error occurred!")
 248         stubInputStream = "".byteInputStream(StandardCharsets.UTF_8) // no data available in the stream...
 249         // Run the command
 250         val communicator = NetconfDeviceCommunicator(
 251             stubInputStream, mockOutputStream,
 252             genDeviceInfo(), netconfSessionListener, replies
 253         )
 254         val completableFuture = communicator.sendMessage(msgPayload, msgId)
 255         // verify
 256         verify { mockOutputStream.write(any(), any(), any()) }
 257         verify(exactly = 0) { mockOutputStream.flush() }
 258         assertTrue { completableFuture.isCompletedExceptionally }
 259     }
 260
 261     private fun genDeviceInfo(): DeviceInfo {
 262         return DeviceInfo().apply {
 263             username = "user"
 264             password = "pass"
 265             ipAddress = "localhost"
 266             port = 4567
 267         }
 268     }
 269 }