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
|
import multiprocessing
import subprocess
import sys
import time
import unittest
if sys.platform == "darwin" and multiprocessing.get_start_method(True) is None:
multiprocessing.set_start_method("fork")
import odil
class TestEchoSCP(object):
def test_echo_scp_release(self):
process = multiprocessing.Process(target=lambda: self.run_server(False))
process.start()
time.sleep(0.5)
client_code = self.run_client(False)
process.join(2)
server_code = process.exitcode
self.assertEqual(client_code, 0)
self.assertEqual(server_code, 0)
def test_echo_scp_abort(self):
process = multiprocessing.Process(target=lambda: self.run_server(True))
process.start()
time.sleep(0.5)
client_code = self.run_client(True)
process.join(2)
server_code = process.exitcode
self.assertEqual(client_code, 0)
self.assertEqual(server_code, 0)
def run_client(self, use_abort):
command = ["echoscu", "-q"]
if use_abort:
command.append("--abort")
command.extend(["localhost", "11113"])
return subprocess.call(command)
def run_server(self, use_abort):
called = [False]
def echo_callback(message):
called[0] = True
return 0
association = odil.Association()
association.set_tcp_timeout(1)
association.receive_association("v4", 11113)
echo_scp = odil.EchoSCP(association)
echo_scp.set_callback(echo_callback)
message = association.receive_message()
echo_scp(message)
termination_ok = False
try:
association.receive_message()
except odil.AssociationReleased:
if not use_abort:
termination_ok = True
except odil.AssociationAborted:
if use_abort:
termination_ok = True
if called[0] and termination_ok:
sys.exit(0)
else:
sys.exit(1)
if __name__ == "__main__":
unittest.main()
|
