test_server_backend.py

import subprocess
import tempfile
import unittest
from pathlib import Path
from unittest import mock

import server


UNIT = """[Service]
Environment=STEWARD_ENGINE=claude
Environment=STEWARD_ENGINE_BIN=/opt/bin/claude
Environment=STEWARD_MODEL=claude-opus-5
Environment=PATH=/opt/bin:/usr/bin
ExecStart=/repo/tick.sh
"""


class BackendConfigTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.unit_dir = Path(self.temp.name)
        self.service = self.unit_dir / "repo-steward.service"
        self.service.write_text(UNIT, encoding="utf-8")
        self.unit_patch = mock.patch.object(server, "UNIT_DIR", self.unit_dir)
        self.unit_patch.start()

    def tearDown(self):
        self.unit_patch.stop()
        self.temp.cleanup()

    @mock.patch.object(server.shutil, "which")
    def test_current_backend_reports_unit_and_available_options(self, which):
        which.side_effect = lambda value: f"/opt/bin/{value}" if value in {"claude", "codex"} else None

        result = server.current_backend()

        self.assertEqual(result["value"], "claude")
        self.assertEqual(result["label"], "Claude Code")
        self.assertEqual(result["model"], "claude-opus-5")
        availability = {item["value"]: item["available"] for item in result["options"]}
        self.assertEqual(availability, {
            "claude": True, "codex": True, "gemini": False, "opencode": False,
        })

    @mock.patch.object(server.subprocess, "run")
    @mock.patch.object(server.shutil, "which")
    def test_switch_updates_binary_and_clears_provider_specific_model(self, which, run):
        which.side_effect = lambda value: f"/new/bin/{value}"
        run.return_value = subprocess.CompletedProcess([], 0, "", "")

        ok, result = server.set_backend("codex")

        self.assertTrue(ok)
        self.assertEqual(result["value"], "codex")
        text = self.service.read_text(encoding="utf-8")
        self.assertIn("Environment=STEWARD_ENGINE=codex\n", text)
        self.assertIn("Environment=STEWARD_ENGINE_BIN=/new/bin/codex\n", text)
        self.assertNotIn("STEWARD_MODEL", text)
        run.assert_called_once_with(
            ["systemctl", "--user", "daemon-reload"], capture_output=True, text=True)

    @mock.patch.object(server.subprocess, "run")
    @mock.patch.object(server.shutil, "which", return_value="/new/bin/codex")
    def test_failed_reload_restores_the_original_unit(self, _which, run):
        run.side_effect = [
            subprocess.CompletedProcess([], 1, "", "reload failed"),
            subprocess.CompletedProcess([], 0, "", ""),
        ]

        ok, error = server.set_backend("codex")

        self.assertFalse(ok)
        self.assertEqual(error, "reload failed")
        self.assertEqual(self.service.read_text(encoding="utf-8"), UNIT)

    @mock.patch.object(server.shutil, "which", return_value=None)
    def test_uninstalled_backend_is_refused_without_writing(self, _which):
        ok, error = server.set_backend("gemini")

        self.assertFalse(ok)
        self.assertIn("not installed", error)
        self.assertEqual(self.service.read_text(encoding="utf-8"), UNIT)

    @mock.patch.object(server.subprocess, "run")
    def test_preconfigured_custom_backend_can_be_selected(self, run):
        self.service.write_text(
            UNIT + 'Environment=STEWARD_ENGINE_CMD=my-agent --prompt "$PROMPT"\n',
            encoding="utf-8")
        run.return_value = subprocess.CompletedProcess([], 0, "", "")

        ok, result = server.set_backend("custom")

        self.assertTrue(ok)
        self.assertEqual(result["value"], "custom")
        self.assertEqual(result["label"], "Custom command")
        text = self.service.read_text(encoding="utf-8")
        self.assertNotIn("STEWARD_ENGINE_BIN", text)
        self.assertIn("STEWARD_ENGINE_CMD", text)


class MissingConfigTests(unittest.TestCase):
    """A server started before config.yaml exists must not serve empty pages."""

    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        (self.root / "config.example.yaml").write_text(
            "mode: draft\nrepos:\n  - name: owner/repo\n", encoding="utf-8")
        (self.root / "dashboard-first-run.html").write_text(
            (Path(server.__file__).parent / "dashboard-first-run.html").read_text())
        patcher = mock.patch.object(server, "ROOT", self.root)
        patcher.start()
        self.addCleanup(patcher.stop)

    def test_missing_config_is_created_from_the_example(self):
        self.assertTrue(server.ensure_config())
        self.assertEqual((self.root / "config.yaml").read_text(),
                         (self.root / "config.example.yaml").read_text())
        self.assertEqual([r["name"] for r in server.repos_config()], ["owner/repo"])
        self.assertIn("owner/repo", server.first_run_page())

    def test_existing_config_is_left_alone(self):
        (self.root / "config.yaml").write_text("mode: live\n", encoding="utf-8")
        self.assertFalse(server.ensure_config())
        self.assertEqual((self.root / "config.yaml").read_text(), "mode: live\n")


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