baiyue-logistics-chat/tests/test_maintenance_scripts.py

"""Maintenance script regression tests using temporary SQLite databases."""

import os
import sqlite3
import subprocess
import sys
import tempfile
import unittest
from contextlib import closing
from pathlib import Path
from unittest.mock import patch

from scripts import close_active_sessions, inspect_human_sessions


ROOT = Path(__file__).resolve().parents[1]


class MaintenanceScriptTests(unittest.TestCase):
    def setUp(self):
        self.directory = tempfile.TemporaryDirectory()
        self.addCleanup(self.directory.cleanup)
        self.sandbox = Path(self.directory.name)
        self.database = self.sandbox / "history with spaces.db"
        self.other_directory = self.sandbox / "elsewhere"
        self.other_directory.mkdir()
        with closing(sqlite3.connect(self.database)) as connection:
            connection.executescript(
                "CREATE TABLE human_sessions(id INTEGER PRIMARY KEY, status TEXT);"
                "CREATE TABLE human_messages(id INTEGER PRIMARY KEY, content TEXT);"
                "INSERT INTO human_sessions VALUES"
                "(1, 'waiting'), (2, 'active'), (3, 'closed'), (4, 'active');"
            )
            connection.executemany(
                "INSERT INTO human_messages VALUES (?, ?)",
                [(number, f"message {number}") for number in range(1, 9)],
            )
            connection.commit()

    def run_script(self, name, database):
        return subprocess.run(
            [sys.executable, str(ROOT / "scripts" / name), "--db", str(database)],
            cwd=self.other_directory,
            capture_output=True,
            text=True,
            encoding="utf-8",
            timeout=15,
        )

    def test_inspect_returns_recent_messages_and_open_sessions_without_changes(self):
        original = self.database.read_bytes()
        messages, sessions = inspect_human_sessions.inspect_sessions(self.database)
        self.assertEqual([row[0] for row in messages], [8, 7, 6, 5, 4])
        self.assertEqual(sessions, [(1, "waiting"), (2, "active"), (4, "active")])
        self.assertEqual(self.database.read_bytes(), original)

    def test_close_only_changes_active_sessions_and_can_be_repeated(self):
        self.assertEqual(close_active_sessions.close_active_sessions(self.database), 2)
        with closing(sqlite3.connect(self.database)) as connection:
            sessions = connection.execute(
                "SELECT id, status FROM human_sessions ORDER BY id"
            ).fetchall()
            messages = connection.execute("SELECT COUNT(*) FROM human_messages").fetchone()[0]
        self.assertEqual(sessions, [(1, "waiting"), (2, "closed"), (3, "closed"), (4, "closed")])
        self.assertEqual(messages, 8)
        self.assertEqual(close_active_sessions.close_active_sessions(self.database), 0)

    def test_missing_database_is_not_created(self):
        missing = self.sandbox / "missing.db"
        for operation in (
            inspect_human_sessions.inspect_sessions,
            close_active_sessions.close_active_sessions,
        ):
            with self.subTest(operation=operation.__name__):
                with self.assertRaises(sqlite3.OperationalError):
                    operation(missing)
                self.assertFalse(missing.exists())

    def test_default_database_is_independent_of_working_directory(self):
        original_directory = Path.cwd()
        try:
            os.chdir(self.other_directory)
            with (
                patch.object(inspect_human_sessions, "DEFAULT_DB", self.database),
                patch.object(close_active_sessions, "DEFAULT_DB", self.database),
            ):
                self.assertEqual(len(inspect_human_sessions.inspect_sessions()[0]), 5)
                self.assertEqual(close_active_sessions.close_active_sessions(), 2)
            self.assertFalse((self.other_directory / "history.db").exists())
        finally:
            os.chdir(original_directory)

    def test_cli_supports_other_working_directories_and_paths_with_spaces(self):
        inspection = self.run_script("inspect_human_sessions.py", self.database)
        self.assertEqual(inspection.returncode, 0, inspection.stderr)
        self.assertIn("message 8", inspection.stdout)
        closing_result = self.run_script("close_active_sessions.py", self.database)
        self.assertEqual(closing_result.returncode, 0, closing_result.stderr)
        self.assertIn("Closed 2 active session(s).", closing_result.stdout)
        self.assertFalse((self.other_directory / "history.db").exists())

    def test_cli_reports_missing_database_without_creating_it(self):
        missing = self.sandbox / "missing.db"
        for script in ("inspect_human_sessions.py", "close_active_sessions.py"):
            with self.subTest(script=script):
                result = self.run_script(script, missing)
                self.assertEqual(result.returncode, 1)
                self.assertIn("Database error:", result.stderr)
                self.assertFalse(missing.exists())

    def test_importing_scripts_has_no_database_side_effects(self):
        for script in ("inspect_human_sessions.py", "close_active_sessions.py"):
            with self.subTest(script=script):
                result = subprocess.run(
                    [sys.executable, "-c", "import runpy, sys; runpy.run_path(sys.argv[1])",
                     str(ROOT / "scripts" / script)],
                    cwd=self.other_directory,
                    capture_output=True,
                    text=True,
                    timeout=15,
                )
                self.assertEqual(result.returncode, 0, result.stderr)
                self.assertEqual(result.stdout, "")
        self.assertEqual(list(self.other_directory.iterdir()), [])


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