"""Java reference implementation interop tests. Spins up a podman pod with the geti2p/i2p Java router and a Python container. Extracts the Java router's NTCP2 info from its RouterInfo file, then attempts an NTCP2 handshake. Requirements: - podman available on the host - docker.io/geti2p/i2p:latest image pulled - python:3.12-slim image available - This test is slow (~60-90s) due to Java router startup time Extracted from the monolith (i2p-python/tests/test_integration/test_java_interop.py). """ import json import os import subprocess import tempfile import time import unittest import pytest def _run(cmd, **kwargs): """Run a command and return stdout.""" result = subprocess.run( cmd, shell=True, capture_output=True, text=True, timeout=120, **kwargs ) if result.returncode != 0: raise RuntimeError( f"Command failed: {cmd}\nstdout: {result.stdout}\nstderr: {result.stderr}" ) return result.stdout.strip() def _run_quiet(cmd, **kwargs): """Run a command, ignoring errors.""" subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=60, **kwargs) POD_NAME = "i2p-interop-test" # Integration package root (where scripts/ lives) INTEGRATION_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) # Router package (bundles all i2p_* namespace packages) ROUTER_DIR = os.path.abspath(os.path.join(INTEGRATION_DIR, "..", "i2p-py-router")) PIP_CMD = "pip install -q cryptography && pip install -q --no-deps -e /deps/router" @pytest.mark.slow @pytest.mark.podman class TestJavaInterop(unittest.TestCase): """Interop tests connecting Python NTCP2 to the Java I2P reference router.""" @classmethod def setUpClass(cls): """Check that podman and the Java I2P image are available.""" try: _run("podman --version") except (FileNotFoundError, RuntimeError): raise unittest.SkipTest("podman not available") if not os.path.isdir(ROUTER_DIR): raise unittest.SkipTest(f"Router package not found at {ROUTER_DIR}") # Check if the Java image exists result = subprocess.run( "podman image exists docker.io/geti2p/i2p:latest", shell=True, capture_output=True ) if result.returncode != 0: raise unittest.SkipTest("geti2p/i2p:latest image not available (run: podman pull docker.io/geti2p/i2p:latest)") def setUp(self): """Create shared tmpdir and podman pod.""" self._shared_dir = tempfile.mkdtemp(prefix="i2p-interop-test-") os.chmod(self._shared_dir, 0o777) _run_quiet(f"podman pod rm -f {POD_NAME}") _run(f"podman pod create --name {POD_NAME}") def tearDown(self): """Remove the pod and shared dir.""" _run_quiet(f"podman pod rm -f {POD_NAME}") import shutil shutil.rmtree(self._shared_dir, ignore_errors=True) def _wait_for_file(self, path, timeout=90): """Wait for a file to appear and have content.""" deadline = time.time() + timeout while time.time() < deadline: if os.path.exists(path) and os.path.getsize(path) > 0: try: with open(path) as f: return json.load(f) except (json.JSONDecodeError, IOError): pass time.sleep(1) # Get logs for debugging for name in ["java-router", "py-extractor", "py-connector"]: logs = subprocess.run( f"podman logs {name} 2>&1", shell=True, capture_output=True, text=True ) print(f"\n--- {name} logs ---\n{logs.stdout}\n{logs.stderr}") raise TimeoutError(f"File {path} not ready after {timeout}s") def _wait_for_binary_file(self, path, min_size=100, timeout=90): """Wait for a binary file to appear with minimum size.""" deadline = time.time() + timeout while time.time() < deadline: if os.path.exists(path) and os.path.getsize(path) >= min_size: return True time.sleep(1) return False def test_connect_to_java_router(self): """Start Java router, extract NTCP2 info, attempt handshake.""" shared = self._shared_dir # Start Java router with minimal config i2p_config_dir = os.path.join(shared, "i2p_config") os.makedirs(i2p_config_dir, exist_ok=True) os.chmod(i2p_config_dir, 0o777) _run( f"podman run -d --pod {POD_NAME} --name java-router " f"-v {i2p_config_dir}:/i2p/.i2p:Z " f"-e JVM_XMX=256m " f"docker.io/geti2p/i2p:latest" ) # Wait for the router to generate its RouterInfo file ri_path = os.path.join(i2p_config_dir, "router.info") print(f"Waiting for Java router to generate {ri_path}...", flush=True) ri_ready = self._wait_for_binary_file(ri_path, min_size=100, timeout=90) if not ri_ready: if os.path.exists(i2p_config_dir): files = os.listdir(i2p_config_dir) print(f"Files in config dir: {files}", flush=True) self.fail("Java router did not generate router.info in time") # Give the router a moment to finish writing and start NTCP2 listener time.sleep(5) # Extract NTCP2 info using our Python parser _run( f"podman run -d --pod {POD_NAME} --name py-extractor " f"-v {ROUTER_DIR}:/deps/router:Z " f"-v {INTEGRATION_DIR}:/work:Z -v {shared}:/shared:Z " f"python:3.12-slim bash -c '" f"{PIP_CMD} && python /work/scripts/extract_java_ri.py " f"--ri-file /shared/i2p_config/router.info " f"--output /shared/java_ntcp2_info.json'" ) # Wait for extraction result info = self._wait_for_file(os.path.join(shared, "java_ntcp2_info.json"), timeout=30) print(f"Extraction result: {json.dumps(info, indent=2)}", flush=True) if info.get("status") != "ok": self.skipTest( f"Could not extract NTCP2 info from Java RouterInfo: {info.get('error', 'unknown')}" ) # Now attempt the NTCP2 handshake _run( f"podman run -d --pod {POD_NAME} --name py-connector " f"-v {ROUTER_DIR}:/deps/router:Z " f"-v {INTEGRATION_DIR}:/work:Z -v {shared}:/shared:Z " f"python:3.12-slim bash -c '" f"{PIP_CMD} && python /work/scripts/connect_to_java.py " f"--info-file /shared/java_ntcp2_info.json " f"--result-file /shared/interop_result.json'" ) # Wait for the interop result result = self._wait_for_file(os.path.join(shared, "interop_result.json"), timeout=60) print(f"Interop result: {json.dumps(result, indent=2)}", flush=True) if result.get("status") == "ok": self.assertEqual(result["handshake"], "complete") print("SUCCESS: NTCP2 handshake with Java router completed!", flush=True) if result.get("frames_received"): print(f"Received {len(result['frames_received'])} frames from Java router", flush=True) else: print(f"INTEROP NOTE: Handshake did not complete: {result.get('error')}", flush=True) self.skipTest( f"Java interop handshake incomplete (expected): {result.get('error')}" ) def test_parse_java_router_info(self): """Parse a Java-generated RouterInfo file and verify basic structure.""" shared = self._shared_dir # Start Java router i2p_config_dir = os.path.join(shared, "i2p_config") os.makedirs(i2p_config_dir, exist_ok=True) os.chmod(i2p_config_dir, 0o777) _run( f"podman run -d --pod {POD_NAME} --name java-router " f"-v {i2p_config_dir}:/i2p/.i2p:Z " f"-e JVM_XMX=256m " f"docker.io/geti2p/i2p:latest" ) ri_path = os.path.join(i2p_config_dir, "router.info") print(f"Waiting for {ri_path}...", flush=True) ri_ready = self._wait_for_binary_file(ri_path, min_size=100, timeout=90) if not ri_ready: self.fail("Java router did not generate router.info") time.sleep(3) # Parse using our Python code _run( f"podman run -d --pod {POD_NAME} --name py-parser " f"-v {ROUTER_DIR}:/deps/router:Z " f"-v {INTEGRATION_DIR}:/work:Z -v {shared}:/shared:Z " f"python:3.12-slim bash -c '" f"{PIP_CMD} && python /work/scripts/extract_java_ri.py " f"--ri-file /shared/i2p_config/router.info " f"--output /shared/parse_result.json'" ) result = self._wait_for_file(os.path.join(shared, "parse_result.json"), timeout=30) print(f"Parse result: {json.dumps(result, indent=2)}", flush=True) if result.get("status") == "ok": self.assertIn("host", result) self.assertIn("port", result) self.assertIn("static_key_hex", result) self.assertEqual(result["static_key_len"], 32, f"Expected 32-byte X25519 key, got {result['static_key_len']}") print(f"Successfully parsed Java RouterInfo: " f"NTCP2 at {result['host']}:{result['port']}", flush=True) else: self.skipTest( f"Could not parse Java RouterInfo: {result.get('error')}" ) if __name__ == "__main__": unittest.main()