Merge pull request 'test: isolate CLI output capture with monkeypatch/redirect to fix stdout corruption in full suite (Issue #178)' (#180) from feat/issue-178-fix-test-suite-stdout-capture into master
This commit was merged in pull request #180.
This commit is contained in:
@@ -11,6 +11,7 @@ import json
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
import contextlib
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
# The module under test lives in the repo root, not a package.
|
||||
@@ -38,7 +39,8 @@ class TestArgParsing(unittest.TestCase):
|
||||
@patch("create_issue.api_request", return_value={"number": 1, "html_url": "http://x/1"})
|
||||
@patch("create_issue.get_credentials", return_value=FAKE_CREDS)
|
||||
def test_minimal_args(self, _cred, _api):
|
||||
rc = create_issue.main(["--title", "Hello"])
|
||||
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
|
||||
rc = create_issue.main(["--title", "Hello"])
|
||||
self.assertEqual(rc, 0)
|
||||
|
||||
def test_missing_title_exits(self):
|
||||
@@ -50,7 +52,8 @@ class TestArgParsing(unittest.TestCase):
|
||||
@patch("create_issue.get_credentials", return_value=FAKE_CREDS)
|
||||
def test_remote_choices(self, _cred, _api):
|
||||
for remote in ("dadeschools", "prgs"):
|
||||
rc = create_issue.main(["--remote", remote, "--title", "X"])
|
||||
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
|
||||
rc = create_issue.main(["--remote", remote, "--title", "X"])
|
||||
self.assertEqual(rc, 0, f"--remote {remote} should be accepted")
|
||||
|
||||
def test_invalid_remote_exits(self):
|
||||
@@ -138,7 +141,8 @@ class TestAuthFailure(unittest.TestCase):
|
||||
|
||||
@patch("create_issue.get_credentials", return_value=("", ""))
|
||||
def test_no_credentials_returns_1(self, _cred):
|
||||
rc = create_issue.main(["--title", "T"])
|
||||
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
|
||||
rc = create_issue.main(["--title", "T"])
|
||||
self.assertEqual(rc, 1)
|
||||
|
||||
|
||||
@@ -152,7 +156,8 @@ class TestAPIError(unittest.TestCase):
|
||||
def test_api_error_returns_1(self, _cred):
|
||||
with patch("create_issue.api_request",
|
||||
side_effect=RuntimeError("HTTP 422: duplicate")):
|
||||
rc = create_issue.main(["--title", "Dup"])
|
||||
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
|
||||
rc = create_issue.main(["--title", "Dup"])
|
||||
self.assertEqual(rc, 1)
|
||||
|
||||
|
||||
|
||||
@@ -7,6 +7,7 @@ import io
|
||||
import json
|
||||
import sys
|
||||
import unittest
|
||||
import contextlib
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
@@ -34,7 +35,8 @@ class TestArgParsing(unittest.TestCase):
|
||||
@patch("create_pr.urllib.request.urlopen", return_value=_mock_urlopen())
|
||||
@patch("create_pr.get_credentials", return_value=FAKE_CREDS)
|
||||
def test_minimal_required_args(self, _cred, _url):
|
||||
rc = create_pr.main(["--title", "PR Title", "--head", "feat/branch"])
|
||||
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
|
||||
rc = create_pr.main(["--title", "PR Title", "--head", "feat/branch"])
|
||||
self.assertEqual(rc, 0)
|
||||
|
||||
def test_missing_title_exits(self):
|
||||
|
||||
@@ -2,9 +2,11 @@
|
||||
|
||||
All API calls are mocked — no real network or keychain access.
|
||||
"""
|
||||
import io
|
||||
import json
|
||||
import sys
|
||||
import unittest
|
||||
import contextlib
|
||||
from unittest.mock import MagicMock, call, patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
@@ -33,7 +35,8 @@ class TestLabelCreation(unittest.TestCase):
|
||||
|
||||
# Patch sys.argv to avoid --dry
|
||||
with patch.object(sys, "argv", ["manage_labels.py"]):
|
||||
manage_labels.main()
|
||||
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
|
||||
manage_labels.main()
|
||||
|
||||
# The GET call happens, but no POST calls for label creation
|
||||
get_calls = [c for c in mock_api.call_args_list if c[0][0] == "GET"]
|
||||
@@ -59,7 +62,8 @@ class TestLabelCreation(unittest.TestCase):
|
||||
|
||||
mock_api.side_effect = side_effect
|
||||
with patch.object(sys, "argv", ["manage_labels.py"]):
|
||||
manage_labels.main()
|
||||
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
|
||||
manage_labels.main()
|
||||
|
||||
post_calls = [
|
||||
c for c in mock_api.call_args_list
|
||||
@@ -79,7 +83,8 @@ class TestDryRun(unittest.TestCase):
|
||||
mock_api.return_value = [] # no existing labels
|
||||
|
||||
with patch.object(sys, "argv", ["manage_labels.py", "--dry"]):
|
||||
manage_labels.main()
|
||||
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
|
||||
manage_labels.main()
|
||||
|
||||
# Only the GET call should be made, no POST or PUT
|
||||
for c in mock_api.call_args_list:
|
||||
@@ -107,7 +112,8 @@ class TestLabelMapping(unittest.TestCase):
|
||||
|
||||
mock_api.side_effect = side_effect
|
||||
with patch.object(sys, "argv", ["manage_labels.py"]):
|
||||
manage_labels.main()
|
||||
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
|
||||
manage_labels.main()
|
||||
|
||||
put_calls = [c for c in mock_api.call_args_list if c[0][0] == "PUT"]
|
||||
self.assertEqual(len(put_calls), len(manage_labels.MAPPING))
|
||||
|
||||
@@ -45,13 +45,17 @@ class TestMergeDisabled(unittest.TestCase):
|
||||
mock_api.assert_not_called()
|
||||
|
||||
def test_message_points_to_gated_workflow(self):
|
||||
from _pytest.monkeypatch import MonkeyPatch
|
||||
import io
|
||||
import contextlib
|
||||
with patch("merge_pr.get_auth_header", return_value=FAKE_CREDS), \
|
||||
patch("merge_pr.api_request") as mock_api:
|
||||
buf = io.StringIO()
|
||||
with contextlib.redirect_stderr(buf):
|
||||
monkeypatch = MonkeyPatch()
|
||||
monkeypatch.setattr(sys, "stderr", buf)
|
||||
try:
|
||||
rc = merge_pr.main(["--pr-number", "81"])
|
||||
finally:
|
||||
monkeypatch.undo()
|
||||
self.assertEqual(rc, 2)
|
||||
mock_api.assert_not_called()
|
||||
msg = buf.getvalue().lower()
|
||||
|
||||
@@ -9,6 +9,8 @@ import shutil
|
||||
from unittest.mock import patch
|
||||
from io import StringIO
|
||||
|
||||
from _pytest.monkeypatch import MonkeyPatch
|
||||
|
||||
# Add project root to sys.path
|
||||
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
if PROJECT_ROOT not in sys.path:
|
||||
@@ -127,24 +129,29 @@ class TestMigrateProfiles(unittest.TestCase):
|
||||
v2_data = migrate_profiles.migrate_v1_to_v2(self.v1_content)
|
||||
self.assertTrue(migrate_profiles.validate_v2_data(v2_data))
|
||||
|
||||
@patch("sys.stdout", new_callable=StringIO)
|
||||
def test_dry_run_default(self, mock_stdout):
|
||||
def test_dry_run_default(self):
|
||||
"""Verify that running without -w prints generated config without modifying files."""
|
||||
output_file = os.path.join(self.temp_dir, "migrated_dry.json")
|
||||
test_args = [
|
||||
"migrate_profiles.py",
|
||||
"-i", self.input_file,
|
||||
"-o", output_file
|
||||
]
|
||||
with patch.object(sys, "argv", test_args):
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
migrate_profiles.main()
|
||||
self.assertEqual(cm.exception.code, 0)
|
||||
monkeypatch = MonkeyPatch()
|
||||
mock_stdout = StringIO()
|
||||
monkeypatch.setattr(sys, "stdout", mock_stdout)
|
||||
try:
|
||||
output_file = os.path.join(self.temp_dir, "migrated_dry.json")
|
||||
test_args = [
|
||||
"migrate_profiles.py",
|
||||
"-i", self.input_file,
|
||||
"-o", output_file
|
||||
]
|
||||
with patch.object(sys, "argv", test_args):
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
migrate_profiles.main()
|
||||
self.assertEqual(cm.exception.code, 0)
|
||||
|
||||
self.assertFalse(os.path.exists(output_file))
|
||||
self.assertFalse(os.path.exists(f"{self.input_file}.bak"))
|
||||
self.assertFalse(os.path.exists(output_file))
|
||||
self.assertFalse(os.path.exists(f"{self.input_file}.bak"))
|
||||
|
||||
stdout_output = mock_stdout.getvalue()
|
||||
stdout_output = mock_stdout.getvalue()
|
||||
finally:
|
||||
monkeypatch.undo()
|
||||
self.assertIn("DRY-RUN MODE", stdout_output)
|
||||
self.assertIn("version", stdout_output)
|
||||
self.assertIn("identities", stdout_output)
|
||||
@@ -165,13 +172,18 @@ class TestMigrateProfiles(unittest.TestCase):
|
||||
json.dump(sensitive, f)
|
||||
|
||||
test_args = ["migrate_profiles.py", "-i", self.input_file]
|
||||
with patch("sys.stdout", new_callable=StringIO) as mock_stdout:
|
||||
monkeypatch = MonkeyPatch()
|
||||
mock_stdout = StringIO()
|
||||
monkeypatch.setattr(sys, "stdout", mock_stdout)
|
||||
try:
|
||||
with patch.object(sys, "argv", test_args):
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
migrate_profiles.main()
|
||||
self.assertEqual(cm.exception.code, 0)
|
||||
|
||||
stdout_output = mock_stdout.getvalue()
|
||||
stdout_output = mock_stdout.getvalue()
|
||||
finally:
|
||||
monkeypatch.undo()
|
||||
self.assertNotIn("super-secret-token-value", stdout_output)
|
||||
self.assertNotIn("token", stdout_output.lower())
|
||||
|
||||
|
||||
+6
-2
@@ -4,6 +4,8 @@ Mocks api_request and credentials.
|
||||
"""
|
||||
import sys
|
||||
import unittest
|
||||
import io
|
||||
import contextlib
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
@@ -24,14 +26,16 @@ class TestListPRs(unittest.TestCase):
|
||||
mock_api.return_value = [
|
||||
{"number": 1, "title": "PR 1", "head": {"ref": "branch1"}, "base": {"ref": "main"}, "html_url": "http://url1", "mergeable": True}
|
||||
]
|
||||
rc = list_prs.main([])
|
||||
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
|
||||
rc = list_prs.main([])
|
||||
self.assertEqual(rc, 0)
|
||||
mock_api.assert_called_once()
|
||||
|
||||
@patch("list_prs.api_request", return_value=[])
|
||||
@patch("list_prs.get_auth_header", return_value=FAKE_CREDS)
|
||||
def test_list_prs_empty(self, _auth, mock_api):
|
||||
rc = list_prs.main(["--state", "closed"])
|
||||
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
|
||||
rc = list_prs.main(["--state", "closed"])
|
||||
self.assertEqual(rc, 0)
|
||||
mock_api.assert_called_once()
|
||||
|
||||
|
||||
@@ -4,6 +4,8 @@ All tests mock credentials and API requests so no real network calls are made.
|
||||
"""
|
||||
import sys
|
||||
import unittest
|
||||
import io
|
||||
import contextlib
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
# The modules under test live in the repo root
|
||||
@@ -31,7 +33,8 @@ class TestCloseIssueCLI(unittest.TestCase):
|
||||
@patch("close_issue.api_request")
|
||||
@patch("close_issue.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_successful_close(self, _auth, mock_api):
|
||||
rc = close_issue.main(["42"])
|
||||
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
|
||||
rc = close_issue.main(["42"])
|
||||
self.assertEqual(rc, 0)
|
||||
mock_api.assert_called_once()
|
||||
url = mock_api.call_args[0][1]
|
||||
@@ -69,7 +72,8 @@ class TestMarkIssueCLI(unittest.TestCase):
|
||||
[{"id": 101, "name": "status:in-progress"}],
|
||||
[{"name": "status:in-progress"}],
|
||||
]
|
||||
rc = mark_issue.main(["15", "start"])
|
||||
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
|
||||
rc = mark_issue.main(["15", "start"])
|
||||
self.assertEqual(rc, 0)
|
||||
self.assertEqual(mock_api.call_count, 2)
|
||||
|
||||
|
||||
@@ -79,15 +79,19 @@ class TestAPIPayload(unittest.TestCase):
|
||||
self.assertEqual(mock_api.call_count, 0)
|
||||
|
||||
def test_merge_flag_message_points_to_gated_workflow(self):
|
||||
from _pytest.monkeypatch import MonkeyPatch
|
||||
import io
|
||||
import contextlib
|
||||
with patch("review_pr.get_auth_header", return_value=FAKE_CREDS), \
|
||||
patch("review_pr.api_request") as mock_api:
|
||||
buf = io.StringIO()
|
||||
with contextlib.redirect_stderr(buf):
|
||||
monkeypatch = MonkeyPatch()
|
||||
monkeypatch.setattr(sys, "stderr", buf)
|
||||
try:
|
||||
rc = review_pr.main([
|
||||
"--pr-number", "81", "--event", "APPROVE", "--merge",
|
||||
])
|
||||
finally:
|
||||
monkeypatch.undo()
|
||||
self.assertEqual(rc, 2)
|
||||
self.assertEqual(mock_api.call_count, 0)
|
||||
msg = buf.getvalue().lower()
|
||||
|
||||
@@ -14,6 +14,7 @@ each precondition of a blind queue review:
|
||||
|
||||
These are the harness assertions from the issue's Required behavior 7.
|
||||
"""
|
||||
import io
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
@@ -463,5 +464,30 @@ class TestFinalReport(unittest.TestCase):
|
||||
self.assertFalse(report["merge_allowed"])
|
||||
|
||||
|
||||
class TestStdoutIsolation(unittest.TestCase):
|
||||
"""Regression test for #178: tests must not close or corrupt stdout/stderr
|
||||
(prevents need for junitxml workaround in full suite runs and review validation).
|
||||
"""
|
||||
|
||||
def test_stdout_remains_usable(self):
|
||||
"""After typical test activity (mocks, redirects, prints from mains), stdout should be usable."""
|
||||
# Verify not closed
|
||||
self.assertFalse(getattr(sys.stdout, "closed", False))
|
||||
# Should be able to write (even if captured by pytest)
|
||||
try:
|
||||
sys.stdout.write("")
|
||||
sys.stdout.flush()
|
||||
except Exception as exc:
|
||||
self.fail(f"stdout write failed after test activity: {exc}")
|
||||
|
||||
def test_stderr_remains_usable(self):
|
||||
self.assertFalse(getattr(sys.stderr, "closed", False))
|
||||
try:
|
||||
sys.stderr.write("")
|
||||
sys.stderr.flush()
|
||||
except Exception as exc:
|
||||
self.fail(f"stderr write failed: {exc}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user