coredesktop/server/report-receiver/test_receiver.py

"""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 contextlib
import gzip
import http.server
import io
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):
        # Keep what the receiver logs ("filed as issue #1", ...) out of the
        # test results. It's in self.stderr for a test that wants to check it.
        self.stderr = io.StringIO()
        self.enterContext(contextlib.redirect_stderr(self.stderr))
        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()