PopenPlugin

PopenPlugin

PopenPlugin(verifier)

Bases: StateMachinePlugin

Popen interception plugin.

Replaces subprocess.Popen with _FakePopen at activate() time and restores the original at deactivate() time. Uses reference counting so nested sandboxes work correctly.

States: created -> running -> terminated

Coexists with SubprocessPlugin: SubprocessPlugin patches subprocess.run and shutil.which; PopenPlugin patches subprocess.Popen. Both plugins target independent names in the subprocess module and restore correctly.

Source code in src/tripwire/plugins/popen_plugin.py

def __init__(self, verifier: "StrictVerifier") -> None:
    super().__init__(verifier)
    self._spawn_sentinel = _StepSentinel(_SOURCE_SPAWN)
    self._communicate_sentinel = _StepSentinel(_SOURCE_COMMUNICATE)
    self._wait_sentinel = _StepSentinel(_SOURCE_WAIT)

install_patches

install_patches()

Install subprocess.Popen patch.

Source code in src/tripwire/plugins/popen_plugin.py

def install_patches(self) -> None:
    """Install subprocess.Popen patch."""
    global _tripwire_popen_class

    PopenPlugin._original_popen = subprocess.Popen
    _tripwire_popen_class = _FakePopen
    setattr(subprocess, "Popen", _FakePopen)

restore_patches

restore_patches()

Restore original subprocess.Popen.

Source code in src/tripwire/plugins/popen_plugin.py

def restore_patches(self) -> None:
    """Restore original subprocess.Popen."""
    global _tripwire_popen_class

    if PopenPlugin._original_popen is not None:
        setattr(subprocess, "Popen", PopenPlugin._original_popen)
        PopenPlugin._original_popen = None
    _tripwire_popen_class = None

check_conflicts

check_conflicts()

Verify subprocess.Popen has not been patched by a third party.

Source code in src/tripwire/plugins/popen_plugin.py

def check_conflicts(self) -> None:
    """Verify subprocess.Popen has not been patched by a third party."""
    current_popen: Any = subprocess.Popen
    if current_popen is not _ORIGINAL_POPEN and current_popen is not _FakePopen:
        mod = getattr(current_popen, "__module__", None) or ""
        qualname = getattr(current_popen, "__qualname__", None) or ""
        if "unittest.mock" in mod or "MagicMock" in qualname:
            patcher = "unittest.mock"
        elif "pytest_mock" in mod:
            patcher = "pytest-mock"
        else:
            patcher = "an unknown library"
        raise ConflictError(target="subprocess.Popen", patcher=patcher)

matches

matches(interaction, expected)

Field-by-field comparison with dirty-equals support.

Source code in src/tripwire/plugins/popen_plugin.py

def matches(self, interaction: Interaction, expected: dict[str, Any]) -> bool:
    """Field-by-field comparison with dirty-equals support."""
    try:
        for key, expected_val in expected.items():
            actual_val = interaction.details.get(key)
            if expected_val != actual_val:
                return False
        return True
    except Exception:
        return False

assertable_fields

assertable_fields(interaction)

Return assertable fields for each step type.

Source code in src/tripwire/plugins/popen_plugin.py

def assertable_fields(self, interaction: Interaction) -> frozenset[str]:
    """Return assertable fields for each step type."""
    if interaction.source_id == _SOURCE_WAIT:
        return frozenset()
    return frozenset(interaction.details.keys())