Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
46 changes: 46 additions & 0 deletions src/cronboard_widgets/CronCommand.py
Original file line number Diff line number Diff line change
@@ -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("-")
7 changes: 2 additions & 5 deletions src/cronboard_widgets/CronCreator.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
PathDropdownItem,
)
from cron_descriptor import Options, ExpressionDescriptor
from cronboard_widgets.CronCommand import remote_crontab_write_command


CRON_ALIASES = {
Expand Down Expand Up @@ -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()
Expand Down
13 changes: 6 additions & 7 deletions src/cronboard_widgets/CronDeleteConfirmation.py
Original file line number Diff line number Diff line change
@@ -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]):
Expand All @@ -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
Expand Down Expand Up @@ -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()
Expand Down
27 changes: 10 additions & 17 deletions src/cronboard_widgets/CronTable.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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()

Expand Down Expand Up @@ -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()

Expand All @@ -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()

Expand Down Expand Up @@ -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()
Expand Down
53 changes: 53 additions & 0 deletions tests/CronCommand_test.py
Original file line number Diff line number Diff line change
@@ -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"
)
5 changes: 3 additions & 2 deletions tests/CronDeleteConfirmation_test.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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()

Expand All @@ -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())
4 changes: 1 addition & 3 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()


Expand Down