|
5 | 5 | import io |
6 | 6 | import os |
7 | 7 | import tempfile |
8 | | -from unittest.mock import AsyncMock, mock_open, patch |
| 8 | +from unittest.mock import AsyncMock, Mock, mock_open, patch |
9 | 9 |
|
10 | 10 | import pytest |
11 | 11 |
|
12 | | -from pybricksdev.cli import Run, Tool |
| 12 | +from pybricksdev.cli import Compile, Run, Tool, Udev |
13 | 13 |
|
14 | 14 |
|
15 | 15 | class TestTool: |
@@ -246,3 +246,350 @@ async def test_download_connection_error(self): |
246 | 246 |
|
247 | 247 | # Verify disconnect was not called since connection failed |
248 | 248 | mock_hub.disconnect.assert_not_called() |
| 249 | + |
| 250 | + @pytest.mark.asyncio |
| 251 | + async def test_run_ble(self): |
| 252 | + """Test running a program with BLE connection.""" |
| 253 | + # Create a mock hub |
| 254 | + mock_hub = AsyncMock() |
| 255 | + mock_hub.run = AsyncMock() |
| 256 | + |
| 257 | + # Set up mocks using ExitStack |
| 258 | + with contextlib.ExitStack() as stack: |
| 259 | + # Create and manage temporary file |
| 260 | + temp = stack.enter_context( |
| 261 | + tempfile.NamedTemporaryFile( |
| 262 | + suffix=".py", mode="w+", delete=False, encoding="utf-8" |
| 263 | + ) |
| 264 | + ) |
| 265 | + temp.write("print('test')") |
| 266 | + temp_path = temp.name |
| 267 | + stack.callback(os.unlink, temp_path) |
| 268 | + |
| 269 | + # Create args |
| 270 | + args = argparse.Namespace( |
| 271 | + conntype="ble", |
| 272 | + file=stack.enter_context(open(temp_path, "r", encoding="utf-8")), |
| 273 | + name="MyHub", |
| 274 | + start=True, |
| 275 | + wait=True, |
| 276 | + ) |
| 277 | + |
| 278 | + mock_hub_class = stack.enter_context( |
| 279 | + patch( |
| 280 | + "pybricksdev.connections.pybricks.PybricksHubBLE", |
| 281 | + return_value=mock_hub, |
| 282 | + ) |
| 283 | + ) |
| 284 | + stack.enter_context( |
| 285 | + patch("pybricksdev.ble.find_device", return_value="mock_device") |
| 286 | + ) |
| 287 | + |
| 288 | + # Run the command |
| 289 | + run_cmd = Run() |
| 290 | + await run_cmd.run(args) |
| 291 | + |
| 292 | + # Verify the hub was created and used correctly |
| 293 | + mock_hub_class.assert_called_once_with("mock_device") |
| 294 | + mock_hub.connect.assert_called_once() |
| 295 | + mock_hub.run.assert_called_once_with(temp_path, True) |
| 296 | + mock_hub.disconnect.assert_called_once() |
| 297 | + |
| 298 | + @pytest.mark.asyncio |
| 299 | + async def test_run_usb(self): |
| 300 | + """Test running a program with USB connection.""" |
| 301 | + # Create a mock hub |
| 302 | + mock_hub = AsyncMock() |
| 303 | + mock_hub.run = AsyncMock() |
| 304 | + |
| 305 | + # Set up mocks using ExitStack |
| 306 | + with contextlib.ExitStack() as stack: |
| 307 | + # Create and manage temporary file |
| 308 | + temp = stack.enter_context( |
| 309 | + tempfile.NamedTemporaryFile( |
| 310 | + suffix=".py", mode="w+", delete=False, encoding="utf-8" |
| 311 | + ) |
| 312 | + ) |
| 313 | + temp.write("print('test')") |
| 314 | + temp_path = temp.name |
| 315 | + stack.callback(os.unlink, temp_path) |
| 316 | + |
| 317 | + # Create args |
| 318 | + args = argparse.Namespace( |
| 319 | + conntype="usb", |
| 320 | + file=stack.enter_context(open(temp_path, "r", encoding="utf-8")), |
| 321 | + name=None, |
| 322 | + start=True, |
| 323 | + wait=True, |
| 324 | + ) |
| 325 | + |
| 326 | + mock_hub_class = stack.enter_context( |
| 327 | + patch( |
| 328 | + "pybricksdev.connections.pybricks.PybricksHubUSB", |
| 329 | + return_value=mock_hub, |
| 330 | + ) |
| 331 | + ) |
| 332 | + stack.enter_context(patch("usb.core.find", return_value="mock_device")) |
| 333 | + |
| 334 | + # Run the command |
| 335 | + run_cmd = Run() |
| 336 | + await run_cmd.run(args) |
| 337 | + |
| 338 | + # Verify the hub was created and used correctly |
| 339 | + mock_hub_class.assert_called_once_with("mock_device") |
| 340 | + mock_hub.connect.assert_called_once() |
| 341 | + mock_hub.run.assert_called_once_with(temp_path, True) |
| 342 | + mock_hub.disconnect.assert_called_once() |
| 343 | + |
| 344 | + @pytest.mark.asyncio |
| 345 | + async def test_run_stdin(self): |
| 346 | + """Test running a program from stdin.""" |
| 347 | + # Create a mock hub |
| 348 | + mock_hub = AsyncMock() |
| 349 | + mock_hub.run = AsyncMock() |
| 350 | + |
| 351 | + # Create a mock stdin |
| 352 | + mock_stdin = io.StringIO("print('test')") |
| 353 | + mock_stdin.buffer = io.BytesIO(b"print('test')") |
| 354 | + mock_stdin.name = "<stdin>" |
| 355 | + |
| 356 | + # Create args |
| 357 | + args = argparse.Namespace( |
| 358 | + conntype="ble", |
| 359 | + file=mock_stdin, |
| 360 | + name="MyHub", |
| 361 | + start=True, |
| 362 | + wait=True, |
| 363 | + ) |
| 364 | + |
| 365 | + # Set up mocks using ExitStack |
| 366 | + with contextlib.ExitStack() as stack: |
| 367 | + mock_hub_class = stack.enter_context( |
| 368 | + patch( |
| 369 | + "pybricksdev.connections.pybricks.PybricksHubBLE", |
| 370 | + return_value=mock_hub, |
| 371 | + ) |
| 372 | + ) |
| 373 | + stack.enter_context( |
| 374 | + patch("pybricksdev.ble.find_device", return_value="mock_device") |
| 375 | + ) |
| 376 | + mock_temp = stack.enter_context(patch("tempfile.NamedTemporaryFile")) |
| 377 | + mock_temp.return_value.__enter__.return_value.name = "/tmp/test.py" |
| 378 | + mock_temp.return_value.__enter__.return_value.write = Mock() |
| 379 | + mock_temp.return_value.__enter__.return_value.flush = Mock() |
| 380 | + |
| 381 | + # Run the command |
| 382 | + run_cmd = Run() |
| 383 | + await run_cmd.run(args) |
| 384 | + |
| 385 | + # Verify the hub was created and used correctly |
| 386 | + mock_hub_class.assert_called_once_with("mock_device") |
| 387 | + mock_hub.connect.assert_called_once() |
| 388 | + mock_hub.run.assert_called_once_with("<stdin>", True) |
| 389 | + mock_hub.disconnect.assert_called_once() |
| 390 | + |
| 391 | + @pytest.mark.asyncio |
| 392 | + async def test_run_connection_error(self): |
| 393 | + """Test handling connection errors.""" |
| 394 | + # Create a mock hub that raises an error during connect |
| 395 | + mock_hub = AsyncMock() |
| 396 | + mock_hub.connect.side_effect = RuntimeError("Connection failed") |
| 397 | + |
| 398 | + # Set up mocks using ExitStack |
| 399 | + with contextlib.ExitStack() as stack: |
| 400 | + # Create and manage temporary file |
| 401 | + temp = stack.enter_context( |
| 402 | + tempfile.NamedTemporaryFile( |
| 403 | + suffix=".py", mode="w+", delete=False, encoding="utf-8" |
| 404 | + ) |
| 405 | + ) |
| 406 | + temp.write("print('test')") |
| 407 | + temp_path = temp.name |
| 408 | + stack.callback(os.unlink, temp_path) |
| 409 | + |
| 410 | + # Create args |
| 411 | + args = argparse.Namespace( |
| 412 | + conntype="ble", |
| 413 | + file=stack.enter_context(open(temp_path, "r", encoding="utf-8")), |
| 414 | + name="MyHub", |
| 415 | + start=False, |
| 416 | + wait=True, |
| 417 | + ) |
| 418 | + |
| 419 | + stack.enter_context( |
| 420 | + patch( |
| 421 | + "pybricksdev.connections.pybricks.PybricksHubBLE", |
| 422 | + return_value=mock_hub, |
| 423 | + ) |
| 424 | + ) |
| 425 | + stack.enter_context( |
| 426 | + patch("pybricksdev.ble.find_device", return_value="mock_device") |
| 427 | + ) |
| 428 | + |
| 429 | + # Run the command and verify it raises the error |
| 430 | + run_cmd = Run() |
| 431 | + with pytest.raises(RuntimeError, match="Connection failed"): |
| 432 | + await run_cmd.run(args) |
| 433 | + |
| 434 | + # Verify disconnect was not called since connection failed |
| 435 | + mock_hub.disconnect.assert_not_called() |
| 436 | + |
| 437 | + |
| 438 | +class TestCompile: |
| 439 | + """Tests for the Compile command.""" |
| 440 | + |
| 441 | + def test_add_parser(self): |
| 442 | + """Test that the parser is set up correctly.""" |
| 443 | + # Create a subparsers object |
| 444 | + parser = argparse.ArgumentParser() |
| 445 | + subparsers = parser.add_subparsers() |
| 446 | + |
| 447 | + # Add the compile command |
| 448 | + compile_cmd = Compile() |
| 449 | + compile_cmd.add_parser(subparsers) |
| 450 | + |
| 451 | + # Verify the parser was created with correct arguments |
| 452 | + assert "compile" in subparsers.choices |
| 453 | + parser = subparsers.choices["compile"] |
| 454 | + assert parser.tool is compile_cmd |
| 455 | + |
| 456 | + # Test that required arguments are present |
| 457 | + mock_file = mock_open(read_data="print('test')") |
| 458 | + mock_file.return_value.name = "test.py" |
| 459 | + with patch("builtins.open", mock_file): |
| 460 | + args = parser.parse_args(["test.py"]) |
| 461 | + assert args.file.name == "test.py" |
| 462 | + assert args.abi == 6 # Default ABI version |
| 463 | + |
| 464 | + # Test with custom ABI version |
| 465 | + mock_file = mock_open(read_data="print('test')") |
| 466 | + mock_file.return_value.name = "test.py" |
| 467 | + with patch("builtins.open", mock_file): |
| 468 | + args = parser.parse_args(["test.py", "--abi", "5"]) |
| 469 | + assert args.abi == 5 |
| 470 | + |
| 471 | + # Test that invalid ABI version is rejected |
| 472 | + with pytest.raises(SystemExit): |
| 473 | + parser.parse_args(["test.py", "--abi", "4"]) |
| 474 | + |
| 475 | + @pytest.mark.asyncio |
| 476 | + async def test_compile_file(self): |
| 477 | + """Test compiling a Python file.""" |
| 478 | + # Create a mock compile function |
| 479 | + mock_compile = AsyncMock() |
| 480 | + mock_compile.return_value = b"compiled bytecode" |
| 481 | + |
| 482 | + # Set up mocks using ExitStack |
| 483 | + with contextlib.ExitStack() as stack: |
| 484 | + # Create and manage temporary file |
| 485 | + temp = stack.enter_context( |
| 486 | + tempfile.NamedTemporaryFile( |
| 487 | + suffix=".py", mode="w+", delete=False, encoding="utf-8" |
| 488 | + ) |
| 489 | + ) |
| 490 | + temp.write("print('test')") |
| 491 | + temp_path = temp.name |
| 492 | + stack.callback(os.unlink, temp_path) |
| 493 | + |
| 494 | + # Create args |
| 495 | + args = argparse.Namespace( |
| 496 | + file=stack.enter_context(open(temp_path, "r", encoding="utf-8")), |
| 497 | + abi=6, |
| 498 | + ) |
| 499 | + |
| 500 | + # Mock the compile function |
| 501 | + stack.enter_context( |
| 502 | + patch("pybricksdev.compile.compile_multi_file", mock_compile) |
| 503 | + ) |
| 504 | + mock_print = stack.enter_context(patch("pybricksdev.compile.print_mpy")) |
| 505 | + |
| 506 | + # Run the command |
| 507 | + compile_cmd = Compile() |
| 508 | + await compile_cmd.run(args) |
| 509 | + |
| 510 | + # Verify compilation was called correctly |
| 511 | + mock_compile.assert_called_once_with(temp_path, 6) |
| 512 | + mock_print.assert_called_once_with(b"compiled bytecode") |
| 513 | + |
| 514 | + @pytest.mark.asyncio |
| 515 | + async def test_compile_stdin(self): |
| 516 | + """Test compiling from stdin.""" |
| 517 | + # Create a mock stdin |
| 518 | + mock_stdin = io.StringIO("print('test')") |
| 519 | + mock_stdin.buffer = io.BytesIO(b"print('test')") |
| 520 | + mock_stdin.name = "<stdin>" |
| 521 | + |
| 522 | + # Create a mock compile function |
| 523 | + mock_compile = AsyncMock() |
| 524 | + mock_compile.return_value = b"compiled bytecode" |
| 525 | + |
| 526 | + # Set up mocks using ExitStack |
| 527 | + with contextlib.ExitStack() as stack: |
| 528 | + # Create args |
| 529 | + args = argparse.Namespace( |
| 530 | + file=mock_stdin, |
| 531 | + abi=6, |
| 532 | + ) |
| 533 | + |
| 534 | + # Mock the compile function and tempfile |
| 535 | + stack.enter_context( |
| 536 | + patch("pybricksdev.compile.compile_multi_file", mock_compile) |
| 537 | + ) |
| 538 | + mock_print = stack.enter_context(patch("pybricksdev.compile.print_mpy")) |
| 539 | + mock_temp = stack.enter_context(patch("tempfile.NamedTemporaryFile")) |
| 540 | + mock_temp.return_value.__enter__.return_value.name = "/tmp/test.py" |
| 541 | + mock_temp.return_value.__enter__.return_value.write = Mock() |
| 542 | + mock_temp.return_value.__enter__.return_value.flush = Mock() |
| 543 | + |
| 544 | + # Run the command |
| 545 | + compile_cmd = Compile() |
| 546 | + await compile_cmd.run(args) |
| 547 | + |
| 548 | + # Verify compilation was called correctly |
| 549 | + mock_compile.assert_called_once_with("<stdin>", 6) |
| 550 | + mock_print.assert_called_once_with(b"compiled bytecode") |
| 551 | + |
| 552 | + |
| 553 | +class TestUdev: |
| 554 | + """Tests for the Udev command.""" |
| 555 | + |
| 556 | + def test_add_parser(self): |
| 557 | + """Test that the parser is set up correctly.""" |
| 558 | + # Create a subparsers object |
| 559 | + parser = argparse.ArgumentParser() |
| 560 | + subparsers = parser.add_subparsers() |
| 561 | + |
| 562 | + # Add the udev command |
| 563 | + udev_cmd = Udev() |
| 564 | + udev_cmd.add_parser(subparsers) |
| 565 | + |
| 566 | + # Verify the parser was created with correct arguments |
| 567 | + assert "udev" in subparsers.choices |
| 568 | + parser = subparsers.choices["udev"] |
| 569 | + assert parser.tool is udev_cmd |
| 570 | + |
| 571 | + @pytest.mark.asyncio |
| 572 | + async def test_print_rules(self): |
| 573 | + """Test printing udev rules.""" |
| 574 | + # Mock the read_text function |
| 575 | + mock_rules = ( |
| 576 | + '# Mock udev rules\nSUBSYSTEM=="usb", ATTRS{idVendor}=="0694", MODE="0666"' |
| 577 | + ) |
| 578 | + mock_read_text = Mock(return_value=mock_rules) |
| 579 | + |
| 580 | + # Set up mocks using ExitStack |
| 581 | + with contextlib.ExitStack() as stack: |
| 582 | + # Create args |
| 583 | + args = argparse.Namespace() |
| 584 | + |
| 585 | + # Mock the read_text function |
| 586 | + stack.enter_context(patch("importlib.resources.read_text", mock_read_text)) |
| 587 | + mock_print = stack.enter_context(patch("builtins.print")) |
| 588 | + |
| 589 | + # Run the command |
| 590 | + udev_cmd = Udev() |
| 591 | + await udev_cmd.run(args) |
| 592 | + |
| 593 | + # Verify the rules were printed |
| 594 | + mock_read_text.assert_called_once() |
| 595 | + mock_print.assert_called_once_with(mock_rules) |
0 commit comments