broker/tests/test_mounts.py

import contextlib
import io
import json
import os
import sys
import tempfile
import unittest
import urllib.request
from http.server import ThreadingHTTPServer
from unittest import mock

sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir)))

import coreci_broker


class MountsTests(unittest.TestCase):
    def setUp(self):
        coreci_broker._reset_server_guard()
        tmp = tempfile.TemporaryDirectory()
        self.tmp = tmp.name
        self.env = {
            "BROKER_ENV_PATH": os.path.join(self.tmp, ".env"),
            "BROKER_DB_PATH": os.path.join(self.tmp, "broker.db"),
            "BROKER_AUDIT_LOG_PATH": os.path.join(self.tmp, "broker.audit.log"),
            "BROKER_BIND": "127.0.0.1:0",
        }
        self.addCleanup(tmp.cleanup)
        self.addCleanup(coreci_broker._reset_server_guard)

    def _patched_environ(self):
        return mock.patch.dict(os.environ, self.env, clear=True)

    def _read_env_values(self):
        values = {}
        with open(self.env["BROKER_ENV_PATH"], "r", encoding="utf-8") as handle:
            for raw in handle:
                line = raw.strip()
                if not line or line.startswith("#") or "=" not in line:
                    continue
                key, _, value = line.partition("=")
                values[key.strip()] = value.strip()
        return values

    def _healthz(self, port):
        url = "http://127.0.0.1:%d/healthz" % port
        with urllib.request.urlopen(url, timeout=5) as response:
            return response.status, json.loads(response.read().decode("utf-8"))

    def test_owui_docstring_frontmatter(self):
        doc = coreci_broker.__doc__ or ""
        self.assertTrue(doc.startswith("title: CoreCI Broker"))
        for line in (
            "author: coreci",
            "description: ",
            "version: 0.1.0",
            "requirements: []",
            "licence: MIT",
        ):
            self.assertIn(line, doc)

    def test_source_never_references_dunder_file(self):
        with open(coreci_broker.__file__, "r", encoding="utf-8") as handle:
            source = handle.read()
        self.assertNotIn("__file__", source)

    def test_tools_double_instantiation_starts_exactly_one_server(self):
        serve_calls = []
        real_serve = coreci_broker.serve

        def counting_serve(host, port, broker):
            serve_calls.append((host, port))
            return real_serve(host, port, broker)

        with self._patched_environ():
            with mock.patch.object(coreci_broker, "serve", counting_serve):
                with contextlib.redirect_stdout(io.StringIO()):
                    first = coreci_broker.Tools()
                    port_after_first = coreci_broker._started_server.server_address[1]
                    second = coreci_broker.Tools()
        guard = coreci_broker._started_server
        self.assertEqual(1, len(serve_calls))
        self.assertIsInstance(guard, ThreadingHTTPServer)
        self.assertIs(coreci_broker._started_server, first._server)
        self.assertIs(first._server, second._server)
        self.assertEqual(port_after_first, guard.server_address[1])
        self.assertNotEqual(0, guard.server_address[1])
        self.assertIsInstance(first.valves, coreci_broker.Tools.Valves)
        self.assertIsInstance(second.valves, coreci_broker.Tools.Valves)
        status, payload = self._healthz(port_after_first)
        self.assertEqual(200, status)
        self.assertEqual({"ok": True}, payload)

    def test_tools_status_returns_non_empty_string(self):
        with self._patched_environ():
            with contextlib.redirect_stdout(io.StringIO()):
                tools = coreci_broker.Tools()
                status = tools.status()
        port = tools._server.server_address[1]
        self.assertIsInstance(status, str)
        self.assertTrue(status.strip())
        self.assertIn(str(port), status)

    def test_first_boot_print_discipline_through_tools_boot(self):
        with self._patched_environ():
            first = io.StringIO()
            with contextlib.redirect_stdout(first):
                coreci_broker.Tools()
            second = io.StringIO()
            with contextlib.redirect_stdout(second):
                coreci_broker.Tools()
        stored = self._read_env_values()
        admin_pass = stored["BROKER_ADMIN_PASS"]
        pepper = stored["BROKER_INSTANCE_PEPPER"]
        self.assertEqual(1, first.getvalue().count(admin_pass))
        self.assertNotIn(pepper, first.getvalue())
        self.assertEqual("", second.getvalue())

    def test_run_standalone_boots_serves_and_shuts_down_cleanly(self):
        captured = {}
        real_serve = coreci_broker.serve

        def capturing_serve(host, port, broker):
            server = real_serve(host, port, broker)
            captured["server"] = server
            return server

        def fake_sleep(seconds):
            server = captured["server"]
            status, payload = self._healthz(server.server_address[1])
            captured["healthz"] = (status, payload)
            raise KeyboardInterrupt

        stdout = io.StringIO()
        with mock.patch.object(coreci_broker, "serve", capturing_serve):
            with mock.patch.object(coreci_broker.time, "sleep", fake_sleep):
                with contextlib.redirect_stdout(stdout):
                    coreci_broker.run_standalone(dict(self.env))
        out = stdout.getvalue()
        self.assertEqual((200, {"ok": True}), captured["healthz"])
        self.assertEqual(1, out.count("BROKER_ADMIN_PASS="))
        self.assertIn("listening on http://127.0.0.1:", out)
        self.assertIn("shut down", out)
        self.assertEqual(-1, captured["server"].socket.fileno())


if __name__ == "__main__":
    unittest.main()