tests/test_flowpilot_api.py

from fastapi.testclient import TestClient

from backend.main import app
from backend.storage.database import InMemoryDatabase, db


client = TestClient(app)


def test_in_memory_database_creates_timezone_aware_owner_timestamp():
    db = InMemoryDatabase()

    owner = db.ensure_owner("owner-1", "owner-1@example.com")

    assert owner["created_at"].endswith("+00:00")


def test_status_returns_404_for_unknown_owner_without_creating_placeholder():
    db.reset()

    status_response = client.get("/api/status", params={"owner_id": "missing-owner"})

    assert status_response.status_code == 404
    assert "missing-owner" not in db.owners


def test_analyze_updates_existing_owner_email():
    db.reset()
    db.ensure_owner("sushanth", "sushanth@example.com")

    analyze_response = client.post(
        "/api/analyze",
        json={
            "owner_id": "sushanth",
            "owner_email": "real@example.com",
            "description": "Weekly pricing and orders.",
        },
    )

    assert analyze_response.status_code == 200
    assert db.owners["sushanth"]["email"] == "real@example.com"


def test_owner_flow_builds_and_deploys_workflows():
    db.reset()
    analyze_response = client.post(
        "/api/analyze",
        json={
            "owner_id": "raj",
            "owner_email": "raj@orchard.com",
            "description": (
                "I run an apple orchard. Customers email me orders, I check inventory in my Google Sheet, "
                "reply with pickup details, and every Friday I count weekly orders."
            ),
        },
    )
    assert analyze_response.status_code == 200
    analysis = analyze_response.json()
    assert analysis["tasks"]["fully_automatable"]

    suggest_response = client.post(
        "/api/suggest-workflows",
        json={
            "owner_id": "raj",
            "task_name": "Order processing",
            "task_description": "Read order emails, update sheets, and send confirmations.",
            "category": "fully_automatable",
        },
    )
    assert suggest_response.status_code == 200
    option = suggest_response.json()["options"][0]

    build_response = client.post(
        "/api/build-workflow",
        json={
            "owner_id": "raj",
            "task_name": "Order processing",
            "task_description": "Read order emails, update sheets, and send confirmations.",
            "category": "fully_automatable",
            "selected_option": option,
        },
    )
    assert build_response.status_code == 200
    workflow = build_response.json()["workflow"]
    assert workflow["trigger"]["type"] == "email_received"

    deploy_response = client.post(
        "/api/deploy",
        json={"owner_id": "raj", "workflows": [workflow]},
    )
    assert deploy_response.status_code == 200
    assert deploy_response.json()["status"] == "deployed"


def test_upload_and_status_endpoints():
    db.reset()
    client.post(
        "/api/analyze",
        json={"owner_id": "raj2", "owner_email": "raj2@orchard.com", "description": "Weekly pricing and orders."},
    )
    upload_response = client.post(
        "/api/upload-data",
        json={
            "owner_id": "raj2",
            "filename": "price-list.csv",
            "file_type": "csv",
            "purpose": "price_list",
            "content": "item,price\nHoneycrisp,3.50\nFuji,2.75\n",
        },
    )
    assert upload_response.status_code == 200
    status_response = client.get("/api/status", params={"owner_id": "raj2"})
    assert status_response.status_code == 200
    assert status_response.json()["owner"]["id"] == "raj2"


def test_reset_endpoint_clears_all_in_memory_data():
    db.reset()
    analyze_response = client.post(
        "/api/analyze",
        json={
            "owner_id": "reset-user",
            "owner_email": "reset-user@example.com",
            "description": "Weekly pricing and orders.",
        },
    )
    assert analyze_response.status_code == 200

    reset_response = client.post("/api/reset")

    assert reset_response.status_code == 200
    assert reset_response.json() == {"status": "ok", "message": "In-memory data reset."}
    assert db.owners == {}
    assert db.workflows == {}
    assert db.data_files == {}
    assert db.execution_logs == {}
    assert db.escalations == {}
    assert db.sheets == {}
    assert db.outbound_emails == {}