Quelle  wave_server.py   Sprache: Python

# mypy: allow-untyped-defs

import os
import logging

from . import configuration_loader

from .network.http_handler import HttpHandler
from .network.api.sessions_api_handler import SessionsApiHandler
from .network.api.tests_api_handler import TestsApiHandler
from .network.api.results_api_handler import ResultsApiHandler
from .network.api.devices_api_handler import DevicesApiHandler
from .network.api.general_api_handler import GeneralApiHandler
from .network.static_handler import StaticHandler

from .testing.sessions_manager import SessionsManager
from .testing.results_manager import ResultsManager
from .testing.tests_manager import TestsManager
from .testing.devices_manager import DevicesManager
from .testing.test_loader import TestLoader
from .testing.event_dispatcher import EventDispatcher

VERSION_STRING = "v3.3.0"


class WaveServer:
    def initialize(self,
                   tests,
                   configuration_file_path=None,
                   application_directory_path=None,
                   reports_enabled=None):
        if configuration_file_path is None:
            configuration_file_path = ""
        if application_directory_path is None:
            application_directory_path = ""
        if reports_enabled is None:
            reports_enabled = False

        logger = logging.getLogger("wave-server")

        logger.debug("Loading configuration ...")
        configuration = configuration_loader.load(configuration_file_path)

        # Initialize Managers
        event_dispatcher = EventDispatcher(
            event_cache_duration=configuration["event_cache_duration"]
        )
        sessions_manager = SessionsManager()
        results_manager = ResultsManager()
        tests_manager = TestsManager()
        devices_manager = DevicesManager()
        test_loader = TestLoader()

        sessions_manager.initialize(
            test_loader=test_loader,
            event_dispatcher=event_dispatcher,
            tests_manager=tests_manager,
            results_directory=configuration["results_directory_path"],
            results_manager=results_manager,
            configuration=configuration
        )

        results_manager.initialize(
            results_directory_path=configuration["results_directory_path"],
            sessions_manager=sessions_manager,
            tests_manager=tests_manager,
            import_results_enabled=configuration["import_results_enabled"],
            reports_enabled=reports_enabled,
            persisting_interval=configuration["persisting_interval"]
        )

        tests_manager.initialize(
            test_loader,
            results_manager=results_manager,
            sessions_manager=sessions_manager,
            event_dispatcher=event_dispatcher
        )

        devices_manager.initialize(event_dispatcher)

        exclude_list_file_path = os.path.abspath("./excluded.json")
        include_list_file_path = os.path.abspath("./included.json")
        test_loader.initialize(
            exclude_list_file_path,
            include_list_file_path,
            results_manager=results_manager,
            api_titles=configuration["api_titles"]
        )

        test_loader.load_tests(tests)

        # Initialize HTTP handlers
        static_handler = StaticHandler(
            web_root=configuration["web_root"],
            http_port=configuration["wpt_port"],
            https_port=configuration["wpt_ssl_port"]
        )
        sessions_api_handler = SessionsApiHandler(
            sessions_manager=sessions_manager,
            results_manager=results_manager,
            event_dispatcher=event_dispatcher,
            web_root=configuration["web_root"],
            read_sessions_enabled=configuration["read_sessions_enabled"]
        )
        tests_api_handler = TestsApiHandler(
            tests_manager=tests_manager,
            sessions_manager=sessions_manager,
            wpt_port=configuration["wpt_port"],
            wpt_ssl_port=configuration["wpt_ssl_port"],
            hostname=configuration["hostname"],
            web_root=configuration["web_root"],
            test_loader=test_loader
        )
        devices_api_handler = DevicesApiHandler(
            devices_manager=devices_manager,
            event_dispatcher=event_dispatcher,
            web_root=configuration["web_root"]
        )
        results_api_handler = ResultsApiHandler(
            results_manager,
            sessions_manager,
            web_root=configuration["web_root"]
        )
        general_api_handler = GeneralApiHandler(
            web_root=configuration["web_root"],
            read_sessions_enabled=configuration["read_sessions_enabled"],
            import_results_enabled=configuration["import_results_enabled"],
            reports_enabled=reports_enabled,
            version_string=VERSION_STRING,
            test_type_selection_enabled=configuration["enable_test_type_selection"],
            test_file_selection_enabled=configuration["enable_test_file_selection"]
        )

        # Initialize HTTP server
        http_handler = HttpHandler(
            static_handler=static_handler,
            sessions_api_handler=sessions_api_handler,
            tests_api_handler=tests_api_handler,
            results_api_handler=results_api_handler,
            devices_api_handler=devices_api_handler,
            general_api_handler=general_api_handler,
            http_port=configuration["wpt_port"],
            web_root=configuration["web_root"]
        )
        self.handle_request = http_handler.handle_request