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
139
140
141
142
143
144
145
146
147
148
149
150
151
|
# -*- coding: UTF-8 -*-
from unittest.mock import MagicMock
from librouteros.api import Api, Path, AsyncApi, AyncPath
from librouteros.query import Query, AsyncQuery
import pytest
def test_api_path_returns_Path():
api = Api(protocol=MagicMock())
new = api.path("ip", "address")
assert new.path == "/ip/address"
assert new.api == api
assert isinstance(new, Path)
def test_async_api_path_returns_AsyncPath():
api = AsyncApi(protocol=MagicMock())
new = api.path("ip", "address")
assert new.path == "/ip/address"
assert new.api == api
assert isinstance(new, AyncPath)
class Test_Path:
def setup_method(self):
self.path = Path(
path="/interface",
api=MagicMock(),
)
self.async_path = AyncPath(
path="/interface",
api=MagicMock(),
)
def test_path_str(self):
assert str(self.path) == self.path.path
assert str(self.async_path) == self.async_path.path
def test_join_single_param(self):
assert self.path.join("ethernet").path == self.path.path + "/ethernet"
assert self.async_path.join("ethernet").path == self.async_path.path + "/ethernet"
def test_join_multi_param(self):
assert self.path.join("ethernet", "print").path == self.path.path + "/ethernet/print"
assert self.async_path.join("ethernet", "print").path == self.async_path.path + "/ethernet/print"
def test_join_rstrips_slash(self):
assert self.path.join("ethernet", "print/").path == self.path.path + "/ethernet/print"
assert self.async_path.join("ethernet", "print/").path == self.async_path.path + "/ethernet/print"
def test_select_returns_Query(self):
new = self.path.select("disabled", "name")
assert isinstance(new, Query)
new = self.async_path.select("disabled", "name")
assert isinstance(new, AsyncQuery)
def test_select_Query_has_valid_attributes(self):
new = self.path.select("disabled", "name")
assert new.api == self.path.api
assert new.path == self.path
assert new.keys == ("disabled", "name")
new = self.async_path.select("disabled", "name")
assert new.api == self.async_path.api
assert new.path == self.async_path
assert new.keys == ("disabled", "name")
@pytest.mark.asyncio
async def test_remove(self):
self.path.remove("*1", "*2", "*3")
self.path.api.assert_called_once_with("/interface/remove", **{".id": "*1,*2,*3"})
# Check if returned generator was consumed
assert self.path.api.return_value.__iter__.call_count == 1
await self.async_path.remove("*1", "*2", "*3")
self.async_path.api.assert_called_once_with("/interface/remove", **{".id": "*1,*2,*3"})
# Check if returned generator was consumed
assert self.async_path.api.return_value.__aiter__.call_count == 1
@pytest.mark.asyncio
async def test_add(self):
return_value = ({"ret": "*1"},)
self.path.api.return_value = return_value
new = {"interface": "ether1", "address": "172.1.1.1/24"}
new_id = self.path.add(**new)
self.path.api.assert_called_once_with("/interface/add", **new)
assert new_id == "*1"
# async
# mock async iterable
async def mock_api(*args, **kwargs):
for item in return_value:
yield item
self.async_path.api.side_effect = mock_api
new_id = await self.async_path.add(**new)
self.async_path.api.assert_called_once_with("/interface/add", **new)
assert new_id == "*1"
@pytest.mark.asyncio
async def test_update(self):
args = {"name": "wan", ".id": "*1"}
self.path.update(**args)
self.path.api.assert_called_once_with("/interface/set", **args)
# Check if returned generator was consumed
assert self.path.api.return_value.__iter__.call_count == 1
await self.async_path.update(**args)
self.async_path.api.assert_called_once_with("/interface/set", **args)
assert self.async_path.api.return_value.__aiter__.call_count == 1
@pytest.mark.asyncio
async def test_iter(self):
items = ({".id": "*1"}, {".id": "*2"})
self.path.api.return_value = items
new_items = tuple(self.path)
assert new_items == items
self.path.api.assert_called_once_with("/interface/print")
# async
# mock async iterable
async def mock_api(*args, **kwargs):
for item in items:
yield item
self.async_path.api.side_effect = mock_api
new_items = tuple([r async for r in self.async_path])
assert new_items == items
self.async_path.api.assert_called_once_with("/interface/print")
@pytest.mark.asyncio
async def test_call(self):
self.path.path = "/system/script"
tuple(self.path("run"))
self.path.api.assert_called_once_with("/system/script/run")
# Check if returned generator was consumed
assert self.path.api.return_value.__iter__.call_count == 1
# Async
self.async_path.path = "/system/script"
[r async for r in self.async_path("run")]
self.async_path.api.assert_called_once_with("/system/script/run")
# Check if returned generator was consumed
assert self.async_path.api.return_value.__aiter__.call_count == 1
def test_select_without_keys(self):
query = self.path.select()
assert isinstance(query, Query)
# Async
query = self.async_path.select()
assert isinstance(query, AsyncQuery)
|
