95 lines
2.9 KiB
Python
95 lines
2.9 KiB
Python
|
import json
|
||
|
|
import re
|
||
|
|
|
||
|
|
from fastapi.testclient import TestClient
|
||
|
|
from sqlmodel import Session
|
||
|
|
|
||
|
|
from uptime_pie_api.main import Agent
|
||
|
|
|
||
|
|
|
||
|
|
uuid_re = re.compile(
|
||
|
|
r"^[0-9A-F]{8}-[0-9A-F]{4}-[4][0-9A-F]{3}-[89AB][0-9A-F]{3}-[0-9A-F]{12}$",
|
||
|
|
flags=re.IGNORECASE,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_ping(client: TestClient):
|
||
|
|
response = client.get("/ping")
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert response.text == "OK"
|
||
|
|
|
||
|
|
|
||
|
|
def test_create_agents(client: TestClient):
|
||
|
|
payload = {"name": "test_agent"}
|
||
|
|
response = client.post("/agents", content=json.dumps(payload))
|
||
|
|
json_response = response.json()
|
||
|
|
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert json_response["name"] == "test_agent"
|
||
|
|
assert uuid_re.match(json_response["id"])
|
||
|
|
assert json_response["api_key"] is not None
|
||
|
|
|
||
|
|
|
||
|
|
def test_duplicate_create_agent(client: TestClient):
|
||
|
|
payload = {"name": "test_agent", "api_key": "test_api_key"}
|
||
|
|
response = client.post("/agents", content=json.dumps(payload))
|
||
|
|
assert response.status_code == 200
|
||
|
|
|
||
|
|
response = client.post("/agents", content=json.dumps(payload))
|
||
|
|
json_response = response.json()
|
||
|
|
assert response.status_code == 400
|
||
|
|
assert json_response["detail"] == "Agent already exists"
|
||
|
|
|
||
|
|
|
||
|
|
def test_list_agents(client: TestClient, session: Session):
|
||
|
|
session.add(Agent(name="test_agent", api_key="test_api_key"))
|
||
|
|
session.add(Agent(name="test_agent2", api_key="test_api_key2"))
|
||
|
|
session.commit()
|
||
|
|
|
||
|
|
response = client.get("/agents")
|
||
|
|
json_response = response.json()
|
||
|
|
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert len(json_response) == 2
|
||
|
|
assert json_response[0]["name"] == "test_agent"
|
||
|
|
assert json_response[1]["name"] == "test_agent2"
|
||
|
|
assert uuid_re.match(json_response[0]["id"])
|
||
|
|
assert uuid_re.match(json_response[1]["id"])
|
||
|
|
|
||
|
|
|
||
|
|
def test_get_agent(client: TestClient, session: Session):
|
||
|
|
test_agent = Agent(name="test_agent", api_key="test_api_key")
|
||
|
|
session.add(test_agent)
|
||
|
|
session.commit()
|
||
|
|
session.refresh(test_agent)
|
||
|
|
|
||
|
|
response = client.get(f"/agents/{test_agent.id}")
|
||
|
|
json_response = response.json()
|
||
|
|
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert json_response["name"] == "test_agent"
|
||
|
|
assert json_response["id"] == str(test_agent.id)
|
||
|
|
assert json_response["api_key"] == "test_api_key"
|
||
|
|
|
||
|
|
|
||
|
|
def test_get_agent_not_found(client: TestClient):
|
||
|
|
response = client.get("/agents/00000000-aaaa-1111-bbbb-222222222222")
|
||
|
|
json_response = response.json()
|
||
|
|
|
||
|
|
assert response.status_code == 404
|
||
|
|
assert json_response["detail"] == "Agent not found"
|
||
|
|
|
||
|
|
|
||
|
|
def test_delete_agent(client: TestClient, session: Session):
|
||
|
|
test_agent = Agent(name="test_agent", api_key="test_api_key")
|
||
|
|
session.add(test_agent)
|
||
|
|
session.commit()
|
||
|
|
session.refresh(test_agent)
|
||
|
|
|
||
|
|
response = client.delete(f"/agents/{test_agent.id}")
|
||
|
|
json_response = response.json()
|
||
|
|
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert json_response["message"] == "Agent deleted"
|
||
|
|
assert session.get(Agent, test_agent.id) is None
|