from __future__ import annotations import pytest import pytest_asyncio import asyncio import shutil import re import pathlib import sys import shlex import tempfile import subprocess from typing import Iterator, Dict, AsyncIterator, Any from moonraker import Server from eventloop import EventLoop import utils from fixtures import KlippyProcess, HttpClient, WebsocketClient ASSETS = pathlib.Path(__file__).parent.joinpath("assets") def pytest_addoption(parser: pytest.Parser, pluginmanager): parser.addoption("--klipper-path", action="store", dest="klipper_path") parser.addoption("--klipper-exec", action="store", dest="klipper_exec") def interpolate_config(source_path: pathlib.Path, dest_path: pathlib.Path, keys: Dict[str, Any] ) -> None: def interp(match): return str(keys[match.group(1)]) sub_data = re.sub(r"\${([^}]+)}", interp, source_path.read_text()) dest_path.write_text(sub_data) @pytest.fixture(scope="session", autouse=True) def ssl_certs() -> Iterator[Dict[str, pathlib.Path]]: with tempfile.TemporaryDirectory(prefix="moonraker-certs-") as tmpdir: tmp_path = pathlib.Path(tmpdir) cert_path = tmp_path.joinpath("certificate.pem") key_path = tmp_path.joinpath("privkey.pem") cmd = ( f"openssl req -newkey rsa:4096 -nodes -keyout {key_path} " f"-x509 -days 365 -out {cert_path} -sha256 " "-subj '/C=US/ST=NRW/L=Earth/O=Moonraker/OU=IT/" "CN=www.moonraker-test.com/emailAddress=mail@moonraker-test.com'" ) args = shlex.split(cmd) subprocess.run(args, check=True) yield { "ssl_certificate_path": cert_path, "ssl_key_path": key_path, } @pytest.fixture(scope="class") def event_loop() -> Iterator[asyncio.AbstractEventLoop]: loop = asyncio.get_event_loop_policy().new_event_loop() yield loop loop.close() @pytest.fixture(scope="class") def path_args(request: pytest.FixtureRequest, ssl_certs: Dict[str, pathlib.Path] ) -> Iterator[Dict[str, pathlib.Path]]: path_marker = request.node.get_closest_marker("run_paths") paths = { "moonraker_conf": "base_server.conf", "secrets": "secrets.ini", "printer_cfg": "base_printer.cfg" } if path_marker is not None: paths.update(path_marker.kwargs) moon_cfg_path = ASSETS.joinpath(f"moonraker/{paths['moonraker_conf']}") secrets_path = ASSETS.joinpath(f"moonraker/{paths['secrets']}") pcfg_path = ASSETS.joinpath(f"klipper/{paths['printer_cfg']}") with tempfile.TemporaryDirectory(prefix="moonraker-test") as tmpdir: tmp_path = pathlib.Path(tmpdir) secrets_dest = tmp_path.joinpath(paths['secrets']) shutil.copy(secrets_path, secrets_dest) cfg_path = tmp_path.joinpath("config") cfg_path.mkdir() log_path = tmp_path.joinpath("logs") log_path.mkdir() db_path = tmp_path.joinpath("database") db_path.mkdir() gcode_path = tmp_path.joinpath("gcode_files") gcode_path.mkdir() dest_paths = { "asset_path": ASSETS, "config_path": cfg_path, "database_path": db_path, "log_path": log_path, "gcode_path": gcode_path, "secrets_path": secrets_dest, "klippy_uds_path": tmp_path.joinpath("klippy_uds"), "klippy_pty_path": tmp_path.joinpath("klippy_pty"), "klipper.dict": ASSETS.joinpath("klipper/klipper.dict"), } dest_paths.update(ssl_certs) if "moonraker_log" in paths: dest_paths['moonraker.log'] = log_path.joinpath( paths["moonraker_log"]) moon_cfg_dest = cfg_path.joinpath("moonraker.conf") interpolate_config(moon_cfg_path, moon_cfg_dest, dest_paths) dest_paths['moonraker.conf'] = moon_cfg_dest pcfg_dest = cfg_path.joinpath("printer.cfg") interpolate_config(pcfg_path, pcfg_dest, dest_paths) dest_paths['printer.cfg'] = pcfg_dest if "moonraker_bkp" in paths: bkp_source = ASSETS.joinpath("moonraker/base_server.conf") bkp_dest = cfg_path.joinpath(paths["moonraker_bkp"]) interpolate_config(bkp_source, bkp_dest, dest_paths) yield dest_paths @pytest.fixture(scope="class") def klippy(path_args: Dict[str, pathlib.Path], pytestconfig: pytest.Config) -> Iterator[KlippyProcess]: kpath = pytestconfig.getoption('klipper_path', "~/klipper") kexec = pytestconfig.getoption('klipper_exec', None) if kexec is None: kexec = sys.executable exec = pathlib.Path(kexec).expanduser() klipper_path = pathlib.Path(kpath).expanduser() base_cmd = f"{exec} {klipper_path}/klippy/klippy.py " kproc = KlippyProcess(base_cmd, path_args) kproc.start() yield kproc kproc.stop() @pytest.fixture(scope="class") def base_server(path_args: Dict[str, pathlib.Path], event_loop: asyncio.AbstractEventLoop ) -> Iterator[Server]: evtloop = EventLoop() args = { 'config_file': str(path_args['moonraker.conf']), 'log_file': str(path_args.get("moonraker.log", "")), 'software_version': "moonraker-pytest" } ql = logger = None if args["log_file"]: ql, logger, warning = utils.setup_logging(args) if warning: args["log_warning"] = warning yield Server(args, logger, evtloop) if ql is not None: ql.stop() @pytest_asyncio.fixture(scope="class") async def full_server(base_server: Server) -> AsyncIterator[Server]: base_server.load_components() ret = base_server.server_init(start_server=False) await asyncio.wait_for(ret, 4.) yield base_server if base_server.event_loop.aioloop.is_running(): await base_server._stop_server(exit_reason="terminate") @pytest_asyncio.fixture(scope="class") async def ready_server(full_server: Server, klippy: KlippyProcess): ret = full_server.start_server(connect_to_klippy=False) await asyncio.wait_for(ret, 4.) ret = full_server.klippy_connection.connect() await asyncio.wait_for(ret, 4.) yield full_server @pytest_asyncio.fixture(scope="class") async def http_client() -> AsyncIterator[HttpClient]: client = HttpClient() yield client client.close() @pytest_asyncio.fixture(scope="class") async def websocket_client(request: pytest.FixtureRequest ) -> AsyncIterator[WebsocketClient]: conn_marker = request.node.get_closest_marker("no_ws_connect") client = WebsocketClient() if conn_marker is None: await client.connect() yield client client.close()