flowdeck/tests/test_automations.py

"""FlowDeck — v5.1.0 automations: rules engine (trigger + condition + action).

Covers migration v5, the automations CRUD API, the if-this-then-that engine
(condition matching + action execution on collection-page events), manual/button
runs, and the cron due logic.
"""
import os
import tempfile

import pytest
from conftest import login_test_client
from fastapi.testclient import TestClient


@pytest.fixture
def client():
    db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
    db_path = db_file.name
    db_file.close()

    os.environ["DATABASE_URL"] = f"sqlite:///{db_path}"
    os.environ["APP_SECRET_KEY"] = "test-secret-for-tests"
    os.environ["RATE_LIMIT_ENABLED"] = "false"
    from app.config import settings
    settings.database_url = f"sqlite:///{db_path}"

    from app.db import get_conn, init_db
    from app.main import app
    init_db()
    with get_conn() as conn:
        conn.execute("INSERT OR IGNORE INTO users (id, login, full_name, is_admin) VALUES (1, 'tester', 'Tester', 1)")
        conn.commit()

    tc = TestClient(app)
    yield login_test_client(tc)

    os.unlink(db_path)


def _make_collection(client, name="Autos DB"):
    r = client.post("/db/api", json={"name": name, "schema": [
        {"id": 1, "name": "Status", "prop_type": "select"},
        {"id": 2, "name": "Priority", "prop_type": "text"},
    ]})
    return r.json()["id"]


def _make_page(client, coll_id, title="Page", props=None):
    r = client.post(f"/db/{coll_id}/pages/api", json={"title": title, "properties": props or {}})
    return r.json()["id"]


# ── Migration v5 ──

def test_automations_tables_created_by_migration(client):
    from app.db import get_conn
    with get_conn() as conn:
        tables = {r["name"] for r in conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table'"
        ).fetchall()}
        assert "automations" in tables
        assert "automation_runs" in tables
        cols = {r[1] for r in conn.execute("PRAGMA table_info(automations)").fetchall()}
        assert {"trigger_type", "event", "condition_json", "actions_json", "enabled"} <= cols
        runs = {r[1] for r in conn.execute("PRAGMA table_info(automation_runs)").fetchall()}
        assert {"automation_id", "status", "detail", "trigger_source"} <= runs


# ── Condition matching (service) ──

def test_condition_matching_service():
    from app.services.automations import evaluate_conditions

    assert evaluate_conditions("[]", {"Status": "Done"}) is True
    assert evaluate_conditions('[{"property":"Status","op":"eq","value":"Done"}]', {"Status": "Done"}) is True
    assert evaluate_conditions('[{"property":"Status","op":"eq","value":"Done"}]', {"Status": "Todo"}) is False
    assert evaluate_conditions('[{"property":"Priority","op":"is_empty"}]', {"Status": "Done"}) is True
    assert evaluate_conditions('[{"property":"Priority","op":"is_not_empty"}]', {"Priority": "High"}) is True
    assert evaluate_conditions('[{"property":"Title","op":"contains","value":"urgent"}]', {"Title": "very urgent"}) is True
    # AND semantics
    assert evaluate_conditions(
        '[{"property":"Status","op":"eq","value":"Done"},{"property":"Priority","op":"neq","value":"Low"}]',
        {"Status": "Done", "Priority": "High"},
    ) is True
    assert evaluate_conditions(
        '[{"property":"Status","op":"eq","value":"Done"},{"property":"Priority","op":"neq","value":"Low"}]',
        {"Status": "Done", "Priority": "Low"},
    ) is False
    # changed only valid with before_props
    assert evaluate_conditions('[{"property":"Status","op":"changed"}]', {"Status": "Done"},
                               {"Status": "Todo"}) is True
    assert evaluate_conditions('[{"property":"Status","op":"changed"}]', {"Status": "Done"},
                               {"Status": "Done"}) is False


# ── CRUD API ──

def test_automation_crud(client):
    # create
    r = client.post("/workspace/automations", json={
        "name": "Notify when done",
        "trigger_type": "event",
        "event": "page.updated",
        "collection_id": None,
        "condition_json": '[{"property":"Status","op":"eq","value":"Done"}]',
        "actions_json": '[{"type":"notify","message":"Reacted to Done"}]',
    })
    assert r.status_code == 200, r.text
    aid = r.json()["id"]

    # list
    d = client.get("/workspace/automations").json()
    assert len(d["automations"]) == 1
    assert d["automations"][0]["name"] == "Notify when done"

    # get
    assert client.get(f"/workspace/automations/{aid}").json()["id"] == aid

    # update
    r = client.put(f"/workspace/automations/{aid}", json={
        "name": "Renamed",
        "trigger_type": "event",
        "event": "page.created",
        "collection_id": None,
        "condition_json": "[]",
        "actions_json": "[]",
    })
    assert r.status_code == 200
    assert client.get(f"/workspace/automations/{aid}").json()["name"] == "Renamed"

    # validation: name required
    r = client.post("/workspace/automations", json={"name": "  "})
    assert r.status_code == 400
    # validation: bad actions JSON
    r = client.post("/workspace/automations", json={
        "name": "Bad", "trigger_type": "event", "event": "page.created",
        "actions_json": "not json",
    })
    assert r.status_code == 400

    # delete
    assert client.delete(f"/workspace/automations/{aid}").status_code == 200
    # After delete: the app's HTML 404 handler redirects non-API GETs to /workspaces.
    from app.db import get_conn
    with get_conn() as conn:
        row = conn.execute("SELECT id FROM automations WHERE id=?", (aid,)).fetchone()
    assert row is None


def test_automation_scope_collection_filter(client):
    c1 = _make_collection(client, "C1")
    c2 = _make_collection(client, "C2")
    only_c1 = client.post("/workspace/automations", json={
        "name": "Only C1",
        "trigger_type": "event",
        "event": "page.created",
        "collection_id": c1,
        "condition_json": "[]",
        "actions_json": '[{"type":"notify","message":"page in C1"}]',
    }).json()["id"]

    p_c1 = _make_page(client, c1, props={"Status": "Todo"})
    _make_page(client, c2, props={"Status": "Todo"})

    from app.db import get_conn
    with get_conn() as conn:
        runs_c1 = conn.execute(
            "SELECT * FROM automation_runs WHERE automation_id=? AND status='fired'",
            (only_c1,),
        ).fetchall()
    # Only page in C1 should have triggered the automation.
    assert len(runs_c1) == 1
    assert runs_c1[0]["page_id"] == p_c1


# ── Event engine: fire on collection-page events ──

def test_event_trigger_set_property_action(client):
    coll = _make_collection(client)
    pid = _make_page(client, coll, props={"Status": "Todo"})

    client.post("/workspace/automations", json={
        "name": "Mark archived",
        "trigger_type": "event",
        "event": "page.updated",
        "collection_id": coll,
        "condition_json": '[{"property":"Status","op":"eq","value":"Archived"}]',
        "actions_json": '[{"type":"set_property","property":"Priority","value":"Low"}]',
    })

    assert client.put(f"/db/pages/{pid}/api", json={
        "properties": {"Status": "Archived", "Priority": "High"},
    }).status_code == 200

    from app.db import get_conn
    with get_conn() as conn:
        row = conn.execute(
            "SELECT property_values_json FROM collection_pages WHERE id=?", (pid,)
        ).fetchone()
        props = __import__("json").loads(row["property_values_json"])
    assert props["Priority"] == "Low"  # action overrode the value
    assert props["Status"] == "Archived"  # original value preserved

    # The action should not re-fire (DB-level write, not via API)
    with get_conn() as conn:
        runs = conn.execute(
            "SELECT COUNT(*) AS c FROM automation_runs WHERE status='fired'"
        ).fetchone()["c"]
    assert runs == 1


def test_event_trigger_skipped_when_condition_fails(client):
    coll = _make_collection(client)
    pid = _make_page(client, coll, props={"Status": "Todo"})
    client.post("/workspace/automations", json={
        "name": "Skip unless Done",
        "trigger_type": "event",
        "event": "page.updated",
        "collection_id": coll,
        "condition_json": '[{"property":"Status","op":"eq","value":"Done"}]',
        "actions_json": '[{"type":"notify","message":"done!"}]',
    })
    client.put(f"/db/pages/{pid}/api", json={"properties": {"Status": "Todo"}})
    from app.db import get_conn
    with get_conn() as conn:
        row = conn.execute(
            "SELECT status FROM automation_runs WHERE automation_id=(SELECT id FROM automations LIMIT 1)"
        ).fetchone()
    assert row["status"] == "skipped"


def test_page_deleted_fires_event(client):
    coll = _make_collection(client)
    pid = _make_page(client, coll)
    client.post("/workspace/automations", json={
        "name": "On delete",
        "trigger_type": "event",
        "event": "page.deleted",
        "collection_id": None,
        "condition_json": "[]",
        "actions_json": '[{"type":"notify","message":"page deleted"}]',
    })
    assert client.delete(f"/db/pages/{pid}/api").status_code == 200
    from app.db import get_conn
    with get_conn() as conn:
        runs = conn.execute("SELECT * FROM automation_runs WHERE status='fired'").fetchall()
    assert len(runs) == 1
    assert runs[0]["trigger_source"] == "event"


def test_create_page_action(client):
    coll = _make_collection(client)
    target = _make_collection(client, "Target")
    _make_page(client, coll, props={"Status": "Go"})
    client.post("/workspace/automations", json={
        "name": "Spawn a page",
        "trigger_type": "event",
        "event": "page.updated",
        "collection_id": coll,
        "condition_json": '[{"property":"Status","op":"contains","value":"Go"}]',
        "actions_json": json_str([{"type": "create_page", "collection_id": target,
                                   "title": "Generated from {{title}}",
                                   "properties": {"Status": "Todo"}}]),
    })
    from app.db import get_conn
    with get_conn() as conn:
        src = conn.execute(
            "SELECT id FROM collection_pages WHERE collection_id=? ORDER BY id DESC LIMIT 1",
            (coll,),
        ).fetchone()["id"]
    client.put(f"/db/pages/{src}/api", json={"properties": {"Status": "Go"}})

    with get_conn() as conn:
        rows = conn.execute(
            "SELECT title FROM collection_pages WHERE collection_id=? ORDER BY id DESC",
            (target,),
        ).fetchall()
    assert any("Generated from" in (r["title"] or "") for r in rows)


# ── Manual / button run ──

def test_button_run_endpoint(client):
    coll = _make_collection(client)
    pid = _make_page(client, coll, props={"Status": "Todo"})
    auto_id = client.post("/workspace/automations", json={
        "name": "Button action",
        "trigger_type": "event",
        "event": "page.updated",
        "collection_id": None,
        "condition_json": "[]",
        "actions_json": '[{"type":"set_property","property":"Status","value":"Clicked"}]',
    }).json()["id"]

    r = client.post(f"/api/automations/{auto_id}/run", json={"page_id": pid})
    assert r.status_code == 200
    assert r.json()["status"] == "fired"

    from app.db import get_conn
    with get_conn() as conn:
        props = __import__("json").loads(conn.execute(
            "SELECT property_values_json FROM collection_pages WHERE id=?", (pid,)
        ).fetchone()["property_values_json"])
    assert props["Status"] == "Clicked"

    # history recorded
    d = client.get(f"/workspace/automations/{auto_id}/runs").json()
    assert len(d["runs"]) == 1
    assert d["runs"][0]["trigger_source"] == "button"


def test_manual_run_skipped_for_disabled(client):
    auto_id = client.post("/workspace/automations", json={
        "name": "Disabled",
        "trigger_type": "event",
        "event": "page.created",
        "collection_id": None,
        "condition_json": "[]",
        "actions_json": "[]",
        "enabled": False,
    }).json()["id"]
    r = client.post(f"/workspace/automations/{auto_id}/run", json={})
    assert r.json()["status"] == "skipped"


# ── Cron due logic ──

def test_cron_due():
    from datetime import datetime, timedelta

    from app.services.automations import cron_due

    now = datetime(2026, 9, 7, 10, 15, 0)
    # every 15 minutes: due when >= 15 minutes elapsed, otherwise not
    last = (now - timedelta(minutes=15)).isoformat()
    assert cron_due("*/15 * * * *", last, now) is True
    last = (now - timedelta(minutes=10)).isoformat()
    assert cron_due("*/15 * * * *", last, now) is False
    # no last run → always due
    assert cron_due("*/30 * * * *", None, now) is True
    # fixed minute
    assert cron_due("15 * * * *", None, now) is True
    assert cron_due("45 * * * *", None, now) is False
    # @hourly / @daily
    assert cron_due("@hourly", None, now) is True
    assert cron_due("@daily", (now - timedelta(hours=25)).isoformat(), now) is True
    assert cron_due("@daily", (now - timedelta(hours=2)).isoformat(), now) is False
    assert cron_due("      ", None, now) is False


def json_str(obj):
    import json
    return json.dumps(obj)