diff --git a/packaging/test/e2e_smoke.sh b/packaging/test/e2e_smoke.sh index c6c86a49..72f89024 100755 --- a/packaging/test/e2e_smoke.sh +++ b/packaging/test/e2e_smoke.sh @@ -4,6 +4,7 @@ set -euo pipefail SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" python3 "$SCRIPT_DIR/packaging/test/test_e2e_sse.py" python3 "$SCRIPT_DIR/packaging/test/test_e2e_fixture.py" +python3 "$SCRIPT_DIR/packaging/test/test_e2e_network.py" tmp="$(mktemp -d)" COLLECTOR_PID="" ADVISOR_STUB_PID="" @@ -70,7 +71,7 @@ fi # Create a test overlay config with fast sampling interval for smoke test cat > "$tmp/test_config.yaml" < "$tmp/fake_cursor_main.go" < "$tmp/fake-agent.log" 2>&1 & +"$tmp/sa-e2e-agent" "$tmp" > "$tmp/fake-agent.log" 2>&1 & AGENT_PID=$! FIXTURE_AGENT_PID=$AGENT_PID diff --git a/packaging/test/test_e2e_network.py b/packaging/test/test_e2e_network.py new file mode 100644 index 00000000..2dde7914 --- /dev/null +++ b/packaging/test/test_e2e_network.py @@ -0,0 +1,63 @@ +#!/usr/bin/env python3 +"""Build and run the smoke agent against a real local non-loopback socket.""" +import ipaddress +import os +import pathlib +import re +import select +import subprocess +import tempfile +import time +import unittest + + +class NetworkFixtureTests(unittest.TestCase): + def test_smoke_agent_establishes_non_loopback_connection(self): + source = pathlib.Path(__file__).with_name("e2e_smoke.sh").read_text() + agent_source = source.split('cat > "$tmp/fake_cursor_main.go" < "$tmp/fake-agent.log"', source) + with tempfile.TemporaryDirectory() as directory: + root = pathlib.Path(directory) + (root / "fake_cursor_main.go").write_text(agent_source) + subprocess.run( + ["bash", "-c", 'tmp=$1\n' + build.group(0), "network-fixture", directory], + check=True, capture_output=True, text=True, timeout=60, + ) + process = subprocess.Popen( + [str(root / binary_name), directory], + env={**os.environ, "HOME": directory}, + stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, + ) + output = b"" + try: + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + readable, _, _ = select.select([process.stderr], [], [], max(0, deadline - time.monotonic())) + if not readable: + break + chunk = os.read(process.stderr.fileno(), 8192) + if not chunk: + break + output += chunk + if b"network fixture connected to " in output: + break + self.assertIn(b"network fixture connected to ", output, output.decode()) + endpoint = output.decode().split("network fixture connected to ", 1)[1].splitlines()[0] + host = ipaddress.ip_address(endpoint.rsplit(":", 1)[0]) + self.assertFalse(host.is_loopback, endpoint) + self.assertFalse(host.is_unspecified, endpoint) + self.assertIsNone(process.poll(), output.decode()) + finally: + if process.poll() is None: + process.terminate() + process.communicate(timeout=5) + + +if __name__ == "__main__": + unittest.main()