Skip to content

Fix bogus overwrite imported txt_output to the id function #479

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jonasbardino merged 2 commits into from
Mar 9, 2026
Merged

Fix bogus overwrite imported txt_output to the id function #479

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
31e99a6
Fix bogus overwrite imported `txt_output` to the `id` function. Proba…
jonasbardino Mar 8, 2026
ac09604
Sync event runner testst with cron version where missing log errors f…
jonasbardino Mar 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions mig/lib/events.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
#
Expand Down Expand Up @@ -58,20 +58,20 @@
atjobs_pattern += r"([0-9]{2}) (.*)$"
atjobs_expr = re.compile(atjobs_pattern)

TRIGGER_EVENT = "_trigger_event"

Check failure on line 61 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'TRIGGER_EVENT' (60% confidence)

Check failure on line 61 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'TRIGGER_EVENT' (60% confidence)

# Only cache rule misses for one minute at a time to catch rule updates.
# Run complete expire cycle if miss cache exceeds expire size.

MISS_CACHE_TTL = 60

Check failure on line 66 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'MISS_CACHE_TTL' (60% confidence)

Check failure on line 66 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'MISS_CACHE_TTL' (60% confidence)
CACHE_EXPIRE_SIZE = 10000

Check failure on line 67 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'CACHE_EXPIRE_SIZE' (60% confidence)

Check failure on line 67 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'CACHE_EXPIRE_SIZE' (60% confidence)

# Rate limit helpers

RATE_LIMIT_FIELD, SETTLE_TIME_FIELD = ("rate_limit", "settle_time")

Check failure on line 71 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'SETTLE_TIME_FIELD' (60% confidence)

Check failure on line 71 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'RATE_LIMIT_FIELD' (60% confidence)

Check failure on line 71 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'SETTLE_TIME_FIELD' (60% confidence)

Check failure on line 71 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'RATE_LIMIT_FIELD' (60% confidence)
DEFAULT_PERIOD = "m"

Check failure on line 72 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'DEFAULT_PERIOD' (60% confidence)

Check failure on line 72 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'DEFAULT_PERIOD' (60% confidence)
DEFAULT_TIME = "0"

Check failure on line 73 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'DEFAULT_TIME' (60% confidence)

Check failure on line 73 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'DEFAULT_TIME' (60% confidence)
UNIT_PERIODS = {

Check failure on line 74 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'UNIT_PERIODS' (60% confidence)

Check failure on line 74 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'UNIT_PERIODS' (60% confidence)
"s": 1,
"m": 60,
"h": 60 * 60,
Expand Down Expand Up @@ -499,8 +499,7 @@

# logger.debug('(%s) import main from %s' % (pid, function))

main = id
txt_format = id
main = None
saved_environ = _save_env(os.environ)
try:
main = importlib.import_module(
Expand Down Expand Up @@ -563,7 +562,7 @@
from mig.shared.conf import get_configuration_object

conf = get_configuration_object()
client_id = "/C=DK/ST=NA/L=NA/O=NBI/OU=NA/CN=Jonas Bardino/emailAddress=bardino@nbi.ku.dk"

Check warning on line 565 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

line too long (94 > 80 characters)

Check warning on line 565 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

line too long (94 > 80 characters)
now = datetime.datetime.now()
now = now.replace(second=0, microsecond=0)
trigger_rule = {
Expand All @@ -590,7 +589,7 @@
print(" %s: %s" % (key, val))

crontab_lines = [
"* * * * * pack cront-test.txt cron-test-+SCHEDYEAR+-+SCHEDMONTH+-+SCHEDDAY+.zip"

Check warning on line 592 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

line too long (89 > 80 characters)

Check warning on line 592 in mig/lib/events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

line too long (89 > 80 characters)
]
crontab_rules = parse_crontab_contents(conf, client_id, crontab_lines)
cron_times = [
Expand Down
116 changes: 40 additions & 76 deletions tests/test_mig_lib_events.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
# -*- coding: utf-8 -*-
#
# --- BEGIN_HEADER ---
Expand Down Expand Up @@ -39,12 +39,12 @@
parse_crontab_contents, run_cron_command, run_events_command
from tests.support import MigTestCase, ensure_dirs_exist

DUMMY_USER_DN = "/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test User/emailAddress=test@example.com"

Check warning on line 42 in tests/test_mig_lib_events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

line too long (94 > 80 characters)

Check warning on line 42 in tests/test_mig_lib_events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

line too long (94 > 80 characters)
DUMMY_FULL_NAME = "Test User"

Check failure on line 43 in tests/test_mig_lib_events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'DUMMY_FULL_NAME' (60% confidence)

Check failure on line 43 in tests/test_mig_lib_events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'DUMMY_FULL_NAME' (60% confidence)
DUMMY_ORGANIZATION = "Test Org"

Check failure on line 44 in tests/test_mig_lib_events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'DUMMY_ORGANIZATION' (60% confidence)

Check failure on line 44 in tests/test_mig_lib_events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

unused variable 'DUMMY_ORGANIZATION' (60% confidence)
DUMMY_EMAIL = "test@example.com"
DUMMY_SKIP_EMAIL = ""
DUMMY_CLIENT_DIR = "+C=DK+ST=NA+L=NA+O=Test_Org+OU=NA+CN=Test_User+emailAddress=test@example.com"

Check warning on line 47 in tests/test_mig_lib_events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

line too long (97 > 80 characters)

Check warning on line 47 in tests/test_mig_lib_events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

line too long (97 > 80 characters)
DUMMY_CRON_JOB = {"command": ["touch", "test.txt"], "run_as": DUMMY_USER_DN}
DUMMY_CRONTAB_NAME = "crontab"
DUMMY_ATJOBS_NAME = "atjobs"
Expand Down Expand Up @@ -517,7 +517,7 @@
},
# Get first Friday of current month
now.replace(day=7).replace(
day=4 + now.replace(day=7).day - now.replace(day=7).weekday()),

Check warning on line 520 in tests/test_mig_lib_events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

line too long (83 > 80 characters)

Check warning on line 520 in tests/test_mig_lib_events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

line too long (83 > 80 characters)
),
(
{
Expand All @@ -541,7 +541,7 @@
},
# Get first Friday of current month
now.replace(day=7).replace(
day=4 + now.replace(day=7).day - now.replace(day=7).weekday()),

Check warning on line 544 in tests/test_mig_lib_events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

line too long (83 > 80 characters)

Check warning on line 544 in tests/test_mig_lib_events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

line too long (83 > 80 characters)
),
]
for job, now in test_cases:
Expand Down Expand Up @@ -641,7 +641,7 @@
},
# Get first Friday of current month
now.replace(day=7).replace(
day=4 + now.replace(day=7).day - now.replace(day=7).weekday()),

Check warning on line 644 in tests/test_mig_lib_events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

line too long (83 > 80 characters)

Check warning on line 644 in tests/test_mig_lib_events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

line too long (83 > 80 characters)
),
(
{
Expand All @@ -665,7 +665,7 @@
},
# Get first Friday of current month
now.replace(day=7).replace(
day=4 + now.replace(day=7).day - now.replace(day=7).weekday()),

Check warning on line 668 in tests/test_mig_lib_events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

line too long (83 > 80 characters)

Check warning on line 668 in tests/test_mig_lib_events.py

View workflow job for this annotation

GitHub Actions / Style check python and annotate 

line too long (83 > 80 characters)
),
(
{
Expand Down Expand Up @@ -2694,17 +2694,6 @@
except Exception as exc:
self.fail("run_cron_command raised an exception: %s" % exc)

def test_run_cron_command_with_special_chars(self):
"""Test running cron command with special characters"""
target_path = "/test/path/file@name#with$special&chars.txt"
command_list = ["touch", target_path]
crontab_entry = {"run_as": DUMMY_USER_DN}
# NOTE: touch fails without log in input validation
with self.assertRaises(Exception):
run_cron_command(
command_list, target_path, crontab_entry, self.configuration
)

def test_run_events_command_with_spaces(self):
"""Test running events command with spaces"""
target_path = "test file with spaces.txt"
Expand All @@ -2718,21 +2707,26 @@
except Exception as exc:
self.fail("run_events_command raised an exception: %s" % exc)

def test_run_cron_command_with_special_chars(self):
"""Test running cron command with special characters"""
target_path = "/test/path/file@name#with$special&chars.txt"
command_list = ["touch", target_path]
crontab_entry = {"run_as": DUMMY_USER_DN}
# NOTE: touch fails without log in input validation
with self.assertRaises(Exception):
run_cron_command(
command_list, target_path, crontab_entry, self.configuration
)

def test_run_events_command_with_special_chars(self):
"""Test running events command with special characters"""
target_path = "/test/path/file@name#with$special&chars.txt"
command_list = ["touch", target_path]
rule = {"run_as": DUMMY_USER_DN}
with self.assertLogs(level="ERROR") as log_capture:
with self.assertRaises(Exception):
run_events_command(
command_list, target_path, rule, self.configuration
)
self.assertTrue(
any(
"found invalid character" in msg
for msg in log_capture.output
)
# NOTE: touch fails without log in input validation
with self.assertRaises(Exception):
run_events_command(
command_list, target_path, rule, self.configuration
)

def test_run_cron_command_with_long_command(self):
Expand Down Expand Up @@ -2779,13 +2773,10 @@
target_path = "dummy"
command_list = ["touch"]
rule = {"run_as": DUMMY_USER_DN}
with self.assertLogs(level="ERROR") as log_capture:
with self.assertRaises(Exception):
run_events_command(
command_list, target_path, rule, self.configuration
)
self.assertTrue(
any("path: is required" in msg for msg in log_capture.output)
# NOTE: touch fails without log in input validation
with self.assertRaises(Exception):
run_events_command(
command_list, target_path, rule, self.configuration
)

def test_run_cron_command_with_output_redirection(self):
Expand All @@ -2804,16 +2795,10 @@
target_path = "test.txt"
command_list = ["touch", "test", ">", target_path]
rule = {"run_as": DUMMY_USER_DN}
with self.assertLogs(level="ERROR") as log_capture:
with self.assertRaises(Exception):
run_events_command(
command_list, target_path, rule, self.configuration
)
self.assertTrue(
any(
"found invalid character" in msg
for msg in log_capture.output
)
# NOTE: touch fails without log in input validation
with self.assertRaises(Exception):
run_events_command(
command_list, target_path, rule, self.configuration
)

def test_run_cron_command_with_pipe(self):
Expand All @@ -2832,16 +2817,10 @@
target_path = "test.txt"
command_list = ["touch", "test", "|", "grep", "test"]
rule = {"run_as": DUMMY_USER_DN}
with self.assertLogs(level="ERROR") as log_capture:
with self.assertRaises(Exception):
run_events_command(
command_list, target_path, rule, self.configuration
)
self.assertTrue(
any(
"found invalid character" in msg
for msg in log_capture.output
)
# NOTE: touch fails without log in input validation
with self.assertRaises(Exception):
run_events_command(
command_list, target_path, rule, self.configuration
)

def test_run_cron_command_with_background(self):
Expand Down Expand Up @@ -2891,14 +2870,11 @@
target_path = "test.txt"
command_list = ["touch", '"test with spaces"']
rule = {"run_as": DUMMY_USER_DN}
with self.assertLogs(level="ERROR") as log_capture:
with self.assertRaises(Exception):
run_events_command(
command_list, target_path, rule, self.configuration
)
self.assertTrue(
any("found invalid character" in msg for msg in log_capture.output)
)
# NOTE: touch fails without log in input validation
with self.assertRaises(Exception):
run_events_command(
command_list, target_path, rule, self.configuration
)

def test_run_cron_command_with_environment_variable(self):
"""Test running cron command with environment variable"""
Expand Down Expand Up @@ -2942,16 +2918,10 @@
target_path = "test.txt"
command_list = ["touch", "test\\ with\\ spaces"]
rule = {"run_as": DUMMY_USER_DN}
with self.assertLogs(level="ERROR") as log_capture:
with self.assertRaises(Exception):
run_events_command(
command_list, target_path, rule, self.configuration
)
self.assertTrue(
any(
"found invalid character" in msg
for msg in log_capture.output
)
# NOTE: touch fails without log in input validation
with self.assertRaises(Exception):
run_events_command(
command_list, target_path, rule, self.configuration
)

def test_run_cron_command_with_multiple_commands(self):
Expand Down Expand Up @@ -2996,16 +2966,10 @@
target_path = "test.txt"
command_list = ["chksum", "md5", "<", "input.txt"]
rule = {"run_as": DUMMY_USER_DN}
with self.assertLogs(level="ERROR") as log_capture:
with self.assertRaises(Exception):
run_events_command(
command_list, target_path, rule, self.configuration
)
self.assertTrue(
any(
"found invalid character" in msg
for msg in log_capture.output
)
# NOTE: touch fails without log in input validation
with self.assertRaises(Exception):
run_events_command(
command_list, target_path, rule, self.configuration
)

def test_run_cron_command_with_here_document(self):
Expand Down