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
|
from time import sleep
from os import (
devnull,
)
from random import randint
from subprocess import Popen, check_call
from tempfile import NamedTemporaryFile
import socket
from pathlib import Path
import pytest
import pytest_asyncio
from librouteros import connect, async_connect
from librouteros.exceptions import LibRouterosError
from librouteros.login import (
plain,
token,
async_plain,
async_token,
)
DEV_NULL = open(devnull, "w")
VERSION_LOGIN = {
"6.44.5": {
"sync": plain,
"async": async_plain,
},
"6.33.3": {
"sync": token,
"async": async_token,
},
}
def api_session(port):
last_exc = None
for _ in range(30):
try:
return connect(
host="127.0.0.1",
port=port,
username="admin",
password="",
timeout=60,
)
except (LibRouterosError, socket.error, socket.timeout) as exc:
last_exc = exc
sleep(1)
raise RuntimeError("Could not connect to device. Last exception {}".format(last_exc))
async def api_session_async(port):
last_exc = None
for _ in range(30):
try:
return await async_connect(
host="127.0.0.1",
port=port,
username="admin",
password="",
timeout=60,
)
except (LibRouterosError, socket.error, socket.timeout) as exc:
last_exc = exc
sleep(1)
raise RuntimeError("Could not connect to device. Last exception {}".format(last_exc))
def disk_image(version):
"""Create a temporary disk image backed by original one."""
img = NamedTemporaryFile()
# Path to backing image must be absolute or relative to new image
backing_img = Path().joinpath("images/routeros_{}.qcow2".format(version)).absolute()
cmd = [
"qemu-img",
"create",
"-f",
"qcow2",
"-F",
"qcow2",
"-b",
str(backing_img),
img.name,
]
check_call(cmd, stdout=DEV_NULL)
return img
def routeros_vm(disk_image):
# pylint: disable=redefined-outer-name
port = randint(49152, 65535)
cmd = [
"qemu-system-x86_64",
"-m",
"64",
"-display",
"none",
"-hda",
disk_image.name,
"-net",
"user,hostfwd=tcp::{}-:8728".format(port),
"-net",
"nic,model=virtio",
"-cpu",
"max",
]
proc = Popen(cmd, stdout=DEV_NULL, close_fds=True)
return port, proc
@pytest.fixture(scope="function", params=VERSION_LOGIN.keys())
def routeros_login(request):
# pylint: disable=redefined-outer-name
version = request.param
image = disk_image(version)
port, proc = routeros_vm(image)
request.addfinalizer(proc.kill)
request.addfinalizer(image.close)
def get_login_method(exc_type):
return port, VERSION_LOGIN[version][exc_type]
return get_login_method
@pytest.fixture(scope="function")
def routeros_api(request):
# pylint: disable=redefined-outer-name
version = "6.44.5"
image = disk_image(version)
port, proc = routeros_vm(image)
request.addfinalizer(proc.kill)
request.addfinalizer(image.close)
return api_session(port=port)
@pytest_asyncio.fixture(scope="function")
async def routeros_api_async(request):
# pylint: disable=redefined-outer-name
version = "6.44.5"
image = disk_image(version)
port, proc = routeros_vm(image)
request.addfinalizer(proc.kill)
request.addfinalizer(image.close)
return await api_session_async(port=port)
|
