-
Notifications
You must be signed in to change notification settings - Fork 5
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
d3ab4e7
commit aef0f7d
Showing
1 changed file
with
196 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,196 @@ | ||
import asyncio | ||
from typing import Any | ||
|
||
import requests | ||
from fastapi import status | ||
from pydantic import TypeAdapter | ||
|
||
from blueapi.config import RestConfig | ||
from blueapi.service.interface import get_devices, get_plans | ||
from blueapi.service.model import ( | ||
DeviceModel, | ||
DeviceResponse, | ||
EnvironmentResponse, | ||
PlanModel, | ||
PlanResponse, | ||
TaskResponse, | ||
) | ||
from blueapi.worker.event import WorkerState | ||
from blueapi.worker.task import Task | ||
from blueapi.worker.task_worker import TrackableTask | ||
|
||
ENDPOINT = f"{RestConfig().protocol}://{RestConfig().host}:{RestConfig().port}" | ||
|
||
|
||
def get_response( | ||
url: str, BaseModel: Any, status_code: int = status.HTTP_200_OK | ||
) -> Any: | ||
get_response = requests.get(url) | ||
model = TypeAdapter(BaseModel).validate_python(get_response.json()) | ||
assert get_response.status_code == status_code | ||
return model | ||
|
||
|
||
def post_response( | ||
url: str, json: dict[str, Any], BaseModel: Any, status_code: int | ||
) -> Any: | ||
post_response = requests.post(url, json=json) | ||
assert post_response.status_code == status_code | ||
return TypeAdapter(BaseModel).validate_python(post_response.json()) | ||
|
||
|
||
def delete_response(url: str, BaseModel: Any, status_code: int) -> Any: | ||
delete_response = requests.delete(url) | ||
assert delete_response.status_code == status_code | ||
return TypeAdapter(BaseModel).validate_python(delete_response.json()) | ||
|
||
|
||
def put_response( | ||
url: str, data: dict[str, str], BaseModel: Any, status_code: int | ||
) -> Any: | ||
put_response = requests.put(url, data=data) | ||
assert put_response.status_code == status_code | ||
return TypeAdapter(BaseModel).validate_python(put_response.json()) | ||
|
||
|
||
def test_server_up(): | ||
response = requests.get(ENDPOINT + "/docs") | ||
assert response.status_code == status.HTTP_200_OK | ||
|
||
|
||
def test_get_current_state_of_environment(): | ||
environment = get_response(ENDPOINT + "/environment", EnvironmentResponse) | ||
|
||
assert isinstance( | ||
environment, EnvironmentResponse | ||
) and environment == EnvironmentResponse(initialized=True) | ||
|
||
|
||
async def test_delete_current_environment(): | ||
delete_response = requests.delete(ENDPOINT + "/environment") | ||
assert delete_response.status_code == status.HTTP_200_OK | ||
assert EnvironmentResponse(initialized=False).model_dump() == delete_response.json() | ||
await asyncio.sleep(5) | ||
get_response = requests.get(ENDPOINT + "/environment") | ||
assert get_response.status_code == status.HTTP_200_OK | ||
assert TypeAdapter(EnvironmentResponse).validate_python( | ||
get_response.json() | ||
) == EnvironmentResponse(initialized=True) | ||
|
||
|
||
def test_get_plans(): | ||
response = requests.get(ENDPOINT + "/plans") | ||
assert response.status_code == status.HTTP_200_OK | ||
assert TypeAdapter(PlanResponse).validate_python(response.json()) == PlanResponse( | ||
plans=get_plans() | ||
) | ||
|
||
|
||
def test_get_plans_by_name(): | ||
for plan in get_plans(): | ||
response = requests.get(f"{ENDPOINT}/plans/{plan.name}") | ||
|
||
assert response.status_code == status.HTTP_200_OK | ||
assert TypeAdapter(PlanModel).validate_python(response.json()) == plan | ||
|
||
|
||
def test_get_devices(): | ||
response = requests.get(ENDPOINT + "/devices") | ||
assert response.status_code == status.HTTP_200_OK | ||
assert TypeAdapter(DeviceResponse).validate_python( | ||
response.json() | ||
) == DeviceResponse(devices=get_devices()) | ||
|
||
|
||
def test_get_device_by_name(): | ||
for device in get_devices(): | ||
response = requests.get(f"{ENDPOINT}/devices/{device.name}") | ||
assert response.status_code == status.HTTP_200_OK | ||
assert TypeAdapter(DeviceModel).validate_python(response.json()) == device | ||
|
||
|
||
_SIMPLE_TASK = Task(name="sleep", params={"time": 0.0}) | ||
_LONG_TASK = Task(name="sleep", params={"time": 1.0}) | ||
_INDEFINITE_TASK = Task( | ||
name="set_absolute", | ||
params={"movable": "fake_device", "value": 4.0}, | ||
) | ||
_FAILING_TASK = Task(name="failing_plan", params={}) | ||
|
||
|
||
def test_post_task(): | ||
post_response = requests.post(ENDPOINT + "/tasks", json=_SIMPLE_TASK.model_dump()) | ||
assert post_response.status_code == status.HTTP_201_CREATED | ||
created_task = TypeAdapter(TaskResponse).validate_python(post_response.json()) | ||
|
||
assert created_task | ||
|
||
delete_response = requests.delete(f"{ENDPOINT}/tasks/{created_task.task_id}") | ||
assert delete_response.status_code == status.HTTP_200_OK | ||
assert TypeAdapter(TaskResponse).validate_python(delete_response.json()) | ||
|
||
|
||
def test_get_tasks(): ... | ||
|
||
|
||
def test_delete_task_by_id(): | ||
post_response = requests.post(ENDPOINT + "/tasks", json=_SIMPLE_TASK.model_dump()) | ||
assert post_response.status_code == status.HTTP_201_CREATED | ||
created_task = TypeAdapter(TaskResponse).validate_python(post_response.json()) | ||
|
||
assert created_task | ||
|
||
delete_response = requests.delete(f"{ENDPOINT}/tasks/{created_task.task_id}") | ||
assert delete_response.status_code == status.HTTP_200_OK | ||
assert TypeAdapter(TaskResponse).validate_python(delete_response.json()) | ||
|
||
|
||
def test_get_task_by_id(): | ||
post_response = requests.post(ENDPOINT + "/tasks", json=_SIMPLE_TASK.model_dump()) | ||
assert post_response.status_code == status.HTTP_201_CREATED | ||
created_task = TypeAdapter(TaskResponse).validate_python(post_response.json()) | ||
|
||
assert created_task | ||
|
||
get_response = requests.get(f"{ENDPOINT}/tasks/{created_task.task_id}") | ||
assert get_response.status_code == status.HTTP_200_OK | ||
get_task = TypeAdapter(TrackableTask).validate_python(get_response.json()) | ||
|
||
assert ( | ||
get_task.task_id == created_task.task_id | ||
and get_task.is_pending | ||
and not get_task.is_complete | ||
and len(get_task.errors) == 0 | ||
) | ||
delete_response = requests.delete(f"{ENDPOINT}/tasks/{created_task.task_id}") | ||
assert delete_response.status_code == status.HTTP_200_OK | ||
assert TypeAdapter(TaskResponse).validate_python(delete_response.json()) | ||
|
||
|
||
def test_get_worker_task_by_id(): | ||
created_task = post_response( | ||
url=ENDPOINT + "/tasks", | ||
json=_SIMPLE_TASK.model_dump(), | ||
BaseModel=TaskResponse, | ||
status_code=status.HTTP_201_CREATED, | ||
) | ||
assert created_task | ||
get_worker_state = get_response(ENDPOINT + "/worker/state", WorkerState) | ||
|
||
assert ( | ||
isinstance(get_worker_state, WorkerState) | ||
and get_worker_state is WorkerState.IDLE | ||
) | ||
|
||
delete_response = requests.delete(f"{ENDPOINT}/tasks/{created_task.task_id}") | ||
assert delete_response.status_code == status.HTTP_200_OK | ||
assert TypeAdapter(TaskResponse).validate_python(delete_response.json()) | ||
|
||
|
||
def test_put_worker_task(): ... | ||
|
||
|
||
def test_get_worker_state(): ... | ||
|
||
|
||
def test_put_worker_state(): ... |