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
|
import sys
import pytest
from scrapli.driver.base.sync_driver import BaseDriver
from scrapli.exceptions import ScrapliTypeError
@pytest.mark.parametrize(
"test_data",
(
("comms_prompt_pattern", "new_value"),
("comms_return_char", "new_value"),
),
ids=(
"comms_prompt_pattern",
"comms_return_char",
),
)
def test_comms_properties(base_drivers, test_data):
"""
Assert sync driver properly sets comms property values
"""
attribute, new_value = test_data
original_value = getattr(base_drivers, attribute)
assert getattr(base_drivers._base_channel_args, attribute) == original_value
assert getattr(base_drivers.channel._base_channel_args, attribute) == original_value
setattr(base_drivers, attribute, new_value)
assert getattr(base_drivers, attribute) == new_value
assert getattr(base_drivers._base_channel_args, attribute) == new_value
assert getattr(base_drivers.channel._base_channel_args, attribute) == new_value
@pytest.mark.parametrize(
"test_data",
(
("comms_prompt_pattern", False),
("comms_return_char", False),
),
ids=("comms_prompt_pattern", "comms_return_char"),
)
def test_comms_properties_exception(base_drivers, test_data):
"""Assert sync driver raises an exception for invalid comms property values"""
attribute, new_value = test_data
with pytest.raises(ScrapliTypeError):
setattr(base_drivers, attribute, new_value)
@pytest.mark.parametrize(
"test_data",
(
("timeout_socket", 999.99),
("timeout_transport", 999.99),
),
ids=(
"timeout_socket",
"timeout_transport",
),
)
def test_timeout_properties_transport(base_drivers, test_data):
"""
Assert sync driver properly sets transport related timeout values
"""
attribute, new_value = test_data
original_value = getattr(base_drivers, attribute)
assert getattr(base_drivers._base_transport_args, attribute) == original_value
assert getattr(base_drivers.transport._base_transport_args, attribute) == original_value
setattr(base_drivers, attribute, new_value)
assert getattr(base_drivers, attribute) == new_value
assert getattr(base_drivers._base_transport_args, attribute) == new_value
assert getattr(base_drivers.transport._base_transport_args, attribute) == new_value
def test_timeout_properties_transport_plugin_set_timeout(monkeypatch):
"""
Assert sync driver properly sets transport related timeout values
Specifically for plugins with set_timeout method -- i.e. ssh2/paramiko
"""
monkeypatch.setattr(
"scrapli.transport.plugins.paramiko.transport.ParamikoTransport._set_timeout",
lambda cls, value: None,
)
driver = BaseDriver(host="localhost", transport="paramiko")
assert driver.timeout_transport == 30.0
driver.timeout_transport = 999.99
assert driver.timeout_transport == 999.99
@pytest.mark.parametrize(
"test_data",
(
("timeout_socket", None),
("timeout_transport", None),
),
ids=(
"timeout_socket",
"timeout_transport",
),
)
def test_timeout_properties_transport_exception(base_drivers, test_data):
"""Assert sync driver raises an exception for invalid transport timeout values"""
attribute, new_value = test_data
with pytest.raises(ScrapliTypeError):
setattr(base_drivers, attribute, new_value)
@pytest.mark.parametrize(
"test_data",
(("timeout_ops", 999.99),),
ids=("timeout_ops",),
)
def test_timeout_properties_channel(base_drivers, test_data):
"""
Assert sync driver properly sets transport related timeout values
"""
attribute, new_value = test_data
original_value = getattr(base_drivers, attribute)
assert getattr(base_drivers._base_channel_args, attribute) == original_value
assert getattr(base_drivers.channel._base_channel_args, attribute) == original_value
setattr(base_drivers, attribute, new_value)
assert getattr(base_drivers, attribute) == new_value
assert getattr(base_drivers._base_channel_args, attribute) == new_value
assert getattr(base_drivers.channel._base_channel_args, attribute) == new_value
@pytest.mark.parametrize(
"test_data",
(("timeout_ops", None),),
ids=("timeout_ops",),
)
def test_timeout_properties_channel_exception(base_drivers, test_data):
"""Assert sync driver raises an exception for invalid transport timeout values"""
attribute, new_value = test_data
with pytest.raises(ScrapliTypeError):
setattr(base_drivers, attribute, new_value)
|
