1# Copyright 2015 Cloudbase Solutions Srl
2# All Rights Reserved.
3#
4#    Licensed under the Apache License, Version 2.0 (the "License"); you may
5#    not use this file except in compliance with the License. You may obtain
6#    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, WITHOUT
12#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13#    License for the specific language governing permissions and limitations
14#    under the License.
15
16from unittest import mock
17
18import ddt
19
20from os_win import exceptions
21from os_win.tests.unit import test_base
22from os_win.utils.storage import smbutils
23
24
25@ddt.ddt
26class SMBUtilsTestCase(test_base.OsWinBaseTestCase):
27
28    _autospec_classes = [
29        smbutils.win32utils.Win32Utils,
30    ]
31
32    def setUp(self):
33        super(SMBUtilsTestCase, self).setUp()
34
35        self._smbutils = smbutils.SMBUtils()
36        self._smbutils._smb_conn = mock.Mock()
37        self._mock_run = self._smbutils._win32_utils.run_and_check_output
38        self._smb_conn = self._smbutils._smb_conn
39
40    @mock.patch.object(smbutils.SMBUtils, 'unmount_smb_share')
41    @mock.patch('os.path.exists')
42    def _test_check_smb_mapping(self, mock_exists, mock_unmount_smb_share,
43                                existing_mappings=True, share_available=False):
44        mock_exists.return_value = share_available
45
46        fake_mappings = (
47            [mock.sentinel.smb_mapping] if existing_mappings else [])
48
49        self._smb_conn.Msft_SmbMapping.return_value = fake_mappings
50
51        ret_val = self._smbutils.check_smb_mapping(
52            mock.sentinel.share_path, remove_unavailable_mapping=True)
53
54        self.assertEqual(existing_mappings and share_available, ret_val)
55        if existing_mappings and not share_available:
56            mock_unmount_smb_share.assert_called_once_with(
57                mock.sentinel.share_path, force=True)
58
59    def test_check_mapping(self):
60        self._test_check_smb_mapping()
61
62    def test_remake_unavailable_mapping(self):
63        self._test_check_smb_mapping(existing_mappings=True,
64                                     share_available=False)
65
66    def test_available_mapping(self):
67        self._test_check_smb_mapping(existing_mappings=True,
68                                     share_available=True)
69
70    def test_mount_smb_share(self):
71        fake_create = self._smb_conn.Msft_SmbMapping.Create
72        self._smbutils.mount_smb_share(mock.sentinel.share_path,
73                                       mock.sentinel.username,
74                                       mock.sentinel.password)
75        fake_create.assert_called_once_with(
76            RemotePath=mock.sentinel.share_path,
77            UserName=mock.sentinel.username,
78            Password=mock.sentinel.password)
79
80    def test_mount_smb_share_failed(self):
81        self._smb_conn.Msft_SmbMapping.Create.side_effect = exceptions.x_wmi
82
83        self.assertRaises(exceptions.SMBException,
84                          self._smbutils.mount_smb_share,
85                          mock.sentinel.share_path)
86
87    def _test_unmount_smb_share(self, force=False):
88        fake_mapping = mock.Mock()
89        fake_mapping_attr_err = mock.Mock()
90        fake_mapping_attr_err.side_effect = AttributeError
91        smb_mapping_class = self._smb_conn.Msft_SmbMapping
92        smb_mapping_class.return_value = [fake_mapping, fake_mapping_attr_err]
93
94        self._smbutils.unmount_smb_share(mock.sentinel.share_path,
95                                         force)
96
97        smb_mapping_class.assert_called_once_with(
98            RemotePath=mock.sentinel.share_path)
99        fake_mapping.Remove.assert_called_once_with(Force=force)
100
101    def test_soft_unmount_smb_share(self):
102        self._test_unmount_smb_share()
103
104    def test_force_unmount_smb_share(self):
105        self._test_unmount_smb_share(force=True)
106
107    def test_unmount_smb_share_wmi_exception(self):
108        fake_mapping = mock.Mock()
109        fake_mapping.Remove.side_effect = exceptions.x_wmi
110        self._smb_conn.Msft_SmbMapping.return_value = [fake_mapping]
111
112        self.assertRaises(exceptions.SMBException,
113                          self._smbutils.unmount_smb_share,
114                          mock.sentinel.share_path, force=True)
115
116    def test_get_smb_share_path(self):
117        fake_share = mock.Mock(Path=mock.sentinel.share_path)
118        self._smb_conn.Msft_SmbShare.return_value = [fake_share]
119
120        share_path = self._smbutils.get_smb_share_path(
121            mock.sentinel.share_name)
122
123        self.assertEqual(mock.sentinel.share_path, share_path)
124        self._smb_conn.Msft_SmbShare.assert_called_once_with(
125            Name=mock.sentinel.share_name)
126
127    def test_get_unexisting_smb_share_path(self):
128        self._smb_conn.Msft_SmbShare.return_value = []
129
130        share_path = self._smbutils.get_smb_share_path(
131            mock.sentinel.share_name)
132
133        self.assertIsNone(share_path)
134        self._smb_conn.Msft_SmbShare.assert_called_once_with(
135            Name=mock.sentinel.share_name)
136
137    @ddt.data({'local_ips': [mock.sentinel.ip0, mock.sentinel.ip1],
138               'dest_ips': [mock.sentinel.ip2, mock.sentinel.ip3],
139               'expected_local': False},
140              {'local_ips': [mock.sentinel.ip0, mock.sentinel.ip1],
141               'dest_ips': [mock.sentinel.ip1, mock.sentinel.ip3],
142               'expected_local': True},
143              {'local_ips': [],
144               'dest_ips': ['127.0.0.1'],
145               'expected_local': True})
146    @ddt.unpack
147    @mock.patch('os_win._utils.get_ips')
148    @mock.patch('socket.gethostname')
149    def test_is_local_share(self, mock_gethostname, mock_get_ips,
150                            local_ips, dest_ips, expected_local):
151        fake_share_server = 'fake_share_server'
152        fake_share = '\\\\%s\\fake_share' % fake_share_server
153
154        mock_get_ips.side_effect = (local_ips,
155                                    ['127.0.0.1', '::1'],
156                                    dest_ips)
157        self._smbutils._loopback_share_map = {}
158
159        is_local = self._smbutils.is_local_share(fake_share)
160        self.assertEqual(expected_local, is_local)
161
162        # We ensure that this value is cached, calling it again
163        # and making sure that we have attempted to resolve the
164        # address only once.
165        self._smbutils.is_local_share(fake_share)
166
167        mock_gethostname.assert_called_once_with()
168        mock_get_ips.assert_has_calls(
169            [mock.call(mock_gethostname.return_value),
170             mock.call('localhost'),
171             mock.call(fake_share_server)])
172