142 lines
5.4 KiB
Python
142 lines
5.4 KiB
Python
import contextlib
|
|
import http.server
|
|
import io
|
|
import json
|
|
import threading
|
|
import unittest
|
|
|
|
from tests.fakes import FakeModel, config, pr_ref
|
|
from tests.test_flow import docs_gitea
|
|
from werkrevisor.cli import execute
|
|
from werkrevisor.redact import Redactor
|
|
from werkrevisor.statsclient import StatsClient, StatsError
|
|
from tests.fakes import deps as make_deps
|
|
|
|
|
|
class FakeStats:
|
|
def __init__(self, error=None):
|
|
self.records = []
|
|
self.error = error
|
|
|
|
def report(self, record):
|
|
if self.error:
|
|
raise self.error
|
|
self.records.append(record)
|
|
|
|
|
|
def review(stats, base_repo={"private": False}, publish=True):
|
|
gitea = docs_gitea()
|
|
gitea.base_repo = base_repo
|
|
d = make_deps(gitea, FakeModel(prompt_files=["doc/PR/1.md"]))
|
|
d.stats = stats
|
|
err = io.StringIO()
|
|
with contextlib.redirect_stderr(err):
|
|
outcome = execute(config(gate={"yellow_gate": "block"}), pr_ref(), d, Redactor(("s3cret-token",)), publish=publish)
|
|
return outcome, err.getvalue()
|
|
|
|
|
|
class StatsClientTest(unittest.TestCase):
|
|
def test_run_reports_its_usage_record(self):
|
|
stats = FakeStats()
|
|
outcome, _ = review(stats)
|
|
[record] = stats.records
|
|
self.assertEqual((record["repo"], record["pr"], record["verdict"]), ("team/app", 7, outcome.result.verdict.name))
|
|
self.assertEqual(record["pr_url"], "https://git.example.org/team/app/pulls/7")
|
|
self.assertIs(record["repo_private"], False)
|
|
|
|
def test_a_dry_run_reports_its_usage_record_as_local(self):
|
|
for publish, local in ((True, False), (False, True)):
|
|
with self.subTest(publish=publish):
|
|
stats = FakeStats()
|
|
review(stats, publish=publish)
|
|
self.assertIs(stats.records[0]["local"], local)
|
|
|
|
def test_a_private_repository_is_reported_as_such(self):
|
|
stats = FakeStats()
|
|
review(stats, {"private": True})
|
|
self.assertIs(stats.records[0]["repo_private"], True)
|
|
|
|
def test_a_missing_or_odd_private_flag_is_reported_as_unknown(self):
|
|
for base_repo in (None, {}, {"private": None}, {"private": "false"}, {"private": 0}):
|
|
with self.subTest(base_repo=base_repo):
|
|
stats = FakeStats()
|
|
review(stats, base_repo)
|
|
self.assertIsNone(stats.records[0]["repo_private"])
|
|
|
|
def test_failed_report_never_changes_the_review(self):
|
|
expected, _ = review(None)
|
|
for error in (StatsError("POST /api/usage failed: connection refused"),
|
|
StatsError("POST /api/usage failed with HTTP 500"),
|
|
StatsError("POST /api/usage failed with HTTP 401 for s3cret-token")):
|
|
with self.subTest(error=str(error)):
|
|
outcome, log = review(FakeStats(error))
|
|
self.assertEqual((outcome.result.verdict, outcome.result.exit_code),
|
|
(expected.result.verdict, expected.result.exit_code))
|
|
self.assertIn("werkrevisor: usage not reported to the statistics service: POST /api/usage failed", log)
|
|
self.assertNotIn("s3cret-token", log)
|
|
|
|
|
|
class SlowStats:
|
|
def report(self, record):
|
|
import time
|
|
time.sleep(30)
|
|
|
|
|
|
class DeadlineTest(unittest.TestCase):
|
|
def test_a_hanging_report_ends_after_the_deadline(self):
|
|
from unittest import mock
|
|
from werkrevisor import cli
|
|
with mock.patch.object(cli, "STATS_DEADLINE_SECONDS", 0.2):
|
|
outcome, log = review(SlowStats())
|
|
self.assertIn("usage not reported to the statistics service: no answer within 0.2 s", log)
|
|
|
|
|
|
class Handler(http.server.BaseHTTPRequestHandler):
|
|
status = 201
|
|
received = []
|
|
|
|
def do_POST(self):
|
|
body = self.rfile.read(int(self.headers["Content-Length"]))
|
|
Handler.received.append((self.path, self.headers["Authorization"], json.loads(body)))
|
|
self.send_response(Handler.status)
|
|
self.end_headers()
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
|
|
class HttpClientTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.server = http.server.HTTPServer(("127.0.0.1", 0), Handler)
|
|
threading.Thread(target=cls.server.serve_forever, daemon=True).start()
|
|
cls.url = f"http://127.0.0.1:{cls.server.server_port}"
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.server.shutdown()
|
|
cls.server.server_close()
|
|
|
|
def test_posts_the_record_with_the_source_token(self):
|
|
Handler.status, Handler.received = 201, []
|
|
StatsClient(self.url, "source-token").report({"repo": "team/app", "pr": 7})
|
|
self.assertEqual(Handler.received, [("/api/usage", "Bearer source-token", {"repo": "team/app", "pr": 7})])
|
|
|
|
def test_refused_or_failed_report_raises(self):
|
|
for status in (401, 500):
|
|
with self.subTest(status=status):
|
|
Handler.status = status
|
|
with self.assertRaisesRegex(StatsError, f"HTTP {status}"):
|
|
StatsClient(self.url, "source-token").report({})
|
|
with self.assertRaisesRegex(StatsError, "POST /api/usage failed"):
|
|
StatsClient("http://127.0.0.1:9", "source-token", timeout=2).report({})
|
|
|
|
def test_token_travels_only_over_https_or_localhost(self):
|
|
with self.assertRaisesRegex(StatsError, "must use https"):
|
|
StatsClient("http://stats.example.org", "source-token")
|
|
StatsClient("https://stats.example.org", "source-token")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|