124 lines
4.7 KiB
Python
124 lines
4.7 KiB
Python
"""mDNS autostart, retry и shutdown."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from app import _run_auto_migrations, create_app
|
|
from app.services.mdns_service import (
|
|
ensure_mdns_running,
|
|
mdns_skip_reason_on_boot,
|
|
mdns_status,
|
|
register_mdns_shutdown,
|
|
reset_mdns_state_for_tests,
|
|
schedule_mdns_retry_if_needed,
|
|
should_start_mdns_on_boot,
|
|
start_mdns,
|
|
)
|
|
from config import TestingConfig
|
|
|
|
|
|
class MdnsServiceTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
reset_mdns_state_for_tests()
|
|
self.app = create_app(TestingConfig)
|
|
|
|
def tearDown(self) -> None:
|
|
reset_mdns_state_for_tests()
|
|
|
|
def test_skip_reason_reloader_parent(self) -> None:
|
|
self.app.config["TESTING"] = False
|
|
self.app.config["WESP_MDNS_ENABLED"] = True
|
|
self.app.debug = True
|
|
with patch.dict(os.environ, {}, clear=False):
|
|
os.environ.pop("WERKZEUG_RUN_MAIN", None)
|
|
self.assertEqual(mdns_skip_reason_on_boot(self.app), "Flask reloader (родитель)")
|
|
self.assertFalse(should_start_mdns_on_boot(self.app))
|
|
|
|
def test_skip_reason_when_mdns_disabled(self) -> None:
|
|
self.app.config["TESTING"] = False
|
|
self.app.config["WESP_MDNS_ENABLED"] = False
|
|
self.assertEqual(mdns_skip_reason_on_boot(self.app), "mDNS выключен в настройках")
|
|
|
|
@patch("app.services.mdns_service._ZEROCONF_AVAILABLE", False)
|
|
def test_ensure_mdns_running_logs_when_zeroconf_missing(self) -> None:
|
|
self.app.config["TESTING"] = False
|
|
self.app.config["WESP_MDNS_ENABLED"] = True
|
|
self.app.config["WESP_NETWORK_LOCAL_HOSTNAME"] = "server.local"
|
|
with self.app.app_context():
|
|
ok = ensure_mdns_running(self.app, source="test")
|
|
self.assertFalse(ok)
|
|
|
|
@patch("app.services.mdns_service.start_mdns")
|
|
def test_ensure_mdns_running_calls_start_when_allowed(self, mock_start: MagicMock) -> None:
|
|
self.app.config["TESTING"] = False
|
|
self.app.config["WESP_MDNS_ENABLED"] = True
|
|
self.app.config["WESP_NETWORK_LOCAL_HOSTNAME"] = "server.local"
|
|
self.app.debug = False
|
|
|
|
def _activate(_app) -> None:
|
|
from app.services import mdns_service
|
|
|
|
mdns_service._active = True
|
|
|
|
mock_start.side_effect = _activate
|
|
with self.app.app_context():
|
|
ok = ensure_mdns_running(self.app, source="test")
|
|
self.assertTrue(ok)
|
|
mock_start.assert_called_once()
|
|
|
|
@patch("app.services.mdns_service.start_mdns")
|
|
@patch("app.services.mdns_service._mdns_retry_thread_running", return_value=False)
|
|
@patch("threading.Thread")
|
|
def test_schedule_retry_when_still_inactive(
|
|
self, mock_thread: MagicMock, _mock_running: MagicMock, _mock_start: MagicMock
|
|
) -> None:
|
|
self.app.config["TESTING"] = False
|
|
self.app.config["WESP_MDNS_ENABLED"] = True
|
|
self.app.debug = False
|
|
schedule_mdns_retry_if_needed(self.app)
|
|
mock_thread.assert_called_once()
|
|
self.assertEqual(mock_thread.call_args.kwargs.get("name"), "wesp-mdns-retry")
|
|
|
|
@patch("alembic.command.upgrade")
|
|
def test_run_auto_migrations_clears_wesp_alembic_flag(self, _mock_upgrade: MagicMock) -> None:
|
|
self.app.config["TESTING"] = False
|
|
with patch.dict(os.environ, {}, clear=False):
|
|
os.environ.pop("WESP_ALEMBIC", None)
|
|
_run_auto_migrations(self.app)
|
|
self.assertNotIn("WESP_ALEMBIC", os.environ)
|
|
|
|
def test_register_shutdown_skipped_in_testing(self) -> None:
|
|
register_mdns_shutdown(self.app)
|
|
# повторный вызов не должен падать
|
|
register_mdns_shutdown(self.app)
|
|
|
|
@patch("app.services.mdns_service._register_service_with_timeout")
|
|
@patch("app.services.mdns_service.IPVersion")
|
|
@patch("app.services.mdns_service.Zeroconf")
|
|
@patch("app.services.mdns_service.ServiceInfo")
|
|
@patch("app.services.mdns_service._ZEROCONF_AVAILABLE", True)
|
|
@patch("app.services.mdns_service.detect_lan_ip", return_value="192.168.1.187")
|
|
def test_start_mdns_success(
|
|
self,
|
|
_mock_ip: MagicMock,
|
|
_mock_info_cls: MagicMock,
|
|
mock_zc_cls: MagicMock,
|
|
mock_ip_version: MagicMock,
|
|
mock_register: MagicMock,
|
|
) -> None:
|
|
self.app.config["TESTING"] = False
|
|
self.app.config["WESP_MDNS_ENABLED"] = True
|
|
self.app.config["WESP_NETWORK_LOCAL_HOSTNAME"] = "server.local"
|
|
mock_zc_cls.return_value = MagicMock()
|
|
with self.app.app_context():
|
|
start_mdns(self.app)
|
|
mock_register.assert_called_once()
|
|
self.assertTrue(mdns_status().get("mdns_active"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|