#5 adds pa schedule update command. by: Piotr

Author: caseneuve

cli/schedule.py

@@ -1,3 +1,6 @@
+import logging
+import sys
+from datetime import datetime
 from typing import List
 
 import typer
@@ -232,5 +235,104 @@ def stringify_values(task, attr):
 
 
 @app.command()
-def update():
-    raise NotImplementedError
+def update(
+    task_id: int = typer.Argument(..., metavar="id"),
+    command: str = typer.Option(
+        None,
+        "-c",
+        "--command",
+        help="Changes command to COMMAND (multiword commands should be quoted)"
+    ),
+    hour: int = typer.Option(
+        None,
+        "-o",
+        "--hour",
+        min=0,
+        max=23,
+        help="Changes hour to HOUR (in 24h format)"
+    ),
+    minute: int = typer.Option(
+        None,
+        "-m",
+        "--minute",
+        min=0,
+        max=59,
+        help="Changes minute to MINUTE"
+    ),
+    disable: bool = typer.Option(False, "-d", "--disable", help="Disables task"),
+    enable: bool = typer.Option(False, "-e", "--enable", help="Enables task"),
+    toggle_enabled: bool = typer.Option(
+        False, "-t", "--toggle-enabled", help="Toggles enable/disable state"
+    ),
+    daily: bool = typer.Option(
+        False,
+        "-a",
+        "--daily",
+        help=(
+            "Switches interval to daily "
+            "(when --hour is not provided, sets it automatically to current hour)"
+        )
+    ),
+    hourly: bool = typer.Option(
+        False,
+        "-u",
+        "--hourly",
+        help="Switches interval to hourly (takes precedence over --hour, i.e. sets hour to None)"
+    ),
+    quiet: bool = typer.Option(False, "-q", "--quiet", help="Turns off messages"),
+    porcelain: bool = typer.Option(
+        False, "-p", "--porcelain", help="Prints message in easy-to-parse format"
+    ),
+):
+    """Update a scheduled task.
+
+    Note that logfile name will change after updating the task but it won't be
+    created until first execution of the task.
+    To change interval from hourly to daily use --daily flag and provide --hour.
+    When --daily flag is not accompanied with --hour, new hour for the task
+    will be automatically set to current hour.
+    When changing interval from daily to hourly --hour flag is ignored.
+
+    Example:
+    Change command for a scheduled task 42:
+
+        pa schedule update 42 --command "echo new command"
+
+    Change interval of the task 42 from hourly to daily to be run at 10 am:
+
+        pa schedule update 42 --hour 10
+
+    Change interval of the task 42 from daily to hourly and set new minute:
+
+        pa schedule update 42 --minute 13 --hourly"""
+
+    kwargs = {k: v for k, v in locals().items() if k != "task_id"}
+    logger = get_logger()
+
+    porcelain = kwargs.pop("porcelain")
+    if not kwargs.pop("quiet"):
+        logger.setLevel(logging.INFO)
+
+    if not any(kwargs.values()):
+        msg = "Nothing to update!"
+        logger.warning(msg if porcelain else snakesay(msg))
+        sys.exit(1)
+
+    if kwargs.pop("hourly"):
+        kwargs["interval"] = "hourly"
+    if kwargs.pop("daily"):
+        kwargs["hour"] = kwargs["hour"] if kwargs["hour"] else datetime.now().hour
+        kwargs["interval"] = "daily"
+
+    task = get_task_from_id(task_id)
+
+    enable_opt = [k for k in ["toggle_enabled", "disable", "enable"] if kwargs.pop(k)]
+    params = {k: v for k, v in kwargs.items() if v}
+    if enable_opt:
+        lookup = {"toggle_enabled": not task.enabled, "disable": False, "enable": True}
+        params.update({"enabled": lookup[enable_opt[0]]})
+
+    try:
+        task.update_schedule(params, porcelain=porcelain)
+    except Exception as e:
+        logger.warning(snakesay(str(e)))

tests/test_cli_schedule.py

@@ -273,3 +273,76 @@ def test_warns_when_wrong_format_provided(self, mocker, task_list):
         assert mock_tabulate.call_count == 0
         assert wrong_format not in tabulate_formats
         assert "Table format has to be one of" in result.stdout
+
+
+@pytest.mark.clischeduleupdate
+class TestUpdate:
+    def test_enables_task_and_sets_porcelain(self, mocker):
+        mock_task_from_id = mocker.patch("cli.schedule.get_task_from_id")
+
+        runner.invoke(app, ["update", "42", "--enable", "--porcelain"])
+
+        assert mock_task_from_id.call_args == call(42)
+        assert mock_task_from_id.return_value.method_calls == [
+            call.update_schedule({"enabled": True}, porcelain=True)
+        ]
+
+    def test_turns_off_snakesay(self, mocker):
+        mock_logger = mocker.patch("cli.schedule.get_logger")
+
+        runner.invoke(app, ["update", "42", "--quiet"])
+
+        assert mock_logger.return_value.setLevel.call_count == 0
+
+    def test_warns_when_task_update_schedule_raises(self, mocker):
+        mock_logger = mocker.patch("cli.schedule.get_logger")
+        mock_task_from_id = mocker.patch("cli.schedule.get_task_from_id")
+        mock_task_from_id.return_value.update_schedule.side_effect = Exception("error")
+        mock_snake = mocker.patch("cli.schedule.snakesay")
+
+        runner.invoke(app, ["update", "42", "--disable"])
+
+        assert mock_snake.call_args == call("error")
+        assert mock_logger.return_value.warning.call_args == call(mock_snake.return_value)
+
+    def test_ensures_proper_daily_params(self, mocker):
+        mock_task_from_id = mocker.patch("cli.schedule.get_task_from_id")
+
+        result = runner.invoke(app, ["update", "42", "--hourly"])
+
+        assert mock_task_from_id.return_value.update_schedule.call_args == call(
+            {"interval": "hourly"}, porcelain=False
+        )
+
+    def test_ensures_proper_hourly_params(self, mocker):
+        mock_task_from_id = mocker.patch("cli.schedule.get_task_from_id")
+        mock_datetime = mocker.patch("cli.schedule.datetime")
+
+        runner.invoke(app, ["update", "42", "--daily"])
+
+        assert mock_task_from_id.return_value.update_schedule.call_args == call(
+            {"interval": "daily", "hour": mock_datetime.now.return_value.hour},
+            porcelain=False
+        )
+
+    def test_validates_minute(self):
+        result = runner.invoke(app, ["update", "42", "--minute", "88"])
+        assert "88 is not in the valid range of 0 to 59" in result.stdout
+
+    def test_validates_hour(self):
+        result = runner.invoke(app, ["update", "42", "--daily", "--hour", "33"])
+        assert "33 is not in the valid range of 0 to 23" in result.stdout
+
+    def test_complains_when_no_id_provided(self):
+        result = runner.invoke(app, ["update"])
+        assert "Missing argument 'id'" in result.stdout
+
+    def test_exits_early_when_nothing_to_update(self, mocker):
+        mock_logger = mocker.patch("cli.schedule.get_logger").return_value
+        mock_snakesay = mocker.patch("cli.schedule.snakesay")
+
+        result = runner.invoke(app, ["update", "42"])
+
+        assert mock_snakesay.call_args == call("Nothing to update!")
+        assert mock_logger.warning.call_args == call(mock_snakesay.return_value)
+        assert result.exit_code == 1