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
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
|
import sys
from datetime import datetime
import pytest
from scrapli.exceptions import ScrapliCommandFailure
from scrapli.helper import _textfsm_get_template
from scrapli.response import MultiResponse, Response
def test_response_init():
response = Response("localhost", "ls -al", failed_when_contains="tacocat")
response_start_time = str(datetime.now())[:-7]
assert response.host == "localhost"
assert response.channel_input == "ls -al"
assert str(response.start_time)[:-7] == response_start_time
assert response.failed is True
assert bool(response) is True
assert (
repr(response)
== "Response(host='localhost',channel_input='ls -al',textfsm_platform='',genie_platform='',failed_when_contains=['tacocat'])"
)
assert str(response) == "Response <Success: False>"
assert response.failed_when_contains == ["tacocat"]
with pytest.raises(ScrapliCommandFailure):
response.raise_for_status()
def test_multi_response():
host = "localhost"
response1 = Response(host, "ls -al")
response2 = Response(host, "ls -al")
multi_response = MultiResponse()
assert multi_response.host == ""
multi_response.extend([response1, response2])
assert len(multi_response) == 2
assert multi_response.failed is True
assert (
repr(multi_response)
== "[Response(host='localhost',channel_input='ls -al',textfsm_platform='',genie_platform='',failed_when_contains=None), Response(host='localhost',channel_input='ls -al',textfsm_platform='',genie_platform='',failed_when_contains=None)]"
)
assert str(multi_response) == "MultiResponse <Success: False; Response Elements: 2>"
with pytest.raises(ScrapliCommandFailure):
multi_response.raise_for_status()
multi_response[0].failed = False
multi_response[1].failed = False
assert multi_response.failed is False
assert multi_response.raise_for_status() is None
assert multi_response.host == host
assert (
repr(multi_response)
== "[Response(host='localhost',channel_input='ls -al',textfsm_platform='',genie_platform='',failed_when_contains=None), Response(host='localhost',channel_input='ls -al',textfsm_platform='',genie_platform='',failed_when_contains=None)]"
)
assert str(multi_response) == "MultiResponse <Success: True; Response Elements: 2>"
assert multi_response.result == "ls -al\nls -al\n"
def test_response_record_result():
response = Response("localhost", "ls -al")
response_end_time = str(datetime.now())[:-7]
response_bytes = b"""
ls -al
total 264
drwxr-xr-x 34 carl staff 1088 Jan 27 19:07 ./
drwxr-xr-x 21 carl staff 672 Jan 25 15:56 ../
-rw-r--r-- 1 carl staff 53248 Jan 27 19:07 .coverage
drwxr-xr-x 12 carl staff 384 Jan 27 19:13 .git/"""
response.record_response(response_bytes)
assert str(response.finish_time)[:-7] == response_end_time
assert response.result == response_bytes.decode()
assert response.failed is False
def test_response_record_result_failed_when_failed():
response = Response("localhost", "ls -al", failed_when_contains=["!racecar!"])
response_end_time = str(datetime.now())[:-7]
response_bytes = b"""
ls -al
total 264
drwxr-xr-x 34 carl staff 1088 Jan 27 19:07 ./
drwxr-xr-x 21 carl staff 672 Jan 25 15:56 ../
-rw-r--r-- 1 carl staff 53248 Jan 27 19:07 !racecar!
drwxr-xr-x 12 carl staff 384 Jan 27 19:13 .git/"""
response.record_response(response_bytes)
assert str(response.finish_time)[:-7] == response_end_time
assert response.result == response_bytes.decode()
assert response.failed is True
def test_response_record_result_failed_when_success():
response = Response("localhost", "ls -al", failed_when_contains=["!racecar!"])
response_end_time = str(datetime.now())[:-7]
response_bytes = b"""
ls -al
total 264
drwxr-xr-x 34 carl staff 1088 Jan 27 19:07 ./
drwxr-xr-x 21 carl staff 672 Jan 25 15:56 ../
-rw-r--r-- 1 carl staff 53248 Jan 27 19:07 .coverage
drwxr-xr-x 12 carl staff 384 Jan 27 19:13 .git/"""
response.record_response(response_bytes)
assert str(response.finish_time)[:-7] == response_end_time
assert response.result == response_bytes.decode()
assert response.failed is False
@pytest.mark.parametrize(
"parse_type",
[
(
False,
["Internet", "172.31.254.1", "-", "0000.0c07.acfe", "ARPA", "Vlan254"],
),
(
True,
{
"protocol": "Internet",
"address": "172.31.254.1",
"age": "-",
"mac": "0000.0c07.acfe",
"type": "ARPA",
"interface": "Vlan254",
},
),
],
ids=["to_dict_false", "to_dict_true"],
)
def test_response_parse_textfsm(parse_type):
to_dict = parse_type[0]
expected_result = parse_type[1]
response = Response("localhost", channel_input="show ip arp", textfsm_platform="cisco_ios")
response_bytes = b"""Protocol Address Age (min) Hardware Addr Type Interface
Internet 172.31.254.1 - 0000.0c07.acfe ARPA Vlan254
Internet 172.31.254.2 - c800.84b2.e9c2 ARPA Vlan254
"""
response.record_response(response_bytes)
assert response.textfsm_parse_output(to_dict=to_dict)[0] == expected_result
def test_response_parse_textfsm_string_path():
template = _textfsm_get_template("cisco_ios", "show ip arp").name
response = Response("localhost", channel_input="show ip arp", textfsm_platform="cisco_ios")
response_bytes = b"""Protocol Address Age (min) Hardware Addr Type Interface
Internet 172.31.254.1 - 0000.0c07.acfe ARPA Vlan254
Internet 172.31.254.2 - c800.84b2.e9c2 ARPA Vlan254
"""
response.record_response(response_bytes)
result = response.textfsm_parse_output(template=template)
assert result[0]["address"] == "172.31.254.1"
def test_response_parse_textfsm_fail():
response = Response("localhost", channel_input="show ip arp", textfsm_platform="cisco_ios")
response_bytes = b""
response.record_response(response_bytes)
assert response.textfsm_parse_output() == []
def test_response_parse_textfsm_no_template():
response = Response("localhost", channel_input="show ip arp", textfsm_platform="potato")
response_bytes = b""
response.record_response(response_bytes)
assert response.textfsm_parse_output() == []
@pytest.mark.skipif(
sys.version_info.minor > 10, reason="genie not currently available for python 3.11"
)
def test_response_parse_genie():
response = Response("localhost", channel_input="show ip arp", genie_platform="iosxe")
response_bytes = b"""Protocol Address Age (min) Hardware Addr Type Interface
Internet 172.31.254.1 - 0000.0c07.acfe ARPA Vlan254
Internet 172.31.254.2 - c800.84b2.e9c2 ARPA Vlan254
"""
response.record_response(response_bytes)
result = response.genie_parse_output()
assert (
result["interfaces"]["Vlan254"]["ipv4"]["neighbors"]["172.31.254.1"]["ip"] == "172.31.254.1"
)
@pytest.mark.skipif(
sys.version_info.minor > 10, reason="genie not currently available for python 3.11"
)
def test_response_parse_genie_fail():
response = Response("localhost", channel_input="show ip arp", genie_platform="iosxe")
response_bytes = b""
response.record_response(response_bytes)
assert response.genie_parse_output() == []
def test_response_parse_ttp():
response = Response("localhost", channel_input="show ip arp", genie_platform="iosxe")
# example data lifted straight out of ttp docs
ttp_template = """
interface {{ interface }}
ip address {{ ip }}/{{ mask }}
description {{ description }}
ip vrf {{ vrf }}
"""
response_bytes = b""" interface Loopback0
description Router-id-loopback
ip address 192.168.0.113/24
!
interface Vlan778
description CPE_Acces_Vlan
ip address 2002::fd37/124
ip vrf CPE1
!
"""
response.record_response(response_bytes)
result = response.ttp_parse_output(template=ttp_template)
assert result[0][0]["ip"] == "192.168.0.113"
def test_response_parse_ttp_fail():
response = Response("localhost", channel_input="show ip arp", genie_platform="iosxe")
response_bytes = b""
response.record_response(response_bytes)
assert response.ttp_parse_output(template="blah") == [{}]
def test_record_response_unicodedecodeerror():
# this test validates that we catch unicdedecodeerror when decoding raw result
response = Response("localhost", channel_input="show ip arp", genie_platform="iosxe")
response_bytes = b"Manufacturer name :p\xb67\x038\x93\xa5\x03\x10\n"
response.record_response(response_bytes)
assert repr(response.result) == "'Manufacturer name :p¶7\\x038\\x93¥\\x03\\x10\\n'"
|
