Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
45c5cac2bc | ||
|
|
601c608c58 | ||
|
|
e2bccbafee | ||
|
|
340b585bfd | ||
|
|
ff4ab500df |
@@ -18,6 +18,54 @@ import re
|
||||
|
||||
_FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
|
||||
|
||||
|
||||
# Repo name disambiguation rules for blind PR queue review.
|
||||
# User phrases referencing the "MCP Gitea tool" (or similar) must resolve to
|
||||
# Gitea-Tools repo, not be confused with mcp-control-plane.
|
||||
# If ambiguous (e.g. just "open PRs"), check both configured repos.
|
||||
REPO_ALIASES = {
|
||||
"gitea-tools": "Scaled-Tech-Consulting/Gitea-Tools",
|
||||
"gitea tool": "Scaled-Tech-Consulting/Gitea-Tools",
|
||||
"mcp gitea tool": "Scaled-Tech-Consulting/Gitea-Tools",
|
||||
"gitea mcp tool": "Scaled-Tech-Consulting/Gitea-Tools",
|
||||
"gitea-tools repo": "Scaled-Tech-Consulting/Gitea-Tools",
|
||||
"mcp-control-plane": "Scaled-Tech-Consulting/mcp-control-plane",
|
||||
"mcp control plane": "Scaled-Tech-Consulting/mcp-control-plane",
|
||||
}
|
||||
|
||||
|
||||
def resolve_repos_from_user_reference(
|
||||
reference: str, configured: list[str] | None = None
|
||||
) -> list[str]:
|
||||
"""Resolve a user reference string to the list of target repos to inventory.
|
||||
|
||||
- Exact aliases for Gitea-Tools map only to Gitea-Tools.
|
||||
- mcp-control-plane aliases map only to it.
|
||||
- Empty, ambiguous, or general "open PRs" default to all configured repos
|
||||
(both by default).
|
||||
- Returns subset of configured; never invents new repos.
|
||||
"""
|
||||
if configured is None:
|
||||
configured = [
|
||||
"Scaled-Tech-Consulting/Gitea-Tools",
|
||||
"Scaled-Tech-Consulting/mcp-control-plane",
|
||||
]
|
||||
if not reference or not reference.strip():
|
||||
return list(configured)
|
||||
|
||||
ref_lower = reference.lower()
|
||||
matched = []
|
||||
for alias, full_name in REPO_ALIASES.items():
|
||||
if alias in ref_lower:
|
||||
if full_name not in matched:
|
||||
matched.append(full_name)
|
||||
if matched:
|
||||
# return only the matched ones that are in configured, preserving order
|
||||
return [r for r in configured if r in matched]
|
||||
# no specific alias match → check all (complete inventory required)
|
||||
return list(configured)
|
||||
|
||||
|
||||
SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION = (
|
||||
"evidence missing: report contamination as unknown and "
|
||||
"choose another PR or stop"
|
||||
|
||||
@@ -181,11 +181,21 @@ Worktree folder = branch with `/` replaced by `-`
|
||||
that the diff base is the PR base branch. If `HEAD` does not match the
|
||||
pinned head, **stop before review/merge**
|
||||
(`review_proofs.verify_pinned_head_checkout`).
|
||||
6. **Inventory proof (#173):** a blind queue review must prove listing
|
||||
completeness before claiming "only PRs found": both configured
|
||||
repositories checked, open-PR filters stated, pagination handled or
|
||||
explicitly not needed, and the total open PR count per repo reported
|
||||
(`review_proofs.assess_inventory_completeness`).
|
||||
6. **Inventory proof (#173 + repo disambiguation hardening):** a blind queue
|
||||
review must prove listing completeness before claiming "only PRs found".
|
||||
Use repo-name disambiguation:
|
||||
- "Gitea-Tools" / "gitea tool" / "MCP Gitea tool" / "gitea MCP tool" /
|
||||
"gitea-tools repo" resolve **only** to `Scaled-Tech-Consulting/Gitea-Tools`.
|
||||
- "mcp-control-plane" resolves only to `Scaled-Tech-Consulting/mcp-control-plane`.
|
||||
- Ambiguous ("open PRs", no explicit repo, "MCP Gitea tooling") → inventory
|
||||
**both** configured repos.
|
||||
Report must state exactly which repo(s) were checked. If only one checked:
|
||||
"Only <repo> was checked. Other configured repos were not checked. This is
|
||||
not a complete queue inventory." Never let a single-repo zero hide PRs in
|
||||
the other.
|
||||
Both configured repos must be reported with state filter, pagination proof,
|
||||
and open-PR count (`review_proofs.assess_inventory_completeness` and
|
||||
`resolve_repos_from_user_reference`).
|
||||
7. Inspect the full diff; confirm scope matches the linked issue; flag unrelated files.
|
||||
8. Run the tests. Validation reporting must include the exact command and
|
||||
exact results: pass/fail, counts of tests passed/skipped/failed, any
|
||||
|
||||
@@ -5,6 +5,19 @@ Copy, fill the `<...>` fields, and paste as the task prompt.
|
||||
```text
|
||||
Task: review PR #<pr> for issue #<n>.
|
||||
|
||||
Repo name disambiguation (Gitea-Tools blind review hardening):
|
||||
- "Gitea-Tools", "gitea tool", "MCP Gitea tool", "gitea MCP tool", "gitea-tools repo"
|
||||
→ MUST resolve to `Scaled-Tech-Consulting/Gitea-Tools` (never treat as mcp-control-plane).
|
||||
- "mcp-control-plane", "mcp control plane" → only `Scaled-Tech-Consulting/mcp-control-plane`.
|
||||
- If user says "open PRs", "the queue", "MCP Gitea tooling" without explicit repo,
|
||||
or reference is ambiguous: check BOTH configured repos:
|
||||
`Scaled-Tech-Consulting/Gitea-Tools` and `Scaled-Tech-Consulting/mcp-control-plane`.
|
||||
- In the final report, always state exactly which repo(s) were checked.
|
||||
If only one was checked: explicitly say "Only <repo> was checked. Other
|
||||
configured repos were not checked. This is not a complete queue inventory."
|
||||
- A single-repo "no open PRs" result MUST NOT be reported as global "no open PRs"
|
||||
if the other configured repo was not inventoried.
|
||||
|
||||
Rules (llm-project-workflow):
|
||||
- Review in a SEPARATE detached review worktree, never the author's folder.
|
||||
- You must NOT be the PR author. If the authenticated user == PR author, stop.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -25,6 +26,7 @@ from review_proofs import ( # noqa: E402
|
||||
assess_self_review_contamination,
|
||||
assess_validation_report,
|
||||
build_final_report,
|
||||
resolve_repos_from_user_reference,
|
||||
verify_pinned_head_checkout,
|
||||
)
|
||||
|
||||
@@ -464,6 +466,117 @@ 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}")
|
||||
|
||||
|
||||
class TestRepoNameDisambiguation(unittest.TestCase):
|
||||
"""Harness assertions for repo name disambiguation (new blind-review hardening).
|
||||
|
||||
"MCP Gitea tool" etc. must resolve to Gitea-Tools, not silently default to
|
||||
mcp-control-plane. "open PRs" or ambiguous must check both. Single-repo
|
||||
zero-result must not hide PRs in the other configured repo.
|
||||
"""
|
||||
|
||||
CONFIGURED = [
|
||||
"Scaled-Tech-Consulting/Gitea-Tools",
|
||||
"Scaled-Tech-Consulting/mcp-control-plane",
|
||||
]
|
||||
|
||||
def test_gitea_tools_aliases_resolve_to_gitea_tools_only(self):
|
||||
for ref in [
|
||||
"MCP Gitea tool",
|
||||
"gitea tool",
|
||||
"Gitea-Tools",
|
||||
"gitea mcp tool",
|
||||
"gitea-tools repo",
|
||||
]:
|
||||
result = resolve_repos_from_user_reference(ref, self.CONFIGURED)
|
||||
self.assertEqual(result, ["Scaled-Tech-Consulting/Gitea-Tools"])
|
||||
|
||||
def test_mcp_control_plane_alias_resolves_only_to_it(self):
|
||||
result = resolve_repos_from_user_reference(
|
||||
"mcp-control-plane", self.CONFIGURED
|
||||
)
|
||||
self.assertEqual(result, ["Scaled-Tech-Consulting/mcp-control-plane"])
|
||||
|
||||
def test_ambiguous_or_empty_defaults_to_both(self):
|
||||
self.assertEqual(
|
||||
resolve_repos_from_user_reference("open PRs", self.CONFIGURED),
|
||||
self.CONFIGURED,
|
||||
)
|
||||
self.assertEqual(
|
||||
resolve_repos_from_user_reference("", self.CONFIGURED),
|
||||
self.CONFIGURED,
|
||||
)
|
||||
self.assertEqual(
|
||||
resolve_repos_from_user_reference("review the queue", self.CONFIGURED),
|
||||
self.CONFIGURED,
|
||||
)
|
||||
|
||||
def test_single_repo_zero_result_does_not_hide_other(self):
|
||||
# Simulate a run that only inventoried mcp because of bad alias resolution.
|
||||
# The inventory must still require Gitea-Tools to claim "no open PRs".
|
||||
mcp_only_reports = [
|
||||
{
|
||||
"repo": "Scaled-Tech-Consulting/mcp-control-plane",
|
||||
"state_filter": "open",
|
||||
"pagination_complete": True,
|
||||
"open_pr_count": 0,
|
||||
}
|
||||
]
|
||||
result = assess_inventory_completeness(
|
||||
repo_reports=mcp_only_reports,
|
||||
required_repos=self.CONFIGURED,
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(
|
||||
any("Gitea-Tools" in r for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_full_inventory_of_both_is_required_for_exhaustive_claim(self):
|
||||
both_reports = [
|
||||
{
|
||||
"repo": "Scaled-Tech-Consulting/Gitea-Tools",
|
||||
"state_filter": "open",
|
||||
"pagination_complete": True,
|
||||
"open_pr_count": 1,
|
||||
},
|
||||
{
|
||||
"repo": "Scaled-Tech-Consulting/mcp-control-plane",
|
||||
"state_filter": "open",
|
||||
"pagination_complete": True,
|
||||
"open_pr_count": 0,
|
||||
},
|
||||
]
|
||||
result = assess_inventory_completeness(
|
||||
repo_reports=both_reports, required_repos=self.CONFIGURED
|
||||
)
|
||||
self.assertTrue(result["complete"])
|
||||
self.assertTrue(result["can_claim_exhaustive"])
|
||||
|
||||
|
||||
class TestControllerHandoff(unittest.TestCase):
|
||||
"""Issue #182: final reports must end with a Controller Handoff."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user