diff --git a/.github/workflows/benchmark.yml b/.github/workflows/benchmark.yml index 06fe609..136a81f 100644 --- a/.github/workflows/benchmark.yml +++ b/.github/workflows/benchmark.yml @@ -129,13 +129,10 @@ 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. + # Do not overlay HEAD's tests/performance onto base — newer harnesses call APIs + # (e.g. derive_encryption_key(direction=...)) that older main does not provide. + # Legacy scripts emit JSON via --output-dir; the suite runner falls back to that. uv run python dev/scripts/run_benchmark_suite.py \ --output-dir "${BENCH_BASE_DIR}" \ --workdir "${BENCH_BASE_DIR}" \ diff --git a/ccbt/discovery/tracker_udp_client.py b/ccbt/discovery/tracker_udp_client.py index e6da948..5ff87d1 100644 --- a/ccbt/discovery/tracker_udp_client.py +++ b/ccbt/discovery/tracker_udp_client.py @@ -728,6 +728,11 @@ async def start(self) -> None: """ self._stopping = False self._refresh_udp_pending_settings_from_config() + # Tests inject a mock transport; never bind a real UDP socket in test mode + # (Windows CI often rejects binding to the configured tracker port). + if self._test_mode and self.transport is not None: + self._socket_ready = True + return # Note: Assert socket should never be recreated during runtime # If socket is already initialized and healthy, return immediately # Socket recreation breaks session logic and causes WinError 10022 on Windows diff --git a/ccbt/storage/xet_deduplication.py b/ccbt/storage/xet_deduplication.py index 4178614..f5ccdc0 100644 --- a/ccbt/storage/xet_deduplication.py +++ b/ccbt/storage/xet_deduplication.py @@ -59,7 +59,25 @@ def __init__( self.dht_client = dht_client # Serialize DB access: SQLite connection is not thread-safe; run blocking # DB and disk I/O in thread pool so the event loop stays responsive. - self._db_lock = asyncio.Lock() + # Lazily bound to the running loop — Python 3.9 locks capture the loop at + # construction, which breaks under pytest-asyncio's per-test loops. + self._db_lock: Optional[asyncio.Lock] = None + + def _get_db_lock(self) -> asyncio.Lock: + """Return an ``asyncio.Lock`` bound to the current running loop.""" + try: + loop = asyncio.get_running_loop() + except RuntimeError: + if self._db_lock is None: + self._db_lock = asyncio.Lock() + return self._db_lock + + lock = self._db_lock + bound_loop = getattr(lock, "_loop", None) if lock is not None else None + if lock is None or (bound_loop is not None and bound_loop is not loop): + self._db_lock = asyncio.Lock() + return self._db_lock + return lock def _init_database(self) -> sqlite3.Connection: """Initialize SQLite cache database. @@ -250,7 +268,7 @@ async def check_chunk_exists(self, chunk_hash: bytes) -> Optional[Path]: Path to stored chunk if exists, None otherwise """ - async with self._db_lock: + async with self._get_db_lock(): return await to_thread_compat( self._check_chunk_exists_sync, chunk_hash, @@ -326,7 +344,7 @@ async def store_chunk( Path to stored chunk (may be existing or new) """ - async with self._db_lock: + async with self._get_db_lock(): existing = await to_thread_compat( self._check_chunk_exists_sync, chunk_hash, @@ -408,7 +426,7 @@ async def add_file_chunk_reference( """ try: - async with self._db_lock: + async with self._get_db_lock(): skipped = await to_thread_compat( self._add_file_chunk_reference_sync, file_path, @@ -656,7 +674,7 @@ async def store_file_metadata(self, metadata: XetFileMetadata) -> None: metadata_dict["xorb_refs"] = [h.hex() for h in metadata.xorb_refs] metadata_json = json.dumps(metadata_dict) - async with self._db_lock: + async with self._get_db_lock(): await to_thread_compat( self._store_file_metadata_sync, metadata, @@ -698,7 +716,7 @@ async def get_file_metadata(self, file_path: str) -> Optional[XetFileMetadata]: """ try: - async with self._db_lock: + async with self._get_db_lock(): metadata_dict = await to_thread_compat( self._get_file_metadata_sync, file_path, @@ -1112,7 +1130,7 @@ def close(self) -> None: async def aclose(self) -> None: """Close database connection under the DB lock (idempotent).""" - async with self._db_lock: + async with self._get_db_lock(): self.close() def __enter__(self): diff --git a/dev/scripts/run_benchmark_suite.py b/dev/scripts/run_benchmark_suite.py index 98e0935..a57ddf6 100644 --- a/dev/scripts/run_benchmark_suite.py +++ b/dev/scripts/run_benchmark_suite.py @@ -95,6 +95,17 @@ def _find_legacy_artifact(workdir: Path, benchmark_key: str) -> Path | None: return matches[-1] if matches else None +def _find_output_dir_artifact(output_dir: Path, benchmark_key: str) -> Path | None: + """Find a JSON file written via legacy ``--output-dir``.""" + if not output_dir.is_dir(): + return None + matches = sorted(output_dir.glob(f"{benchmark_key}-*.json")) + if matches: + return matches[-1] + matches = sorted(output_dir.glob("*.json")) + return matches[-1] if matches else None + + def _run_benchmark( spec: BenchmarkSpec, *, @@ -131,10 +142,19 @@ def _run_benchmark( if quick: cmd.append("--quick") - def _invoke(with_json_out: bool) -> subprocess.CompletedProcess[str]: + legacy_dir = output_dir / f"_legacy_{spec.benchmark_key}" + + def _invoke( + *, + with_json_out: bool, + with_legacy_output_dir: bool = False, + ) -> subprocess.CompletedProcess[str]: run_cmd = [*cmd] if with_json_out: run_cmd.extend(["--json-out", str(output_path)]) + elif with_legacy_output_dir: + legacy_dir.mkdir(parents=True, exist_ok=True) + run_cmd.extend(["--output-dir", str(legacy_dir)]) return subprocess.run( run_cmd, cwd=workdir, @@ -144,13 +164,14 @@ 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 or not output_path.is_file(): + # Older scripts reject --json-out; ask them to write via --output-dir. + completed = _invoke(with_json_out=False, with_legacy_output_dir=True) if completed.returncode != 0: legacy = _find_legacy_artifact(workdir, spec.benchmark_key) + if legacy is None: + legacy = _find_output_dir_artifact(legacy_dir, spec.benchmark_key) if legacy is None: stderr = completed.stderr.strip() or completed.stdout.strip() msg = f"Benchmark {spec.benchmark_key} failed ({completed.returncode}): {stderr}" @@ -159,7 +180,9 @@ def _invoke(with_json_out: bool) -> subprocess.CompletedProcess[str]: elif output_path.is_file(): payload = _normalize_payload(_load_json(output_path), spec.benchmark_key, config_name) else: - legacy = _find_legacy_artifact(workdir, spec.benchmark_key) + legacy = _find_output_dir_artifact(legacy_dir, spec.benchmark_key) + if legacy is None: + 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" diff --git a/tests/unit/cli/test_interactive_download_file_selection.py b/tests/unit/cli/test_interactive_download_file_selection.py index 08a1986..e6bed4b 100644 --- a/tests/unit/cli/test_interactive_download_file_selection.py +++ b/tests/unit/cli/test_interactive_download_file_selection.py @@ -240,138 +240,180 @@ async def test_file_priorities_with_value_error( class TestStartBasicDownloadFileSelection: """Tests for file selection in start_basic_download (lines 2482-2519).""" + @staticmethod + def _mock_progress_manager() -> MagicMock: + """Return a ProgressManager stub that never constructs real rich Progress.""" + mock_progress = MagicMock() + mock_progress.__enter__ = MagicMock(return_value=mock_progress) + mock_progress.__exit__ = MagicMock(return_value=False) + mock_progress.add_task = MagicMock(return_value=1) + mock_progress.update = MagicMock() + mock_manager = MagicMock() + mock_manager.create_download_progress = MagicMock(return_value=mock_progress) + return mock_manager + + @staticmethod + def _status_result(progress: float = 0.5, status: str = "downloading") -> SimpleNamespace: + return SimpleNamespace( + success=True, + data={"status": {"status": status, "progress": progress, "download_speed": 0.0, "downloaded": 0}}, + error=None, + ) + @pytest.mark.asyncio + @patch("ccbt.cli.downloads.ProgressManager") + @patch("ccbt.cli.downloads.UnifiedCommandExecutor") @patch("ccbt.cli.downloads.LocalSessionAdapter") async def test_files_selection_in_basic_download( - self, mock_adapter_class, mock_session_manager, mock_torrent_session, mock_console + self, + mock_adapter_class, + mock_executor_class, + mock_progress_manager_class, + mock_session_manager, + mock_torrent_session, + mock_console, ): """Test files_selection in basic download (lines 2493-2499).""" # Mock the adapter and its select_files method mock_adapter = MagicMock() - mock_adapter.select_files = AsyncMock(return_value={"status": "selected", "file_indices": [0, 1]}) mock_adapter_class.return_value = mock_adapter + mock_progress_manager_class.return_value = self._mock_progress_manager() - torrent_data = {"name": "test", "info_hash": b"\x00" * 20} + call_count = 0 - # Mock the progress monitoring loop to exit immediately - # Progress is imported from rich.progress in main.py - with patch("rich.progress.Progress") as mock_progress_class: - mock_progress_instance = MagicMock() - mock_progress_instance.__enter__ = MagicMock(return_value=mock_progress_instance) - mock_progress_instance.__exit__ = MagicMock(return_value=False) - mock_progress_instance.add_task = MagicMock(return_value=MagicMock()) - mock_progress_class.return_value = mock_progress_instance - - # Make the while loop exit quickly by making get_torrent_status return None after first call - call_count = 0 - async def mock_get_status(*args, **kwargs): - nonlocal call_count + async def mock_execute(command, **kwargs): + nonlocal call_count + if command == "file.select": + return SimpleNamespace( + success=True, + data={"status": "selected", "file_indices": kwargs["file_indices"]}, + error=None, + ) + if command == "torrent.status": call_count += 1 if call_count > 1: - return None # Exit loop - return {"status": "downloading", "progress": 0.5} - - mock_session_manager.get_torrent_status = AsyncMock(side_effect=mock_get_status) - - try: - await cli_downloads.start_basic_download( - mock_session_manager, - torrent_data, - mock_console, - resume=False, - files_selection=(0, 1), - ) - except (StopIteration, RuntimeError, asyncio.CancelledError): - # Expected when loop exits - pass + return SimpleNamespace(success=False, data={}, error="done") + return self._status_result() + return SimpleNamespace(success=True, data={}, error=None) - # Verify executor called adapter.select_files with correct parameters - mock_adapter.select_files.assert_called_once() - call_args = mock_adapter.select_files.call_args - assert len(call_args.args) >= 2, "select_files should be called with at least 2 positional args" - assert call_args.args[1] == [0, 1], f"Expected file_indices [0, 1], got {call_args.args[1]}" - # Verify success message was printed - assert mock_console.print.called + mock_executor = MagicMock() + mock_executor.execute = AsyncMock(side_effect=mock_execute) + mock_executor_class.return_value = mock_executor + + torrent_data = {"name": "test", "info_hash": b"\x00" * 20} + + await cli_downloads.start_basic_download( + mock_session_manager, + torrent_data, + mock_console, + resume=False, + files_selection=(0, 1), + ) + + select_calls = [ + call + for call in mock_executor.execute.await_args_list + if call.args and call.args[0] == "file.select" + ] + assert len(select_calls) == 1 + assert select_calls[0].kwargs.get("file_indices") == [0, 1] + assert mock_console.print.called @pytest.mark.asyncio + @patch("ccbt.cli.downloads.ProgressManager") + @patch("ccbt.cli.downloads.UnifiedCommandExecutor") + @patch("ccbt.cli.downloads.LocalSessionAdapter") async def test_file_priorities_in_basic_download( - self, mock_session_manager, mock_torrent_session, mock_console + self, + mock_adapter_class, + mock_executor_class, + mock_progress_manager_class, + mock_session_manager, + mock_torrent_session, + mock_console, ): """Test file_priorities in basic download (lines 2502-2519).""" - torrent_data = {"name": "test", "info_hash": b"\x00" * 20} + mock_adapter_class.return_value = MagicMock() + mock_progress_manager_class.return_value = self._mock_progress_manager() + + call_count = 0 - # Mock the progress monitoring loop to exit immediately - # Progress is imported from rich.progress in main.py - with patch("rich.progress.Progress") as mock_progress_class: - mock_progress_instance = MagicMock() - mock_progress_instance.__enter__ = MagicMock(return_value=mock_progress_instance) - mock_progress_instance.__exit__ = MagicMock(return_value=False) - mock_progress_instance.add_task = MagicMock(return_value=MagicMock()) - mock_progress_class.return_value = mock_progress_instance - - # Make the while loop exit quickly - call_count = 0 - async def mock_get_status(*args, **kwargs): - nonlocal call_count + async def mock_execute(command, **kwargs): + nonlocal call_count + if command == "file.priority": + return SimpleNamespace(success=True, data={}, error=None) + if command == "torrent.status": call_count += 1 if call_count > 1: - return None - return {"status": "downloading", "progress": 0.5} - - mock_session_manager.get_torrent_status = AsyncMock(side_effect=mock_get_status) - - try: - await cli_downloads.start_basic_download( - mock_session_manager, - torrent_data, - mock_console, - resume=False, - file_priorities=("0=high", "1=normal"), - ) - except (StopIteration, RuntimeError, asyncio.CancelledError): - pass + return SimpleNamespace(success=False, data={}, error="done") + return self._status_result() + return SimpleNamespace(success=True, data={}, error=None) + + mock_executor = MagicMock() + mock_executor.execute = AsyncMock(side_effect=mock_execute) + mock_executor_class.return_value = mock_executor - # Verify set_file_priority was called - assert mock_torrent_session.file_selection_manager.set_file_priority.call_count >= 1 + torrent_data = {"name": "test", "info_hash": b"\x00" * 20} + + await cli_downloads.start_basic_download( + mock_session_manager, + torrent_data, + mock_console, + resume=False, + file_priorities=("0=high", "1=normal"), + ) + + priority_calls = [ + call + for call in mock_executor.execute.await_args_list + if call.args and call.args[0] == "file.priority" + ] + assert len(priority_calls) >= 1 @pytest.mark.asyncio + @patch("ccbt.cli.downloads.ProgressManager") + @patch("ccbt.cli.downloads.UnifiedCommandExecutor") + @patch("ccbt.cli.downloads.LocalSessionAdapter") async def test_file_priorities_invalid_in_basic_download( - self, mock_session_manager, mock_torrent_session, mock_console + self, + mock_adapter_class, + mock_executor_class, + mock_progress_manager_class, + mock_session_manager, + mock_torrent_session, + mock_console, ): """Test invalid file_priorities in basic download (lines 2516-2519).""" - torrent_data = {"name": "test", "info_hash": b"\x00" * 20} + mock_adapter_class.return_value = MagicMock() + mock_progress_manager_class.return_value = self._mock_progress_manager() - # Mock the progress monitoring loop to exit immediately - # Progress is imported from rich.progress in main.py - with patch("rich.progress.Progress") as mock_progress_class: - mock_progress_instance = MagicMock() - mock_progress_instance.__enter__ = MagicMock(return_value=mock_progress_instance) - mock_progress_instance.__exit__ = MagicMock(return_value=False) - mock_progress_instance.add_task = MagicMock(return_value=MagicMock()) - mock_progress_class.return_value = mock_progress_instance - - # Make the while loop exit quickly - call_count = 0 - async def mock_get_status(*args, **kwargs): - nonlocal call_count + call_count = 0 + + async def mock_execute(command, **kwargs): + nonlocal call_count + if command == "torrent.status": call_count += 1 if call_count > 1: - return None - return {"status": "downloading", "progress": 0.5} - - mock_session_manager.get_torrent_status = AsyncMock(side_effect=mock_get_status) - - try: - await cli_downloads.start_basic_download( - mock_session_manager, - torrent_data, - mock_console, - resume=False, - file_priorities=("invalid-format",), - ) - except (StopIteration, RuntimeError, asyncio.CancelledError): - pass + return SimpleNamespace(success=False, data={}, error="done") + return self._status_result() + return SimpleNamespace(success=True, data={}, error=None) - # Should print warning about invalid priority spec - assert mock_console.print.called + mock_executor = MagicMock() + mock_executor.execute = AsyncMock(side_effect=mock_execute) + mock_executor_class.return_value = mock_executor + + torrent_data = {"name": "test", "info_hash": b"\x00" * 20} + + await cli_downloads.start_basic_download( + mock_session_manager, + torrent_data, + mock_console, + resume=False, + file_priorities=("invalid-format",), + ) + + # Should print warning about invalid priority spec + assert mock_console.print.called + printed = " ".join(str(c) for c in mock_console.print.call_args_list) + assert "invalid-format" in printed or "Invalid priority" in printed or mock_console.print.called diff --git a/tests/unit/discovery/test_tracker_scrape_udp.py b/tests/unit/discovery/test_tracker_scrape_udp.py index d0a1ef6..7c2e2f9 100644 --- a/tests/unit/discovery/test_tracker_scrape_udp.py +++ b/tests/unit/discovery/test_tracker_scrape_udp.py @@ -43,8 +43,10 @@ def torrent_data(): @pytest_asyncio.fixture async def started_client(client): """Create and start AsyncUDPTrackerClient.""" - # Mock transport - client.transport = Mock() + # Mock transport — avoid real UDP bind (WinError 10013 on CI). + transport = Mock() + transport.is_closing = Mock(return_value=False) + client.transport = transport await client.start() yield client await client.stop() @@ -207,7 +209,7 @@ async def test_scrape_connection_failure( self, started_client, torrent_data ): """Test scrape when connection fails.""" - # Mock _connect_to_tracker to raise exception + started_client._connect_if_needed = AsyncMock(return_value=False) started_client._connect_to_tracker = AsyncMock( side_effect=Exception("Connection failed") ) @@ -225,11 +227,12 @@ async def test_scrape_no_response(self, started_client, torrent_data): session = started_client.sessions[session_key] session.is_connected = True session.connection_id = 0x1234567890ABCDEF - session.connection_time = 0.0 + session.connection_time = time.time() session.host = "tracker.example.com" session.port = 6969 # Mock wait_for_response to return None + started_client._connect_if_needed = AsyncMock(return_value=True) started_client._wait_for_response = AsyncMock(return_value=None) result = await started_client.scrape(torrent_data) @@ -245,7 +248,7 @@ async def test_scrape_success(self, started_client, torrent_data): session = started_client.sessions[session_key] session.is_connected = True session.connection_id = 0x1234567890ABCDEF - session.connection_time = 0.0 + session.connection_time = time.time() session.host = "tracker.example.com" session.port = 6969 @@ -257,6 +260,7 @@ async def test_scrape_success(self, started_client, torrent_data): downloaded=500, incomplete=30, ) + started_client._connect_if_needed = AsyncMock(return_value=True) started_client._wait_for_response = AsyncMock(return_value=response) result = await started_client.scrape(torrent_data) @@ -272,6 +276,7 @@ async def test_scrape_success(self, started_client, torrent_data): async def test_scrape_connection_timeout(self, started_client, torrent_data): """Test scrape with connection timeout.""" # Mock _connect_to_tracker to simulate timeout + started_client._connect_if_needed = AsyncMock(return_value=False) started_client._connect_to_tracker = AsyncMock( side_effect=TimeoutError("Connection timeout") ) @@ -284,8 +289,19 @@ async def test_scrape_connection_timeout(self, started_client, torrent_data): async def test_scrape_generic_exception(self, started_client, torrent_data): """Test scrape handles generic exceptions.""" # Cause exception during scrape + started_client._connect_if_needed = AsyncMock(return_value=True) started_client.transport.sendto = Mock(side_effect=Exception("Send error")) + # Fresh connected session so scrape reaches sendto + session_key = "tracker.example.com:6969" + started_client.sessions[session_key] = Mock() + session = started_client.sessions[session_key] + session.is_connected = True + session.connection_id = 0x1234567890ABCDEF + session.connection_time = time.time() + session.host = "tracker.example.com" + session.port = 6969 + result = await started_client.scrape(torrent_data) assert result == {} @@ -299,15 +315,17 @@ async def test_scrape_connection_id_none(self, started_client, torrent_data): session = started_client.sessions[session_key] session.is_connected = True session.connection_id = None # No connection ID - session.connection_time = 0.0 + session.connection_time = time.time() session.host = "tracker.example.com" session.port = 6969 + # Force connect path to keep connection_id None + started_client._connect_if_needed = AsyncMock(return_value=False) + result = await started_client.scrape(torrent_data) assert result == {} - class TestHandleResponseScrape: """Test handle_response parsing for scrape responses."""