|
| 1 | +"""Exercise core's subprocess refusal contract through the posted comment body.""" |
| 2 | + |
| 3 | +import json |
| 4 | +import os |
| 5 | +import re |
| 6 | +import subprocess |
| 7 | +import sys |
| 8 | +import tempfile |
| 9 | +import unittest |
| 10 | +from pathlib import Path |
| 11 | +from unittest.mock import patch |
| 12 | + |
| 13 | +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts")) |
| 14 | +import analyze_repository as ar |
| 15 | + |
| 16 | +ROOT = Path(__file__).resolve().parent.parent |
| 17 | +PRICING = "https://codeboarding.org/pricing" |
| 18 | + |
| 19 | + |
| 20 | +class QuotaCommentTests(unittest.TestCase): |
| 21 | + def _failure_body(self, report: object, tier: str = "hosted", mode: str = "review") -> tuple[str, str]: |
| 22 | + with tempfile.TemporaryDirectory() as tmp: |
| 23 | + root = Path(tmp) |
| 24 | + outputs = root / "analysis-outputs" |
| 25 | + program = ( |
| 26 | + "import os,sys; from pathlib import Path; " |
| 27 | + f"Path(os.environ['CODEBOARDING_ERROR_FILE']).write_text({json.dumps(json.dumps(report))}); " |
| 28 | + "sys.exit(3)" |
| 29 | + ) |
| 30 | + with ( |
| 31 | + patch.dict(os.environ, {"GITHUB_OUTPUT": str(outputs), "CODEBOARDING_LLM_TIER": tier}), |
| 32 | + self.assertRaises(ar.AnalysisError), |
| 33 | + ): |
| 34 | + ar._run_command([sys.executable, "-c", program], root / "out") |
| 35 | + |
| 36 | + reason = outputs.read_text().removeprefix("failure_reason=").strip() |
| 37 | + summary = root / "summary.md" |
| 38 | + script = "build-review-failure.sh" if mode == "review" else "sync-summary.sh" |
| 39 | + result = subprocess.run( |
| 40 | + [str(ROOT / "scripts" / "action" / script)], |
| 41 | + env={ |
| 42 | + "PATH": os.environ["PATH"], |
| 43 | + "RUNNER_TEMP": tmp, |
| 44 | + "GITHUB_OUTPUT": str(root / "comment-outputs"), |
| 45 | + "GITHUB_STEP_SUMMARY": str(summary), |
| 46 | + "GITHUB_SERVER_URL": "https://github.com", |
| 47 | + "GITHUB_REPOSITORY": "octocat/hello", |
| 48 | + "GITHUB_RUN_ID": "123", |
| 49 | + "FAILURE_REASON": reason, |
| 50 | + }, |
| 51 | + capture_output=True, |
| 52 | + text=True, |
| 53 | + check=True, |
| 54 | + ) |
| 55 | + self.assertEqual(result.stderr, "") |
| 56 | + text = summary.read_text() |
| 57 | + if mode == "review": |
| 58 | + self.assertEqual(text, (root / "review-failure.md").read_text()) |
| 59 | + self.assertIn(f"path={root / 'review-failure.md'}", (root / "comment-outputs").read_text()) |
| 60 | + return text, reason |
| 61 | + |
| 62 | + def test_free_credits_comment_links_to_purchase_and_explains_setup(self): |
| 63 | + body, reason = self._failure_body({"version": 1, "code": "free_quota_exhausted"}) |
| 64 | + self.assertIn("### CodeBoarding review · failed", body) |
| 65 | + self.assertIn("free CodeBoarding credits for this week are exhausted", body) |
| 66 | + self.assertIn(f"[Buy a CodeBoarding license]({PRICING})", body) |
| 67 | + self.assertIn("shared across its repositories", body) |
| 68 | + self.assertIn("`CODEBOARDING_LICENSE`", body) |
| 69 | + self.assertIn("`llm: license`", body) |
| 70 | + self.assertIn("`license_key: ${{ secrets.CODEBOARDING_LICENSE }}`", body) |
| 71 | + self.assertIn("https://github.com/octocat/hello/actions/runs/123", body) |
| 72 | + self.assertNotIn("Traceback", body) |
| 73 | + self.assertIn(reason, body) |
| 74 | + |
| 75 | + def test_sync_reports_the_same_purchase_instructions(self): |
| 76 | + body, reason = self._failure_body({"version": 1, "code": "free_quota_exhausted"}, mode="sync") |
| 77 | + self.assertIn(f"[Buy a CodeBoarding license]({PRICING})", body) |
| 78 | + self.assertIn(reason, body) |
| 79 | + |
| 80 | + def test_supplied_license_is_checked_before_suggesting_another_purchase(self): |
| 81 | + for code in ("license_not_accepted", "free_quota_exhausted"): |
| 82 | + with self.subTest(code=code): |
| 83 | + body, _ = self._failure_body({"version": 1, "code": code}, tier="license") |
| 84 | + self.assertIn("Check that the license is active", body) |
| 85 | + self.assertIn(f"]({PRICING})", body) |
| 86 | + self.assertNotIn("After purchasing", body) |
| 87 | + |
| 88 | + def test_provider_billing_does_not_sell_a_codeboarding_license(self): |
| 89 | + for code in ("provider_quota_exhausted", "hosted_provider_quota_exhausted"): |
| 90 | + with self.subTest(code=code): |
| 91 | + body, _ = self._failure_body({"version": 1, "code": code}) |
| 92 | + self.assertNotIn(PRICING, body) |
| 93 | + self.assertNotIn("Buy", body) |
| 94 | + |
| 95 | + def test_unknown_or_invalid_report_keeps_the_legacy_message(self): |
| 96 | + for report in ( |
| 97 | + [], |
| 98 | + {"version": 2, "code": "free_quota_exhausted"}, |
| 99 | + {"version": 1, "code": []}, |
| 100 | + {"version": 1, "code": "unknown"}, |
| 101 | + ): |
| 102 | + with self.subTest(report=report): |
| 103 | + body, _ = self._failure_body(report) |
| 104 | + self.assertIn("LLM provider's token or credit quota is exhausted", body) |
| 105 | + self.assertNotIn(PRICING, body) |
| 106 | + |
| 107 | + def test_workflow_posts_the_tested_failure_file(self): |
| 108 | + action = (ROOT / "action.yml").read_text() |
| 109 | + steps = re.split(r"\n - name: ", action[action.index("\nruns:\n") :])[1:] |
| 110 | + build = next(step for step in steps if step.startswith("Build review failure\n")) |
| 111 | + post = next(step for step in steps if step.startswith("Post review failure\n")) |
| 112 | + self.assertIn("failure()", build) |
| 113 | + self.assertIn( |
| 114 | + "steps.review_baseline.outputs.failure_reason || steps.review_analyze.outputs.failure_reason", build |
| 115 | + ) |
| 116 | + self.assertIn("build-review-failure.sh", build) |
| 117 | + self.assertIn("marocchino/sticky-pull-request-comment@v2", post) |
| 118 | + self.assertIn("path: ${{ steps.review_failure.outputs.path }}", post) |
| 119 | + self.assertIn("header: ${{ steps.guard.outputs.comment_id }}", post) |
| 120 | + |
| 121 | + |
| 122 | +if __name__ == "__main__": |
| 123 | + unittest.main() |
0 commit comments