|
12 | 12 | from parsl.executors.high_throughput.manager_selector import RandomManagerSelector
|
13 | 13 |
|
14 | 14 |
|
15 |
| -def make_interchange(*, interchange_address: Optional[str], cert_dir: Optional[str]) -> Interchange: |
| 15 | +def make_interchange(*, |
| 16 | + interchange_address: Optional[str], |
| 17 | + cert_dir: Optional[str], |
| 18 | + worker_ports: Optional[tuple[int, int]] = None) -> Interchange: |
16 | 19 | return Interchange(interchange_address=interchange_address,
|
17 | 20 | cert_dir=cert_dir,
|
18 | 21 | client_address="127.0.0.1",
|
19 | 22 | client_ports=(50055, 50056, 50057),
|
20 |
| - worker_ports=None, |
| 23 | + worker_ports=worker_ports, |
21 | 24 | worker_port_range=(54000, 55000),
|
22 | 25 | hub_address=None,
|
23 | 26 | hub_zmq_port=None,
|
@@ -105,3 +108,10 @@ def test_limited_interface_binding(cert_dir: Optional[str]):
|
105 | 108 | assert len(matched_conns) == 1
|
106 | 109 | # laddr.ip can return ::ffff:127.0.0.1 when using IPv6
|
107 | 110 | assert address in matched_conns[0].laddr.ip
|
| 111 | + |
| 112 | + |
| 113 | +@pytest.mark.local |
| 114 | +@pytest.mark.parametrize("encrypted", (True, False), indirect=True) |
| 115 | +def test_fixed_ports(cert_dir: Optional[str]): |
| 116 | + ix = make_interchange(interchange_address=None, cert_dir=cert_dir, worker_ports=(51117, 51118)) |
| 117 | + assert ix.interchange_address == "*" |
0 commit comments