feat(mma): Implement Tier 4 QA interceptor in shell_runner.py
This commit is contained in:
@@ -1,14 +1,16 @@
|
||||
# shell_runner.py
|
||||
import subprocess, shutil
|
||||
from pathlib import Path
|
||||
from typing import Callable, Optional
|
||||
|
||||
TIMEOUT_SECONDS = 60
|
||||
|
||||
def run_powershell(script: str, base_dir: str) -> str:
|
||||
def run_powershell(script: str, base_dir: str, qa_callback: Optional[Callable[[str], str]] = None) -> str:
|
||||
"""
|
||||
Run a PowerShell script with working directory set to base_dir.
|
||||
Returns a string combining stdout, stderr, and exit code.
|
||||
Raises nothing - all errors are captured into the return string.
|
||||
If qa_callback is provided and the command fails or has stderr,
|
||||
the callback is called with the stderr content and its result is appended.
|
||||
"""
|
||||
safe_dir = str(base_dir).replace("'", "''")
|
||||
full_script = f"Set-Location -LiteralPath '{safe_dir}'\n{script}"
|
||||
@@ -25,6 +27,13 @@ def run_powershell(script: str, base_dir: str) -> str:
|
||||
if r.stdout.strip(): parts.append(f"STDOUT:\n{r.stdout.strip()}")
|
||||
if r.stderr.strip(): parts.append(f"STDERR:\n{r.stderr.strip()}")
|
||||
parts.append(f"EXIT CODE: {r.returncode}")
|
||||
|
||||
# QA Interceptor logic
|
||||
if (r.returncode != 0 or r.stderr.strip()) and qa_callback:
|
||||
qa_analysis = qa_callback(r.stderr.strip())
|
||||
if qa_analysis:
|
||||
parts.append(f"\nQA ANALYSIS:\n{qa_analysis}")
|
||||
|
||||
return "\n".join(parts)
|
||||
except subprocess.TimeoutExpired: return f"ERROR: timed out after {TIMEOUT_SECONDS}s"
|
||||
except Exception as e: return f"ERROR: {e}"
|
||||
|
||||
102
tests/test_tier4_interceptor.py
Normal file
102
tests/test_tier4_interceptor.py
Normal file
@@ -0,0 +1,102 @@
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch
|
||||
import subprocess
|
||||
from shell_runner import run_powershell
|
||||
|
||||
def test_run_powershell_qa_callback_on_failure():
|
||||
"""
|
||||
Test that qa_callback is called when a powershell command fails (non-zero exit code).
|
||||
The result of the callback should be appended to the output.
|
||||
"""
|
||||
script = "Write-Error 'something went wrong'; exit 1"
|
||||
base_dir = "."
|
||||
|
||||
# Mocking subprocess.run to simulate failure
|
||||
mock_result = MagicMock()
|
||||
mock_result.stdout = ""
|
||||
mock_result.stderr = "something went wrong"
|
||||
mock_result.returncode = 1
|
||||
|
||||
qa_callback = MagicMock(return_value="QA ANALYSIS: This looks like a syntax error.")
|
||||
|
||||
with patch("subprocess.run", return_value=mock_result), \
|
||||
patch("shutil.which", return_value="powershell.exe"):
|
||||
|
||||
# We expect run_powershell to accept qa_callback
|
||||
output = run_powershell(script, base_dir, qa_callback=qa_callback)
|
||||
|
||||
# Verify callback was called with stderr
|
||||
qa_callback.assert_called_once_with("something went wrong")
|
||||
|
||||
# Verify output contains the callback result
|
||||
assert "QA ANALYSIS: This looks like a syntax error." in output
|
||||
assert "STDERR:\nsomething went wrong" in output
|
||||
assert "EXIT CODE: 1" in output
|
||||
|
||||
def test_run_powershell_qa_callback_on_stderr_only():
|
||||
"""
|
||||
Test that qa_callback is called when a command has stderr even if exit code is 0.
|
||||
"""
|
||||
script = "Write-Error 'non-fatal error'"
|
||||
base_dir = "."
|
||||
|
||||
mock_result = MagicMock()
|
||||
mock_result.stdout = "Success"
|
||||
mock_result.stderr = "non-fatal error"
|
||||
mock_result.returncode = 0
|
||||
|
||||
qa_callback = MagicMock(return_value="QA ANALYSIS: Ignorable warning.")
|
||||
|
||||
with patch("subprocess.run", return_value=mock_result), \
|
||||
patch("shutil.which", return_value="powershell.exe"):
|
||||
|
||||
output = run_powershell(script, base_dir, qa_callback=qa_callback)
|
||||
|
||||
qa_callback.assert_called_once_with("non-fatal error")
|
||||
assert "QA ANALYSIS: Ignorable warning." in output
|
||||
assert "STDOUT:\nSuccess" in output
|
||||
|
||||
def test_run_powershell_no_qa_callback_on_success():
|
||||
"""
|
||||
Test that qa_callback is NOT called when the command succeeds without stderr.
|
||||
"""
|
||||
script = "Write-Output 'All good'"
|
||||
base_dir = "."
|
||||
|
||||
mock_result = MagicMock()
|
||||
mock_result.stdout = "All good"
|
||||
mock_result.stderr = ""
|
||||
mock_result.returncode = 0
|
||||
|
||||
qa_callback = MagicMock()
|
||||
|
||||
with patch("subprocess.run", return_value=mock_result), \
|
||||
patch("shutil.which", return_value="powershell.exe"):
|
||||
|
||||
output = run_powershell(script, base_dir, qa_callback=qa_callback)
|
||||
|
||||
qa_callback.assert_not_called()
|
||||
assert "STDOUT:\nAll good" in output
|
||||
assert "EXIT CODE: 0" in output
|
||||
assert "QA ANALYSIS" not in output
|
||||
|
||||
def test_run_powershell_optional_qa_callback():
|
||||
"""
|
||||
Test that run_powershell still works without providing a qa_callback.
|
||||
"""
|
||||
script = "Write-Error 'error'"
|
||||
base_dir = "."
|
||||
|
||||
mock_result = MagicMock()
|
||||
mock_result.stdout = ""
|
||||
mock_result.stderr = "error"
|
||||
mock_result.returncode = 1
|
||||
|
||||
with patch("subprocess.run", return_value=mock_result), \
|
||||
patch("shutil.which", return_value="powershell.exe"):
|
||||
|
||||
# Should not raise TypeError even if qa_callback is not provided
|
||||
output = run_powershell(script, base_dir)
|
||||
|
||||
assert "STDERR:\nerror" in output
|
||||
assert "EXIT CODE: 1" in output
|
||||
Reference in New Issue
Block a user