Something went wrong. Try again.
Integration tests
Something went wrong. Try again.
9.7 kB · 246 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247"""Java reference implementation interop tests.
Spins up a podman pod with the geti2p/i2p Java router and a Pythoncontainer. Extracts the Java router's NTCP2 info from its RouterInfofile, 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 jsonimport osimport subprocessimport tempfileimport timeimport 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.podmanclass 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()