netconf-executor: NetconfSessionImplTest

Issue-ID: CCSDK-1126
patchset 6: test: undid most of NetconfSessionImpl and muted some tests it affected
Change-Id: Iff406705fa06a00c2629d248010192354d9e2a41
Signed-off-by: Oleg Mitsura <oleg.mitsura@amdocs.com>

2 files changed, 37 insertions, 16 deletions

diff --git a/ms/blueprintsprocessor/functions/netconf-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfDeviceCommunicator.kt b/ms/blueprintsprocessor/functions/netconf-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfDeviceCommunicator.kt
index 12e3b83da..6ef4f41fb 100644
--- a/ms/blueprintsprocessor/functions/netconf-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfDeviceCommunicator.kt
+++ b/ms/blueprintsprocessor/functions/netconf-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfDeviceCommunicator.kt
@@ -30,6 +30,7 @@ import java.io.OutputStream
 import java.io.OutputStreamWriter
 import java.nio.charset.StandardCharsets
 import java.util.concurrent.CompletableFuture
+import java.util.concurrent.TimeUnit
 
 class NetconfDeviceCommunicator(private var inputStream: InputStream,
                                 private var out: OutputStream,
@@ -232,4 +233,21 @@ class NetconfDeviceCommunicator(private var inputStream: InputStream,
             NetconfMessageUtils.getMsgId(deviceReply),
             deviceInfo))
     }
+
+    /**
+     * Gets the value of the {@link CompletableFuture} from {@link NetconfDeviceCommunicator#sendMessage}
+     * This function is used by NetconfSessionImpl. Needed to wrap exception testing in NetconfSessionImpl.
+     * @param fut {@link CompletableFuture} object
+     * @param timeout the maximum time to wait
+     * @param timeUnit the time unit of the timeout argument
+     * @return the result value
+     * @throws CancellationException if this future was cancelled
+     * @throws ExecutionException if this future completed exceptionally
+     * @throws InterruptedException if the current thread was interrupted while waiting
+     * @throws TimeoutException if the wait timed outStream
+     */
+    internal fun getFutureFromSendMessage(
+        fut: CompletableFuture<String>, timeout: Long, timeUnit: TimeUnit): String {
+        return fut.get(timeout, timeUnit)
+    }
 }
diff --git a/ms/blueprintsprocessor/functions/netconf-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfSessionImpl.kt b/ms/blueprintsprocessor/functions/netconf-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfSessionImpl.kt
index 12eb43f45..7e56e3e51 100644
--- a/ms/blueprintsprocessor/functions/netconf-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfSessionImpl.kt
+++ b/ms/blueprintsprocessor/functions/netconf-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfSessionImpl.kt
@@ -34,22 +34,21 @@ import org.onap.ccsdk.cds.blueprintsprocessor.functions.netconf.executor.utils.R
 import org.onap.ccsdk.cds.blueprintsprocessor.functions.netconf.executor.utils.RpcStatus
 import org.slf4j.LoggerFactory
 import java.io.IOException
-import java.util.*
+import java.util.Collections
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.ExecutionException
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.TimeoutException
-import java.util.concurrent.atomic.AtomicReference
 
 class NetconfSessionImpl(private val deviceInfo: DeviceInfo, private val rpcService: NetconfRpcService) :
     NetconfSession {
 
     private val log = LoggerFactory.getLogger(NetconfSessionImpl::class.java)
 
-    private val errorReplies: MutableList<String> = Collections.synchronizedList(listOf())
+    private val errorReplies: MutableList<String> = Collections.synchronizedList(mutableListOf())
     private val replies: MutableMap<String, CompletableFuture<String>> = ConcurrentHashMap()
-    private val deviceCapabilities = setOf<String>()
+    private val deviceCapabilities = mutableSetOf<String>()
 
     private var connectionTimeout: Long = 0
     private var replyTimeout: Int = 0
@@ -117,11 +116,8 @@ class NetconfSessionImpl(private val deviceInfo: DeviceInfo, private val rpcServ
             } catch (ioe: IOException) {
                 log.warn("$deviceInfo: Error closing session($sessionId) for host($deviceInfo)", ioe)
             }
-
-//            NetconfReceivedEvent(NetconfReceivedEvent.Type.SESSION_CLOSED, "",
-//                "Closed due to unexpected error " + e.cause, "-1", deviceInfo)
-            errorReplies.clear() // move to cleanUp()?
-            replies.clear()
+            clearErrorReplies()
+            clearReplies()
 
             throw NetconfException("$deviceInfo: Closing session $sessionId for request $formattedRequest", e)
         }
@@ -144,27 +140,26 @@ class NetconfSessionImpl(private val deviceInfo: DeviceInfo, private val rpcServ
         try {
             if (client.isClosed) {
                 log.info("Trying to restart the whole SSH connection with {}", deviceInfo)
-                replies.clear()
+                clearReplies()
                 startConnection()
             } else if (session.isClosed) {
                 log.info("Trying to restart the session with {}", deviceInfo)
-                replies.clear()
+                clearReplies()
                 startSession()
             } else if (channel.isClosed) {
                 log.info("Trying to reopen the channel with {}", deviceInfo)
-                replies.clear()
+                clearReplies()
                 openChannel()
             } else {
                 return
             }
         } catch (e: IOException) {
-            log.error("Can't reopen connection for device {}", e.message)
+            log.error("Can't reopen connection for device {} error: {}", deviceInfo, e.message)
             throw NetconfException(String.format("Cannot re-open the connection with device (%s)", deviceInfo), e)
         } catch (e: IllegalStateException) {
-            log.error("Can't reopen connection for device {}", e.message)
+            log.error("Can't reopen connection for device {} error: {}", deviceInfo, e.message)
             throw NetconfException(String.format("Cannot re-open the connection with device (%s)", deviceInfo), e)
         }
-
     }
 
     override fun getDeviceInfo(): DeviceInfo {
@@ -191,8 +186,13 @@ class NetconfSessionImpl(private val deviceInfo: DeviceInfo, private val rpcServ
 
     }
 
-    private fun startClient() {
+    //Needed to unit test connect method interacting with client.start in startClient() below
+    private fun setupNewSSHClient() {
         client = SshClient.setUpDefaultClient()
+    }
+
+    private fun startClient() {
+        setupNewSSHClient()
 
         client.properties.putIfAbsent(FactoryManager.IDLE_TIMEOUT, TimeUnit.SECONDS.toMillis(idleTimeout.toLong()))
         client.properties.putIfAbsent(FactoryManager.NIO2_READ_TIMEOUT, TimeUnit.SECONDS.toMillis(idleTimeout + 15L))
@@ -304,4 +304,7 @@ class NetconfSessionImpl(private val deviceInfo: DeviceInfo, private val rpcServ
 
     internal fun clearErrorReplies() = errorReplies.clear()
     internal fun clearReplies() = replies.clear()
+    internal fun setClient(client: SshClient) { this.client = client }
+    internal fun setSession(session: ClientSession) { this.session = session }
+    internal fun setChannel(channel: ClientChannel) { this.channel = channel }
 }
\ No newline at end of file