mirror of
https://framagit.org/framasoft/framaspace/argos.git
synced 2025-04-28 18:02:41 +02:00
Alexis Métaireau
83f57c6e47
- The web interface is exposed at /, and the api at /api. - Include picocss for a minimal CSS framework - Added some queries and models.Task properties to access the latest results
45 lines
1.3 KiB
Python
45 lines
1.3 KiB
Python
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from argos.schemas import AgentResult
|
|
from argos.server import app, models
|
|
|
|
|
|
@pytest.fixture()
|
|
def test_db():
|
|
models.Base.metadata.create_all(bind=app.state.engine)
|
|
yield app.state.db
|
|
models.Base.metadata.drop_all(bind=app.state.engine)
|
|
|
|
|
|
def test_read_tasks_requires_auth():
|
|
with TestClient(app) as client:
|
|
response = client.get("/api/tasks")
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_tasks_retrieval_and_results(test_db):
|
|
with TestClient(app) as client:
|
|
token = app.state.config.service.secrets[0]
|
|
client.headers = {"Authorization": f"Bearer {token}"}
|
|
response = client.get("/api/tasks")
|
|
assert response.status_code == 200
|
|
|
|
tasks = response.json()
|
|
assert len(tasks) == 2
|
|
|
|
results = []
|
|
for task in tasks:
|
|
results.append(
|
|
AgentResult(task_id=task["id"], status="success", context={})
|
|
)
|
|
|
|
data = [r.model_dump() for r in results]
|
|
response = client.post("/api/results", json=data)
|
|
|
|
assert response.status_code == 201
|
|
assert test_db.query(models.Result).count() == 2
|
|
|
|
# The list of tasks should be empty now
|
|
response = client.get("/api/tasks")
|
|
assert len(response.json()) == 0
|