import io import json import unittest import urllib.error from contextlib import redirect_stderr, redirect_stdout from tests.test_gitea import FakeOpener, FakeResponse from werkrevisor import secretprobe VALUE = "probe-secret-0123456789abcdef" def answer(body): return lambda request: FakeResponse(json.dumps(body).encode()) def status(code): def raise_it(request): raise urllib.error.HTTPError(request.full_url, code, "error", {}, None) return raise_it def probe(key, *handlers): queue = list(handlers) opener = FakeOpener(lambda request: queue.pop(0)(request)) out = io.StringIO() with redirect_stdout(out): code = secretprobe.main([key], io.StringIO(VALUE + "\n"), opener) assert VALUE not in out.getvalue() return code, out.getvalue(), opener.requests USER = answer({"login": "ai-reviewer"}) REPO = answer({"permissions": {"admin": False, "push": True}}) class GiteaProbeTest(unittest.TestCase): def test_the_bot_with_write_access_and_write_scopes(self): code, out, requests = probe("GITEA_BOT_TOKEN", USER, REPO, status(404), status(422)) self.assertEqual((code, out), (0, "GITEA_BOT_TOKEN: accepted: ai-reviewer, with write access to mi/werkprobe " "and the scopes to write repository and issue\n")) self.assertEqual([(r.get_method(), r.full_url) for r in requests], [ ("GET", "https://git.javagil.de/api/v1/user"), ("GET", "https://git.javagil.de/api/v1/repos/mi/werkprobe"), ("POST", "https://git.javagil.de/api/v1/repos/mi/werkprobe/pulls/0/reviews"), ("POST", "https://git.javagil.de/api/v1/repos/mi/werkprobe/issues/0/comments")]) self.assertTrue(all(r.get_header("Authorization") == f"token {VALUE}" for r in requests)) def test_a_token_with_read_scopes_only(self): self.assertEqual(probe("GITEA_BOT_TOKEN", USER, REPO, status(403))[:2], (1, "GITEA_BOT_TOKEN: ai-reviewer, but the token may not write repository (HTTP 403)\n")) self.assertEqual(probe("GITEA_BOT_TOKEN", USER, REPO, status(404), status(403))[:2], (1, "GITEA_BOT_TOKEN: ai-reviewer, but the token may not write issue (HTTP 403)\n")) def test_a_write_that_succeeds_is_no_probe(self): self.assertEqual(probe("GITEA_BOT_TOKEN", USER, REPO, answer({"id": 1}))[:2], (2, "GITEA_BOT_TOKEN: HTTP 200 on writing repository, the probe did not complete\n")) def test_a_token_of_another_account(self): code, out, _ = probe("GITEA_BOT_TOKEN", answer({"login": "mi"})) self.assertEqual((code, out), (1, "GITEA_BOT_TOKEN: the token belongs to 'mi', not to ai-reviewer\n")) def test_an_account_without_write_access(self): code, out, _ = probe("GITEA_BOT_TOKEN", USER, answer({"permissions": {"push": False}})) self.assertEqual((code, out), (1, "GITEA_BOT_TOKEN: ai-reviewer, but without write access to mi/werkprobe\n")) def test_a_revoked_token(self): self.assertEqual(probe("GITEA_BOT_TOKEN", status(401))[:2], (1, "GITEA_BOT_TOKEN: refused with HTTP 401\n")) def test_an_outage_is_no_refusal(self): self.assertEqual(probe("GITEA_BOT_TOKEN", status(502))[:2], (2, "GITEA_BOT_TOKEN: HTTP 502, the probe did not complete\n")) self.assertEqual(probe("GITEA_BOT_TOKEN", USER, status(503))[:2], (2, "GITEA_BOT_TOKEN: HTTP 503, the probe did not complete\n")) def unreachable(request): raise urllib.error.URLError("Name or service not known") self.assertEqual(probe("GITEA_BOT_TOKEN", unreachable)[:2], (2, "GITEA_BOT_TOKEN: GET /api/v1/user failed: Name or service not known\n")) class OpenRouterProbeTest(unittest.TestCase): def test_the_key_info_without_its_label(self): code, out, requests = probe("OPENROUTER_API_KEY", answer({"data": { "label": "sk-or-v1-abc...xyz", "limit": 100, "limit_remaining": 63.214, "limit_reset": "monthly"}})) self.assertEqual((code, out), (0, "OPENROUTER_API_KEY: accepted: limit 100.00 USD, 63.21 USD left, " "limit reset monthly\n")) self.assertEqual(requests[0].get_header("Authorization"), f"Bearer {VALUE}") def test_a_key_without_limit(self): self.assertEqual(probe("OPENROUTER_API_KEY", answer({"data": {"limit": None, "limit_remaining": None}}))[1], "OPENROUTER_API_KEY: accepted: limit none, none left, limit reset never\n") def test_a_revoked_key(self): self.assertEqual(probe("OPENROUTER_API_KEY", status(401))[:2], (1, "OPENROUTER_API_KEY: refused with HTTP 401\n")) class TypeSafeProbeTest(unittest.TestCase): def test_the_model_list_names_jev(self): code, out, requests = probe("TYPESAFE_API_KEY", answer({"models": [{"name": "jev-latest"}, {"name": "jev-preview"}]})) self.assertEqual((code, out), (0, "TYPESAFE_API_KEY: accepted: the account has jev-latest\n")) self.assertEqual((requests[0].full_url, requests[0].get_method()), ("https://api.typesafe.ai/v1/models", "GET")) self.assertEqual(requests[0].get_header("Authorization"), f"Bearer {VALUE}") def test_an_account_without_jev(self): self.assertEqual(probe("TYPESAFE_API_KEY", answer({"models": []}))[:2], (1, "TYPESAFE_API_KEY: accepted, but the account lacks jev-latest\n")) def test_a_revoked_key(self): self.assertEqual(probe("TYPESAFE_API_KEY", status(401))[:2], (1, "TYPESAFE_API_KEY: refused with HTTP 401\n")) class StatsProbeTest(unittest.TestCase): def test_a_source_token_gets_the_empty_record_refused(self): code, out, requests = probe("WERKREVISOR_STATS_TOKEN", status(400)) self.assertEqual((code, out), (0, "WERKREVISOR_STATS_TOKEN: accepted as source token; " "the empty record was refused, nothing stored\n")) self.assertEqual((requests[0].full_url, requests[0].data), ("https://werkrevisor.javagil.de/api/usage", b"{}")) def test_an_unknown_token(self): self.assertEqual(probe("WERKREVISOR_STATS_TOKEN", status(401))[:2], (1, "WERKREVISOR_STATS_TOKEN: refused: no source token of the service\n")) def test_a_stored_record_would_be_no_probe(self): self.assertEqual(probe("WERKREVISOR_STATS_TOKEN", answer({"stored": True}))[0], 2) class MalformedValueTest(unittest.TestCase): def test_a_value_with_a_line_break_never_reaches_a_header_or_the_output(self): # The real opener: http.client rejects the header before it connects, with the value in its message. for value in ("first-line-0123456789\nsecond-line", "with space 0123456789", "x1y2z3"): out, err = io.StringIO(), io.StringIO() with redirect_stdout(out), redirect_stderr(err): code = secretprobe.main(["GITEA_BOT_TOKEN"], io.StringIO(value + "\n")) self.assertEqual((code, out.getvalue()), (2, ""), value) self.assertEqual(err.getvalue(), "GITEA_BOT_TOKEN: the value is too short, too long or has characters " "other than A-Z a-z 0-9 . _ ~ + / = -\n") for part in value.split(): self.assertNotIn(part, err.getvalue()) class UsageTest(unittest.TestCase): def test_unknown_key_or_empty_value(self): with redirect_stderr(io.StringIO()) as err: self.assertEqual(secretprobe.main(["GITEA_CLAUDE_CODE_TOKEN"], io.StringIO(VALUE)), 2) self.assertEqual(secretprobe.main(["GITEA_BOT_TOKEN"], io.StringIO("\n")), 2) self.assertNotIn(VALUE, err.getvalue()) if __name__ == "__main__": unittest.main()