1 | """Async TCP Server tests.""" |
||
2 | |||
3 | import asyncio |
||
4 | import errno |
||
5 | import logging |
||
6 | from unittest.mock import MagicMock, patch |
||
7 | |||
8 | from kytos.core.atcp_server import (KytosServer, KytosServerProtocol, |
||
9 | exception_handler) |
||
10 | |||
11 | # Using "nettest" TCP port as a way to avoid conflict with a running |
||
12 | # Kytos server on 6653. |
||
13 | TEST_ADDRESS = ('127.0.0.1', 4138) |
||
14 | |||
15 | |||
16 | class TestKytosServer: |
||
17 | """Test if a Kytos Server will go up and receive connections.""" |
||
18 | |||
19 | def setup(self): |
||
20 | """Start new asyncio loop and a test TCP server.""" |
||
21 | # pylint: disable=attribute-defined-outside-init |
||
22 | self.loop = asyncio.new_event_loop() |
||
23 | asyncio.set_event_loop(None) |
||
24 | self.server = KytosServer(TEST_ADDRESS, KytosServerProtocol, |
||
25 | None, 'openflow', loop=self.loop) |
||
26 | self.server.serve_forever() |
||
27 | |||
28 | def test_connection_to_server(self): |
||
29 | """Test if we really can connect to TEST_ADDRESS.""" |
||
30 | @asyncio.coroutine |
||
31 | def wait_and_go(): |
||
32 | """Wait a little for the server to go up, then connect.""" |
||
33 | yield from asyncio.sleep(0.01, loop=self.loop) |
||
34 | # reader, writer = ... |
||
35 | _ = yield from asyncio.open_connection( |
||
36 | *TEST_ADDRESS, loop=self.loop) |
||
37 | |||
38 | self.loop.run_until_complete(wait_and_go()) |
||
39 | |||
40 | def test_exception_handler_oserror(self, caplog): |
||
41 | """Test the TCP Server Exception Handler. |
||
42 | |||
43 | 1. create mock OSError/TimeoutError instances |
||
44 | 2. call exception_handler with them |
||
45 | 3. ensure log is OK |
||
46 | """ |
||
47 | caplog.set_level(logging.INFO) |
||
48 | |||
49 | exception = TimeoutError() |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
Loading history...
|
|||
50 | context = {"exception": exception, |
||
51 | "transport": "unit_tests"} |
||
52 | exception_handler(self.loop, context) |
||
53 | |||
54 | exception2 = OSError(errno.EBADF, "Bad file descriptor") |
||
55 | context2 = {"exception": exception2, |
||
56 | "transport": "unit_tests"} |
||
57 | exception_handler(self.loop, context2) |
||
58 | |||
59 | assert caplog.record_tuples == [ |
||
60 | ("kytos.core.atcp_server", |
||
61 | logging.INFO, |
||
62 | "Socket timeout: 'unit_tests'"), |
||
63 | ("kytos.core.atcp_server", |
||
64 | logging.INFO, |
||
65 | "Socket closed: 'unit_tests'"), |
||
66 | ] |
||
67 | |||
68 | |||
69 | class TestKytosServerProtocol: |
||
70 | """KytosServerProtocol tests.""" |
||
71 | |||
72 | def setup(self): |
||
73 | """Instantiate a KytosServerProtocol.""" |
||
74 | # pylint: disable=attribute-defined-outside-init |
||
75 | loop = asyncio.new_event_loop() |
||
76 | asyncio.set_event_loop(loop) |
||
77 | |||
78 | self.connection = MagicMock() |
||
79 | self.connection.address = 'addr' |
||
80 | self.connection.port = 123 |
||
81 | |||
82 | self.server_protocol = KytosServerProtocol() |
||
83 | self.server_protocol.server = MagicMock() |
||
84 | self.server_protocol.connection = self.connection |
||
85 | |||
86 | @patch('kytos.core.atcp_server.KytosEvent') |
||
87 | def test_data_received(self, mock_kytos_event): |
||
88 | """Test data_received method.""" |
||
89 | buffers = self.server_protocol.server.controller.buffers |
||
90 | self.connection.protocol.name = 'protocol' |
||
91 | |||
92 | self.server_protocol.data_received(b'data') |
||
93 | |||
94 | expected_content = {'source': self.connection, 'new_data': b'data'} |
||
95 | expected_name = 'kytos/core.protocol.raw.in' |
||
96 | mock_kytos_event.assert_called_with(content=expected_content, |
||
97 | name=expected_name) |
||
98 | buffers.raw.aput.assert_called_with(mock_kytos_event.return_value) |
||
99 | |||
100 | @patch('kytos.core.atcp_server.KytosEvent') |
||
101 | def test_connection_lost(self, mock_kytos_event): |
||
102 | """Test connection_lost method.""" |
||
103 | buffers = self.server_protocol.server.controller.buffers |
||
104 | self.connection.protocol.name = 'protocol' |
||
105 | |||
106 | self.server_protocol.connection_lost('exc') |
||
107 | |||
108 | self.connection.close.assert_called() |
||
109 | expected_content = {'source': self.connection, 'exception': 'exc'} |
||
110 | expected_name = 'kytos/core.protocol.connection.lost' |
||
111 | mock_kytos_event.assert_called_with(content=expected_content, |
||
112 | name=expected_name) |
||
113 | buffers.app.aput.assert_called_with(mock_kytos_event.return_value) |
||
114 |