add variables module to monitor utils

Author: GabrielSalla

docs/monitor.md

@@ -56,9 +56,10 @@ To create a monitor, import specific dependencies from `monitor_utils`. Availabl
 - [`ValueRule`](#value-rule)
 - [`PriorityLevels`](#priority-levels)
 
-**Functions**
+**Available functions and modules**
 - [`query`](#query)
 - [`read_file`](#read-file)
+- [`variables`](#variables)
 
 To get started with a simple monitor, use the following imports:
 
@@ -360,5 +361,37 @@ The function takes 2 parameters:
 content = read_file("search_query.sql")
 ```
 
+## Variables
+The `variables` module allows storing and retrieving variables that can be used across executions of the monitor. This is useful for maintaining state or configuration information.
+
+Available functions are:
+**`get_variable`**
+
+The function takes one parameter:
+- `name`: The name of the variable to retrieve.
+
+Return the value of a variable. If the variable does not exist, returns `None`.
+
+**`set_variable`**
+
+The function takes two parameters:
+- `name`: The name of the variable to set.
+- `value`: The value to assign to the variable.
+
+Sets the value of a variable. If the variable does not exist yet, it's created.
+
+Both functions must be called from functions defined in the monitor base code. If they're called from any other Python file, they will raise an error as they won't be able to identify the monitor that's calling it.
+
+```python
+from monitor_utils import variables
+
+async def search() -> list[IssueDataType] | None:
+    # Set a variable
+    await variables.set_variable("my_var", "some_value")
+
+async def update(issues_data: list[IssueDataType]) -> list[IssueDataType] | None:
+    value = await variables.get_variable("my_var")
+```
+
 # Registering
 After creating the monitor, the next step is to register it on Sentinela. Check the [Registering a monitor](monitor_registering.md) documentation for more information.

src/monitor_utils/__init__.py

@@ -16,6 +16,7 @@
 from databases import query
 from models.utils.priority import AlertPriority
 
+from . import variables
 from .read_file import read_file
 
 __all__ = [
@@ -31,4 +32,5 @@
     "ReactionOptions",
     "read_file",
     "ValueRule",
+    "variables",
 ]

src/monitor_utils/variables.py

@@ -0,0 +1,38 @@
+from types import ModuleType
+from typing import cast
+
+from models import Variable
+from utils.stack import get_caller
+
+"""The functions 'set_variable' and 'get_variable' must be called from inside a monitor to be able
+to identify the monitor ID. If called outside a monitor, they will raise an error."""
+
+
+def _get_monitor_id(monitor_module: ModuleType) -> int:
+    """Get the monitor ID from the monitor module"""
+    monitor_id = getattr(monitor_module, "SENTINELA_MONITOR_ID", None)
+    if monitor_id is None:
+        raise ValueError("Function called outside a monitor or the monitor was not loaded properly")
+
+    return cast(int, monitor_id)
+
+
+async def set_variable(name: str, value: str | None) -> None:
+    """Set a variable for the monitor"""
+    monitor_module, _ = get_caller()
+    monitor_id = _get_monitor_id(monitor_module)
+
+    variable = await Variable.get_or_create(monitor_id=monitor_id, name=name)
+    await variable.set(value)
+
+
+async def get_variable(name: str) -> str | None:
+    """Get a variable for the monitor, or None if it does not exist"""
+    monitor_module, _ = get_caller()
+    monitor_id = _get_monitor_id(monitor_module)
+
+    variable = await Variable.get(Variable.monitor_id == monitor_id, Variable.name == name)
+    if variable is None:
+        return None
+
+    return variable.value

tests/monitor_utils/test_variables.py

@@ -0,0 +1,212 @@
+import pytest
+
+from models import Monitor, Variable
+from monitor_utils.variables import _get_monitor_id, get_variable, set_variable
+
+pytestmark = pytest.mark.asyncio(loop_scope="session")
+
+
+@pytest.mark.parametrize("monitor_id", [123, 456, 789])
+async def test_get_monitor_id(monitor_id):
+    """'_get_monitor_id' should return the monitor ID from the module if SENTINELA_MONITOR_ID is
+    set"""
+
+    class Module:
+        SENTINELA_MONITOR_ID = monitor_id
+
+    result = _get_monitor_id(Module())
+    assert result == monitor_id
+
+
+async def test_get_monitor_id_with_none():
+    """'_get_monitor_id' should raise ValueError if SENTINELA_MONITOR_ID is None"""
+
+    class Module:
+        SENTINELA_MONITOR_ID = None
+
+    expected_error_message = (
+        "Function called outside a monitor or the monitor was not loaded properly"
+    )
+    with pytest.raises(ValueError, match=expected_error_message):
+        _get_monitor_id(Module())
+
+
+async def test_get_monitor_id_without_attribute():
+    """'_get_monitor_id' should raise ValueError if SENTINELA_MONITOR_ID is not defined"""
+
+    class Module:
+        pass
+
+    expected_error_message = (
+        "Function called outside a monitor or the monitor was not loaded properly"
+    )
+    with pytest.raises(ValueError, match=expected_error_message):
+        _get_monitor_id(Module())
+
+
+@pytest.mark.parametrize(
+    "variable_name, variable_value",
+    [("test_var", "test_value"), ("another_var", "another_value"), ("empty_var", None)],
+)
+async def test_set_variable(sample_monitor: Monitor, variable_name, variable_value):
+    """'set_variable' should set a variable for the monitor"""
+    call_function = sample_monitor.code.call_function  # type: ignore[attr-defined]
+
+    async def f() -> None:
+        await call_function(set_variable, variable_name, variable_value)
+
+    await f()
+
+    variables = await Variable.get_all(Variable.monitor_id == sample_monitor.id)
+    assert len(variables) == 1
+
+    assert variables[0].monitor_id == sample_monitor.id
+    assert variables[0].name == variable_name
+    assert variables[0].value == variable_value
+
+
+async def test_set_variable_multiple_variables(sample_monitor: Monitor):
+    """'set_variable' should set multiple different variables for the same monitor"""
+    call_function = sample_monitor.code.call_function  # type: ignore[attr-defined]
+
+    async def f() -> None:
+        await call_function(set_variable, "var1", "value1")
+        await call_function(set_variable, "var2", "value2")
+        await call_function(set_variable, "var3", None)
+
+    await f()
+
+    variables = await Variable.get_all(Variable.monitor_id == sample_monitor.id)
+    assert len(variables) == 3
+
+    variables_dict = {var.name: var.value for var in variables}
+    expected_variables = {"var1": "value1", "var2": "value2", "var3": None}
+    assert variables_dict == expected_variables
+
+
+async def test_set_variable_update_existing(sample_monitor: Monitor):
+    """'set_variable' should update an existing variable if it already exists"""
+    await Variable.create(
+        monitor_id=sample_monitor.id,
+        name="test_var",
+        value="initial_value",
+    )
+
+    call_function = sample_monitor.code.call_function  # type: ignore[attr-defined]
+
+    async def f() -> None:
+        await call_function(set_variable, "test_var", "updated_value")
+
+    await f()
+
+    variables = await Variable.get_all(Variable.monitor_id == sample_monitor.id)
+    assert len(variables) == 1
+    assert variables[0].name == "test_var"
+    assert variables[0].value == "updated_value"
+
+
+async def test_set_variable_monitor_id_none(monkeypatch, sample_monitor: Monitor):
+    """'set_variable' should raise 'ValueError' if 'SENTINELA_MONITOR_ID' is None"""
+    monkeypatch.setattr(sample_monitor.code, "SENTINELA_MONITOR_ID", None, raising=False)
+
+    call_function = sample_monitor.code.call_function  # type: ignore[attr-defined]
+
+    async def f() -> None:
+        await call_function(set_variable, "test_var", "test_value")
+
+    expected_error_message = (
+        "Function called outside a monitor or the monitor was not loaded properly"
+    )
+    with pytest.raises(ValueError, match=expected_error_message):
+        await f()
+
+
+@pytest.mark.parametrize(
+    "variable_name, variable_value",
+    [("test_var", "test_value"), ("another_var", "another_value"), ("empty_var", None)],
+)
+async def test_get_variable(sample_monitor: Monitor, variable_name, variable_value):
+    """'get_variable' should get a variable for the monitor"""
+    await Variable.create(
+        monitor_id=sample_monitor.id,
+        name=variable_name,
+        value=variable_value,
+    )
+
+    call_function = sample_monitor.code.call_function  # type: ignore[attr-defined]
+
+    async def f() -> None:
+        result = await call_function(get_variable, variable_name)
+        assert result == variable_value
+
+    await f()
+
+
+async def test_get_variable_none_value(sample_monitor: Monitor):
+    """'get_variable' should be able to get a variable with 'None' value"""
+    await Variable.create(
+        monitor_id=sample_monitor.id,
+        name="test_var",
+        value=None,
+    )
+
+    call_function = sample_monitor.code.call_function  # type: ignore[attr-defined]
+
+    async def f() -> None:
+        assert await call_function(get_variable, "test_var") is None
+
+    await f()
+
+
+async def test_get_variable_not_exists(sample_monitor: Monitor):
+    """'get_variable' should not raise an error if the variable does not exist and return None"""
+    variable = await Variable.get(
+        Variable.monitor_id == sample_monitor.id,
+        Variable.name == "non_existent_var",
+    )
+    assert variable is None
+
+    call_function = sample_monitor.code.call_function  # type: ignore[attr-defined]
+
+    async def f() -> None:
+        assert await call_function(get_variable, "non_existent_var") is None
+
+    await f()
+
+
+async def test_get_variable_multiple_variables(sample_monitor: Monitor):
+    """'get_variable' should correctly retrieve specific variables when multiple exist"""
+    await Variable.create(
+        monitor_id=sample_monitor.id,
+        name="var1",
+        value="value1",
+    )
+    await Variable.create(
+        monitor_id=sample_monitor.id,
+        name="var2",
+        value="value2",
+    )
+
+    call_function = sample_monitor.code.call_function  # type: ignore[attr-defined]
+
+    async def f() -> None:
+        assert await call_function(get_variable, "var1") == "value1"
+        assert await call_function(get_variable, "var2") == "value2"
+
+    await f()
+
+
+async def test_get_variable_monitor_id_none(monkeypatch, sample_monitor: Monitor):
+    """'get_variable' should raise ValueError if SENTINELA_MONITOR_ID is None"""
+    monkeypatch.setattr(sample_monitor.code, "SENTINELA_MONITOR_ID", None)
+
+    call_function = sample_monitor.code.call_function  # type: ignore[attr-defined]
+
+    async def f() -> None:
+        await call_function(get_variable, "test_var")
+
+    expected_error_message = (
+        "Function called outside a monitor or the monitor was not loaded properly"
+    )
+    with pytest.raises(ValueError, match=expected_error_message):
+        await f()

tests/sample_monitor_code.py

@@ -1,4 +1,4 @@
-from typing import TypedDict
+from typing import Any, Callable, TypedDict
 
 from monitor_utils import IssueOptions, MonitorOptions
 
@@ -21,3 +21,9 @@ class IssueDataType(TypedDict):
 async def search() -> list[IssueDataType] | None: ...
 async def update(issues_data: list[IssueDataType]) -> list[IssueDataType] | None: ...
 def is_solved(issue_data: IssueDataType) -> bool: ...
+
+
+async def call_function(function: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
+    """Call a function and return the result. This function is used in tests to mock calls that must
+    come from inside the monitor"""
+    return await function(*args, **kwargs)