Create monitor validate route

src/commands/__init__.py

@@ -5,6 +5,7 @@
     disable_monitor,
     enable_monitor,
     issue_drop,
+    monitor_code_validate,
     monitor_register,
 )
 
@@ -14,7 +15,7 @@
     "alert_solve",
     "disable_monitor",
     "enable_monitor",
-    "get_message_request",
     "issue_drop",
+    "monitor_code_validate",
     "monitor_register",
 ]

src/commands/requests.py

@@ -1,8 +1,19 @@
+import random
+import string
+import time
+
 import components.monitors_loader as monitors_loader
 import message_queue as message_queue
 from models import Monitor
 
 
+async def monitor_code_validate(monitor_code: str) -> None:
+    """Validate a monitor code without registering it"""
+    timestamp_string = str(int(time.time()))
+    random_string = "".join(random.choice(string.ascii_lowercase) for _ in range(8))
+    monitors_loader.check_monitor(f"monitor_{timestamp_string}_{random_string}", monitor_code)
+
+
 async def monitor_register(
     monitor_name: str, monitor_code: str, additional_files: dict[str, str]
 ) -> Monitor:

src/components/http_server/monitor_routes.py

@@ -108,13 +108,58 @@ async def monitor_enable(request: Request) -> Response:
         return web.json_response(error_response, status=400)
 
 
+@monitor_routes.post(base_route + "/validate")
+@monitor_routes.post(base_route + "/validate/")
+async def monitor_validate(request: Request) -> Response:
+    """Route to check a monitor without registering it"""
+    request_data = await request.json()
+    monitor_code = request_data.get("monitor_code")
+
+    error_response: dict[str, str | list[Any]]
+
+    if monitor_code is None:
+        error_response = {"status": "error", "message": "'monitor_code' parameter is required"}
+        return web.json_response(error_response, status=400)
+
+    try:
+        await commands.monitor_code_validate(monitor_code)
+    except pydantic.ValidationError as e:
+        error_response = {
+            "status": "error",
+            "message": "Type validation error",
+            "error": [
+                {
+                    "loc": list(error["loc"]),
+                    "type": error["type"],
+                    "msg": error["msg"],
+                }
+                for error in e.errors()
+            ],
+        }
+        return web.json_response(error_response, status=400)
+    except MonitorValidationError as e:
+        error_response = {
+            "status": "error",
+            "message": "Module didn't pass check",
+            "error": e.get_error_message(),
+        }
+        return web.json_response(error_response, status=400)
+    except Exception as e:
+        error_response = {"status": "error", "error": str(e)}
+        _logger.error(traceback.format_exc().strip())
+        return web.json_response(error_response, status=400)
+
+    success_response = {"status": "monitor_validated"}
+    return web.json_response(success_response)
+
+
 @monitor_routes.post(base_route + "/register/{monitor_name}")
 @monitor_routes.post(base_route + "/register/{monitor_name}/")
 async def monitor_register(request: Request) -> Response:
     """Route to register a monitor"""
-    request_data = await request.json()
-
     monitor_name = request.match_info["monitor_name"]
+
+    request_data = await request.json()
     monitor_code = request_data.get("monitor_code")
     additional_files = request_data.get("additional_files", {})
 

src/components/monitors_loader/__init__.py

@@ -1,14 +1,14 @@
 from .monitors_loader import (
     MonitorValidationError,
-    _register_monitors,
+    check_monitor,
     init,
     register_monitor,
     wait_stop,
 )
 
 __all__ = [
     "MonitorValidationError",
-    "_register_monitors",
+    "check_monitor",
     "init",
     "register_monitor",
     "wait_stop",

src/components/monitors_loader/monitors_loader.py

@@ -62,6 +62,43 @@ def _file_has_extension(file: str, extensions: list[str]) -> bool:
     return any(file.endswith(f".{extension}") for extension in extensions)
 
 
+def check_monitor(
+    monitor_name: str, monitor_code: str, base_path: str | None = None, log_error: bool = False
+) -> None:
+    """Check if a monitor module is valid without registering it"""
+    module_path, module = module_loader.load_module_from_string(
+        module_name=monitor_name, module_code=monitor_code, base_path=base_path
+    )
+
+    errors = module_loader.check_module(module=module)
+    module_loader.remove_module(module_name=module_loader.make_module_name(module_path))
+    if len(errors) > 0:
+        exception = MonitorValidationError(monitor_name=monitor_name, errors_found=errors)
+        if log_error:
+            _logger.error(exception.get_error_message())
+        raise exception
+
+
+async def register_monitor(
+    monitor_name: str,
+    monitor_code: str,
+    base_path: str | None = None,
+    additional_files: dict[str, str] | None = None,
+) -> Monitor:
+    """Register a monitor and its additional files"""
+    check_monitor(
+        base_path=base_path, monitor_name=monitor_name, monitor_code=monitor_code, log_error=True
+    )
+
+    monitor = await Monitor.get_or_create(name=monitor_name)
+    code_module = await CodeModule.get_or_create(monitor_id=monitor.id)
+    code_module.code = monitor_code
+    code_module.additional_files = additional_files or {}
+    await code_module.save()
+
+    return monitor
+
+
 def _get_monitors_files_from_path(
     path: str, additional_file_extensions: list[str] | None = None
 ) -> Generator[MonitorFiles, None, None]:
@@ -96,37 +133,6 @@ def _get_monitors_files_from_path(
             yield monitor_files
 
 
-async def register_monitor(
-    monitor_name: str,
-    monitor_code: str,
-    base_path: str | None = None,
-    additional_files: dict[str, str] | None = None,
-) -> Monitor:
-    """Register a monitor and its additional files"""
-    if base_path is None:
-        base_path = MONITORS_LOAD_PATH
-
-    monitor_path = module_loader.create_module_files(
-        monitor_name, monitor_code, base_path=base_path
-    )
-    module = module_loader.load_module_from_file(monitor_path)
-
-    # Check the monitor module
-    errors = module_loader.check_module(module=module)
-    if len(errors) > 0:
-        exception = MonitorValidationError(monitor_name=monitor_name, errors_found=errors)
-        _logger.warning(exception.get_error_message())
-        raise exception
-
-    monitor = await Monitor.get_or_create(name=monitor_name)
-    code_module = await CodeModule.get_or_create(monitor_id=monitor.id)
-    code_module.code = monitor_code
-    code_module.additional_files = additional_files or {}
-    await code_module.save()
-
-    return monitor
-
-
 async def _register_monitors_from_path(
     path: str, internal: bool = False, additional_file_extensions: list[str] | None = None
 ) -> None:

src/module_loader/__init__.py

@@ -1,8 +1,17 @@
 from .checker import check_module
-from .loader import create_module_files, load_module_from_file
+from .loader import (
+    create_module_files,
+    load_module_from_file,
+    load_module_from_string,
+    make_module_name,
+    remove_module,
+)
 
 __all__ = [
     "check_module",
     "create_module_files",
     "load_module_from_file",
+    "load_module_from_string",
+    "make_module_name",
+    "remove_module",
 ]

src/module_loader/loader.py

@@ -1,6 +1,7 @@
 import importlib
 import logging
 import os
+import re
 import sys
 import time
 from functools import cache
@@ -59,16 +60,28 @@ def create_module_files(
     return module_path.relative_to(RELATIVE_PATH)
 
 
-def load_module_from_file(module_path: Path) -> ModuleType:
-    """Load a module from a path, returning the module"""
-    module_name = module_path.as_posix().replace("/", ".").strip(".py")
-    monitor_name = module_path.stem
+def make_module_name(module_path: Path) -> str:
+    """Make a module name from a path"""
+    return re.sub(r"\.py$", "", module_path.as_posix().replace("/", "."))
 
-    start_time = time.time()
 
+def remove_module(module_name: str) -> None:
+    """Remove a module from the loaded_modules. The module name should be relative to the 'src'
+    folder. Example: '_monitors.test'"""
     if module_name in sys.modules:
         del sys.modules[module_name]
 
+
+def load_module_from_file(module_path: Path) -> ModuleType:
+    """Load a module from a path, returning the module. If this function is called for the same
+    path in a short time frame, the module won't be reloaded from the files and will be the same
+    that was previously loaded"""
+    module_name = make_module_name(module_path)
+    monitor_name = module_path.stem
+
+    start_time = time.time()
+
+    remove_module(module_name)
     module = importlib.import_module(module_name)
     _logger.info(f"Monitor '{monitor_name}' loaded")
 
@@ -80,3 +93,16 @@ def load_module_from_file(module_path: Path) -> ModuleType:
         _logger.warning(f"Monitor '{monitor_name}' took {total_time} seconds to load")
 
     return module
+
+
+def load_module_from_string(
+    module_name: str, module_code: str, base_path: str | None = None
+) -> tuple[Path, ModuleType]:
+    """Load a module from a code string"""
+    if base_path is None:
+        base_path = MODULES_PATH
+
+    module_path = create_module_files(
+        module_name=module_name, module_code=module_code, base_path=base_path
+    )
+    return module_path, load_module_from_file(module_path)

tests/commands/test_requests.py

@@ -1,5 +1,6 @@
 import json
-from unittest.mock import AsyncMock
+import re
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
@@ -11,6 +12,24 @@
 pytestmark = pytest.mark.asyncio(loop_scope="session")
 
 
+async def test_monitor_code_validate(mocker):
+    """'monitor_code_validate' function should validate a monitor code"""
+    check_monitor_spy: MagicMock = mocker.spy(monitors_loader, "check_monitor")
+
+    with open("tests/sample_monitors/others/monitor_1/monitor_1.py", "r") as file:
+        monitor_code = file.read()
+
+    await requests.monitor_code_validate(monitor_code)
+
+    check_monitor_spy.assert_called_once()
+
+    call_args = check_monitor_spy.call_args
+    assert len(call_args.args) == 2
+    monitor_name_regex = r"monitor_\d{10}_[a-z]{8}"
+    assert re.match(monitor_name_regex, call_args.args[0]) is not None
+    assert call_args.args[1] == monitor_code
+
+
 async def test_monitor_register(mocker):
     """'monitor_register' function should register a monitor with the provided name and module
     code"""