Spaces:
Sleeping
Sleeping
File size: 4,379 Bytes
83fe4f9 426e9e3 83fe4f9 426e9e3 f0c7697 83fe4f9 f0c7697 83fe4f9 f0c7697 83fe4f9 426e9e3 f0c7697 426e9e3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | 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 == {}
|