|
6 | 6 | import time
|
7 | 7 | from collections.abc import Callable, Generator
|
8 | 8 | from typing import TYPE_CHECKING
|
| 9 | +from enum import auto |
9 | 10 |
|
10 | 11 | import pytest
|
11 | 12 |
|
@@ -1424,3 +1425,215 @@ def test_wait_for_all_content_mismatched_match_types(wait_pane: Pane) -> None:
|
1424 | 1425 | [ContentMatchType.CONTAINS, ContentMatchType.EXACT],
|
1425 | 1426 | timeout=0.1,
|
1426 | 1427 | )
|
| 1428 | + |
| 1429 | + |
| 1430 | +def test_wait_for_pane_unsupported_match_type_returns_none(wait_pane: Pane) -> None: |
| 1431 | + """Test behavior when an unsupported match type is passed.""" |
| 1432 | + # Import the module directly to patch |
| 1433 | + import libtmux.test.waiter |
| 1434 | + |
| 1435 | + # Use monkeypatch to replace the retry_until_extended function |
| 1436 | + with pytest.MonkeyPatch.context() as monkeypatch: |
| 1437 | + |
| 1438 | + def mock_retry(*args: object, **kwargs: object) -> tuple[bool, Exception]: |
| 1439 | + """Mock version that simulates a TypeError.""" |
| 1440 | + return False, TypeError("Unsupported match type") |
| 1441 | + |
| 1442 | + # Patch at module level |
| 1443 | + monkeypatch.setattr( |
| 1444 | + libtmux.test.waiter, |
| 1445 | + "retry_until_extended", # Correct function name |
| 1446 | + mock_retry, |
| 1447 | + ) |
| 1448 | + |
| 1449 | + # Now run the function with the patched dependency |
| 1450 | + result = wait_for_pane_content( |
| 1451 | + wait_pane, |
| 1452 | + "test pattern", |
| 1453 | + ContentMatchType.CONTAINS, |
| 1454 | + timeout=0.1, |
| 1455 | + raises=False, |
| 1456 | + ) |
| 1457 | + |
| 1458 | + # Check expected failures |
| 1459 | + assert not result.success |
| 1460 | + assert result.error is not None |
| 1461 | + assert "Unsupported match type" in result.error |
| 1462 | + |
| 1463 | + |
| 1464 | +def test_wait_for_all_content_predicate_match(wait_pane: Pane) -> None: |
| 1465 | + """Test wait_for_all_content with predicate matching.""" |
| 1466 | + # Add some content to the pane |
| 1467 | + wait_pane.send_keys("echo 'Line 1'", enter=True) |
| 1468 | + wait_pane.send_keys("echo 'Line 2'", enter=True) |
| 1469 | + wait_pane.send_keys("echo 'Line 3'", enter=True) |
| 1470 | + |
| 1471 | + # Define multiple predicates that should match |
| 1472 | + def has_line_1(lines: list[str]) -> bool: |
| 1473 | + return any("Line 1" in line for line in lines) |
| 1474 | + |
| 1475 | + def has_line_2(lines: list[str]) -> bool: |
| 1476 | + return any("Line 2" in line for line in lines) |
| 1477 | + |
| 1478 | + # Wait for both predicates to match |
| 1479 | + result = wait_for_all_content( |
| 1480 | + wait_pane, |
| 1481 | + [has_line_1, has_line_2], |
| 1482 | + [ContentMatchType.PREDICATE, ContentMatchType.PREDICATE], |
| 1483 | + timeout=3, |
| 1484 | + ) |
| 1485 | + |
| 1486 | + assert result.success |
| 1487 | + assert result.matched_content is not None |
| 1488 | + assert isinstance(result.matched_content, list) |
| 1489 | + assert len(result.matched_content) == 2 |
| 1490 | + assert result.matched_content[0] == "predicate_function_0" |
| 1491 | + assert result.matched_content[1] == "predicate_function_1" |
| 1492 | + |
| 1493 | + |
| 1494 | +def test_wait_for_all_content_regex_match(wait_pane: Pane) -> None: |
| 1495 | + """Test wait_for_all_content with regex matching.""" |
| 1496 | + # Add some content to the pane |
| 1497 | + wait_pane.send_keys("echo 'ID: ABC-123'", enter=True) |
| 1498 | + wait_pane.send_keys("echo 'Code: XYZ-789'", enter=True) |
| 1499 | + |
| 1500 | + # Define regex patterns that should match |
| 1501 | + pattern1 = re.compile(r"ID: [A-Z]+-\d+") |
| 1502 | + pattern2 = re.compile(r"Code: [A-Z]+-\d+") |
| 1503 | + |
| 1504 | + # Create a type-compatible list of patterns - cast to expected type |
| 1505 | + patterns: list[str | re.Pattern[str] | Callable[[list[str]], bool]] = [ |
| 1506 | + pattern1, |
| 1507 | + pattern2, |
| 1508 | + ] |
| 1509 | + |
| 1510 | + # Wait for both patterns to match |
| 1511 | + result = wait_for_all_content( |
| 1512 | + wait_pane, |
| 1513 | + patterns, |
| 1514 | + [ContentMatchType.REGEX, ContentMatchType.REGEX], |
| 1515 | + timeout=3, |
| 1516 | + ) |
| 1517 | + |
| 1518 | + assert result.success |
| 1519 | + assert result.matched_content is not None |
| 1520 | + assert isinstance(result.matched_content, list) |
| 1521 | + assert len(result.matched_content) == 2 |
| 1522 | + # The matched content should contain the patterns |
| 1523 | + assert pattern1.pattern in result.matched_content |
| 1524 | + assert pattern2.pattern in result.matched_content |
| 1525 | + |
| 1526 | + |
| 1527 | +def test_wait_for_any_content_edge_case_behaviors(wait_pane: Pane) -> None: |
| 1528 | + """Test various edge case behaviors in wait_for_any_content function.""" |
| 1529 | + # Test raising TypeErrors correctly |
| 1530 | + # This will trigger the error branch that's not covered |
| 1531 | + with pytest.raises(TypeError): |
| 1532 | + # Pass a non-callable as predicate |
| 1533 | + wait_for_any_content( |
| 1534 | + wait_pane, |
| 1535 | + [123], # type: ignore |
| 1536 | + [ContentMatchType.PREDICATE], |
| 1537 | + timeout=0.1, |
| 1538 | + ) |
| 1539 | + |
| 1540 | + with pytest.raises(TypeError): |
| 1541 | + # Pass a non-string as exact match |
| 1542 | + wait_for_any_content( |
| 1543 | + wait_pane, |
| 1544 | + [123], # type: ignore |
| 1545 | + [ContentMatchType.EXACT], |
| 1546 | + timeout=0.1, |
| 1547 | + ) |
| 1548 | + |
| 1549 | + with pytest.raises(TypeError): |
| 1550 | + # Pass a non-string as contains match |
| 1551 | + wait_for_any_content( |
| 1552 | + wait_pane, |
| 1553 | + [123], # type: ignore |
| 1554 | + [ContentMatchType.CONTAINS], |
| 1555 | + timeout=0.1, |
| 1556 | + ) |
| 1557 | + |
| 1558 | + with pytest.raises(TypeError): |
| 1559 | + # Pass a non-string, non-pattern as regex match |
| 1560 | + wait_for_any_content( |
| 1561 | + wait_pane, |
| 1562 | + [123], # type: ignore |
| 1563 | + [ContentMatchType.REGEX], |
| 1564 | + timeout=0.1, |
| 1565 | + ) |
| 1566 | + |
| 1567 | + |
| 1568 | +def test_wait_for_all_content_exception_path(wait_pane: Pane) -> None: |
| 1569 | + """Test exception handling path in wait_for_all_content.""" |
| 1570 | + # Import the target module directly |
| 1571 | + import libtmux.test.waiter |
| 1572 | + |
| 1573 | + # Define the error message |
| 1574 | + error_msg = "Simulated error" |
| 1575 | + |
| 1576 | + # Monkeypatch the retry function at the module level |
| 1577 | + with pytest.MonkeyPatch.context() as monkeypatch: |
| 1578 | + |
| 1579 | + def mock_retry(*args: object, **kwargs: object) -> tuple[bool, Exception]: |
| 1580 | + """Mock version that simulates a runtime error.""" |
| 1581 | + return False, RuntimeError(error_msg) |
| 1582 | + |
| 1583 | + # Apply the patch to the correct function |
| 1584 | + monkeypatch.setattr( |
| 1585 | + libtmux.test.waiter, |
| 1586 | + "retry_until_extended", # Correct function name |
| 1587 | + mock_retry, |
| 1588 | + ) |
| 1589 | + |
| 1590 | + # Test with raises=False to catch the exception |
| 1591 | + result = wait_for_all_content( |
| 1592 | + wait_pane, |
| 1593 | + ["test pattern"], |
| 1594 | + ContentMatchType.CONTAINS, |
| 1595 | + timeout=0.5, |
| 1596 | + interval=0.1, |
| 1597 | + raises=False, |
| 1598 | + ) |
| 1599 | + |
| 1600 | + assert not result.success |
| 1601 | + assert result.error is not None |
| 1602 | + assert error_msg in result.error |
| 1603 | + |
| 1604 | + |
| 1605 | +def test_wait_for_any_content_exception_path(wait_pane: Pane) -> None: |
| 1606 | + """Test exception handling path in wait_for_any_content.""" |
| 1607 | + # Import the target module directly |
| 1608 | + import libtmux.test.waiter |
| 1609 | + |
| 1610 | + # Define the error message |
| 1611 | + error_msg = "Simulated error" |
| 1612 | + |
| 1613 | + # Monkeypatch the retry function at the module level |
| 1614 | + with pytest.MonkeyPatch.context() as monkeypatch: |
| 1615 | + |
| 1616 | + def mock_retry(*args: object, **kwargs: object) -> tuple[bool, Exception]: |
| 1617 | + """Mock version that simulates a runtime error.""" |
| 1618 | + return False, RuntimeError(error_msg) |
| 1619 | + |
| 1620 | + # Apply the patch to the correct function |
| 1621 | + monkeypatch.setattr( |
| 1622 | + libtmux.test.waiter, |
| 1623 | + "retry_until_extended", # Correct function name |
| 1624 | + mock_retry, |
| 1625 | + ) |
| 1626 | + |
| 1627 | + # Test with raises=False to catch the exception |
| 1628 | + result = wait_for_any_content( |
| 1629 | + wait_pane, |
| 1630 | + ["test pattern"], |
| 1631 | + ContentMatchType.CONTAINS, |
| 1632 | + timeout=0.5, |
| 1633 | + interval=0.1, |
| 1634 | + raises=False, |
| 1635 | + ) |
| 1636 | + |
| 1637 | + assert not result.success |
| 1638 | + assert result.error is not None |
| 1639 | + assert error_msg in result.error |
0 commit comments