def test_add_known_host_exists_outdated(
         self, check_output, host_key, rm, known_hosts):
     check_output.return_value = '|1|= fookey'
     host_key.return_value = '|1|= fookey_old'
     with patch_open() as (_open, _file):
         utils.add_known_host('foohost', None, None)
         rm.assert_called_with('foohost', None, None)

 def test_add_known_host_exists(self, check_output, host_key, rm):
     check_output.return_value = '|1|= fookey'
     host_key.return_value = '|1|= fookey'
     with patch_open() as (_open, _file):
         utils.add_known_host('foohost')
         self.assertFalse(rm.called)
         self.assertFalse(_file.write.called)

    def test__set_sriov_numvfs(self, mock_sysnet_ints):
        mock_sysnet_ints.side_effect = [{
            'interface': 'eth2',
            'mac_address': 'a8:9d:21:cf:93:fc',
            'pci_address': '0000:10:00.0',
            'state': 'up',
            'sriov': True,
            'sriov_totalvfs': 7,
            'sriov_numvfs': 0
        }], [{
            'interface': 'eth2',
            'mac_address': 'a8:9d:21:cf:93:fc',
            'pci_address': '0000:10:00.0',
            'state': 'up',
            'sriov': True,
            'sriov_totalvfs': 7,
            'sriov_numvfs': 4
        }]
        dev = pci.PCINetDevice('0000:10:00.0')
        self.assertEqual('eth2', dev.interface_name)
        self.assertTrue(dev.sriov)
        self.assertEqual(7, dev.sriov_totalvfs)
        self.assertEqual(0, dev.sriov_numvfs)

        with patch_open() as (mock_open, mock_file):
            dev._set_sriov_numvfs(4)
            mock_open.assert_called_with(
                '/sys/class/net/eth2/device/sriov_numvfs', 'w')
            mock_file.write.assert_called_with("4")
            self.assertTrue(dev.sriov)
            self.assertEqual(7, dev.sriov_totalvfs)
            self.assertEqual(4, dev.sriov_numvfs)

 def test_import_keystone_cert(self, check_call):
     self.relation_get.return_value = 'Zm9vX2NlcnQK'
     with patch_open() as (_open, _file):
         utils.import_keystone_ca_cert()
         _open.assert_called_with(utils.CA_CERT_PATH, 'wb')
         _file.write.assert_called_with('foo_cert\n')
     check_call.assert_called_with(['update-ca-certificates'])

 def test_config_changed_upgrade_available(self):
     self.openstack_upgrade_available.return_value = True
     self.relations_of_type.return_value = False
     with patch_open() as (_open, _file):
         _file.read.return_value = "foo"
         hooks.config_changed()
     self.assertTrue(self.do_openstack_upgrade.called)
     self.assertTrue(self.CONFIGS.write_all.called)

 def test_get_sysnet_mac(self, _update):
     with patch_open() as (_open, _file):
         super_fh = mocked_filehandle()
         _file.readlines = MagicMock()
         _open.side_effect = super_fh._setfilename
         _file.read.side_effect = super_fh._getfilecontents_read
         macaddr = pci.get_sysnet_mac('/sys/class/net/eth3')
     self.assertEqual(macaddr, 'a8:9d:21:cf:93:fd')

 def test_secret_retrieved(self, _path):
     _path.exists.return_value = True
     with patch_open() as (_open, _file):
         _file.read.return_value = 'secret_thing\n'
         self.assertEquals(context.get_shared_secret(),
                           'secret_thing')
         _open.assert_called_with(
             context.SHARED_SECRET.format('quantum'), 'r')

 def test_get_sysnet_device_state(self, _update):
     with patch_open() as (_open, _file):
         super_fh = mocked_filehandle()
         _file.readlines = MagicMock()
         _open.side_effect = super_fh._setfilename
         _file.read.side_effect = super_fh._getfilecontents_read
         state = pci.get_sysnet_device_state('/sys/class/net/eth3')
     self.assertEqual(state, 'down')

 def test_config_changed_nrpe_master(self):
     self.openstack_upgrade_available.return_value = False
     self.relations_of_type.return_value = True
     with patch_open() as (_open, _file):
         _file.read.return_value = "foo"
         hooks.config_changed()
     self.assertTrue(self.CONFIGS.write_all.called)
     self.assertTrue(self.setup_rsync.called)
     self.assertTrue(self.update_nrpe_config.called)

 def test_config_changed_ipv6(self, mock_assert_charm_supports_ipv6):
     self.test_config.set('prefer-ipv6', True)
     self.openstack_upgrade_available.return_value = False
     self.relations_of_type.return_value = False
     with patch_open() as (_open, _file):
         _file.read.return_value = "foo"
         hooks.config_changed()
     self.assertTrue(self.CONFIGS.write_all.called)
     self.assertTrue(self.setup_rsync.called)

 def test_find_block_devices(self):
     self.is_block_device.return_value = True
     with patch_open() as (_open, _file):
         _file.read.return_value = PROC_PARTITIONS
         _file.readlines = MagicMock()
         _file.readlines.return_value = PROC_PARTITIONS.split('\n')
         result = swift_utils.find_block_devices()
     ex = ['/dev/sdb', '/dev/vdb', '/dev/cciss/c1d0']
     self.assertEquals(ex, result)

 def test_secret_created_stored(self, _uuid4, _path):
     _path.exists.return_value = False
     _uuid4.return_value = 'secret_thing'
     with patch_open() as (_open, _file):
         self.assertEquals(context.get_shared_secret(),
                           'secret_thing')
         _open.assert_called_with(
             context.SHARED_SECRET.format('quantum'), 'w')
         _file.write.assert_called_with('secret_thing')

    def test_config_changed_with_openstack_upgrade_action(self):
        self.openstack_upgrade_available.return_value = True
        self.test_config.set('action-managed-upgrade', True)

        with patch_open() as (_open, _file):
            _file.read.return_value = "foo"
            hooks.config_changed()

        self.assertFalse(self.do_openstack_upgrade.called)

 def test_add_known_host_exists_added(
         self, check_output, host_key, rm, known_hosts):
     check_output.return_value = '|1|= fookey'
     host_key.return_value = None
     with patch_open() as (_open, _file):
         _file.write = MagicMock()
         utils.add_known_host('foohost')
         self.assertFalse(rm.called)
         _file.write.assert_called_with('|1|= fookey\n')

 def test_rsync_enable_rsync(self):
     with patch_open() as (_open, _file):
         ctxt = swift_context.RsyncContext()
         _file.read.return_value = 'RSYNC_ENABLE=false'
         ctxt.enable_rsyncd()
         _file.write.assert_called_with('RSYNC_ENABLE=true')
         _file.read.return_value = '#foo'
         ctxt.enable_rsyncd()
         _file.write.assert_called_with('RSYNC_ENABLE=true\n')

 def test_numa_node_cores(self, _parse_cpu_list):
     self.glob.glob.return_value = [
         '/sys/devices/system/node/node0'
     ]
     with patch_open() as (_, mock_file):
         mock_file.read.return_value = TEST_CPULIST_1
         self.assertEqual(context.numa_node_cores(),
                          {'0': [0, 1, 2, 3]})
     self.glob.glob.assert_called_with('/sys/devices/system/node/node*')
     _parse_cpu_list.assert_called_with(TEST_CPULIST_1)

 def test_upgrade_havana_icehouse_source(self, _do_openstack_upgrade):
     "Verify get_step_upgrade_source() for havana->icehouse"
     self.config.side_effect = None
     self.config.return_value = 'cloud:precise-icehouse'
     with patch_open() as (_open, _file):
         _file.read = MagicMock()
         _file.readline.return_value = "deb url precise-updates/havana main"
         utils.do_openstack_upgrade(self.register_configs())
         expected = [call('cloud:precise-icehouse')]
         self.assertEquals(_do_openstack_upgrade.call_args_list, expected)

 def test_find_block_devices_real_world(self):
     self.is_block_device.return_value = True
     side_effect = lambda x: x in ["/dev/sdb", "/dev/sdb1"] # flake8: noqa
     self.is_device_mounted.side_effect = side_effect
     with patch_open() as (_open, _file):
         _file.read.return_value = REAL_WORLD_PARTITIONS
         _file.readlines = MagicMock()
         _file.readlines.return_value = REAL_WORLD_PARTITIONS.split('\n')
         result = swift_utils.find_block_devices()
     expected = ["/dev/sda"]
     self.assertEquals(expected, result)

    def _test_import_authorized_keys_base(self, getpwnam, prefix=None,
                                          auth_key_path='/home/foo/.ssh/'
                                                        'authorized_keys'):
        getpwnam.return_value = self.fake_user('foo')
        self.relation_get.side_effect = [
            3,          # relation_get('known_hosts_max_index')
            'k_h_0',    # relation_get_('known_hosts_0')
            'k_h_1',    # relation_get_('known_hosts_1')
            'k_h_2',    # relation_get_('known_hosts_2')
            3,          # relation_get('authorized_keys_max_index')
            'auth_0',   # relation_get('authorized_keys_0')
            'auth_1',   # relation_get('authorized_keys_1')
            'auth_2',   # relation_get('authorized_keys_2')
        ]

        ex_open = [
            call('/home/foo/.ssh/known_hosts', 'wb'),
            call(auth_key_path, 'wb')
        ]
        ex_write = [
            call('k_h_0\n'),
            call('k_h_1\n'),
            call('k_h_2\n'),
            call('auth_0\n'),
            call('auth_1\n'),
            call('auth_2\n')
        ]

        with patch_open() as (_open, _file):
            utils.import_authorized_keys(user='foo', prefix=prefix)
            self.assertEqual(ex_open, _open.call_args_list)
            self.assertEqual(ex_write, _file.write.call_args_list)
            authkey_root = 'authorized_keys_'
            known_hosts_root = 'known_hosts_'
            if prefix:
                authkey_root = prefix + '_authorized_keys_'
                known_hosts_root = prefix + '_known_hosts_'
            expected_relations = [
                call(known_hosts_root + 'max_index'),
                call(known_hosts_root + '0'),
                call(known_hosts_root + '1'),
                call(known_hosts_root + '2'),
                call(authkey_root + 'max_index'),
                call(authkey_root + '0'),
                call(authkey_root + '1'),
                call(authkey_root + '2')
            ]
            self.assertEqual(sorted(self.relation_get.call_args_list),
                             sorted(expected_relations))

    def eth_int(self, pci_address, _osrealpath, _osislink, subproc_map=None):
        self.glob.glob.side_effect = mocked_globs
        _osislink.side_effect = mocked_islink
        _osrealpath.side_effect = mocked_realpath
        self.subprocess.check_output.side_effect = mocked_subprocess(
            subproc_map=subproc_map)

        with patch_open() as (_open, _file):
            super_fh = mocked_filehandle()
            _file.readlines = MagicMock()
            _open.side_effect = super_fh._setfilename
            _file.read.side_effect = super_fh._getfilecontents_read
            _file.readlines.side_effect = super_fh._getfilecontents_readlines
            netint = pci.PCINetDevice(pci_address)
        return netint