From 3f3d89c9dd2db883b13f847b3a198d8c6461e69f Mon Sep 17 00:00:00 2001 From: Joseph Pollack Date: Tue, 28 Jul 2026 17:24:04 +0200 Subject: [PATCH] Fix remaining PR #3 CI flakes and base benchmark JSON. Isolate XET temp stores, harden MSE/timeout tests, and overlay HEAD bench harness onto older main worktrees so hash_verify emits CI artifacts. --- .github/workflows/benchmark.yml | 6 + dev/scripts/run_benchmark_suite.py | 5 + .../test_mse_tcp_server_pe_first.py | 14 ++- .../resilience/test_resilience_timeout.py | 6 +- tests/unit/security/test_mse_handshake.py | 104 +++++++++++------- tests/unit/storage/conftest.py | 15 +++ .../unit/storage/test_xet_data_aggregator.py | 21 +--- tests/unit/storage/test_xet_deduplication.py | 44 +++----- .../storage/test_xet_defrag_prevention.py | 19 ---- .../storage/test_xet_file_deduplication.py | 19 ---- 10 files changed, 123 insertions(+), 130 deletions(-) create mode 100644 tests/unit/storage/conftest.py diff --git a/.github/workflows/benchmark.yml b/.github/workflows/benchmark.yml index 434a623..06fe609 100644 --- a/.github/workflows/benchmark.yml +++ b/.github/workflows/benchmark.yml @@ -129,6 +129,12 @@ jobs: fi echo "Using base benchmark config: ${CONFIG_FILE}" (cd "${BENCH_BASE_DIR}" && uv sync --dev) + # Older main lacks --json-out / json_out support on bench scripts. + # Overlay HEAD's tests/performance harness into the base worktree so + # measurements still import base ccbt (__file__ under base) but emit + # CI JSON artifacts the suite runner expects. + mkdir -p "${BENCH_BASE_DIR}/tests/performance" + cp -a "${{ github.workspace }}/tests/performance/." "${BENCH_BASE_DIR}/tests/performance/" # Use HEAD's runner script (has current CLI), but execute benchmarks in base workdir. uv run python dev/scripts/run_benchmark_suite.py \ --output-dir "${BENCH_BASE_DIR}" \ diff --git a/dev/scripts/run_benchmark_suite.py b/dev/scripts/run_benchmark_suite.py index f7e61cc..98e0935 100644 --- a/dev/scripts/run_benchmark_suite.py +++ b/dev/scripts/run_benchmark_suite.py @@ -145,6 +145,8 @@ def _invoke(with_json_out: bool) -> subprocess.CompletedProcess[str]: completed = _invoke(with_json_out=True) if completed.returncode != 0: + # Older scripts may reject --json-out; retry without it and look for + # legacy artifact paths or an explicit --output-dir write. completed = _invoke(with_json_out=False) if completed.returncode != 0: @@ -159,7 +161,10 @@ def _invoke(with_json_out: bool) -> subprocess.CompletedProcess[str]: else: legacy = _find_legacy_artifact(workdir, spec.benchmark_key) if legacy is None: + detail = (completed.stderr or completed.stdout or "").strip() msg = f"Benchmark {spec.benchmark_key} produced no JSON artifact" + if detail: + msg = f"{msg}: {detail}" raise RuntimeError(msg) payload = _normalize_payload(_load_json(legacy), spec.benchmark_key, config_name) diff --git a/tests/integration/test_mse_tcp_server_pe_first.py b/tests/integration/test_mse_tcp_server_pe_first.py index f7d0d46..ed963c7 100644 --- a/tests/integration/test_mse_tcp_server_pe_first.py +++ b/tests/integration/test_mse_tcp_server_pe_first.py @@ -51,6 +51,7 @@ async def _run_loopback_mse_handshake( port: int, *, attempts: int = 2, + accept_mock: AsyncMock | None = None, ) -> None: """Run an outbound MSE handshake against an already-started loopback server.""" last_error: str | None = None @@ -71,6 +72,8 @@ async def _run_loopback_mse_handshake( finally: writer.close() await writer.wait_closed() + if accept_mock is not None and accept_mock.await_count > 0: + break if attempt + 1 < attempts: await asyncio.sleep(0.05) assert False, last_error or "MSE handshake failed" @@ -107,7 +110,9 @@ async def _close_incoming_connection( try: await asyncio.sleep(0.05) port = tcp_server.sockets[0].getsockname()[1] - await _run_loopback_mse_handshake(info_hash, outbound_payload, port) + await _run_loopback_mse_handshake( + info_hash, outbound_payload, port, accept_mock=accept_incoming_encrypted + ) assert accept_incoming_encrypted.await_count == 1 accepted = accept_incoming_encrypted.await_args.args @@ -156,7 +161,12 @@ async def _close_incoming_connection( try: await asyncio.sleep(0.05) port = tcp_server.sockets[0].getsockname()[1] - await _run_loopback_mse_handshake(target_info_hash, outbound_payload, port) + await _run_loopback_mse_handshake( + target_info_hash, + outbound_payload, + port, + accept_mock=accept_incoming_target, + ) accept_incoming_ignored.assert_not_awaited() accept_incoming_target.assert_awaited_once() diff --git a/tests/unit/resilience/test_resilience_timeout.py b/tests/unit/resilience/test_resilience_timeout.py index af4ad8f..cc0a089 100644 --- a/tests/unit/resilience/test_resilience_timeout.py +++ b/tests/unit/resilience/test_resilience_timeout.py @@ -25,14 +25,18 @@ def sync_function(): def test_sync_function_timeout_failure(self): """Test synchronous function with timeout - timeout case.""" + gate = threading.Event() + @with_timeout(0.1) def slow_sync_function(): - time.sleep(0.2) # Longer than timeout + gate.wait(timeout=1.0) return "should_not_reach_here" with pytest.raises(TimeoutError, match="Operation timed out after 0.1 seconds"): slow_sync_function() + gate.set() + def test_sync_function_timeout_exception(self): """Test synchronous function with timeout - exception case.""" @with_timeout(1.0) diff --git a/tests/unit/security/test_mse_handshake.py b/tests/unit/security/test_mse_handshake.py index 886d220..42f17f9 100644 --- a/tests/unit/security/test_mse_handshake.py +++ b/tests/unit/security/test_mse_handshake.py @@ -592,29 +592,40 @@ async def responder( server = await asyncio.start_server(responder, "127.0.0.1", 0) try: server_port = server.sockets[0].getsockname()[1] - initiator_reader, initiator_writer = await asyncio.open_connection( - "127.0.0.1", server_port - ) - try: - initiator = MSEHandshake(prefer_rc4=True) - initiator_result = await initiator.initiate_as_initiator( - initiator_reader, - initiator_writer, - info_hash, - timeout=_MSE_INTEGRATION_TIMEOUT, - ) - receiver_result = await asyncio.wait_for( - responder_results.get(), timeout=_MSE_INTEGRATION_TIMEOUT + last_error: str | None = None + initiator_result: MSEHandshakeResult | None = None + receiver_result: MSEHandshakeResult | None = None + for attempt in range(2): + initiator_reader, initiator_writer = await asyncio.open_connection( + "127.0.0.1", server_port ) - finally: - initiator_writer.close() - await initiator_writer.wait_closed() + try: + initiator = MSEHandshake(prefer_rc4=True) + initiator_result = await initiator.initiate_as_initiator( + initiator_reader, + initiator_writer, + info_hash, + timeout=_MSE_INTEGRATION_TIMEOUT, + ) + receiver_result = await asyncio.wait_for( + responder_results.get(), timeout=_MSE_INTEGRATION_TIMEOUT + ) + if initiator_result.success and receiver_result.success: + break + last_error = initiator_result.error or receiver_result.error + finally: + initiator_writer.close() + await initiator_writer.wait_closed() + if attempt == 0: + await asyncio.sleep(0.05) finally: server.close() await server.wait_closed() - assert initiator_result.success is True - assert receiver_result.success is True + assert initiator_result is not None + assert receiver_result is not None + assert initiator_result.success is True, last_error + assert receiver_result.success is True, last_error assert initiator_result.cipher is not None assert receiver_result.cipher is not None @@ -1266,7 +1277,7 @@ async def responder(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) reader=reader, writer=writer, info_hash=info_hash, - timeout=1.0, + timeout=_MSE_INTEGRATION_TIMEOUT, initial_payload_size=0, info_hash_candidates=[info_hash], ) @@ -1277,30 +1288,41 @@ async def responder(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) server = await asyncio.start_server(responder, "127.0.0.1", 0) try: server_port = server.sockets[0].getsockname()[1] - initiator_reader, initiator_writer = await asyncio.open_connection( - "127.0.0.1", server_port - ) - try: - initiator_handshake = MSEHandshake() - initiator_result = await initiator_handshake.initiate_as_initiator( - initiator_reader, - initiator_writer, - info_hash, - timeout=1.0, - initial_payload=initial_payload, - ) - responder_result = await asyncio.wait_for( - responder_results.get(), timeout=1.0 + last_error: str | None = None + initiator_result: MSEHandshakeResult | None = None + responder_result: MSEHandshakeResult | None = None + for attempt in range(2): + initiator_reader, initiator_writer = await asyncio.open_connection( + "127.0.0.1", server_port ) + try: + initiator_handshake = MSEHandshake() + initiator_result = await initiator_handshake.initiate_as_initiator( + initiator_reader, + initiator_writer, + info_hash, + timeout=_MSE_INTEGRATION_TIMEOUT, + initial_payload=initial_payload, + ) + responder_result = await asyncio.wait_for( + responder_results.get(), timeout=_MSE_INTEGRATION_TIMEOUT + ) + if initiator_result.success and responder_result.success: + break + last_error = initiator_result.error or responder_result.error + finally: + initiator_writer.close() + await initiator_writer.wait_closed() + if attempt == 0: + await asyncio.sleep(0.05) - assert initiator_result.success is True - assert initiator_result.resolved_info_hash == info_hash - assert responder_result.success is True - assert responder_result.decrypted_initial_data == initial_payload - assert responder_result.resolved_info_hash == info_hash - finally: - initiator_writer.close() - await initiator_writer.wait_closed() + assert initiator_result is not None + assert responder_result is not None + assert initiator_result.success is True, last_error + assert initiator_result.resolved_info_hash == info_hash + assert responder_result.success is True, last_error + assert responder_result.decrypted_initial_data == initial_payload + assert responder_result.resolved_info_hash == info_hash finally: server.close() await server.wait_closed() diff --git a/tests/unit/storage/conftest.py b/tests/unit/storage/conftest.py new file mode 100644 index 0000000..d64743c --- /dev/null +++ b/tests/unit/storage/conftest.py @@ -0,0 +1,15 @@ +"""Shared fixtures for storage unit tests.""" + +from __future__ import annotations + +import tempfile +from pathlib import Path + +import pytest + + +@pytest.fixture +def temp_db_path() -> str: + """Provide an isolated SQLite cache path with a dedicated chunk store directory.""" + with tempfile.TemporaryDirectory() as tmpdir: + yield str(Path(tmpdir) / "cache.db") diff --git a/tests/unit/storage/test_xet_data_aggregator.py b/tests/unit/storage/test_xet_data_aggregator.py index 16fed07..aee3d23 100644 --- a/tests/unit/storage/test_xet_data_aggregator.py +++ b/tests/unit/storage/test_xet_data_aggregator.py @@ -5,7 +5,7 @@ from __future__ import annotations -import tempfile +import hashlib import pytest @@ -17,29 +17,12 @@ def _chunk_hash(label: bytes) -> bytes: """Build a deterministic 32-byte chunk hash for tests.""" - return (label * 16)[:32] + return hashlib.sha256(label).digest() class TestXetDataAggregator: """Test XetDataAggregator class.""" - @pytest.fixture - def temp_db_path(self): - """Create temporary database path for testing.""" - with tempfile.NamedTemporaryFile(delete=False, suffix=".db") as f: - db_path = f.name - yield db_path - # Cleanup - import os - import time - for _ in range(5): - try: - if os.path.exists(db_path): - os.unlink(db_path) - break - except (PermissionError, OSError): - time.sleep(0.1) - @pytest.fixture def dedup(self, temp_db_path): """Create XetDeduplication instance for testing.""" diff --git a/tests/unit/storage/test_xet_deduplication.py b/tests/unit/storage/test_xet_deduplication.py index d32c089..2ecaf45 100644 --- a/tests/unit/storage/test_xet_deduplication.py +++ b/tests/unit/storage/test_xet_deduplication.py @@ -21,22 +21,6 @@ class TestXetDeduplication: """Test XetDeduplication class.""" - @pytest.fixture - def temp_db_path(self): - """Create temporary database path for testing.""" - with tempfile.NamedTemporaryFile(delete=False, suffix=".db") as f: - db_path = f.name - yield db_path - # Cleanup - try multiple times on Windows - import time - for _ in range(5): - try: - if os.path.exists(db_path): - os.unlink(db_path) - break - except (PermissionError, OSError): - time.sleep(0.1) - @pytest.fixture def dedup(self, temp_db_path): """Create XetDeduplication instance for testing.""" @@ -48,20 +32,22 @@ def dedup(self, temp_db_path): def test_initialization(self, temp_db_path): """Test deduplication cache initialization.""" dedup = XetDeduplication(cache_db_path=temp_db_path) + try: + # Database should be created + assert os.path.exists(temp_db_path) + + # Check table exists + conn = sqlite3.connect(temp_db_path) + cursor = conn.cursor() + cursor.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='chunks'" + ) + result = cursor.fetchone() + conn.close() - # Database should be created - assert os.path.exists(temp_db_path) - - # Check table exists - conn = sqlite3.connect(temp_db_path) - cursor = conn.cursor() - cursor.execute( - "SELECT name FROM sqlite_master WHERE type='table' AND name='chunks'" - ) - result = cursor.fetchone() - conn.close() - - assert result is not None + assert result is not None + finally: + dedup.close() @pytest.mark.asyncio async def test_check_chunk_not_exists(self, dedup): diff --git a/tests/unit/storage/test_xet_defrag_prevention.py b/tests/unit/storage/test_xet_defrag_prevention.py index 2dc3320..cd27a1a 100644 --- a/tests/unit/storage/test_xet_defrag_prevention.py +++ b/tests/unit/storage/test_xet_defrag_prevention.py @@ -5,8 +5,6 @@ from __future__ import annotations -import tempfile - import pytest from ccbt.storage.xet_deduplication import XetDeduplication @@ -18,23 +16,6 @@ class TestXetDefragPrevention: """Test XetDefragPrevention class.""" - @pytest.fixture - def temp_db_path(self): - """Create temporary database path for testing.""" - with tempfile.NamedTemporaryFile(delete=False, suffix=".db") as f: - db_path = f.name - yield db_path - # Cleanup - import os - import time - for _ in range(5): - try: - if os.path.exists(db_path): - os.unlink(db_path) - break - except (PermissionError, OSError): - time.sleep(0.1) - @pytest.fixture def dedup(self, temp_db_path): """Create XetDeduplication instance for testing.""" diff --git a/tests/unit/storage/test_xet_file_deduplication.py b/tests/unit/storage/test_xet_file_deduplication.py index 1f11bc6..e06707e 100644 --- a/tests/unit/storage/test_xet_file_deduplication.py +++ b/tests/unit/storage/test_xet_file_deduplication.py @@ -6,8 +6,6 @@ from __future__ import annotations -import tempfile - import pytest from ccbt.models import XetFileMetadata @@ -20,23 +18,6 @@ class TestXetFileDeduplication: """Test XetFileDeduplication class.""" - @pytest.fixture - def temp_db_path(self): - """Create temporary database path for testing.""" - with tempfile.NamedTemporaryFile(delete=False, suffix=".db") as f: - db_path = f.name - yield db_path - # Cleanup - import os - import time - for _ in range(5): - try: - if os.path.exists(db_path): - os.unlink(db_path) - break - except (PermissionError, OSError): - time.sleep(0.1) - @pytest.fixture def dedup(self, temp_db_path): """Create XetDeduplication instance for testing."""