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
|
import asyncio
import time
from concurrent.futures import ThreadPoolExecutor
import pytest
from run import sync_run_servers
from scrapli.driver.core import (
AsyncEOSDriver,
AsyncIOSXEDriver,
AsyncIOSXRDriver,
AsyncJunosDriver,
AsyncNXOSDriver,
EOSDriver,
IOSXEDriver,
IOSXRDriver,
JunosDriver,
NXOSDriver,
)
DEVICES = {
"common": {
"host": "localhost",
"auth_username": "scrapli",
"auth_password": "scrapli",
"auth_secondary": "scrapli",
"auth_strict_key": False,
},
"cisco_iosxe": {
"driver": IOSXEDriver,
"async_driver": AsyncIOSXEDriver,
"port": 2221,
},
"cisco_nxos": {
"driver": NXOSDriver,
"async_driver": AsyncNXOSDriver,
"port": 2222,
},
"cisco_iosxr": {
"driver": IOSXRDriver,
"async_driver": AsyncIOSXRDriver,
"port": 2223,
},
"arista_eos": {
"driver": EOSDriver,
"async_driver": AsyncEOSDriver,
"port": 2224,
},
"juniper_junos": {
"driver": JunosDriver,
"async_driver": AsyncJunosDriver,
"port": 2225,
},
}
@pytest.fixture(scope="module", autouse=True)
def mock_ssh_servers():
with ThreadPoolExecutor(max_workers=1) as pool:
loop = asyncio.new_event_loop()
pool.submit(sync_run_servers, loop)
time.sleep(1.5)
# yield to let all the tests run, then we can deal w/ cleaning up the thread/loop
# we need to have this scoped to module so it starts/stops just for integration tests
yield
loop.call_soon_threadsafe(loop.stop)
# seems a little sleep before and after starting/stopping makes things a little smoother...
time.sleep(1.5)
@pytest.fixture(
scope="function",
params=["cisco_iosxe", "cisco_nxos", "cisco_iosxr", "arista_eos", "juniper_junos"],
)
def device_type(request):
yield request.param
@pytest.fixture(scope="function", params=["system", "ssh2", "paramiko"])
def transport(request):
yield request.param
@pytest.fixture(scope="function")
def sync_conn(device_type, transport):
driver = DEVICES[device_type]["driver"]
port = DEVICES[device_type]["port"]
if transport == "telnet":
port = port + 1
conn = driver(
port=port,
transport=transport,
**DEVICES["common"],
)
conn.open()
yield conn
try:
conn.close()
except EOFError:
# sometimes paramiko has a fit and raises EOF... havent been able to tell why, but tired of
# builds failing for this one tiny issue
pass
|
