split-trainer/server/tests/test_main.py

"""Тесты точки входа: коды возврата при неверных настройках и healthcheck контейнера."""
import os
import socket
import subprocess
import sys
import tempfile
import threading
import unittest

from splitapi.config import from_env
from splitapi.http_api import make_server
from splitapi.mailer import MemoryMailer
from splitapi.ratelimit import RateLimiter
from splitapi.service import Service
from splitapi.store import Store

SERVER_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))


def run_module(env_overrides, *args):
    env = {k: v for k, v in os.environ.items() if not k.startswith(("PUBLIC_URL", "SMTP_", "MAIL_", "PORT", "DB_PATH", "TRUST_PROXY"))}
    env.update(env_overrides)
    return subprocess.run([sys.executable, "-m", "splitapi", *args], cwd=SERVER_DIR, env=env, capture_output=True, text=True, timeout=60)


def free_port():
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


class StartupTests(unittest.TestCase):
    def test_missing_public_url_exits_with_code_2_and_explains(self):
        result = run_module({})
        self.assertEqual(result.returncode, 2)
        self.assertIn("PUBLIC_URL", result.stderr)

    def test_public_site_without_smtp_refuses_to_start(self):
        result = run_module({"PUBLIC_URL": "https://trainer.example.com"})
        self.assertEqual(result.returncode, 2)
        self.assertIn("SMTP", result.stderr)

    def test_explicit_console_mode_passes_configuration_checks_and_serves(self):
        with tempfile.TemporaryDirectory() as tmp:
            port = free_port()
            env = {**{k: v for k, v in os.environ.items()}, "PUBLIC_URL": "https://trainer.example.com", "MAIL_BACKEND": "console",
                   "DB_PATH": os.path.join(tmp, "m.sqlite3"), "PORT": str(port)}
            proc = subprocess.Popen([sys.executable, "-m", "splitapi"], cwd=SERVER_DIR, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
            try:
                import time
                deadline, ok = time.monotonic() + 20, False
                while time.monotonic() < deadline and not ok:
                    ok = run_module({"PORT": str(port)}, "--healthcheck").returncode == 0
                    time.sleep(0.2)
                self.assertTrue(ok, "процесс не начал отвечать на healthcheck")
            finally:
                proc.terminate()
                try:
                    _, err = proc.communicate(timeout=10)
                except subprocess.TimeoutExpired:
                    proc.kill()
                    _, err = proc.communicate()
            self.assertIn("SMTP не настроен", err)          # предупреждение о режиме разработки видно в логе


class SmtpWarningTests(unittest.TestCase):
    def _start_and_collect_log(self, extra_env):
        with tempfile.TemporaryDirectory() as tmp:
            port = free_port()
            env = {**os.environ, "PUBLIC_URL": "https://trainer.example.com", "DB_PATH": os.path.join(tmp, "w.sqlite3"), "PORT": str(port), **extra_env}
            proc = subprocess.Popen([sys.executable, "-m", "splitapi"], cwd=SERVER_DIR, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
            try:
                import time
                deadline, ok = time.monotonic() + 20, False
                while time.monotonic() < deadline and not ok:
                    ok = run_module({"PORT": str(port)}, "--healthcheck").returncode == 0
                    time.sleep(0.2)
                self.assertTrue(ok)
            finally:
                proc.terminate()
                try:
                    _, err = proc.communicate(timeout=10)
                except subprocess.TimeoutExpired:
                    proc.kill()
                    _, err = proc.communicate()
        return err

    def test_unencrypted_smtp_to_a_remote_host_is_called_out(self):
        err = self._start_and_collect_log({"SMTP_HOST": "mail.example.com", "SMTP_SECURITY": "none"})
        self.assertIn("без шифрования", err)

    def test_encrypted_smtp_gives_no_such_warning(self):
        err = self._start_and_collect_log({"SMTP_HOST": "mail.example.com", "SMTP_SECURITY": "starttls"})
        self.assertNotIn("без шифрования", err)


class HealthcheckTests(unittest.TestCase):
    def test_fails_when_nothing_is_listening(self):
        self.assertEqual(run_module({"PORT": str(free_port())}, "--healthcheck").returncode, 1)

    def test_succeeds_against_a_running_server(self):
        with tempfile.TemporaryDirectory() as tmp:
            cfg = from_env({"PUBLIC_URL": "http://127.0.0.1:8080", "DB_PATH": os.path.join(tmp, "h.sqlite3")})
            server = make_server(cfg, Service(cfg, Store(cfg.db_path), MemoryMailer(), RateLimiter()), host="127.0.0.1", port=0)
            thread = threading.Thread(target=lambda: server.serve_forever(poll_interval=0.01), daemon=True)
            thread.start()
            try:
                self.assertEqual(run_module({"PORT": str(server.server_address[1])}, "--healthcheck").returncode, 0)
            finally:
                server.shutdown()
                server.server_close()


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