"""Tests for the pre-create issue content gate (Issue #582).""" import sys import unittest from unittest.mock import patch sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) from issue_content_gate import ( # noqa: E402 assess_issue_content, find_vague_reference, has_durable_source_pointer, normalize_text, pre_create_issue_content_gate, ) from mcp_server import gitea_create_issue # noqa: E402 FAKE_AUTH = "Basic dGVzdDp0ZXN0" ISSUE_WRITE_ENV = { "GITEA_ALLOWED_OPERATIONS": "gitea.issue.create", } class TestNormalizeAndPointers(unittest.TestCase): def test_normalize_strips_punctuation_and_case(self): self.assertEqual(normalize_text(" The Drafted-Issue! "), "the drafted issue") def test_issue_ref_is_durable_pointer(self): self.assertTrue(has_durable_source_pointer("See #402 for the spec")) def test_file_path_is_durable_pointer(self): self.assertTrue(has_durable_source_pointer("Draft in task_capability_map.py")) def test_scratchpad_is_durable_pointer(self): self.assertTrue(has_durable_source_pointer("Full text in scratchpad/draft.md")) def test_url_is_durable_pointer(self): self.assertTrue( has_durable_source_pointer("https://gitea.prgs.cc/issues/1 has it") ) def test_plain_text_has_no_durable_pointer(self): self.assertFalse(has_durable_source_pointer("just some words here")) class TestFindVagueReference(unittest.TestCase): def test_bare_vague_phrases_are_flagged(self): for text in ( "the drafted issue", "The prepared issue", "the issue we discussed", "the previous draft", "Create the drafted issue", "please file the prepared issue", ): with self.subTest(text=text): self.assertIsNotNone(find_vague_reference(text)) def test_durable_pointer_defeats_vague_reference(self): # Same vague phrase but with a durable pointer -> not vague. self.assertIsNone(find_vague_reference("the drafted issue in #402")) self.assertIsNone( find_vague_reference("the prepared issue: see scratchpad/draft.md") ) def test_long_specific_body_with_incidental_phrase_is_not_vague(self): body = ( "As discussed, the create_issue tool must reject vague references. " "Add a preflight check that lists the exact missing fields and " "returns BLOCKED + DIAGNOSE with concrete acceptance criteria." ) self.assertIsNone(find_vague_reference(body)) def test_concrete_short_body_is_not_vague(self): self.assertIsNone(find_vague_reference("Fix null deref in auth handler")) def test_empty_is_not_vague(self): self.assertIsNone(find_vague_reference("")) class TestAssessIssueContent(unittest.TestCase): def test_missing_title_is_incomplete(self): result = assess_issue_content("", "some body") self.assertFalse(result["complete"]) self.assertIn("title", result["missing_fields"]) def test_vague_body_is_incomplete_and_diagnosed(self): result = assess_issue_content("Add the thing", "the drafted issue") self.assertFalse(result["complete"]) self.assertIn("body", result["vague_fields"]) self.assertIn("vague reference", result["diagnose"]) def test_vague_title_is_incomplete(self): result = assess_issue_content("the prepared issue", "") self.assertFalse(result["complete"]) self.assertIn("title", result["vague_fields"]) def test_complete_content_passes(self): result = assess_issue_content( "Block vague issue creation", "Add a preflight gate that rejects placeholder references.", ) self.assertTrue(result["complete"]) self.assertEqual(result["missing_fields"], []) self.assertEqual(result["vague_fields"], []) def test_title_only_is_allowed(self): # Empty body must NOT block: title-only creation stays allowed. result = assess_issue_content("A concrete specific title", "") self.assertTrue(result["complete"]) def test_durable_pointer_body_passes(self): result = assess_issue_content( "Implement the drafted change", "the drafted issue is fully specified in #582", ) self.assertTrue(result["complete"]) def test_allow_incomplete_override(self): result = assess_issue_content( "the drafted issue", "the drafted issue", allow_incomplete=True ) self.assertTrue(result["complete"]) self.assertTrue(result["override_applied"]) def test_gate_wrapper_sets_performed(self): blocked = pre_create_issue_content_gate("Add x", "the drafted issue") self.assertFalse(blocked["performed"]) ok = pre_create_issue_content_gate("Add x", "concrete actionable body") self.assertTrue(ok["performed"]) class TestCreateIssueContentMCPGate(unittest.TestCase): @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop") @patch("mcp_server.api_request") @patch("mcp_server.api_get_all") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_blocks_vague_reference_without_creating( self, _auth, mock_get_all, mock_api, mock_role_check ): mock_role_check.return_value = (True, []) with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True): result = gitea_create_issue( title="Create the drafted issue", body="the drafted issue", remote="prgs", ) self.assertFalse(result["performed"]) self.assertIn("content_gate", result) self.assertIn("body", result["content_gate"]["vague_fields"]) self.assertTrue(any("BLOCKED + DIAGNOSE" in r for r in result["reasons"])) mock_api.assert_not_called() mock_get_all.assert_not_called() @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop") @patch("mcp_server.api_request") @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_creates_when_content_is_concrete( self, _auth, _get_all, mock_api, mock_role_check ): mock_role_check.return_value = (True, []) mock_api.return_value = {"number": 7, "html_url": "https://x/issues/7"} with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True): result = gitea_create_issue( title="Block vague issue creation", body="Add a preflight gate rejecting placeholder references.", remote="prgs", ) self.assertEqual(result["number"], 7) @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop") @patch("mcp_server.api_request") @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_override_creates_despite_vague_content( self, _auth, _get_all, mock_api, mock_role_check ): mock_role_check.return_value = (True, []) mock_api.return_value = {"number": 8, "html_url": "https://x/issues/8"} with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True): result = gitea_create_issue( title="the drafted issue", body="the drafted issue", remote="prgs", allow_incomplete_content=True, ) self.assertEqual(result["number"], 8) if __name__ == "__main__": unittest.main()