site/WESP_REL/tests/test_plymouth_theme.py

"""Тесты Plymouth: рендер кадров и API админки."""

from __future__ import annotations

import tempfile
import threading
import unittest
from pathlib import Path
from unittest.mock import patch

from app import create_app, db
from app.routes.auth import _ensure_default_superuser
from app.services import plymouth_frame_renderer as renderer
from app.services import plymouth_theme_service as plymouth_svc
from app.services.plymouth_theme_service import build_and_install_theme, get_plymouth_status
from app.services.setup_state import mark_setup_complete
from config import TestingConfig


class PlymouthTestConfig(TestingConfig):
    _TMP = tempfile.mkdtemp(prefix="wesp-plymouth-test-")
    BASE_DIR = _TMP
    DATA_DIR = f"{_TMP}/data"
    SQLALCHEMY_DATABASE_URI = f"sqlite:///{_TMP}/recipes_test.db"
    SQLALCHEMY_BINDS = {"reports": f"sqlite:///{_TMP}/reports_test.db"}
    AUTH_LOGIN = "admin"
    AUTH_PASSWORD = "admin-secret"
    SECRET_KEY = "unit-test-secret-key-not-default-32chars"


class PlymouthThemeTests(unittest.TestCase):
    def setUp(self) -> None:
        self.app = create_app(PlymouthTestConfig)
        self.client = self.app.test_client()
        self.ctx = self.app.app_context()
        self.ctx.push()
        db.create_all()
        _ensure_default_superuser()
        mark_setup_complete(self.app)
        self._login()

    def _login(self) -> None:
        r = self.client.post(
            "/api/auth/login",
            json={"login": "admin", "password": "admin-secret"},
        )
        self.assertEqual(r.status_code, 200)

    def tearDown(self) -> None:
        db.session.remove()
        db.drop_all()
        self.ctx.pop()

    def test_frame_count_matches_duration(self) -> None:
        self.assertGreater(renderer.frame_count(6000, 25), 100)
        self.assertEqual(renderer.animation_end_ms(6000), renderer.wave2_end_ms(6000) + 60)

    def test_status_endpoint_requires_superuser(self) -> None:
        anon = create_app(PlymouthTestConfig)
        with anon.app_context():
            db.create_all()
            r = anon.test_client().get("/api/admin/plymouth/status")
            self.assertIn(r.status_code, (401, 403, 302))

    def test_status_endpoint_ok(self) -> None:
        r = self.client.get("/api/admin/plymouth/status")
        self.assertEqual(r.status_code, 200)
        data = r.get_json()
        self.assertEqual(data.get("status"), "success")
        self.assertEqual(data.get("resolution"), "800x600")
        self.assertIn("job", data)

    @patch("app.services.plymouth_theme_service.build_and_install_theme")
    def test_install_starts_background_job(self, mock_build) -> None:
        r = self.client.post("/api/admin/plymouth/install")
        self.assertEqual(r.status_code, 200)
        self.assertTrue(r.get_json().get("ok"))
        mock_build.assert_called_once()

    @patch("app.services.plymouth_theme_service.build_and_install_theme")
    def test_install_busy_when_running(self, mock_build) -> None:
        from app.services import plymouth_theme_service as svc

        with svc._lock:
            svc._job_state["running"] = True
        try:
            r = self.client.post("/api/admin/plymouth/install")
            self.assertEqual(r.status_code, 409)
            mock_build.assert_not_called()
        finally:
            with svc._lock:
                svc._job_state["running"] = False

    def test_get_status_in_app(self) -> None:
        with self.app.app_context():
            st = get_plymouth_status(self.app)
        self.assertIn("plymouth_installed", st)
        self.assertIn("build_dir", st)

    @patch("app.services.plymouth_theme_service._run_install_subprocess")
    @patch("app.services.plymouth_theme_service.render_all_frames")
    @patch("app.services.plymouth_theme_service.pillow_available", return_value=True)
    def test_build_thread_has_app_context(
        self, _pillow: object, mock_render: object, mock_install: object
    ) -> None:
        from flask import current_app
        root = Path(__file__).resolve().parents[1]
        logo = root / "static" / "logo2.png"
        if not logo.is_file():
            self.skipTest("logo2.png missing")

        mock_render.return_value = {"frames": 2, "output_dir": "animation"}
        mock_install.return_value = {"ok": True, "message": "Тема установлена"}

        errors: list[BaseException] = []

        def worker() -> None:
            try:
                with self.app.test_request_context():
                    build_and_install_theme(current_app)
            except BaseException as exc:
                errors.append(exc)

        thread = threading.Thread(target=worker, name="test-plymouth")
        thread.start()
        thread.join(timeout=10)
        self.assertFalse(thread.is_alive(), "plymouth build thread hung")
        self.assertEqual(errors, [], errors[0] if errors else None)
        with plymouth_svc._lock:
            phase = plymouth_svc._job_state.get("phase")
        self.assertIn(phase, ("done", "error"))
        if phase == "error":
            self.assertNotIn(
                "application context",
                str(plymouth_svc._job_state.get("error", "")).lower(),
            )


@unittest.skipUnless(renderer.pillow_available(), "Pillow required")
class PlymouthFrameRenderTests(unittest.TestCase):
    def test_render_single_frame(self) -> None:
        root = Path(__file__).resolve().parents[1]
        logo = root / "static" / "logo2.png"
        if not logo.is_file():
            self.skipTest("logo2.png missing")
        logo_img = renderer._load_logo_buffer(logo)
        cells = renderer._build_logo_cells(logo_img, renderer.INITIAL_CELL)
        layout = renderer._compute_logo_layout(800, 600, embedded=True)
        frame = renderer.render_frame(logo_img, cells, layout, 0)
        self.assertEqual(frame.size, (800, 600))

    def test_render_all_frames_small(self) -> None:
        root = Path(__file__).resolve().parents[1]
        logo = root / "static" / "logo2.png"
        if not logo.is_file():
            self.skipTest("logo2.png missing")
        with tempfile.TemporaryDirectory() as tmp:
            out = Path(tmp)
            summary = renderer.render_all_frames(
                logo,
                out,
                width=320,
                height=240,
                duration_ms=400,
                fps=10,
            )
            self.assertGreater(summary["frames"], 2)
            anim = out / "animation"
            self.assertTrue((anim / "0.png").is_file())


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