List tasks

src/hipercow/cli.py

@@ -1,7 +1,7 @@
 import click
 
 from hipercow import root
-from hipercow.task import task_status
+from hipercow.task import task_list, task_status
 from hipercow.task_create import task_create
 from hipercow.task_eval import task_eval
 
@@ -22,24 +22,31 @@ def task():
     pass  # pragma: no cover
 
 
-@task.command()
+@task.command("status")
 @click.argument("task_id")
-def status(task_id: str):
+def cli_task_status(task_id: str):
     r = root.open_root()
     click.echo(task_status(r, task_id))
 
 
-@task.command()
+@task.command("list")
+def cli_task_list():
+    r = root.open_root()
+    for task_id in task_list(r):
+        click.echo(task_id)
+
+
+@task.command("create")
 @click.argument("cmd", nargs=-1)
-def create(cmd: tuple[str]):
+def cli_task_create(cmd: tuple[str]):
     r = root.open_root()
     data = {"cmd": list(cmd)}
     task_id = task_create(r, "shell", data, {})
     click.echo(task_id)
 
 
-@task.command()
+@task.command("eval")
 @click.argument("task_id")
-def eval(task_id: str):
+def cli_task_eval(task_id: str):
     r = root.open_root()
     task_eval(r, task_id)

src/hipercow/task.py

@@ -90,3 +90,8 @@ def task_data_write(root: Root, data: TaskData) -> None:
 def task_data_read(root: Root, task_id: str) -> TaskData:
     with root.path_task_data(task_id).open("rb") as f:
         return pickle.load(f)
+
+
+def task_list(root: Root) -> list[str]:
+    contents = (root.path / "tasks").rglob("data")
+    return ["".join(el.parts[-3:-1]) for el in contents if el.is_file()]

tests/test_cli.py

@@ -20,7 +20,7 @@ def test_can_create_task(tmp_path):
     with runner.isolated_filesystem(temp_dir=tmp_path):
         root.init(".")
         r = root.open_root()
-        res = runner.invoke(cli.create, ["echo", "hello", "world"])
+        res = runner.invoke(cli.cli_task_create, ["echo", "hello", "world"])
         assert res.exit_code == 0
         task_id = res.stdout.strip()
         assert task.task_status(r, task_id) == task.TaskStatus.CREATED
@@ -31,19 +31,33 @@ def test_can_run_task(tmp_path):
     with runner.isolated_filesystem(temp_dir=tmp_path):
         root.init(".")
         r = root.open_root()
-        res = runner.invoke(cli.create, ["echo", "hello", "world"])
+        res = runner.invoke(cli.cli_task_create, ["echo", "hello", "world"])
         assert res.exit_code == 0
         task_id = res.stdout.strip()
 
-        res = runner.invoke(cli.status, task_id)
+        res = runner.invoke(cli.cli_task_status, task_id)
         assert res.exit_code == 0
         assert res.output.strip() == "created"
 
-        res = runner.invoke(cli.eval, task_id)
+        res = runner.invoke(cli.cli_task_eval, task_id)
         assert res.exit_code == 0
         # It would be good to test that we get the expected output
         # here, and empirically we do.  However we don't seem to get
         # it captured in the runner output, though it is swallowed by
         # something.  I've checked with the capsys fixture and that
         # does not seem to have it either.
         assert task.task_status(r, task_id) == task.TaskStatus.SUCCESS
+
+
+def test_can_list_tasks(tmp_path):
+    runner = CliRunner()
+    with runner.isolated_filesystem(temp_dir=tmp_path):
+        root.init(".")
+        r = root.open_root()
+        res = runner.invoke(cli.cli_task_create, ["echo", "hello", "world"])
+        assert res.exit_code == 0
+        task_id = res.stdout.strip()
+
+        res = runner.invoke(cli.cli_task_list, [])
+        assert res.exit_code == 0
+        assert res.output.strip() == task_id

tests/test_task.py

@@ -1,6 +1,6 @@
 from hipercow import root
 from hipercow import task_create as tc
-from hipercow.task import TaskStatus, set_task_status, task_status
+from hipercow.task import TaskStatus, set_task_status, task_list, task_status
 from hipercow.util import transient_working_directory
 
 
@@ -35,3 +35,15 @@ def test_that_missing_tasks_have_missing_status(tmp_path):
 
 def test_can_convert_to_nice_string():
     assert str(TaskStatus.CREATED) == "created"
+
+
+def test_can_list_tasks(tmp_path):
+    root.init(tmp_path)
+    r = root.open_root(tmp_path)
+    assert task_list(r) == []
+    with transient_working_directory(tmp_path):
+        t1 = tc.task_create_shell(["echo", "hello world"])
+    assert task_list(r) == [t1]
+    with transient_working_directory(tmp_path):
+        t2 = tc.task_create_shell(["echo", "hello world"])
+    assert set(task_list(r)) == {t1, t2}