| 178 | |
| 179 | |
| 180 | async def test_wireguard(tdata, monkeypatch, caplog): |
| 181 | caplog.set_level("DEBUG") |
| 182 | |
| 183 | monkeypatch.setattr(ConnectionHandler, "handle_client", _echo_server) |
| 184 | |
| 185 | system = platform.system() |
| 186 | if system == "Linux": |
| 187 | test_client_name = "linux-x86_64" |
| 188 | elif system == "Darwin": |
| 189 | test_client_name = "macos-x86_64" |
| 190 | elif system == "Windows": |
| 191 | test_client_name = "windows-x86_64.exe" |
| 192 | else: |
| 193 | return pytest.skip("Unsupported platform for wg-test-client.") |
| 194 | |
| 195 | arch = platform.machine() |
| 196 | if arch != "AMD64" and arch != "x86_64": |
| 197 | return pytest.skip("Unsupported architecture for wg-test-client.") |
| 198 | |
| 199 | test_client_path = tdata.path(f"wg-test-client/{test_client_name}") |
| 200 | test_conf = tdata.path(f"wg-test-client/test.conf") |
| 201 | |
| 202 | with taddons.context(Proxyserver()): |
| 203 | inst = WireGuardServerInstance.make(f"wireguard:{test_conf}@0", MagicMock()) |
| 204 | |
| 205 | await inst.start() |
| 206 | assert "WireGuard server listening" in caplog.text |
| 207 | |
| 208 | _, port = inst.listen_addrs[0] |
| 209 | |
| 210 | assert inst.is_running |
| 211 | proc = await asyncio.create_subprocess_exec( |
| 212 | test_client_path, |
| 213 | str(port), |
| 214 | stdout=asyncio.subprocess.PIPE, |
| 215 | stderr=asyncio.subprocess.PIPE, |
| 216 | ) |
| 217 | stdout, stderr = await proc.communicate() |
| 218 | |
| 219 | try: |
| 220 | assert proc.returncode == 0 |
| 221 | except AssertionError: |
| 222 | print(stdout) |
| 223 | print(stderr) |
| 224 | raise |
| 225 | |
| 226 | await inst.stop() |
| 227 | assert "stopped" in caplog.text |
| 228 | |
| 229 | |
| 230 | @pytest.mark.parametrize("host", ["127.0.0.1", "::1"]) |