111 lines
3.5 KiB
Python
111 lines
3.5 KiB
Python
"""Tests for Group CRUD and bulk action endpoints.
|
|
|
|
Follows TDD: tests written first, then implementation.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_config():
|
|
os.environ["CONFIG_PATH"] = os.path.join(os.path.dirname(__file__), "..", "config.json")
|
|
from server.config import _load_config
|
|
import server.config
|
|
server.config._config = None
|
|
yield
|
|
|
|
|
|
@pytest.fixture
|
|
def groups_file():
|
|
f = tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False)
|
|
json.dump({"groups": []}, f)
|
|
f.close()
|
|
os.environ["GROUPS_PATH"] = f.name
|
|
yield f.name
|
|
os.unlink(f.name)
|
|
|
|
|
|
@pytest.fixture
|
|
def client(groups_file):
|
|
from server.group_store import GroupStore
|
|
GroupStore._instance = None
|
|
from server.main import app
|
|
return TestClient(app)
|
|
|
|
|
|
class TestListGroups:
|
|
def test_list_empty(self, client):
|
|
resp = client.get("/api/groups")
|
|
assert resp.status_code == 200
|
|
assert resp.json() == []
|
|
|
|
|
|
class TestCreateGroup:
|
|
def test_create(self, client):
|
|
resp = client.post("/api/groups", json={
|
|
"id": "test", "name": "Test", "services": [{"node": "hf-pdocker-01", "container": "test"}]
|
|
})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["id"] == "test"
|
|
|
|
def test_create_then_list(self, client):
|
|
client.post("/api/groups", json={"id": "g1", "name": "G1", "services": []})
|
|
resp = client.get("/api/groups")
|
|
assert len(resp.json()) == 1
|
|
assert resp.json()[0]["name"] == "G1"
|
|
|
|
def test_create_duplicate_409(self, client):
|
|
client.post("/api/groups", json={"id": "g1", "name": "G1", "services": []})
|
|
resp = client.post("/api/groups", json={"id": "g1", "name": "Dup", "services": []})
|
|
assert resp.status_code == 409
|
|
|
|
|
|
class TestUpdateGroup:
|
|
def test_update(self, client):
|
|
client.post("/api/groups", json={"id": "g1", "name": "G1", "services": []})
|
|
resp = client.put("/api/groups/g1", json={
|
|
"id": "g1", "name": "Updated", "services": [{"node": "bart", "container": "jellyfin"}]
|
|
})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["name"] == "Updated"
|
|
assert len(resp.json()["services"]) == 1
|
|
|
|
def test_update_nonexistent_404(self, client):
|
|
resp = client.put("/api/groups/nope", json={"id": "nope", "name": "N", "services": []})
|
|
assert resp.status_code == 404
|
|
|
|
|
|
class TestDeleteGroup:
|
|
def test_delete(self, client):
|
|
client.post("/api/groups", json={"id": "g1", "name": "G1", "services": []})
|
|
resp = client.delete("/api/groups/g1")
|
|
assert resp.status_code == 200
|
|
assert len(client.get("/api/groups").json()) == 0
|
|
|
|
def test_delete_nonexistent_404(self, client):
|
|
resp = client.delete("/api/groups/nope")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
class TestPersistence:
|
|
def test_persists_to_file(self, client, groups_file):
|
|
client.post("/api/groups", json={"id": "g1", "name": "G1", "services": []})
|
|
with open(groups_file) as f:
|
|
data = json.load(f)
|
|
assert len(data["groups"]) == 1
|
|
|
|
|
|
class TestBulkActions:
|
|
def test_start_nonexistent_group_404(self, client):
|
|
assert client.post("/api/groups/nope/start").status_code == 404
|
|
|
|
def test_stop_nonexistent_group_404(self, client):
|
|
assert client.post("/api/groups/nope/stop").status_code == 404
|
|
|
|
def test_restart_nonexistent_group_404(self, client):
|
|
assert client.post("/api/groups/nope/restart").status_code == 404
|