From f0d8519b8005856287d4acee2447892c35b513e6 Mon Sep 17 00:00:00 2001 From: volcano303 Date: Wed, 29 Apr 2026 14:12:36 +0200 Subject: [PATCH] Add fcron command fallback --- src/cronboard_widgets/CronCommand.py | 46 ++++++++++++++++ src/cronboard_widgets/CronCreator.py | 7 +-- .../CronDeleteConfirmation.py | 13 +++-- src/cronboard_widgets/CronTable.py | 27 ++++------ tests/CronCommand_test.py | 53 +++++++++++++++++++ tests/CronDeleteConfirmation_test.py | 5 +- tests/conftest.py | 4 +- 7 files changed, 121 insertions(+), 34 deletions(-) create mode 100644 src/cronboard_widgets/CronCommand.py create mode 100644 tests/CronCommand_test.py diff --git a/src/cronboard_widgets/CronCommand.py b/src/cronboard_widgets/CronCommand.py new file mode 100644 index 0000000..f15f13b --- /dev/null +++ b/src/cronboard_widgets/CronCommand.py @@ -0,0 +1,46 @@ +import shlex +from shutil import which + +import crontab as crontab_module +from crontab import CronTab + + +def get_local_crontab_command() -> str: + """Return the installed crontab-compatible command.""" + for command in ("crontab", "fcrontab"): + command_path = which(command) + if command_path: + return command_path + return crontab_module.CRON_COMMAND + + +def create_user_crontab() -> CronTab: + """Create a user CronTab using crontab or fcrontab when available.""" + original_command = crontab_module.CRON_COMMAND + crontab_module.CRON_COMMAND = get_local_crontab_command() + try: + return CronTab(user=True) + finally: + crontab_module.CRON_COMMAND = original_command + + +def remote_crontab_command(*args: str) -> str: + quoted_args = " ".join(shlex.quote(str(arg)) for arg in args if arg) + suffix = f" {quoted_args}" if quoted_args else "" + return ( + "if command -v crontab >/dev/null 2>&1; " + f"then crontab{suffix}; " + f"else fcrontab{suffix}; fi" + ) + + +def remote_crontab_list_command(crontab_user: str | None = None) -> str: + if crontab_user: + return remote_crontab_command("-u", crontab_user, "-l") + return remote_crontab_command("-l") + + +def remote_crontab_write_command(crontab_user: str | None = None) -> str: + if crontab_user: + return remote_crontab_command("-u", crontab_user, "-") + return remote_crontab_command("-") diff --git a/src/cronboard_widgets/CronCreator.py b/src/cronboard_widgets/CronCreator.py index 3b395d8..51c13ac 100644 --- a/src/cronboard_widgets/CronCreator.py +++ b/src/cronboard_widgets/CronCreator.py @@ -15,6 +15,7 @@ PathDropdownItem, ) from cron_descriptor import Options, ExpressionDescriptor +from cronboard_widgets.CronCommand import remote_crontab_write_command CRON_ALIASES = { @@ -350,11 +351,7 @@ def write_cron_changes(self): if self.remote and self.ssh_client: try: new_crontab_content = self.cron.render() - crontab_cmd = ( - f"crontab -u {self.crontab_user} -" - if self.crontab_user - else "crontab -" - ) + crontab_cmd = remote_crontab_write_command(self.crontab_user) stdin, _, stderr = self.ssh_client.exec_command(crontab_cmd) stdin.write(new_crontab_content) stdin.channel.shutdown_write() diff --git a/src/cronboard_widgets/CronDeleteConfirmation.py b/src/cronboard_widgets/CronDeleteConfirmation.py index 22462b2..c0c8a5c 100644 --- a/src/cronboard_widgets/CronDeleteConfirmation.py +++ b/src/cronboard_widgets/CronDeleteConfirmation.py @@ -1,9 +1,12 @@ from textual.app import ComposeResult -from crontab import CronTab from textual.binding import Binding from textual.widgets import Button, Label from textual.containers import Grid, Horizontal, Vertical from textual.screen import ModalScreen +from cronboard_widgets.CronCommand import ( + create_user_crontab, + remote_crontab_write_command, +) class CronDeleteConfirmation(ModalScreen[bool]): @@ -22,7 +25,7 @@ def __init__( super().__init__() self.server = server self.job = job - self.cron = cron if cron else CronTab(user=True) + self.cron = cron if cron else create_user_crontab() self.remote = remote self.ssh_client = ssh_client self.message = message @@ -79,11 +82,7 @@ def write_remote_crontab(self): try: new_crontab_content = self.cron.render() - crontab_cmd = ( - f"crontab -u {self.crontab_user} -" - if self.crontab_user - else "crontab -" - ) + crontab_cmd = remote_crontab_write_command(self.crontab_user) stdin, _, stderr = self.ssh_client.exec_command(crontab_cmd) stdin.write(new_crontab_content) stdin.channel.shutdown_write() diff --git a/src/cronboard_widgets/CronTable.py b/src/cronboard_widgets/CronTable.py index 84f2450..d15b3c0 100644 --- a/src/cronboard_widgets/CronTable.py +++ b/src/cronboard_widgets/CronTable.py @@ -5,6 +5,11 @@ from datetime import datetime from rich.text import Text from cronboard_widgets.CronInputSearch import CronInputSearch +from cronboard_widgets.CronCommand import ( + create_user_crontab, + remote_crontab_list_command, + remote_crontab_write_command, +) class CronTable(DataTable): @@ -35,17 +40,13 @@ def __init__(self, remote=False, ssh_client=None, crontab_user=None, **kwargs): self._search_query: str = "" def on_mount(self) -> None: - self.cron: CronTab = CronTab(user=True) + self.cron: CronTab = create_user_crontab() self.add_columns( "ID", "Expression", "Command", "Last Run", "Next Run", "Status" ) if self.remote and self.ssh_client: - crontab_cmd = ( - f"crontab -u {self.crontab_user} -l" - if self.crontab_user - else "crontab -l" - ) + crontab_cmd = remote_crontab_list_command(self.crontab_user) _, stdout, _ = self.ssh_client.exec_command(crontab_cmd) exit_status = stdout.channel.recv_exit_status() @@ -170,11 +171,7 @@ def action_delete_cronjob_keybind(self, job) -> None: def action_refresh(self) -> None: """Refresh the cronjob list.""" if self.remote and self.ssh_client: - crontab_cmd = ( - f"crontab -u {self.crontab_user} -l" - if self.crontab_user - else "crontab -l" - ) + crontab_cmd = remote_crontab_list_command(self.crontab_user) _, stdout, _ = self.ssh_client.exec_command(crontab_cmd) exit_status = stdout.channel.recv_exit_status() @@ -185,7 +182,7 @@ def action_refresh(self) -> None: self.ssh_cron = CronTab(tab=self.crontab_content) else: - self.cron = CronTab(user=True) + self.cron = create_user_crontab() self.load_crontabs() self.refresh_bindings() @@ -350,11 +347,7 @@ def write_remote_crontab(self): try: new_crontab_content = self.ssh_cron.render() - crontab_cmd = ( - f"crontab -u {self.crontab_user} -" - if self.crontab_user - else "crontab -" - ) + crontab_cmd = remote_crontab_write_command(self.crontab_user) stdin, _, stderr = self.ssh_client.exec_command(crontab_cmd) stdin.write(new_crontab_content) stdin.channel.shutdown_write() diff --git a/tests/CronCommand_test.py b/tests/CronCommand_test.py new file mode 100644 index 0000000..fbb4135 --- /dev/null +++ b/tests/CronCommand_test.py @@ -0,0 +1,53 @@ +import crontab as crontab_module + +from cronboard_widgets import CronCommand + + +def test_get_local_crontab_command_prefers_crontab(mocker): + mocker.patch( + "cronboard_widgets.CronCommand.which", + side_effect=lambda command: f"/usr/bin/{command}" + if command == "crontab" + else None, + ) + + assert CronCommand.get_local_crontab_command() == "/usr/bin/crontab" + + +def test_get_local_crontab_command_uses_fcrontab_when_crontab_missing(mocker): + mocker.patch( + "cronboard_widgets.CronCommand.which", + side_effect=lambda command: f"/usr/bin/{command}" + if command == "fcrontab" + else None, + ) + + assert CronCommand.get_local_crontab_command() == "/usr/bin/fcrontab" + + +def test_create_user_crontab_uses_detected_command_and_restores_default(mocker): + mocker.patch( + "cronboard_widgets.CronCommand.get_local_crontab_command", + return_value="/usr/bin/fcrontab", + ) + cron_tab = mocker.patch("cronboard_widgets.CronCommand.CronTab") + original_command = crontab_module.CRON_COMMAND + + CronCommand.create_user_crontab() + + cron_tab.assert_called_once_with(user=True) + assert crontab_module.CRON_COMMAND == original_command + + +def test_remote_crontab_list_command_uses_crontab_with_fcrontab_fallback(): + assert CronCommand.remote_crontab_list_command() == ( + "if command -v crontab >/dev/null 2>&1; " + "then crontab -l; else fcrontab -l; fi" + ) + + +def test_remote_crontab_write_command_quotes_user(): + assert CronCommand.remote_crontab_write_command("cron user") == ( + "if command -v crontab >/dev/null 2>&1; " + "then crontab -u 'cron user' -; else fcrontab -u 'cron user' -; fi" + ) diff --git a/tests/CronDeleteConfirmation_test.py b/tests/CronDeleteConfirmation_test.py index 994f20a..58d1356 100644 --- a/tests/CronDeleteConfirmation_test.py +++ b/tests/CronDeleteConfirmation_test.py @@ -1,5 +1,6 @@ import pytest from cronboard_widgets.CronDeleteConfirmation import CronDeleteConfirmation +from cronboard_widgets.CronCommand import remote_crontab_write_command from .conftest import create_event, create_job_and_cron, make_remote_command from cronboard.app import CronBoard from pytest_mock import MockerFixture @@ -75,7 +76,7 @@ def test_write_remote_crontab(mocker: MockerFixture): result = modal.write_remote_crontab() assert result is True - ssh_client.exec_command.assert_called_once_with("crontab -u root -") + ssh_client.exec_command.assert_called_once_with(remote_crontab_write_command("root")) stdin.write.assert_called_once_with("* * * * * echo hello") stdin.channel.shutdown_write.assert_called_once_with() @@ -92,4 +93,4 @@ def test_write_remote_crontab_error(mocker: MockerFixture): result = modal.write_remote_crontab() assert result is False - ssh_client.exec_command.assert_called_once_with("crontab -") + ssh_client.exec_command.assert_called_once_with(remote_crontab_write_command()) diff --git a/tests/conftest.py b/tests/conftest.py index 1e06782..1fe0a1e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -25,9 +25,7 @@ def app(mocker: MockerFixture): fake_cron = mocker.MagicMock() fake_cron.__iter__ = mocker.MagicMock(side_effect=lambda: iter([fake_job])) mocker.patch("cronboard_widgets.CronTable.CronTab", return_value=fake_cron) - mocker.patch( - "cronboard_widgets.CronDeleteConfirmation.CronTab", return_value=fake_cron - ) + mocker.patch("cronboard_widgets.CronCommand.CronTab", return_value=fake_cron) yield CronBoard()