307 lines
14 KiB
Python
307 lines
14 KiB
Python
import io
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
from werkrevisor import draftreview, reviewreport
|
|
|
|
SETTINGS = {"base_url": "https://git.example.org", "repository": "team/app", "config": "config/werkrevisor.toml",
|
|
"read_token_variable": "EXAMPLE_READ_TOKEN"}
|
|
REPORT = """## AI review of `6d1842aa` (werkrevisor and Codex)
|
|
|
|
### werkrevisor: 🟠 ORANGE / FINDINGS
|
|
|
|
### Findings
|
|
1. **[medium] Missing check** at `src/main/java/net/example/App.java:12-14`
|
|
|
|
ORANGE; Jev: real 0.86
|
|
|
|
### Codex
|
|
"""
|
|
|
|
|
|
def write(path, text):
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
f.write(text)
|
|
|
|
|
|
class LinkTest(unittest.TestCase):
|
|
def test_a_link_to_the_tool_finds_the_package_of_the_checkout(self):
|
|
"""Scenario#132.01"""
|
|
root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
link = os.path.join(tmp, "hs-review")
|
|
os.symlink(os.path.join(root, "tools", "draft-review"), link)
|
|
proc = subprocess.run([sys.executable, link, "7"], capture_output=True, text=True, cwd=tmp,
|
|
env={"PATH": os.environ.get("PATH", ""),
|
|
draftreview.CONFIG_VARIABLE: os.path.join(tmp, "none.toml")})
|
|
|
|
self.assertNotIn("ModuleNotFoundError", proc.stderr)
|
|
self.assertIn("draft-review: no local config", proc.stderr)
|
|
|
|
|
|
class SettingsTest(unittest.TestCase):
|
|
def setUp(self):
|
|
tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(tmp.cleanup)
|
|
self.path = os.path.join(tmp.name, "draft-review.toml")
|
|
|
|
def test_the_local_config_names_server_repository_config_and_tokens(self):
|
|
"""Scenario#132.01"""
|
|
write(self.path, 'base_url = "https://git.example.org/"\nrepository = "team/app"\n'
|
|
'config = "config/werkrevisor.toml"\nread_token_variable = "EXAMPLE_READ_TOKEN"\n')
|
|
|
|
self.assertEqual(draftreview.load_settings(self.path)["base_url"], "https://git.example.org")
|
|
|
|
def test_a_missing_or_incomplete_local_config_is_named(self):
|
|
with self.assertRaisesRegex(draftreview.DraftReviewError, "no local config"):
|
|
draftreview.load_settings(self.path)
|
|
write(self.path, 'base_url = "http://git.example.org"\nrepository = "app"\nextra = 1\n')
|
|
with self.assertRaisesRegex(draftreview.DraftReviewError, "unknown keys: extra; missing: config, read_token_variable"):
|
|
draftreview.load_settings(self.path)
|
|
|
|
def test_secrets_are_read_like_a_shell_would(self):
|
|
write(self.path, '# tokens\nexport A="one"\nB=\'two\'\nC=three\n\nnot a line\n')
|
|
|
|
self.assertEqual(draftreview.read_secrets(self.path), {"A": "one", "B": "two", "C": "three"})
|
|
|
|
|
|
class ResolveTest(unittest.TestCase):
|
|
def test_a_number_means_a_pr_of_the_default_repository(self):
|
|
"""Scenario#132.01"""
|
|
self.assertEqual(draftreview.resolve("343", SETTINGS),
|
|
("https://git.example.org/team/app/pulls/343", "team", "app", 343))
|
|
|
|
def test_a_url_may_name_another_repository_of_the_configured_server(self):
|
|
url = "https://git.example.org/team/other/pulls/7"
|
|
self.assertEqual(draftreview.resolve(url, SETTINGS), (url, "team", "other", 7))
|
|
|
|
def test_a_pr_on_another_server_is_refused(self):
|
|
"""Scenario#132.02"""
|
|
for pr in ("https://git.example.org.evil.test/team/app/pulls/7", "https://other.example/team/app/pulls/7",
|
|
"http://git.example.org/team/app/pulls/7"):
|
|
with self.subTest(pr), self.assertRaises(draftreview.DraftReviewError):
|
|
draftreview.resolve(pr, SETTINGS)
|
|
|
|
|
|
class SummaryTest(unittest.TestCase):
|
|
def test_the_summary_names_the_verdict_and_its_strongest_finding(self):
|
|
self.assertEqual(draftreview.summary(REPORT), "verdict: ORANGE — [medium] Missing check (App.java:12-14)")
|
|
|
|
def test_a_finding_without_a_judgment_of_jev_is_named_too(self):
|
|
self.assertEqual(draftreview.summary(REPORT.replace("ORANGE; Jev: real 0.86", "ORANGE")),
|
|
"verdict: ORANGE — [medium] Missing check (App.java:12-14)")
|
|
|
|
def test_a_failed_review_names_its_cause(self):
|
|
text = "### werkrevisor\n\nThe review did not complete: werkrevisor: cannot read the PR\n"
|
|
self.assertEqual(draftreview.summary(text),
|
|
"verdict: none — The review did not complete: werkrevisor: cannot read the PR")
|
|
|
|
|
|
class HtmlTest(unittest.TestCase):
|
|
def test_the_report_gets_an_html_page_with_clickable_diagrams(self):
|
|
"""Scenario#132.03"""
|
|
calls = []
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
md = os.path.join(tmp, "report.md")
|
|
def fake_pandoc(cmd, **kw):
|
|
calls.append(cmd)
|
|
write(cmd[cmd.index("--output") + 1], "<p>report</p>")
|
|
html = draftreview.write_html(md, "Draft review", fake_pandoc, lambda name: "/usr/bin/pandoc")
|
|
leftovers = os.listdir(tmp)
|
|
report = reviewreport.compose("6d1842aa", None, None, codex_error="x", diagrams=["pr-lens-flow.svg"])
|
|
|
|
self.assertEqual(html, os.path.join(tmp, "report.html"))
|
|
self.assertEqual(calls[0][:2], ["pandoc", md])
|
|
self.assertEqual(leftovers, ["report.html"])
|
|
self.assertIn("[](pr-lens-flow.svg)", report)
|
|
|
|
def test_the_page_runs_no_script_of_the_draft(self):
|
|
"""Scenario#132.04"""
|
|
calls, templates = [], []
|
|
|
|
def fake_pandoc(cmd, **kw):
|
|
calls.append(cmd)
|
|
with open(cmd[cmd.index("--template") + 1], encoding="utf-8") as f:
|
|
templates.append(f.read())
|
|
write(cmd[cmd.index("--output") + 1], '<a href="javascript:alert(1)">x</a><a href="pr-lens-a.svg">a</a>')
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
html = draftreview.write_html(os.path.join(tmp, "report.md"), "t", fake_pandoc,
|
|
lambda name: "/usr/bin/pandoc")
|
|
with open(html, encoding="utf-8") as f:
|
|
page = f.read()
|
|
|
|
self.assertEqual(page, '<a href="#blocked-javascript:alert(1)">x</a><a href="pr-lens-a.svg">a</a>')
|
|
cmd, template = calls[0], templates[0]
|
|
self.assertEqual(cmd[cmd.index("--from") + 1], "markdown_strict-raw_html+pipe_tables")
|
|
head = template.split("</head>")[0]
|
|
self.assertLess(head.index("Content-Security-Policy"), head.index("<title>"))
|
|
self.assertIn("default-src 'none'", head)
|
|
self.assertNotIn("script-src", head)
|
|
self.assertEqual(sorted(set(re.findall(r"\$(\w+)\$", template))), ["body", "pagetitle"])
|
|
|
|
@unittest.skipUnless(shutil.which("pandoc"), "pandoc is not installed")
|
|
def test_a_draft_cannot_inject_a_script_with_pandoc(self):
|
|
"""Scenario#132.04"""
|
|
finding = ("1. **bad** <script>alert(0)</script>\n\n"
|
|
"---\npagetitle: '`</title><script>alert(1)</script><title>`{=html}'\n"
|
|
"css: '`x\" /><script>alert(3)</script><link x=\"`{=html}'\n---\n\n"
|
|
"`<script>alert(4)</script>`{=html} [x](javascript:alert(2))\n")
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
md = os.path.join(tmp, "report.md")
|
|
write(md, reviewreport.compose("6d1842aa", ("YELLOW", finding, ""), None, codex_error="x"))
|
|
with open(draftreview.write_html(md, "t"), encoding="utf-8") as f:
|
|
page = f.read()
|
|
|
|
self.assertNotIn("<script", page)
|
|
self.assertNotIn('href="javascript:', page)
|
|
|
|
def test_without_pandoc_there_is_no_html(self):
|
|
self.assertIsNone(draftreview.write_html("/tmp/x/report.md", "t", None, lambda name: None))
|
|
|
|
|
|
class MainTest(unittest.TestCase):
|
|
def setUp(self):
|
|
tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(tmp.cleanup)
|
|
self.config = os.path.join(tmp.name, "draft-review.toml")
|
|
secrets = os.path.join(tmp.name, "secrets.env")
|
|
write(secrets, "EXAMPLE_READ_TOKEN=read-secret\nEXAMPLE_WRITE_TOKEN=write-secret\n")
|
|
write(self.config, f'base_url = "https://git.example.org"\nrepository = "team/app"\n'
|
|
f'config = "config/werkrevisor.toml"\nsecrets = "{secrets}"\n'
|
|
f'read_token_variable = "EXAMPLE_READ_TOKEN"\nwrite_token_variable = "EXAMPLE_WRITE_TOKEN"\n')
|
|
self.report = os.path.join(tmp.name, "report.md")
|
|
write(self.report, REPORT)
|
|
|
|
def run_main(self, *argv, attached=None, html=None, opener=None):
|
|
calls, self.opened = [], []
|
|
|
|
def fake_report(args, env):
|
|
calls.append((args, env["GITEA_TOKEN"]))
|
|
if attached:
|
|
print(f"attached: {attached}")
|
|
print(f"review-report: {self.report}")
|
|
return 0
|
|
out, err = io.StringIO(), io.StringIO()
|
|
with mock.patch.object(draftreview.reviewreport, "main", fake_report), \
|
|
mock.patch.object(draftreview, "GiteaClient", side_effect=OSError("offline")), \
|
|
mock.patch.object(draftreview, "write_html", return_value=html):
|
|
code = draftreview.main(list(argv), {draftreview.CONFIG_VARIABLE: self.config,
|
|
"WERKREVISOR_CONFIG_TOML": "[model]"}, out, err,
|
|
opener or (lambda target, settings: self.opened.append((target, settings.get("browser")))))
|
|
return code, calls, out.getvalue(), err.getvalue()
|
|
|
|
def test_a_review_reads_with_the_read_token_and_prints_the_verdict(self):
|
|
"""Scenario#132.01"""
|
|
code, calls, out, _ = self.run_main("343")
|
|
|
|
(args, token), = calls
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(args[:2], ["https://git.example.org/team/app/pulls/343", "config/werkrevisor.toml"])
|
|
self.assertNotIn("--attach", args)
|
|
self.assertEqual(token, "read-secret")
|
|
self.assertIn("verdict: ORANGE", out)
|
|
self.assertTrue(out.rstrip().endswith(self.report))
|
|
self.assertNotIn("read-secret", out)
|
|
|
|
def test_attach_writes_with_the_write_token(self):
|
|
"""Scenario#130.01"""
|
|
code, calls, _, _ = self.run_main("343", "--attach")
|
|
|
|
(args, token), = calls
|
|
self.assertEqual((code, token), (0, "write-secret"))
|
|
self.assertIn("--attach", args)
|
|
|
|
def test_open_shows_the_report_in_the_browser_of_the_local_config(self):
|
|
"""Scenario#134.01"""
|
|
with open(self.config, "a", encoding="utf-8") as f:
|
|
f.write('browser = "brave-browser"\n')
|
|
|
|
code, _, out, _ = self.run_main("343", "--open", html="/tmp/x/report.html")
|
|
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(self.opened, [("/tmp/x/report.html", "brave-browser")])
|
|
self.assertIn("/tmp/x/report.html", out)
|
|
|
|
def test_without_open_no_browser_starts(self):
|
|
self.run_main("343", html="/tmp/x/report.html")
|
|
|
|
self.assertEqual(self.opened, [])
|
|
|
|
def test_open_with_attach_shows_the_posted_comment(self):
|
|
"""Scenario#134.03"""
|
|
url = "https://git.example.org/team/app/pulls/343#issuecomment-9"
|
|
|
|
self.run_main("343", "--attach", "--open", attached=url)
|
|
|
|
self.assertEqual(self.opened, [(url, None)])
|
|
|
|
def test_a_browser_that_does_not_start_leaves_the_posted_comment_done(self):
|
|
"""Scenario#134.03"""
|
|
def missing(target, settings):
|
|
raise FileNotFoundError(2, "No such file or directory", "xdg-open")
|
|
|
|
code, _, out, err = self.run_main("343", "--attach", "--open", attached="https://g/c#9", opener=missing)
|
|
|
|
self.assertEqual(code, 0)
|
|
self.assertIn("browser not started", err)
|
|
self.assertIn("open https://g/c#9 yourself", err)
|
|
self.assertTrue(out.rstrip().endswith(self.report))
|
|
|
|
def test_a_broken_browser_command_is_refused_when_the_config_is_read(self):
|
|
with open(self.config, "a", encoding="utf-8") as f:
|
|
f.write('browser = "brave-browser \'--profile"\n')
|
|
|
|
code, calls, _, err = self.run_main("343", "--open")
|
|
|
|
self.assertEqual((code, calls), (2, []))
|
|
self.assertIn("browser is no valid command", err)
|
|
|
|
def test_the_browser_command_defaults_to_xdg_open(self):
|
|
calls = []
|
|
command = draftreview.open_in_browser("/tmp/r.html", {}, lambda cmd, **kw: calls.append(kw))
|
|
self.assertEqual(command, ["xdg-open", "/tmp/r.html"])
|
|
self.assertTrue(calls[0]["start_new_session"])
|
|
self.assertEqual(draftreview.open_in_browser("/r.html", {"browser": "brave-browser --new-window"},
|
|
lambda cmd, **kw: None),
|
|
["brave-browser", "--new-window", "/r.html"])
|
|
|
|
def test_checked_passes_its_reason_to_attach(self):
|
|
"""Scenario#136.01"""
|
|
code, calls, _, _ = self.run_main("343", "--attach", "--checked", "-1 is a typo; finding 2 does not hold")
|
|
|
|
(args, token), = calls
|
|
self.assertEqual((code, token), (0, "write-secret"))
|
|
self.assertIn("--checked=-1 is a typo; finding 2 does not hold", args)
|
|
|
|
def test_checked_file_reads_a_multiline_reason_from_stdin(self):
|
|
"""Scenario#136.01"""
|
|
with mock.patch("sys.stdin", io.StringIO("line one\nline two\n")):
|
|
code, calls, _, _ = self.run_main("343", "--attach", "--checked-file", "-")
|
|
|
|
(args, _), = calls
|
|
self.assertEqual(code, 0)
|
|
self.assertIn("--checked=line one\nline two\n", args)
|
|
|
|
def test_checked_needs_attach(self):
|
|
code, calls, _, err = self.run_main("343", "--checked", "reason")
|
|
|
|
self.assertEqual((code, calls), (2, []))
|
|
self.assertIn("need --attach", err)
|
|
|
|
def test_a_pr_on_another_server_gets_no_token(self):
|
|
"""Scenario#132.02"""
|
|
code, calls, _, err = self.run_main("https://other.example/team/app/pulls/7")
|
|
|
|
self.assertEqual((code, calls), (2, []))
|
|
self.assertIn("expected a PR number or https://git.example.org/", err)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|