Function exec_command

shared_lib/system.py:26–64

Standardized subprocess execution wrapper.

(
    cmd: List[str],
    cwd: Optional[str] = None,
    capture_output: bool = False,
    env: Optional[Mapping[str, str]] = None,
)

def exec_command(
    cmd: List[str],
    cwd: Optional[str] = None,
    capture_output: bool = False,
    env: Optional[Mapping[str, str]] = None,
) -> subprocess.CompletedProcess:
    """Standardized subprocess execution wrapper."""
    log.debug("Executing command", hypothesisId="SYS", cmd=" ".join(cmd), cwd=cwd)
    try:
        # Merge current env with provided env if exists
        full_env = os.environ.copy()
        if env:
            full_env.update(env)

        result = subprocess.run(
            cmd,
            cwd=cwd,
            capture_output=capture_output,
            text=True,
            check=False,
            env=full_env,
        )
        if result.returncode != 0:
            log.debug(
                "Command failed",
                hypothesisId="SYS",
                cmd=" ".join(cmd),
                exit_code=result.returncode,
                stderr=result.stderr if capture_output else "Check container logs",
            )
        return result
    except Exception as e:
        log.debug(
            "Command execution error",
            hypothesisId="SYS",
            cmd=" ".join(cmd),
            error=str(e),
        )
        raise e


def csv_to_dict(filename: str) -> Dict[str, List[str]]: