docker-images/python-tools/tests/test_web.py

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
    assert b"data-execution-mode" in page.data
    assert b'class="split-button"' 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."}