Restore generic Python tools documentation and tests

python-tools/tests/test_discovery.py

@@ -0,0 +1,217 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from app.discovery import Group, TaskCatalog
+
+from .conftest import write_task
+
+
+def test_discovers_only_direct_non_hidden_python_tasks(tmp_path: Path) -> None:
+    root = tmp_path / "tasks"
+    write_task(root, "media_tasks", "check_one.py", '"""Checks one thing."""\n')
+    write_task(root, "media_tasks", "with-hyphen.py", '"""Keeps its hyphen."""\n')
+    write_task(root, "space group", "already named.py", '"""Keeps spaces."""\n')
+    write_task(root, "media_tasks", "internal/private.py", '"""Not exposed."""\n')
+    write_task(root, "media_tasks", "_hidden.py", '"""Not exposed."""\n')
+    (root / "_hidden_group").mkdir(parents=True)
+    (root / "media_tasks" / "notes.txt").write_text("not a task", encoding="utf-8")
+
+    catalog = TaskCatalog(root)
+
+    groups = catalog.groups()
+
+    assert [group.id for group in groups] == ["media_tasks", "space group"]
+    assert groups[0].display_name == "media tasks"
+    assert [task.filename for task in groups[0].tasks] == ["check_one.py", "with-hyphen.py"]
+    assert [task.display_name for task in groups[0].tasks] == ["check one", "with-hyphen"]
+    assert groups[1].tasks[0].display_name == "already named"
+    assert groups[0].tasks[0].description == "Checks one thing."
+
+
+def test_ignores_symlinked_groups_and_tasks(tmp_path: Path) -> None:
+    root = tmp_path / "tasks"
+    external = tmp_path / "external"
+    write_task(external, "outside", "outside.py", '"""Outside."""\n')
+    root.mkdir()
+    (root / "linked_group").symlink_to(external / "outside", target_is_directory=True)
+    group = root / "safe"
+    group.mkdir()
+    (group / "linked.py").symlink_to(external / "outside" / "outside.py")
+
+    assert TaskCatalog(root).groups() == (
+        Group(id="safe", path=group, display_name="safe", tasks=()),
+    )
+
+
+def test_loads_typed_inputs_from_group_manifest(tmp_path: Path) -> None:
+    root = tmp_path / "tasks"
+    group = root / "examples"
+    write_task(root, "examples", "queue.py", '"""Queues work."""\n')
+    (group / "task-inputs.json").write_text(
+        """{
+          "version": 1,
+          "tasks": {
+            "queue.py": {
+              "inputs": [
+                {"name": "identifier", "label": "Identifier", "type": "text", "required": true, "pattern": "[A-Z]{2}[0-9]+"},
+                {"name": "retries", "label": "Retries", "type": "integer", "minimum": 1, "maximum": 5, "default": 2},
+                {"name": "source", "label": "Source", "type": "choice", "options": [{"value": "api", "label": "Service API"}]},
+                {"name": "alerts", "label": "Alerts", "type": "multi_choice", "options": ["email", "webhook"], "default": ["email"]}
+              ],
+              "wait_for_result": true
+            }
+          }
+        }""",
+        encoding="utf-8",
+    )
+
+    task = TaskCatalog(root).task("examples/queue.py")
+
+    assert task is not None
+    assert [(field.name, field.type, field.default) for field in task.inputs] == [
+        ("identifier", "text", None),
+        ("retries", "integer", 2),
+        ("source", "choice", None),
+        ("alerts", "multi_choice", ("email",)),
+    ]
+    assert task.inputs[2].options[0].label == "Service API"
+    assert task.wait_for_result is True
+
+
+def test_loads_sensitive_input_from_group_manifest(tmp_path: Path) -> None:
+    root = tmp_path / "tasks"
+    group = root / "examples"
+    write_task(root, "examples", "prepare.py", '"""Prepares."""\n')
+    (group / "task-inputs.json").write_text(
+        '''{"version":1,"tasks":{"prepare.py":{"inputs":[{"name":"secret","label":"Secret","type":"text","sensitive":true}]}}}''',
+        encoding="utf-8",
+    )
+
+    task = TaskCatalog(root).task("examples/prepare.py")
+
+    assert task is not None
+    assert task.inputs[0].sensitive is True
+
+
+def test_manifest_orders_listed_tasks_and_allows_empty_declarations(tmp_path: Path) -> None:
+    root = tmp_path / "tasks"
+    group = root / "tools"
+    write_task(root, "tools", "apply.py", '"""Applies."""\n')
+    write_task(root, "tools", "review.py", '"""Reviews."""\n')
+    write_task(root, "tools", "status.py", '"""Shows status."""\n')
+    (group / "task-inputs.json").write_text(
+        '''{
+          "version": 1,
+          "tasks": {
+            "review.py": {},
+            "apply.py": {"inputs": []}
+          }
+        }''',
+        encoding="utf-8",
+    )
+
+    tasks = TaskCatalog(root).group("tools")
+
+    assert tasks is not None
+    assert [task.filename for task in tasks.tasks] == ["review.py", "apply.py", "status.py"]
+
+
+def test_rejects_manifest_for_unknown_task(tmp_path: Path) -> None:
+    root = tmp_path / "tasks"
+    group = root / "examples"
+    write_task(root, "examples", "queue.py", '"""Queues work."""\n')
+    (group / "task-inputs.json").write_text(
+        '{"version":1,"tasks":{"missing.py":{"inputs":[]}}}',
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="unknown tasks: missing.py"):
+        TaskCatalog(root).groups()
+
+
+def test_discovers_only_declared_internal_background_tasks(tmp_path: Path) -> None:
+    root = tmp_path / "tasks"
+    group = root / "examples"
+    write_task(root, "examples", "visible.py", '"""Visible."""\n')
+    write_task(root, "examples", "internal/poll.py", '"""Hidden."""\n')
+    (group / "background-tasks.json").write_text(
+        '''{
+          "version": 1,
+          "tasks": [
+            {"id": "poll", "script": "internal/poll.py", "interval_seconds": 60}
+          ]
+        }''',
+        encoding="utf-8",
+    )
+
+    tasks = TaskCatalog(root).background_tasks()
+
+    assert [(task.id, task.group_id, task.path.name, task.interval_seconds) for task in tasks] == [
+        ("examples/poll", "examples", "poll.py", 60)
+    ]
+
+
+def test_loads_file_inputs_and_download_artifact_behavior(tmp_path: Path) -> None:
+    root = tmp_path / "tasks"
+    group = root / "reviews"
+    write_task(root, "reviews", "export.py", '"""Exports a review."""\n')
+    (group / "task-inputs.json").write_text(
+        '''{
+          "version": 1,
+          "tasks": {
+            "export.py": {
+              "wait_for_result": true,
+              "download_artifacts": true,
+              "inputs": [
+                {"name": "review_csv", "label": "Review CSV", "type": "file", "required": true, "accept": [".csv"], "maximum_bytes": 1024}
+              ]
+            }
+          }
+        }''',
+        encoding="utf-8",
+    )
+
+    task = TaskCatalog(root).task("reviews/export.py")
+
+    assert task is not None
+    assert task.wait_for_result is True
+    assert task.download_artifacts is True
+    assert task.inputs[0].accept == (".csv",)
+    assert task.inputs[0].maximum_bytes == 1024
+
+
+def test_execution_declaration_generates_an_action_choice(tmp_path: Path) -> None:
+    root = tmp_path / "tasks"
+    group = root / "examples"
+    write_task(root, "examples", "migrate.py", '"""Migrates settings."""\n')
+    (group / "task-inputs.json").write_text(
+        '''{
+          "version": 1,
+          "tasks": {
+            "migrate.py": {
+              "execution": {
+                "field": "mode",
+                "label": "Action",
+                "dry_run_value": "report",
+                "dry_run_label": "Report only",
+                "execute_value": "apply",
+                "execute_label": "Apply changes"
+              }
+            }
+          }
+        }''',
+        encoding="utf-8",
+    )
+
+    task = TaskCatalog(root).task("examples/migrate.py")
+
+    assert task is not None
+    assert task.execution is not None
+    assert [(field.name, field.default) for field in task.inputs] == [("mode", "report")]
+    assert [(option.value, option.label) for option in task.inputs[0].options] == [
+        ("report", "Report only"),
+        ("apply", "Apply changes"),
+    ]