n2vc/tests/unit/test_provisioner.py

   1 # Copyright 2020 Canonical Ltd.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #     Unless required by applicable law or agreed to in writing, software
  10 #     distributed under the License is distributed on an "AS IS" BASIS,
  11 #     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 #     See the License for the specific language governing permissions and
  13 #     limitations under the License.
  14
  15 from unittest import TestCase, mock
  16
  17 from mock import mock_open
  18 from n2vc.provisioner import SSHProvisioner
  19 from paramiko.ssh_exception import SSHException
  20
  21
  22 class ProvisionerTest(TestCase):
  23     def setUp(self):
  24         self.provisioner = SSHProvisioner(None, None, None)
  25
  26     @mock.patch("n2vc.provisioner.os.path.exists")
  27     @mock.patch("n2vc.provisioner.paramiko.RSAKey")
  28     @mock.patch("n2vc.provisioner.paramiko.SSHClient")
  29     @mock.patch("builtins.open", new_callable=mock_open, read_data="data")
  30     def test__get_ssh_client(self, _mock_open, mock_sshclient, _mock_rsakey, _mock_os):
  31         mock_instance = mock_sshclient.return_value
  32         sshclient = self.provisioner._get_ssh_client()
  33         self.assertEqual(mock_instance, sshclient)
  34         self.assertEqual(
  35             1,
  36             mock_instance.set_missing_host_key_policy.call_count,
  37             "Missing host key call count",
  38         )
  39         self.assertEqual(1, mock_instance.connect.call_count, "Connect call count")
  40
  41     @mock.patch("n2vc.provisioner.os.path.exists")
  42     @mock.patch("n2vc.provisioner.paramiko.RSAKey")
  43     @mock.patch("n2vc.provisioner.paramiko.SSHClient")
  44     @mock.patch("builtins.open", new_callable=mock_open, read_data="data")
  45     def test__get_ssh_client_no_connection(
  46         self, _mock_open, mock_sshclient, _mock_rsakey, _mock_os
  47     ):
  48
  49         mock_instance = mock_sshclient.return_value
  50         mock_instance.method_inside_someobject.side_effect = ["something"]
  51         mock_instance.connect.side_effect = SSHException()
  52
  53         self.assertRaises(SSHException, self.provisioner._get_ssh_client)
  54         self.assertEqual(
  55             1,
  56             mock_instance.set_missing_host_key_policy.call_count,
  57             "Missing host key call count",
  58         )
  59         self.assertEqual(1, mock_instance.connect.call_count, "Connect call count")
  60
  61     @mock.patch("n2vc.provisioner.os.path.exists")
  62     @mock.patch("n2vc.provisioner.paramiko.RSAKey")
  63     @mock.patch("n2vc.provisioner.paramiko.SSHClient")
  64     @mock.patch("builtins.open", new_callable=mock_open, read_data="data")
  65     def test__get_ssh_client_bad_banner(
  66         self, _mock_open, mock_sshclient, _mock_rsakey, _mock_os
  67     ):
  68
  69         mock_instance = mock_sshclient.return_value
  70         mock_instance.method_inside_someobject.side_effect = ["something"]
  71         mock_instance.connect.side_effect = [
  72             SSHException("Error reading SSH protocol banner"),
  73             None,
  74             None,
  75         ]
  76
  77         sshclient = self.provisioner._get_ssh_client()
  78         self.assertEqual(mock_instance, sshclient)
  79         self.assertEqual(
  80             1,
  81             mock_instance.set_missing_host_key_policy.call_count,
  82             "Missing host key call count",
  83         )
  84         self.assertEqual(
  85             3, mock_instance.connect.call_count, "Should attempt 3 connections"
  86         )
  87
  88     @mock.patch("time.sleep", autospec=True)
  89     @mock.patch("n2vc.provisioner.os.path.exists")
  90     @mock.patch("n2vc.provisioner.paramiko.RSAKey")
  91     @mock.patch("n2vc.provisioner.paramiko.SSHClient")
  92     @mock.patch("builtins.open", new_callable=mock_open, read_data="data")
  93     def test__get_ssh_client_unable_to_connect(
  94         self, _mock_open, mock_sshclient, _mock_rsakey, _mock_os, _mock_sleep
  95     ):
  96
  97         mock_instance = mock_sshclient.return_value
  98         mock_instance.connect.side_effect = Exception("Unable to connect to port")
  99
 100         self.assertRaises(Exception, self.provisioner._get_ssh_client)
 101         self.assertEqual(
 102             1,
 103             mock_instance.set_missing_host_key_policy.call_count,
 104             "Missing host key call count",
 105         )
 106         self.assertEqual(
 107             11, mock_instance.connect.call_count, "Should attempt 11 connections"
 108         )
 109
 110     @mock.patch("time.sleep", autospec=True)
 111     @mock.patch("n2vc.provisioner.os.path.exists")
 112     @mock.patch("n2vc.provisioner.paramiko.RSAKey")
 113     @mock.patch("n2vc.provisioner.paramiko.SSHClient")
 114     @mock.patch("builtins.open", new_callable=mock_open, read_data="data")
 115     def test__get_ssh_client_unable_to_connect_once(
 116         self, _mock_open, mock_sshclient, _mock_rsakey, _mock_os, _mock_sleep
 117     ):
 118
 119         mock_instance = mock_sshclient.return_value
 120         mock_instance.connect.side_effect = [
 121             Exception("Unable to connect to port"),
 122             None,
 123         ]
 124
 125         sshclient = self.provisioner._get_ssh_client()
 126         self.assertEqual(mock_instance, sshclient)
 127         self.assertEqual(
 128             1,
 129             mock_instance.set_missing_host_key_policy.call_count,
 130             "Missing host key call count",
 131         )
 132         self.assertEqual(
 133             2, mock_instance.connect.call_count, "Should attempt 2 connections"
 134         )
 135
 136     @mock.patch("n2vc.provisioner.os.path.exists")
 137     @mock.patch("n2vc.provisioner.paramiko.RSAKey")
 138     @mock.patch("n2vc.provisioner.paramiko.SSHClient")
 139     @mock.patch("builtins.open", new_callable=mock_open, read_data="data")
 140     def test__get_ssh_client_other_exception(
 141         self, _mock_open, mock_sshclient, _mock_rsakey, _mock_os
 142     ):
 143
 144         mock_instance = mock_sshclient.return_value
 145         mock_instance.connect.side_effect = Exception()
 146
 147         self.assertRaises(Exception, self.provisioner._get_ssh_client)
 148         self.assertEqual(
 149             1,
 150             mock_instance.set_missing_host_key_policy.call_count,
 151             "Missing host key call count",
 152         )
 153         self.assertEqual(
 154             1, mock_instance.connect.call_count, "Should only attempt 1 connection"
 155         )
 156
 157
 158 #