Spaces:
Sleeping
Sleeping
File size: 3,070 Bytes
4ccc966 9fd90c1 4ccc966 | 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 | from __future__ import annotations
import httpx
import pytest
from enterprise_finance_env._compat import create_app
from enterprise_finance_env.client import EnterpriseFinanceClient
from enterprise_finance_env.models import EnterpriseFinanceActionPayload, EnterpriseFinanceObservation
from enterprise_finance_env.server.enterprise_env import EnterpriseFinanceEnv
@pytest.mark.asyncio
async def test_api_smoke_flow() -> None:
app = create_app(
lambda: EnterpriseFinanceEnv("easy"),
EnterpriseFinanceActionPayload,
EnterpriseFinanceObservation,
env_name="enterprise_finance_env_test",
)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
home = await client.get("/")
assert home.status_code == 200
assert "Enterprise Finance OpenEnv" in home.text
health = await client.get("/health")
assert health.status_code == 200
schema = await client.get("/schema")
assert schema.status_code == 200
assert "action" in schema.json()
reset = await client.post("/reset", json={})
assert reset.status_code == 200
assert "structured_ledgers" in reset.json()["observation"]
first_row = reset.json()["observation"]["structured_ledgers"][0]
step = await client.post(
"/step",
json={
"action": {
"type": "query_subledger",
"entity": first_row["entity_id"],
"account_code": first_row["account_code"],
"date_range": ["2026-01-01", "2026-01-31"],
}
},
)
assert step.status_code == 200
state = await client.get("/state")
assert state.status_code == 200
assert state.json()["difficulty"] == "easy"
@pytest.mark.asyncio
async def test_http_client_flow_uses_rest_endpoints() -> None:
app = create_app(
lambda: EnterpriseFinanceEnv("easy"),
EnterpriseFinanceActionPayload,
EnterpriseFinanceObservation,
env_name="enterprise_finance_env_test_client",
)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as http_client:
async with EnterpriseFinanceClient(
base_url="http://testserver",
async_client=http_client,
) as client:
reset = await client.reset()
assert reset.done is False
first_row = reset.observation.structured_ledgers[0]
result = await client.step(
{
"type": "query_subledger",
"entity": first_row.entity_id,
"account_code": first_row.account_code,
"date_range": ["2026-01-01", "2026-01-31"],
}
)
assert result.done is False
state = await client.state()
assert state.difficulty == "easy"
|