Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
# Copyright 2019 Dom Sekotill <dom.sekotill@kodo.org.uk>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Test cases for wpa_supplicant.client.master.MasterClient
"""
import pathlib
import unittest
from unittest.mock import patch
import anyio
from tests.unit import anyio as anyio_mock
from wpa_supplicant.client import interfaces, master
class InterfaceMethodsTests(unittest.TestCase):
"""
Tests for the *_interface(s?) methods
"""
def setUp(self):
async def _make_client():
return master.MasterClient()
self.client = client = anyio.run(_make_client)
client.sock = anyio_mock.AsyncMock()
client.sock.send.side_effect = len
@anyio_mock.with_anyio()
async def test_connect(self):
"""
Check that connect sets ctrl_dir
"""
client1 = master.MasterClient()
client2 = master.MasterClient()
with patch("wpa_supplicant.client.base.BaseClient.connect",
new_callable=anyio_mock.AsyncMock):
await client1.connect("/tmp/foo/bar")
await client2.connect(pathlib.Path("/tmp/foo/bar"))
self.assertIsInstance(client1.ctrl_dir, pathlib.Path)
self.assertIsInstance(client2.ctrl_dir, pathlib.Path)
assert client1.ctrl_dir == pathlib.Path("/tmp/foo")
assert client2.ctrl_dir == pathlib.Path("/tmp/foo")
@anyio_mock.with_anyio()
async def test_managed_interfaces(self):
"""
Check managed_interfaces() processes lines of names in a list
"""
async with self.client as client:
client.sock.recv.return_value = (
b"enp0s0\n"
b"enp1s0\n"
b"wlp2s0\n"
)
self.assertListEqual(
await client.managed_interfaces(),
["enp0s0", "enp1s0", "wlp2s0"]
)
client.sock.send.assert_called_once_with(b"INTERFACES")
@anyio_mock.with_anyio()
async def test_add_interface(self):
"""
Check add_interface() sends the correct arguments
"""
async with self.client as client:
client.ctrl_dir = pathlib.Path('/tmp')
client.sock.recv.return_value = b"OK"
assert await client.add_interface('enp1s0', driver='wired') is None
client.sock.send.assert_called_once()
args = client.sock.send.call_args[0]
assert args[0].startswith(b"INTERFACE_ADD enp1s0\t\twired\t/tmp\t")
@patch('wpa_supplicant.client.interfaces.InterfaceClient.connect',
new_callable=anyio_mock.AsyncMock)
@anyio_mock.with_anyio()
async def test_connect_interface(self, connect_mock):
"""
Check connect_interface() returns a connected InterfaceClient
"""
async with self.client as client:
client.ctrl_dir = pathlib.Path('/tmp')
client.sock.recv.side_effect = [
b"enp1s0\n", # Response to INTERFACES
]
ifclient = await client.connect_interface('enp1s0')
self.assertIsInstance(ifclient, interfaces.InterfaceClient)
connect_mock.assert_called_once_with("/tmp/enp1s0")
# Check only INTERFACES was sent
client.sock.send.assert_called_once_with(b"INTERFACES")
@patch('wpa_supplicant.client.interfaces.InterfaceClient.connect',
new_callable=anyio_mock.AsyncMock)
@anyio_mock.with_anyio()
async def test_connect_interface_with_add(self, connect_mock):
"""
Check connect_interface() adds the interface when not already managed
"""
async with self.client as client:
client.ctrl_dir = pathlib.Path('/tmp')
client.sock.recv.side_effect = [
b"", # Response to INTERFACES
b"OK", # Response to INTERFACE_ADD
]
ifclient = await client.connect_interface('enp1s0')
self.assertIsInstance(ifclient, interfaces.InterfaceClient)
connect_mock.assert_called_once_with("/tmp/enp1s0")
# Check INTERFACE_ADD sent after INTERFACES
args = client.sock.send.call_args_list
self.assertTupleEqual(args[0][0], (b"INTERFACES",))
assert args[1][0][0].startswith(b"INTERFACE_ADD enp1s0\t")