/
dynamic_server_ports.test
106 lines (78 loc) · 2.87 KB
/
dynamic_server_ports.test
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
The storage server can be told to bind to port 0, allowing the OS to
pick a port dynamically. For this to be useful, there needs to be a
way to tell someone. For this reason, the server posts events to
ZODB.notify.
>>> import ZODB.event
>>> old_notify = ZODB.event.notify
>>> last_event = None
>>> def notify(event):
... global last_event
... last_event = event
>>> ZODB.event.notify = notify
Now, let's start a server and verify that we get a serving event:
>>> import ZEO.StorageServer, ZODB.MappingStorage
>>> server = ZEO.StorageServer.StorageServer(
... ('127.0.0.1', 0), {'1': ZODB.MappingStorage.MappingStorage()})
>>> isinstance(last_event, ZEO.StorageServer.Serving)
True
>>> last_event.server is server
True
>>> last_event.address[0], last_event.address[1] > 0
('127.0.0.1', True)
If the host part pf the address passed to the constructor is not an
empty string. then the server addr attribute is the same as the
address attribute of the event:
>>> server.addr == last_event.address
True
Let's run the server in a thread, to make sure we can connect.
>>> server.start_thread()
>>> client = ZEO.client(last_event.address)
>>> client.is_connected()
True
If we close the server, we'll get a closed event:
>>> server.close()
>>> isinstance(last_event, ZEO.StorageServer.Closed)
True
>>> last_event.server is server
True
>>> wait_until(lambda : not client.is_connected(test=True))
>>> client.close()
If we pass an empty string as the host part of the server address, we
can't really assign a single address, so the server addr attribute is
left alone:
>>> server = ZEO.StorageServer.StorageServer(
... ('', 0), {'1': ZODB.MappingStorage.MappingStorage()})
>>> isinstance(last_event, ZEO.StorageServer.Serving)
True
>>> last_event.server is server
True
>>> last_event.address[1] > 0
True
If the host part pf the address passed to the constructor is not an
empty string. then the server addr attribute is the same as the
address attribute of the event:
>>> server.addr
('', 0)
>>> server.close()
The runzeo module provides some process support, including getting the
server configuration via a ZConfig configuration file. To spell a
dynamic port using ZConfig, you'd use a hostname by itself. In this
case, ZConfig passes None as the port.
>>> import ZEO.runzeo
>>> r = open('conf', 'w').write("""
... <zeo>
... address 127.0.0.1
... </zeo>
... <mappingstorage>
... </mappingstorage>
... """)
>>> options = ZEO.runzeo.ZEOOptions()
>>> options.realize('-C conf'.split())
>>> options.address
('127.0.0.1', None)
>>> rs = ZEO.runzeo.ZEOServer(options)
>>> rs.check_socket()
>>> options.address
('127.0.0.1', 0)
.. cleanup
>>> ZODB.event.notify = old_notify