-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_server.py
97 lines (73 loc) · 2.9 KB
/
test_server.py
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
from pyccs import Server
import pytest
from unittest.mock import patch, Mock
import ctypes
#make sure server is running before running these tests
@pytest.fixture
def connected_server():
server = Server('127.0.0.1', 10000)
server.connect()
yield server
def test_construction():
import ipaddress
s = Server('192.168.0.1', 500)
assert s is not None
addr = ipaddress.IPv4Address('192.168.0.1')
s = Server(addr, 500)
assert s is not None
addr = ipaddress.IPv6Address('2001:db8:3333:4444:5555:6666:7777:8888')
s = Server(addr, 500)
assert s is not None
def test_connect():
server = Server('127.0.0.1', 10000)
retval = server.connect()
assert retval == 0
def test_sendrecv(connected_server):
send_hdlr = "ping\0".encode('ascii')
send_msg = "ping\0".encode('ascii')
retval = connected_server.send_request(send_hdlr, 0, send_msg)
assert retval == 0
response = connected_server.receive_response(35,1)
assert response == b'hello ping from processor 0\n\x00\x00\x00\x00\x00\x00\x00'
@pytest.mark.skip(reason="Server currently has no functionality to handle broadcast")
def test_broadcast(connected_server):
send_hdlr = "ping\0".encode('ascii')
send_msg = "ping\0".encode('ascii')
retval = connected_server.send_broadcast_request(send_hdlr, send_msg)
assert retval == 0
response = connected_server.receive_response(35,1)
assert response == b'hello ping from processor 0\n\x00\x00\x00\x00\x00\x00\x00'
response = connected_server.receive_response(35,1)
assert response == b'hello ping from processor 1\n\x00\x00\x00\x00\x00\x00\x00'
def test_metadata(connected_server):
n_pes = connected_server.num_pes()
n_nodes = connected_server.num_nodes()
node_size = connected_server.node_size(0)
assert n_pes == 2
assert n_nodes == 2
assert node_size == 1
def test_receive_response_msg(connected_server):
send_hdlr = "ping\0".encode('ascii')
send_msg = "steven\0".encode('ascii')
retval = connected_server.send_request(send_hdlr, 0, send_msg)
assert retval == 0
responsemsg = connected_server.receive_response_message()
assert responsemsg == b'hello steven from processor 0\n\x00'
retval = connected_server.send_request(send_hdlr, 1, send_msg)
assert retval == 0
responsemsg = connected_server.receive_response_message()
assert responsemsg == b'hello steven from processor 1\n\x00'
#testing with a larger message
send_msg = "a"*800
expected_res = f"hello {send_msg} from processor 0\n\x00".encode('ascii')
send_msg = "a"*800 + "\0"
retval = connected_server.send_request(send_hdlr, 0, send_msg.encode('ascii'))
assert retval == 0
responsemsg = connected_server.receive_response_message()
assert responsemsg == expected_res
#testing with an invalid PE
try:
retval = connected_server.send_request(send_hdlr, 10000, "invalid message\0".encode('ascii'))
assert retval != 0
except Exception as e:
print(f"Excepted error for invalid PE: {e}")