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
|
# SPDX-FileCopyrightText: 2024 Greenbone AG
#
# SPDX-License-Identifier: GPL-3.0-or-later
import unittest
from gvm.errors import GvmError
from gvm.protocols.core import Connection, InvalidStateError
class RequestMock:
def __init__(self, data: bytes) -> None:
self._data = data
def __bytes__(self) -> bytes:
return self._data
class ConnectionTestCase(unittest.TestCase):
def test_send_request(self) -> None:
request = RequestMock(b"<request/>")
connection = Connection()
data = connection.send(request)
self.assertEqual(data, b"<request/>")
def test_receive_data(self) -> None:
request = RequestMock(b"<request/>")
connection = Connection()
connection.send(request)
response = connection.receive_data(b"<response")
self.assertIsNone(response)
response = connection.receive_data(b' status="200"/>')
self.assertIsNotNone(response)
self.assertTrue(response.is_success) # type: ignore
another_request = RequestMock(b"<another_request/>")
connection.send(another_request)
response = connection.receive_data(b"<another_response")
self.assertIsNone(response)
response = connection.receive_data(b' status="200"/>')
self.assertIsNotNone(response)
self.assertTrue(response.is_success) # type: ignore
def test_receive_invalid_data(self) -> None:
request = RequestMock(b"<request/>")
connection = Connection()
connection.send(request)
with self.assertRaisesRegex(
GvmError,
"^Cannot parse XML response. Response data read b'<response<>'$",
):
connection.receive_data(b"<response<>")
def test_error_state_close(self) -> None:
request = RequestMock(b"<request/>")
connection = Connection()
connection.send(request)
with self.assertRaisesRegex(
GvmError,
"^Cannot parse XML response. Response data read b'<response<>'$",
):
connection.receive_data(b"<response<>")
with self.assertRaisesRegex(
InvalidStateError,
"^The connection is in an error state. Please close the connection.$",
):
connection.receive_data(b"<response/>")
with self.assertRaisesRegex(
InvalidStateError,
"^The connection is in an error state. Please close the connection.$",
):
connection.send(request)
connection.close()
connection.send(request)
def test_send_when_receiving_data(self) -> None:
request = RequestMock(b"<request/>")
connection = Connection()
connection.send(request)
connection.receive_data(b"<response")
with self.assertRaisesRegex(
InvalidStateError,
"^Invalid State$",
):
connection.send(request)
def test_close_before_send(self) -> None:
request = RequestMock(b"<request/>")
connection = Connection()
connection.close()
data = connection.send(request)
self.assertEqual(data, b"<request/>")
def test_close_after_send(self) -> None:
request = RequestMock(b"<request/>")
connection = Connection()
data = connection.send(request)
self.assertEqual(data, b"<request/>")
connection.close()
with self.assertRaisesRegex(
InvalidStateError,
"^Invalid State$",
):
connection.receive_data(b"<response")
data = connection.send(request)
self.assertEqual(data, b"<request/>")
def test_close_after_receive_data(self) -> None:
request = RequestMock(b"<request/>")
connection = Connection()
data = connection.send(request)
self.assertEqual(data, b"<request/>")
connection.receive_data(b"<response")
connection.close()
with self.assertRaisesRegex(
InvalidStateError,
"^Invalid State$",
):
connection.receive_data(b' status="200"/>')
data = connection.send(request)
self.assertEqual(data, b"<request/>")
|
