|
1 | 1 | from __future__ import annotations |
2 | 2 |
|
3 | 3 | import sys |
4 | | -from typing import Optional |
| 4 | +from typing import Any, Optional |
5 | 5 | from unittest.mock import AsyncMock, MagicMock, patch |
6 | 6 |
|
7 | 7 | import pytest |
@@ -287,21 +287,28 @@ def _get_version_map(self) -> dict[str, str]: |
287 | 287 | async def test_uv_pip_install() -> None: |
288 | 288 | runs_calls: list[list[str]] = [] |
289 | 289 |
|
290 | | - class MockUvPackageManager(UvPackageManager): |
291 | | - def run( |
292 | | - self, |
293 | | - command: list[str], |
294 | | - log_callback: Optional[LogCallback] = None, |
295 | | - ) -> bool: |
296 | | - del log_callback |
297 | | - runs_calls.append(command) |
298 | | - return True |
| 290 | + with ( |
| 291 | + patch("subprocess.Popen") as mock_popen, |
| 292 | + patch("sys.stdout.buffer.write"), |
| 293 | + ): |
| 294 | + mock_proc = MagicMock() |
| 295 | + mock_proc.stdout.readline.return_value = b"" |
| 296 | + mock_proc.wait.return_value = 0 |
| 297 | + mock_popen.return_value = mock_proc |
299 | 298 |
|
300 | | - pm = MockUvPackageManager() |
301 | | - await pm._install("foo", upgrade=False) |
302 | | - assert runs_calls == [ |
303 | | - ["uv", "pip", "install", "--compile", "foo", "-p", PY_EXE], |
304 | | - ] |
| 299 | + def capture_command(*args: Any, **kwargs: Any): |
| 300 | + del kwargs |
| 301 | + runs_calls.append(args[0]) |
| 302 | + return mock_proc |
| 303 | + |
| 304 | + mock_popen.side_effect = capture_command |
| 305 | + |
| 306 | + pm = UvPackageManager() |
| 307 | + await pm._install("foo", upgrade=False) |
| 308 | + |
| 309 | + assert runs_calls == [ |
| 310 | + ["uv", "pip", "install", "--compile", "foo", "-p", PY_EXE], |
| 311 | + ] |
305 | 312 |
|
306 | 313 |
|
307 | 314 | def test_log_callback_type() -> None: |
@@ -438,28 +445,30 @@ async def test_uv_install_with_log_callback() -> None: |
438 | 445 | def log_callback(log_line: str) -> None: |
439 | 446 | captured_logs.append(log_line) |
440 | 447 |
|
441 | | - class MockUvPackageManager(UvPackageManager): |
442 | | - def run( |
443 | | - self, |
444 | | - command: list[str], |
445 | | - log_callback: Optional[LogCallback] = None, |
446 | | - ) -> bool: |
447 | | - del command |
448 | | - if log_callback: |
449 | | - log_callback("Resolving dependencies...\n") |
450 | | - log_callback("Installing packages...\n") |
451 | | - return True |
| 448 | + mock_stdout_lines = [ |
| 449 | + b"Resolving dependencies...\n", |
| 450 | + b"Installing packages...\n", |
| 451 | + ] |
452 | 452 |
|
453 | | - pm = MockUvPackageManager() |
454 | | - result = await pm._install( |
455 | | - "pandas", upgrade=False, log_callback=log_callback |
456 | | - ) |
| 453 | + with ( |
| 454 | + patch("subprocess.Popen") as mock_popen, |
| 455 | + patch("sys.stdout.buffer.write"), |
| 456 | + ): |
| 457 | + mock_proc = MagicMock() |
| 458 | + mock_proc.stdout.readline.side_effect = mock_stdout_lines + [b""] |
| 459 | + mock_proc.wait.return_value = 0 |
| 460 | + mock_popen.return_value = mock_proc |
457 | 461 |
|
458 | | - assert result is True |
459 | | - assert captured_logs == [ |
460 | | - "Resolving dependencies...\n", |
461 | | - "Installing packages...\n", |
462 | | - ] |
| 462 | + pm = UvPackageManager() |
| 463 | + result = await pm._install( |
| 464 | + "pandas", upgrade=False, log_callback=log_callback |
| 465 | + ) |
| 466 | + |
| 467 | + assert result is True |
| 468 | + assert captured_logs == [ |
| 469 | + "Resolving dependencies...\n", |
| 470 | + "Installing packages...\n", |
| 471 | + ] |
463 | 472 |
|
464 | 473 |
|
465 | 474 | async def test_micropip_install_with_log_callback() -> None: |
|
0 commit comments