davidandClaude Opus 5.5
480e6ea18d
shellcheck / shellcheck (push) Failing after 16s
The first test checked that a report was no longer pending as soon as its log reached the fake Gitea, a moment before the receiver marks it filed, so it could fail; wait for the mark instead. The tests now close their servers and error responses. When Gitea refuses an issue or attachment, the receiver closes the reply and logs Gitea's reason, such as a token without the right permission. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
218 lines
9.8 KiB
Python
218 lines
9.8 KiB
Python
"""Tests for receiver.py. Run with: python3 -I -m unittest discover -s server/report-receiver
|
|
|
|
They start the receiver on a free port with a fake Gitea, and send it reports
|
|
with the real `report` client, the way a CoreDesktop machine would."""
|
|
|
|
import gzip
|
|
import http.server
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
import receiver # noqa: E402
|
|
|
|
REPO = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
class FakeGitea:
|
|
"""Records the issues and attachments it's given; fail=True answers 500."""
|
|
|
|
def __init__(self):
|
|
self.issues = []
|
|
self.attachments = []
|
|
self.fail = False
|
|
fake = self
|
|
|
|
class Handler(http.server.BaseHTTPRequestHandler):
|
|
def do_POST(self):
|
|
data = self.rfile.read(int(self.headers["Content-Length"]))
|
|
if fake.fail:
|
|
self.send_response(500)
|
|
self.end_headers()
|
|
return
|
|
if self.path.endswith("/issues"):
|
|
fake.issues.append({"path": self.path, "auth": self.headers["Authorization"], **json.loads(data)})
|
|
reply = {"number": len(fake.issues)}
|
|
else:
|
|
fields = receiver.parse_multipart(self.headers["Content-Type"], data)
|
|
fake.attachments.append({"path": self.path, "text": fields["attachment"].decode()})
|
|
reply = {"id": len(fake.attachments)}
|
|
body = json.dumps(reply).encode()
|
|
self.send_response(201)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
self.url = f"http://127.0.0.1:{self.server.server_port}"
|
|
threading.Thread(target=self.server.serve_forever, daemon=True).start()
|
|
|
|
def close(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
|
|
|
|
class ReceiverTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = Path(tempfile.mkdtemp())
|
|
self.gitea = FakeGitea()
|
|
self.store = receiver.Store(self.tmp / "state", receiver.Gitea(self.gitea.url, "Federated_Computer/coredesktop-reports", "secret"))
|
|
self.limiter = receiver.RateLimiter()
|
|
handler = receiver.make_handler(self.store, self.limiter)
|
|
handler.log_message = lambda *args: None
|
|
self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler)
|
|
self.url = f"http://127.0.0.1:{self.server.server_port}/report"
|
|
threading.Thread(target=self.server.serve_forever, daemon=True).start()
|
|
|
|
# A CoreDesktop home with one failed update's log
|
|
self.home = self.tmp / "home"
|
|
logs = self.home / ".local/state/coredesktop/logs"
|
|
logs.mkdir(parents=True)
|
|
self.log = logs / "update-2026-11-08-110307.log"
|
|
self.log.write_text(f"Updating CoreDesktop...\nCloning into '{self.home}/x'...\n"
|
|
"Done, with errors in: 71-siblings.sh\n")
|
|
|
|
def tearDown(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.gitea.close()
|
|
shutil.rmtree(self.tmp)
|
|
|
|
def send(self, *args):
|
|
"""Run the report client, answering yes, and return its output."""
|
|
tty = self.tmp / "tty"
|
|
tty.write_text("y\n")
|
|
env = {"HOME": str(self.home), "USER": "alice", "PATH": os.environ["PATH"],
|
|
"TMPDIR": str(self.tmp), "COREDESKTOP_REPORT_URL": self.url, "COREDESKTOP_TTY": str(tty)}
|
|
result = subprocess.run(["bash", str(REPO / "report"), *args], env=env,
|
|
capture_output=True, text=True, timeout=60)
|
|
return result.stdout + result.stderr
|
|
|
|
def wait_for(self, condition):
|
|
for _ in range(100):
|
|
if condition():
|
|
return
|
|
time.sleep(0.05)
|
|
self.fail("timed out")
|
|
|
|
def test_a_report_from_the_client_is_filed_with_its_log(self):
|
|
output = self.send("--offer", str(self.log), "71-siblings.sh", "46-neovim.sh")
|
|
self.assertIn("Your report's ID is CD-", output)
|
|
report_id = output.split("Your report's ID is ")[1].split(";")[0]
|
|
# Filed means attached and no longer pending, which happens just after
|
|
self.wait_for(lambda: not (self.tmp / "state" / report_id / "pending").exists())
|
|
|
|
issue = self.gitea.issues[0]
|
|
self.assertEqual(issue["path"], "/api/v1/repos/Federated_Computer/coredesktop-reports/issues")
|
|
self.assertEqual(issue["auth"], "token secret")
|
|
self.assertTrue(issue["title"].startswith("Update failed in 71-siblings.sh, 46-neovim.sh"))
|
|
self.assertIn(report_id, issue["body"])
|
|
self.assertIn("Cloning into '~/x'...", issue["body"])
|
|
self.assertNotIn(str(self.home), issue["body"])
|
|
attachment = self.gitea.attachments[0]
|
|
self.assertEqual(attachment["path"], "/api/v1/repos/Federated_Computer/coredesktop-reports/issues/1/assets")
|
|
self.assertIn("Done, with errors in: 71-siblings.sh", attachment["text"])
|
|
self.assertFalse((self.tmp / "state" / report_id / "pending").exists())
|
|
|
|
def test_a_report_gitea_refuses_is_kept_and_filed_later(self):
|
|
self.gitea.fail = True
|
|
output = self.send(str(self.log))
|
|
self.assertIn("Your report's ID is CD-", output)
|
|
report_id = output.split("Your report's ID is ")[1].split(";")[0]
|
|
self.wait_for(lambda: not self.store.lock.locked() and (self.tmp / "state" / report_id / "pending").exists())
|
|
time.sleep(0.2)
|
|
self.assertEqual(self.gitea.issues, [])
|
|
|
|
self.gitea.fail = False
|
|
self.store.retry_and_prune()
|
|
self.assertEqual(len(self.gitea.issues), 1)
|
|
self.assertEqual(len(self.gitea.attachments), 1)
|
|
self.assertFalse((self.tmp / "state" / report_id / "pending").exists())
|
|
|
|
def test_an_attachment_that_failed_is_retried_without_a_second_issue(self):
|
|
report_id = self.store.save({"kind": "update", "version": "", "failed": "x", "kernel": "", "siblings": ""}, "log\n")
|
|
original = self.store.gitea.attach
|
|
self.store.gitea.attach = lambda *args: (_ for _ in ()).throw(OSError("Gitea is down"))
|
|
self.assertFalse(self.store.deliver(report_id))
|
|
self.store.gitea.attach = original
|
|
self.assertTrue(self.store.deliver(report_id))
|
|
self.assertEqual(len(self.gitea.issues), 1)
|
|
self.assertEqual(len(self.gitea.attachments), 1)
|
|
|
|
def post(self, body, content_type="multipart/form-data; boundary=x"):
|
|
request = urllib.request.Request(self.url, data=body, method="POST", headers={"Content-Type": content_type})
|
|
try:
|
|
with urllib.request.urlopen(request) as response:
|
|
return response.status, response.read().decode()
|
|
except urllib.error.HTTPError as error:
|
|
with error:
|
|
return error.code, error.read().decode()
|
|
|
|
def test_things_that_arent_reports_are_turned_away(self):
|
|
self.assertEqual(self.post(b"hello", "text/plain")[0], 400)
|
|
no_log = b'--x\r\nContent-Disposition: form-data; name="kind"\r\n\r\nupdate\r\n--x--\r\n'
|
|
self.assertEqual(self.post(no_log)[0], 400)
|
|
not_gzip = b'--x\r\nContent-Disposition: form-data; name="log"; filename="log.gz"\r\n\r\nplain\r\n--x--\r\n'
|
|
self.assertEqual(self.post(not_gzip), (400, "The log in this report couldn't be read.\n"))
|
|
self.assertEqual(self.gitea.issues, [])
|
|
|
|
def test_a_gzip_bomb_is_refused(self):
|
|
bomb = gzip.compress(b"\0" * (receiver.MAX_LOG_BYTES + 1))
|
|
body = (b'--x\r\nContent-Disposition: form-data; name="log"; filename="log.gz"\r\n'
|
|
b'Content-Type: application/gzip\r\n\r\n' + bomb + b'\r\n--x--\r\n')
|
|
self.assertEqual(self.post(body)[0], 413)
|
|
|
|
def test_too_many_reports_from_one_address_are_refused(self):
|
|
for _ in range(5):
|
|
self.assertIn("Your report's ID is", self.send(str(self.log)))
|
|
output = self.send(str(self.log))
|
|
self.assertIn("Couldn't send the report: Too many reports from this address", output)
|
|
|
|
def test_other_paths_are_not_found(self):
|
|
request = urllib.request.Request(self.url.replace("/report", "/other"), data=b"x", method="POST")
|
|
with self.assertRaises(urllib.error.HTTPError) as caught:
|
|
urllib.request.urlopen(request)
|
|
caught.exception.close()
|
|
self.assertEqual(caught.exception.code, 404)
|
|
|
|
|
|
class PartsTest(unittest.TestCase):
|
|
def test_the_rate_limiter_counts_per_address_and_forgets_old_reports(self):
|
|
limiter = receiver.RateLimiter(per_address=[(3600, 2)], daily=3)
|
|
limiter.check("a", now=0)
|
|
limiter.check("a", now=1)
|
|
with self.assertRaises(receiver.Rejected):
|
|
limiter.check("a", now=2)
|
|
limiter.check("a", now=3700)
|
|
with self.assertRaises(receiver.Rejected): # the daily limit for everyone
|
|
limiter.check("b", now=3701)
|
|
|
|
def test_issue_text_cant_break_out_of_its_table_or_code_block(self):
|
|
details = {"kind": "update", "version": "v | 1", "failed": "a\nb", "kernel": "", "siblings": ""}
|
|
body = receiver.issue_body("CD-TEST23", details, "````\nlog\n", "now")
|
|
self.assertIn("| CoreDesktop | v \\| 1 |", body)
|
|
self.assertIn("| Failed | a<br>b |", body)
|
|
self.assertIn("`````\n````\nlog\n`````", body)
|
|
self.assertEqual(receiver.issue_title(details), "Update failed in a b (v | 1)")
|
|
|
|
def test_ids_are_easy_to_read_out(self):
|
|
for _ in range(100):
|
|
report_id = receiver.new_id()
|
|
self.assertRegex(report_id, r"^CD-[2-9A-HJ-NP-Z]{6}$")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|