fix(macros): no module-level assignment, or the loader refuses the file
CI for csaxs_bec / test (push) Canceled after 0s
CI for csaxs_bec / test (pull_request) Successful in 1m53s

The previous commit added `logger = bec_logger.logger` at module level. The
macro loader rejects any module-level ast.Assign
(bec_lib.macro_update_handler.has_executable_code) and then refuses the whole
file, so run_cont_grid_scan_for_table_row stopped being loaded at all:

    Macro file .../run_cont_grid_scan_for_table_row.py contains executable code
    at module level (line 16) and will not be loaded for security reasons.

Imports, defs, classes, annotated assignments and docstrings are permitted;
plain assignments are not. The logger is now fetched inside the functions.

Adds a test that runs the real has_executable_code over every macro, so this is
caught by the suite instead of by a WARNING in the log stream. Verified to fail
when a module-level assignment is reintroduced.

Noticed only because bec-log-monitor happened to be running at the time -- a
refused macro is otherwise indistinguishable from one that was never installed.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0148tn6uK6oiTH25mzLfJcyc
This commit is contained in:
2026-09-04 20:27:51 +02:00
co-authored by Claude Opus 5
parent cb21186e9d
commit d85da70730
2 changed files with 24 additions and 4 deletions
@@ -13,7 +13,9 @@ import numpy as np
from bec_lib.logger import bec_logger
from bec_lib.scan_repeat import scan_repeat
logger = bec_logger.logger
# NOTE No module-level assignments in a macro: the loader rejects any ast.Assign at module
# level (bec_lib.macro_update_handler.has_executable_code) and refuses the whole file, so
# the logger is fetched inside the functions rather than bound to a module global.
def _row_context(row: dict) -> str:
@@ -52,7 +54,7 @@ def _report_failure(info_msg: str, exc: Exception, row: dict, phone_numbers) ->
detail = traceback.format_exc()
print(f"{info_msg}\n {context}\n {type(exc).__name__}: {exc}\n{detail}")
logger.error(f"{info_msg} [{context}]\n{detail}")
bec_logger.logger.error(f"{info_msg} [{context}]\n{detail}")
try:
msg = bec.messaging.scilog.new()
@@ -65,7 +67,7 @@ def _report_failure(info_msg: str, exc: Exception, row: dict, phone_numbers) ->
except Exception: # pylint: disable=broad-except
# Never let the reporting channel become the reported error.
print(f"Additionally, posting this failure to SciLog failed:\n{traceback.format_exc()}")
logger.warning(f"SciLog post failed while reporting: {info_msg}")
bec_logger.logger.warning(f"SciLog post failed while reporting: {info_msg}")
if phone_numbers:
try:
@@ -74,7 +76,7 @@ def _report_failure(info_msg: str, exc: Exception, row: dict, phone_numbers) ->
msg.send(scope=phone_numbers)
except Exception: # pylint: disable=broad-except
print(f"Additionally, the SMS notification failed:\n{traceback.format_exc()}")
logger.warning(f"SMS notification failed while reporting: {info_msg}")
bec_logger.logger.warning(f"SMS notification failed while reporting: {info_msg}")
# TODO Default True here means that ANY error will be retried up to 3 times. This is dangerous in the sense that
@@ -10,6 +10,7 @@ import ast
import pathlib
import pytest
from bec_lib.macro_update_handler import has_executable_code
MACRO_DIR = pathlib.Path(__file__).resolve().parents[2] / "csaxs_bec" / "macros"
MACROS = sorted(p for p in MACRO_DIR.glob("*.py") if p.name != "__init__.py")
@@ -42,3 +43,20 @@ def test_macros_do_not_go_through_dev_devices(macro: pathlib.Path):
f"{macro.name}: line(s) {offenders} use dev.devices -- `dev` is the device "
"container itself, so this raises DeviceConfigError at runtime. Use dev[<name>]."
)
@pytest.mark.parametrize("macro", MACROS, ids=lambda p: p.name)
def test_macros_are_loadable_by_the_macro_loader(macro: pathlib.Path):
"""A module-level statement makes the loader refuse the whole file, silently.
bec_lib.macro_update_handler.has_executable_code rejects any module-level node that
is not an import, a def, a class, an annotated assignment or a docstring -- a plain
`logger = bec_logger.logger` is enough. The macro is then simply not loaded, and the
only trace is a WARNING in the log stream, so a broken macro looks like a macro that
was never there. The real checker is used here rather than a re-implementation.
"""
blocked, lineno = has_executable_code(macro.read_text())
assert not blocked, (
f"{macro.name}: module-level executable code at line {lineno}; the macro loader "
"will refuse the file. Move it inside a function, or use an annotated assignment."
)