diff --git a/python-tools/README.md b/python-tools/README.md new file mode 100644 index 0000000..bc18b0a --- /dev/null +++ b/python-tools/README.md @@ -0,0 +1,117 @@ +# Python Tools + +This image is a Python 3 task runner with ExifTool and a small web control panel +for running known, mounted Python maintenance tasks. It starts the control panel +with `python -m app`. + +## Control Panel + +The application discovers direct `*.py` files in each non-hidden directory +under `/opt/tasks`. Each directory is a task group. Tasks in one group run +sequentially; different groups may run concurrently. It records run history and +combined output in SQLite under `/var/lib/server-maintenance`, and supports +cancellation of active task process groups. + +The image contains application code only. Mount these paths at runtime: + +- `/opt/tasks` read-only: reviewed task scripts and their input manifests. +- `/opt/credentials` read-only: named credential files used by those scripts. +- `/var/lib/server-maintenance`: persistent SQLite data, logs, and task state. + +Task processes receive a minimal environment. To forward non-secret container +settings to tasks, list their names in `SERVER_MAINTENANCE_TASK_ENV`, separated +by commas. This allowlist prevents unrelated container configuration from +being exposed to every task; credentials should remain in `/opt/credentials`. + +Tasks are never imported during discovery. Their optional description is the +first line of their module docstring. A task must not require stdin or command +line arguments, should write useful output, and must return a non-zero exit +code on failure. + +## Task inputs + +An optional `task-inputs.json` beside a group's task files declares the fields +for its web forms and its task order. It is parsed as data only and uses version +`1`: + +```json +{ + "version": 1, + "tasks": { + "task.py": { + "inputs": [ + {"name": "identifier", "label": "Identifier", "type": "text", "required": true}, + {"name": "attempts", "label": "Attempts", "type": "integer", "minimum": 1, "maximum": 5, "default": 2}, + {"name": "mode", "label": "Mode", "type": "choice", "options": ["standard", "extended"]}, + {"name": "alerts", "label": "Alerts", "type": "multi_choice", "options": ["email", "webhook"]} + ] + } + } +} +``` + +Supported field types are `text`, `integer`, `date`, `datetime`, `choice`, +`multi_choice`, and `file`. Text may also define a regular-expression +`pattern`; integer fields accept `minimum`, `maximum`, and `step`. Choices may +use strings or objects with `value` and `label`. File fields accept a list of +lowercase filename extensions in `accept` and a byte limit in `maximum_bytes`. +An input can define `required` and `default`. +Set `sensitive` to `true` for a value or upload that must not appear in run +history. The form masks sensitive text, the database records `[redacted]`, and +the original value is available only through a private execution input file. +Tasks listed in the manifest appear in JSON order. A listed task may use `{}` +when it only needs ordering; unlisted scripts appear afterward in alphabetical +order. +Set `wait_for_result` to `true` for a short task whose form should wait for its +final success or failure status. This disables the form and warns before +leaving the page while the task is running; tasks without it keep the normal +queued-run behavior. + +For a task with a review-first operation, declare `execution` with the input +field and its review and execution values. The control panel generates the +dropdown. A successful review run offers a button that repeats the same saved +inputs using the execution value; it does not reuse a stale filesystem +snapshot, so the task must still validate its inputs before changing anything. + +The server validates every submitted value before a run is queued. It records +the validated object in the run database and writes it to a per-run file. +Sensitive fields are redacted in the database and use a private `0600` +execution file instead. Tasks read the applicable file path from +`SERVER_MAINTENANCE_INPUT`; no user value is appended to the command line. A +task without an entry in the manifest gets an empty JSON object. Uploaded files +are stored in a private per-run directory; their absolute paths are provided in +the input JSON and the files are removed when the task finishes, fails, or is +cancelled. + +## Downloads + +Tasks can write downloadable files to the directory named by +`SERVER_MAINTENANCE_ARTIFACTS`. Set `download_artifacts` to `true` alongside +`wait_for_result` to start the first artifact download after a successful web +run. Run-detail pages also show available downloads. The server deletes an +artifact after its HTTP transfer ends; any undownloaded artifact is removed +after 24 hours. A browser cannot confirm that a downloaded file was retained +on the client device. + +## Background tasks + +An optional `background-tasks.json` beside a group's task files declares +internal recurring scripts. These scripts must live below the group's +`internal/` directory, are not exposed in the UI, and are launched at most once +per configured interval. Their latest output is retained below the persistent +state directory rather than being added to normal run history. + +```json +{ + "version": 1, + "tasks": [ + {"id": "poll", "script": "internal/poll.py", "interval_seconds": 60} + ] +} +``` + +`compose.example.yaml` is suitable for adding the service to an existing Compose +stack. Set `PYTHON_TOOLS_VOLUME_ROOT` to the host directory that will hold +`state`, `tasks`, and `credentials`; set +`SERVER_MAINTENANCE_ALLOWED_PROXY_IPS` to the reverse proxy's direct peer +address. Keep the application behind an authenticated reverse proxy. diff --git a/python-tools/compose.example.yaml b/python-tools/compose.example.yaml new file mode 100644 index 0000000..7aaeea7 --- /dev/null +++ b/python-tools/compose.example.yaml @@ -0,0 +1,19 @@ +services: + python-tools: + container_name: python-tools + image: git.ajpanton.se/ajp_anton/python-tools:3 + user: "${PYTHON_TOOLS_UID:-1000}:${PYTHON_TOOLS_GID:-1000}" + ports: + - "${PYTHON_TOOLS_PORT:-8080}:8080" + environment: + SERVER_MAINTENANCE_ALLOWED_PROXY_IPS: "${SERVER_MAINTENANCE_ALLOWED_PROXY_IPS:?Set the reverse proxy's direct peer address}" + # Optional task settings. Only names listed here are forwarded to task + # processes; keep credentials in the mounted credentials directory. + # SERVER_MAINTENANCE_TASK_ENV: TASK_SETTING_A,TASK_SETTING_B + volumes: + - /etc/localtime:/etc/localtime:ro + - ${PYTHON_TOOLS_VOLUME_ROOT:?Set the persistent volume root}/state:/var/lib/server-maintenance + - ${PYTHON_TOOLS_VOLUME_ROOT:?Set the persistent volume root}/tasks:/opt/tasks:ro + - ${PYTHON_TOOLS_VOLUME_ROOT:?Set the persistent volume root}/credentials:/opt/credentials:ro + # Add task-specific bind mounts here when required. + restart: always diff --git a/python-tools/tests/test_discovery.py b/python-tools/tests/test_discovery.py new file mode 100644 index 0000000..364976a --- /dev/null +++ b/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"), + ] diff --git a/python-tools/tests/test_runner.py b/python-tools/tests/test_runner.py new file mode 100644 index 0000000..023ddf9 --- /dev/null +++ b/python-tools/tests/test_runner.py @@ -0,0 +1,340 @@ +from __future__ import annotations + +import time +from dataclasses import replace +from io import BytesIO +from pathlib import Path + +from app.discovery import TaskCatalog +from app.runner import RunManager +from app.inputs import PendingUpload +from app.scheduler import BackgroundScheduler + +from .conftest import write_task + + +def wait_for(manager: RunManager, run_id: int, statuses: set[str], timeout: float = 3) -> str: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + run = manager.store.get(run_id) + assert run is not None + if run.status in statuses: + return run.status + time.sleep(0.01) + raise AssertionError(f"run {run_id} did not reach {statuses}") + + +def wait_for_output(path: Path, expected: str, timeout: float = 3) -> None: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if path.exists() and expected in path.read_text(): + return + time.sleep(0.01) + raise AssertionError(f"{expected!r} was not written to {path}") + + +SLEEP_TASK = '''"""Sleeps after writing a marker.""" +import os +import time +from pathlib import Path + +state = Path(os.environ["SERVER_MAINTENANCE_STATE"]) +state.mkdir(parents=True, exist_ok=True) +(state / "events.txt").open("a").write(f"start {Path(__file__).name}\\n") +time.sleep(0.25) +(state / "events.txt").open("a").write(f"end {Path(__file__).name}\\n") +''' + + +def test_tasks_in_one_group_are_serialized_and_groups_overlap(manager: RunManager) -> None: + root = manager.settings.task_root + write_task(root, "first_group", "first.py", SLEEP_TASK) + write_task(root, "first_group", "second.py", SLEEP_TASK) + write_task(root, "second_group", "other.py", SLEEP_TASK) + catalog = TaskCatalog(root) + + first_task = catalog.task("first_group/first.py") + second_task = catalog.task("first_group/second.py") + other_task = catalog.task("second_group/other.py") + assert first_task is not None and second_task is not None and other_task is not None + first, _ = manager.run_task(first_task) + second, _ = manager.run_task(second_task) + other, _ = manager.run_task(other_task) + + wait_for(manager, first.id, {"running"}) + wait_for(manager, other.id, {"running"}) + assert manager.store.get(second.id).status == "queued" + + assert wait_for(manager, first.id, {"succeeded"}) == "succeeded" + assert wait_for(manager, second.id, {"succeeded"}) == "succeeded" + assert wait_for(manager, other.id, {"succeeded"}) == "succeeded" + events = (manager.settings.state_root / "tasks" / "first_group" / "events.txt").read_text() + assert events.splitlines() == [ + "start first.py", + "end first.py", + "start second.py", + "end second.py", + ] + + +def test_group_stops_after_failed_task(manager: RunManager) -> None: + root = manager.settings.task_root + write_task(root, "workflow", "01_ok.py", '"""Works."""\nprint("ok")\n') + write_task(root, "workflow", "02_fail.py", '"""Fails."""\nraise SystemExit(3)\n') + write_task(root, "workflow", "03_never.py", '"""Must not run."""\nraise SystemExit(4)\n') + group = TaskCatalog(root).group("workflow") + assert group is not None + + parent, _ = manager.run_group(group) + + assert wait_for(manager, parent.id, {"failed"}) == "failed" + children = manager.store.children(parent.id) + assert [(child.target_path, child.status, child.exit_code) for child in children] == [ + ("workflow/01_ok.py", "succeeded", 0), + ("workflow/02_fail.py", "failed", 3), + ] + + +def test_duplicate_task_request_returns_the_existing_run(manager: RunManager) -> None: + root = manager.settings.task_root + write_task( + root, + "long_task", + "wait.py", + '"""Waits."""\nimport time\ntime.sleep(30)\n', + ) + task = TaskCatalog(root).task("long_task/wait.py") + assert task is not None + + first, first_created = manager.run_task(task) + second, second_created = manager.run_task(task) + + assert first_created is True + assert second_created is False + assert second.id == first.id + manager.request_cancel(first.id) + assert wait_for(manager, first.id, {"cancelled"}) == "cancelled" + + +def test_cancellation_signals_the_task_process_group(manager: RunManager) -> None: + root = manager.settings.task_root + write_task( + root, + "long_task", + "wait.py", + '"""Waits."""\nimport time\nprint("started", flush=True)\ntime.sleep(30)\n', + ) + task = TaskCatalog(root).task("long_task/wait.py") + assert task is not None + run, _ = manager.run_task(task) + + wait_for(manager, run.id, {"running"}) + wait_for_output(manager.settings.state_root / "runs" / f"{run.id}.log", "started") + manager.request_cancel(run.id) + + assert wait_for(manager, run.id, {"cancelled"}) == "cancelled" + completed = manager.store.get(run.id) + assert completed is not None + assert completed.cancel_requested_at is not None + assert completed.log_path is not None + assert "started" in Path(completed.log_path).read_text() + + +def test_task_receives_validated_input_as_json_file(manager: RunManager) -> None: + root = manager.settings.task_root + write_task( + root, + "input_group", + "read_input.py", + '''"""Reads supplied input.""" +import json +import os +from pathlib import Path + +state = Path(os.environ["SERVER_MAINTENANCE_STATE"]) +data = json.loads(Path(os.environ["SERVER_MAINTENANCE_INPUT"]).read_text()) +(state / "input.json").write_text(json.dumps(data, sort_keys=True)) +''', + ) + task = TaskCatalog(root).task("input_group/read_input.py") + assert task is not None + + run, _ = manager.run_task(task, {"identifier": "AB123", "attempts": 2}) + + assert wait_for(manager, run.id, {"succeeded"}) == "succeeded" + assert (manager.settings.state_root / "tasks" / "input_group" / "input.json").read_text() == ( + '{"attempts": 2, "identifier": "AB123"}' + ) + stored = manager.store.get(run.id) + assert stored is not None + assert stored.input_json == '{"attempts":2,"identifier":"AB123"}' + + +def test_task_receives_only_explicitly_forwarded_environment( + manager: RunManager, + monkeypatch, +) -> None: + monkeypatch.setenv("SERVER_MAINTENANCE_TASK_ENV", "TASK_SETTING") + monkeypatch.setenv("TASK_SETTING", "available") + monkeypatch.setenv("UNRELATED_SETTING", "hidden") + root = manager.settings.task_root + write_task( + root, + "environment", + "read_environment.py", + '''"""Reads forwarded configuration.""" +import os +from pathlib import Path + +state = Path(os.environ["SERVER_MAINTENANCE_STATE"]) +(state / "environment.txt").write_text( + f'{os.environ.get("TASK_SETTING", "")}:{os.environ.get("UNRELATED_SETTING", "")}') +''', + ) + task = TaskCatalog(root).task("environment/read_environment.py") + assert task is not None + + run, _ = manager.run_task(task) + + assert wait_for(manager, run.id, {"succeeded"}) == "succeeded" + assert (manager.settings.state_root / "tasks" / "environment" / "environment.txt").read_text() == ( + "available:" + ) + + +def test_sensitive_input_is_redacted_and_removed_after_execution(manager: RunManager) -> None: + root = manager.settings.task_root + task_path = write_task( + root, + "input_group", + "read_secret.py", + '''"""Reads a sensitive input.""" +import json +import os +from pathlib import Path + +state = Path(os.environ["SERVER_MAINTENANCE_STATE"]) +data = json.loads(Path(os.environ["SERVER_MAINTENANCE_INPUT"]).read_text()) +(state / "secret.txt").write_text(data["source_url"]) +''', + ) + (task_path.parent / "task-inputs.json").write_text( + '''{ + "version": 1, + "tasks": { + "read_secret.py": { + "inputs": [ + {"name": "source_url", "label": "Source URL", "type": "text", "sensitive": true} + ] + } + } + }''', + encoding="utf-8", + ) + task = TaskCatalog(root).task("input_group/read_secret.py") + assert task is not None + + run, _ = manager.run_task(task, {"source_url": "https://example.test/?passkey=secret"}) + + assert wait_for(manager, run.id, {"succeeded"}) == "succeeded" + assert (manager.settings.state_root / "tasks" / "input_group" / "secret.txt").read_text() == ( + "https://example.test/?passkey=secret" + ) + stored = manager.store.get(run.id) + assert stored is not None + assert stored.input_json == '{"source_url":"[redacted]"}' + assert not (manager.settings.state_root / "runs" / f"{run.id}.execution-input.json").exists() + + +def test_declared_background_task_runs_without_creating_run_history(manager: RunManager) -> None: + root = manager.settings.task_root + task = write_task( + root, + "background", + "internal/poll.py", + '"""Internal."""\nprint("background complete")\n', + ) + (task.parent.parent / "background-tasks.json").write_text( + '''{ + "version": 1, + "tasks": [ + {"id": "poll", "script": "internal/poll.py", "interval_seconds": 60} + ] + }''', + encoding="utf-8", + ) + scheduler = BackgroundScheduler(manager.catalog, manager) + scheduler.start() + + log_path = manager.settings.state_root / "background" / "background" / "poll.log" + wait_for_output(log_path, "background complete") + scheduler.stop() + + assert manager.store.history("task", "background/poll") == [] + + +def test_uploaded_file_is_available_to_task_then_removed(manager: RunManager) -> None: + root = manager.settings.task_root + write_task( + root, + "imports", + "read_csv.py", + '''"""Reads an uploaded CSV.""" +import json +import os +from pathlib import Path + +state = Path(os.environ["SERVER_MAINTENANCE_STATE"]) +values = json.loads(Path(os.environ["SERVER_MAINTENANCE_INPUT"]).read_text()) +(state / "csv.txt").write_bytes(Path(values["review_csv"]).read_bytes()) +''', + ) + task = TaskCatalog(root).task("imports/read_csv.py") + assert task is not None + + run, created = manager.run_task( + task, + uploads={ + "review_csv": PendingUpload( + field_name="review_csv", + extension=".csv", + maximum_bytes=1024, + stream=BytesIO(b"answer,left_cluster_id\nsame,cluster\n"), + ) + }, + ) + + assert created is True + assert wait_for(manager, run.id, {"succeeded"}) == "succeeded" + assert (manager.settings.state_root / "tasks" / "imports" / "csv.txt").read_bytes() == ( + b"answer,left_cluster_id\nsame,cluster\n" + ) + assert not (manager.settings.state_root / "runs" / f"{run.id}.uploads").exists() + + +def test_upload_is_removed_when_task_process_cannot_start(manager: RunManager) -> None: + root = manager.settings.task_root + write_task(root, "imports", "apply.py", '\"\"\"Applies a CSV.\"\"\"\n') + task = TaskCatalog(root).task("imports/apply.py") + assert task is not None + broken_manager = RunManager( + replace(manager.settings, python_executable="/missing/python"), + TaskCatalog(root), + manager.store, + ) + + run, created = broken_manager.run_task( + task, + uploads={ + "review_csv": PendingUpload( + field_name="review_csv", + extension=".csv", + maximum_bytes=1024, + stream=BytesIO(b"answer,left_cluster_id\n"), + ) + }, + ) + + assert created is True + assert wait_for(broken_manager, run.id, {"failed"}) == "failed" + assert not (broken_manager.settings.state_root / "runs" / f"{run.id}.uploads").exists() diff --git a/python-tools/tests/test_web.py b/python-tools/tests/test_web.py new file mode 100644 index 0000000..8b6edd9 --- /dev/null +++ b/python-tools/tests/test_web.py @@ -0,0 +1,320 @@ +from __future__ import annotations + +from io import BytesIO +import time + +from app.web import create_app + +from .conftest import write_task + + +def test_index_discovers_task_and_run_endpoint(settings) -> None: + write_task( + settings.task_root, + "status_checks", + "show_status.py", + '"""Shows status."""\nprint("healthy")\n', + ) + app = create_app(settings) + client = app.test_client() + + page = client.get("/") + assert page.status_code == 200 + assert b"status checks" in page.data + assert b"Shows status." in page.data + + response = client.post("/tasks/status_checks/show_status.py/run") + assert response.status_code == 302 + assert "/runs/" in response.headers["Location"] + assert client.get(f"{response.headers['Location']}/state").status_code == 200 + + +def test_proxy_allowlist_rejects_other_direct_peers(settings) -> None: + restricted = settings.__class__( + **{**settings.__dict__, "allowed_proxy_ips": frozenset({"192.0.2.1"})} + ) + app = create_app(restricted) + + assert app.test_client().get("/").status_code == 403 + assert app.test_client().get("/", environ_base={"REMOTE_ADDR": "192.0.2.1"}).status_code == 200 + + +def test_task_form_validates_and_persists_typed_inputs(settings) -> None: + task = write_task( + settings.task_root, + "examples", + "queue.py", + '"""Queues work."""\n', + ) + (task.parent / "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": "days", "label": "Days", "type": "integer", "minimum": 1, "maximum": 7}, + {"name": "mode", "label": "Mode", "type": "choice", "required": true, "options": ["standard", "extended"]}, + {"name": "alerts", "label": "Alerts", "type": "multi_choice", "options": ["email", "webhook"]} + ] + } + } + }""", + encoding="utf-8", + ) + app = create_app(settings) + client = app.test_client() + + page = client.get("/") + assert b'type="number"' in page.data + assert b'type="checkbox"' in page.data + + invalid = client.post( + "/tasks/examples/queue.py/run", + data={"identifier": "bad", "days": "0", "mode": "other"}, + ) + assert invalid.status_code == 400 + assert b"required format" in invalid.data + assert b"at least 1" in invalid.data + assert b"available options" in invalid.data + + response = client.post( + "/tasks/examples/queue.py/run", + data={ + "identifier": "AB123", + "days": "2", + "mode": "standard", + "alerts": ["email", "webhook"], + }, + ) + assert response.status_code == 302 + run_id = int(response.headers["Location"].rsplit("/", 1)[1]) + run = app.extensions["run_store"].get(run_id) + assert run is not None + assert run.input_json == '{"alerts":["email","webhook"],"days":2,"identifier":"AB123","mode":"standard"}' + + +def test_sensitive_input_uses_a_password_field_and_is_redacted(settings) -> None: + task = write_task(settings.task_root, "examples", "prepare.py", '"""Prepares."""\n') + (task.parent / "task-inputs.json").write_text( + '''{ + "version": 1, + "tasks": { + "prepare.py": { + "inputs": [ + {"name": "secret", "label": "Secret", "type": "text", "sensitive": true} + ] + } + } + }''', + encoding="utf-8", + ) + app = create_app(settings) + client = app.test_client() + + assert b'name="secret"' in client.get("/").data + assert b'type="password"' in client.get("/").data + response = client.post("/tasks/examples/prepare.py/run", data={"secret": "private-value"}) + run_id = int(response.headers["Location"].rsplit("/", 1)[1]) + run = app.extensions["run_store"].get(run_id) + assert run is not None + assert run.input_json == '{"secret":"[redacted]"}' + + +def test_group_with_typed_tasks_cannot_run_as_a_group(settings) -> None: + task = write_task( + settings.task_root, + "examples", + "schedule.py", + '"""Schedules work."""\n', + ) + (task.parent / "task-inputs.json").write_text( + '''{ + "version": 1, + "tasks": { + "schedule.py": { + "inputs": [ + {"name": "identifier", "label": "Identifier", "type": "text"} + ] + } + } + }''', + encoding="utf-8", + ) + app = create_app(settings) + client = app.test_client() + + assert b"Run group" not in client.get("/").data + assert client.post("/groups/examples/run").status_code == 404 + + +def test_wait_for_result_task_uses_json_run_endpoint(settings) -> None: + task = write_task( + settings.task_root, + "examples", + "schedule.py", + '"""Schedules work."""\nprint("scheduled")\n', + ) + (task.parent / "task-inputs.json").write_text( + '''{ + "version": 1, + "tasks": { + "schedule.py": { + "wait_for_result": true, + "inputs": [ + {"name": "identifier", "label": "Identifier", "type": "text", "required": true} + ] + } + } + }''', + encoding="utf-8", + ) + app = create_app(settings) + client = app.test_client() + + assert b"data-wait-for-result" in client.get("/").data + + response = client.post( + "/tasks/examples/schedule.py/run", + data={"identifier": "AB123"}, + headers={"Accept": "application/json"}, + ) + + assert response.status_code == 202 + payload = response.get_json() + assert payload["detail_url"].startswith("/runs/") + assert payload["state_url"].endswith("/state") + + invalid = client.post( + "/tasks/examples/schedule.py/run", + data={}, + headers={"Accept": "application/json"}, + ) + assert invalid.status_code == 400 + assert invalid.get_json()["errors"] == {"identifier": "This field is required."} + + +def test_successful_dry_run_can_be_repeated_with_execution_enabled(settings) -> None: + task = write_task( + settings.task_root, + "examples", + "migrate.py", + '''"""Migrates settings.""" +import os +from pathlib import Path + +print(Path(os.environ["SERVER_MAINTENANCE_INPUT"]).read_text()) +''', + ) + (task.parent / "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", + ) + app = create_app(settings) + client = app.test_client() + + page = client.get("/") + assert b"Report only" in page.data + assert b"Apply changes" in page.data + response = client.post("/tasks/examples/migrate.py/run") + run_id = int(response.headers["Location"].rsplit("/", 1)[1]) + deadline = time.monotonic() + 3 + while time.monotonic() < deadline: + run = app.extensions["run_store"].get(run_id) + assert run is not None + if run.status == "succeeded": + break + time.sleep(0.01) + else: + raise AssertionError("dry-run task did not finish") + + detail = client.get(f"/runs/{run_id}") + assert b"Apply changes" in detail.data + execute = client.post(f"/runs/{run_id}/execute") + assert execute.status_code == 302 + executed_id = int(execute.headers["Location"].rsplit("/", 1)[1]) + executed = app.extensions["run_store"].get(executed_id) + assert executed is not None + assert executed.input_json == '{"mode":"apply"}' + + +def test_successful_task_artifact_can_be_downloaded_once(settings) -> None: + task = write_task( + settings.task_root, + "exports", + "review.py", + '''"""Creates a download.""" +import os +from pathlib import Path + +artifact = Path(os.environ["SERVER_MAINTENANCE_ARTIFACTS"]) / "review.zip" +artifact.write_bytes(b"archive") +print("archive created") +''', + ) + app = create_app(settings) + client = app.test_client() + + response = client.post("/tasks/exports/review.py/run") + run_id = int(response.headers["Location"].rsplit("/", 1)[1]) + deadline = time.monotonic() + 3 + while time.monotonic() < deadline: + run = app.extensions["run_store"].get(run_id) + assert run is not None + if run.status == "succeeded": + break + time.sleep(0.01) + else: + raise AssertionError("artifact task did not finish") + + assert client.get(f"/runs/{run_id}/output").get_json() == {"output": "archive created\n"} + assert b"archive created" in client.get(f"/runs/{run_id}").data + + listed = client.get(f"/runs/{run_id}/artifacts").get_json() + assert listed["artifacts"] == [{"name": "review.zip", "url": f"/runs/{run_id}/artifacts/review.zip"}] + download = client.get(listed["artifacts"][0]["url"]) + assert download.data == b"archive" + download.close() + assert app.extensions["run_manager"].artifact_path(run_id, "review.zip") is None + + +def test_file_input_rejects_the_wrong_extension(settings) -> None: + task = write_task(settings.task_root, "imports", "apply.py", '"""Applies a CSV."""\n') + (task.parent / "task-inputs.json").write_text( + '''{ + "version": 1, + "tasks": { + "apply.py": { + "inputs": [ + {"name": "review_csv", "label": "Review CSV", "type": "file", "required": true, "accept": [".csv"], "maximum_bytes": 1024} + ] + } + } + }''', + encoding="utf-8", + ) + app = create_app(settings) + client = app.test_client() + + response = client.post( + "/tasks/imports/apply.py/run", + data={"review_csv": (BytesIO(b"not csv"), "review.txt")}, + headers={"Accept": "application/json"}, + ) + + assert response.status_code == 400 + assert response.get_json()["errors"] == {"review_csv": "Choose a file with one of: .csv."}