423 lines
13 KiB
Python
423 lines
13 KiB
Python
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
from fastapi import status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth.dependencies import get_current_user
|
|
from app.routers.groups import router as groups_router
|
|
from app.utils.database import get_db
|
|
|
|
# Import mocks and fixtures
|
|
from tests.utils.auth_test_fixtures import (
|
|
admin_user_client,
|
|
db_session,
|
|
non_admin_user_client,
|
|
)
|
|
from tests.utils.db_mocks import MockChannelDB, MockGroup, SQLiteUUID
|
|
|
|
# --- Test Cases For Group Creation ---
|
|
|
|
|
|
def test_create_group_success(db_session: Session, admin_user_client):
|
|
group_data = {"name": "Test Group", "sort_order": 1}
|
|
response = admin_user_client.post("/groups/", json=group_data)
|
|
assert response.status_code == status.HTTP_201_CREATED
|
|
data = response.json()
|
|
assert data["name"] == "Test Group"
|
|
assert data["sort_order"] == 1
|
|
assert "id" in data
|
|
assert "created_at" in data
|
|
assert "updated_at" in data
|
|
|
|
# Verify in DB
|
|
db_group = (
|
|
db_session.query(MockGroup).filter(MockGroup.name == "Test Group").first()
|
|
)
|
|
assert db_group is not None
|
|
assert db_group.sort_order == 1
|
|
|
|
|
|
def test_create_group_duplicate(db_session: Session, admin_user_client):
|
|
# Create initial group
|
|
initial_group = MockGroup(
|
|
id=uuid.uuid4(),
|
|
name="Duplicate Group",
|
|
sort_order=1,
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
db_session.add(initial_group)
|
|
db_session.commit()
|
|
|
|
# Attempt to create duplicate
|
|
response = admin_user_client.post(
|
|
"/groups/", json={"name": "Duplicate Group", "sort_order": 2}
|
|
)
|
|
assert response.status_code == status.HTTP_409_CONFLICT
|
|
assert "already exists" in response.json()["detail"]
|
|
|
|
|
|
def test_create_group_forbidden_for_non_admin(
|
|
db_session: Session, non_admin_user_client
|
|
):
|
|
response = non_admin_user_client.post(
|
|
"/groups/", json={"name": "Forbidden Group", "sort_order": 1}
|
|
)
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
assert "required roles" in response.json()["detail"]
|
|
|
|
|
|
# --- Test Cases For Get Group ---
|
|
|
|
|
|
def test_get_group_success(db_session: Session, admin_user_client):
|
|
# Create a group first
|
|
test_group = MockGroup(
|
|
id=uuid.uuid4(),
|
|
name="Get Me Group",
|
|
sort_order=1,
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
db_session.add(test_group)
|
|
db_session.commit()
|
|
|
|
response = admin_user_client.get(f"/groups/{test_group.id}")
|
|
assert response.status_code == status.HTTP_200_OK
|
|
data = response.json()
|
|
assert data["id"] == str(test_group.id)
|
|
assert data["name"] == "Get Me Group"
|
|
assert data["sort_order"] == 1
|
|
|
|
|
|
def test_get_group_not_found(db_session: Session, admin_user_client):
|
|
random_uuid = uuid.uuid4()
|
|
response = admin_user_client.get(f"/groups/{random_uuid}")
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert "Group not found" in response.json()["detail"]
|
|
|
|
|
|
# --- Test Cases For Update Group ---
|
|
|
|
|
|
def test_update_group_success(db_session: Session, admin_user_client):
|
|
# Create initial group
|
|
group_id = uuid.uuid4()
|
|
test_group = MockGroup(
|
|
id=group_id,
|
|
name="Update Me",
|
|
sort_order=1,
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
db_session.add(test_group)
|
|
db_session.commit()
|
|
|
|
update_data = {"name": "Updated Name", "sort_order": 2}
|
|
response = admin_user_client.put(f"/groups/{group_id}", json=update_data)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
data = response.json()
|
|
assert data["name"] == "Updated Name"
|
|
assert data["sort_order"] == 2
|
|
|
|
# Verify in DB
|
|
db_group = db_session.query(MockGroup).filter(MockGroup.id == group_id).first()
|
|
assert db_group.name == "Updated Name"
|
|
assert db_group.sort_order == 2
|
|
|
|
|
|
def test_update_group_conflict(db_session: Session, admin_user_client):
|
|
# Create two groups
|
|
group1 = MockGroup(
|
|
id=uuid.uuid4(),
|
|
name="Group One",
|
|
sort_order=1,
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
group2 = MockGroup(
|
|
id=uuid.uuid4(),
|
|
name="Group Two",
|
|
sort_order=2,
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
db_session.add_all([group1, group2])
|
|
db_session.commit()
|
|
|
|
# Try to rename group2 to conflict with group1
|
|
response = admin_user_client.put(f"/groups/{group2.id}", json={"name": "Group One"})
|
|
assert response.status_code == status.HTTP_409_CONFLICT
|
|
assert "already exists" in response.json()["detail"]
|
|
|
|
|
|
def test_update_group_not_found(db_session: Session, admin_user_client):
|
|
random_uuid = uuid.uuid4()
|
|
response = admin_user_client.put(
|
|
f"/groups/{random_uuid}", json={"name": "Non-existent"}
|
|
)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert "Group not found" in response.json()["detail"]
|
|
|
|
|
|
def test_update_group_forbidden_for_non_admin(
|
|
db_session: Session, non_admin_user_client, admin_user_client
|
|
):
|
|
# Create group with admin
|
|
group_id = uuid.uuid4()
|
|
test_group = MockGroup(
|
|
id=group_id,
|
|
name="Admin Created",
|
|
sort_order=1,
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
db_session.add(test_group)
|
|
db_session.commit()
|
|
|
|
# Attempt update with non-admin
|
|
response = non_admin_user_client.put(
|
|
f"/groups/{group_id}", json={"name": "Non-Admin Update"}
|
|
)
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
assert "required roles" in response.json()["detail"]
|
|
|
|
|
|
# --- Test Cases For Delete Group ---
|
|
|
|
|
|
def test_delete_group_success(db_session: Session, admin_user_client):
|
|
# Create group
|
|
group_id = uuid.uuid4()
|
|
test_group = MockGroup(
|
|
id=group_id,
|
|
name="Delete Me",
|
|
sort_order=1,
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
db_session.add(test_group)
|
|
db_session.commit()
|
|
|
|
# Verify exists before delete
|
|
assert (
|
|
db_session.query(MockGroup).filter(MockGroup.id == group_id).first() is not None
|
|
)
|
|
|
|
response = admin_user_client.delete(f"/groups/{group_id}")
|
|
assert response.status_code == status.HTTP_204_NO_CONTENT
|
|
|
|
# Verify deleted
|
|
assert db_session.query(MockGroup).filter(MockGroup.id == group_id).first() is None
|
|
|
|
|
|
def test_delete_group_with_channels_fails(db_session: Session, admin_user_client):
|
|
# Create group with channel
|
|
group_id = uuid.uuid4()
|
|
test_group = MockGroup(
|
|
id=group_id,
|
|
name="Group With Channels",
|
|
sort_order=1,
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
db_session.add(test_group)
|
|
|
|
# Create channel in this group
|
|
test_channel = MockChannelDB(
|
|
id=uuid.uuid4(),
|
|
tvg_id="channel1.tv",
|
|
name="Channel 1",
|
|
group_id=group_id,
|
|
tvg_name="Channel1",
|
|
tvg_logo="logo.png",
|
|
)
|
|
db_session.add(test_channel)
|
|
db_session.commit()
|
|
|
|
response = admin_user_client.delete(f"/groups/{group_id}")
|
|
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
|
assert "existing channels" in response.json()["detail"]
|
|
|
|
# Verify group still exists
|
|
assert (
|
|
db_session.query(MockGroup).filter(MockGroup.id == group_id).first() is not None
|
|
)
|
|
|
|
|
|
def test_delete_group_not_found(db_session: Session, admin_user_client):
|
|
random_uuid = uuid.uuid4()
|
|
response = admin_user_client.delete(f"/groups/{random_uuid}")
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert "Group not found" in response.json()["detail"]
|
|
|
|
|
|
def test_delete_group_forbidden_for_non_admin(
|
|
db_session: Session, non_admin_user_client, admin_user_client
|
|
):
|
|
# Create group with admin
|
|
group_id = uuid.uuid4()
|
|
test_group = MockGroup(
|
|
id=group_id,
|
|
name="Admin Created",
|
|
sort_order=1,
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
db_session.add(test_group)
|
|
db_session.commit()
|
|
|
|
# Attempt delete with non-admin
|
|
response = non_admin_user_client.delete(f"/groups/{group_id}")
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
assert "required roles" in response.json()["detail"]
|
|
|
|
# Verify group still exists
|
|
assert (
|
|
db_session.query(MockGroup).filter(MockGroup.id == group_id).first() is not None
|
|
)
|
|
|
|
|
|
# --- Test Cases For List Groups ---
|
|
|
|
|
|
def test_list_groups_empty(db_session: Session, admin_user_client):
|
|
response = admin_user_client.get("/groups/")
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert response.json() == []
|
|
|
|
|
|
def test_list_groups_with_data(db_session: Session, admin_user_client):
|
|
# Create some groups
|
|
groups = [
|
|
MockGroup(
|
|
id=uuid.uuid4(),
|
|
name=f"Group {i}",
|
|
sort_order=i,
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
for i in range(3)
|
|
]
|
|
db_session.add_all(groups)
|
|
db_session.commit()
|
|
|
|
response = admin_user_client.get("/groups/")
|
|
assert response.status_code == status.HTTP_200_OK
|
|
data = response.json()
|
|
assert len(data) == 3
|
|
assert data[0]["sort_order"] == 0 # Should be sorted by sort_order
|
|
assert data[1]["sort_order"] == 1
|
|
assert data[2]["sort_order"] == 2
|
|
|
|
|
|
# --- Test Cases For Sort Order Updates ---
|
|
|
|
|
|
def test_update_group_sort_order_success(db_session: Session, admin_user_client):
|
|
# Create group
|
|
group_id = uuid.uuid4()
|
|
test_group = MockGroup(
|
|
id=group_id,
|
|
name="Sort Me",
|
|
sort_order=1,
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
db_session.add(test_group)
|
|
db_session.commit()
|
|
|
|
response = admin_user_client.put(f"/groups/{group_id}/sort", json={"sort_order": 5})
|
|
assert response.status_code == status.HTTP_200_OK
|
|
data = response.json()
|
|
assert data["sort_order"] == 5
|
|
|
|
# Verify in DB
|
|
db_group = db_session.query(MockGroup).filter(MockGroup.id == group_id).first()
|
|
assert db_group.sort_order == 5
|
|
|
|
|
|
def test_update_group_sort_order_not_found(db_session: Session, admin_user_client):
|
|
"""Test that updating sort order for non-existent group returns 404"""
|
|
random_uuid = uuid.uuid4()
|
|
response = admin_user_client.put(
|
|
f"/groups/{random_uuid}/sort", json={"sort_order": 5}
|
|
)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert "Group not found" in response.json()["detail"]
|
|
|
|
|
|
def test_bulk_update_sort_orders_success(db_session: Session, admin_user_client):
|
|
# Create groups
|
|
groups = [
|
|
MockGroup(
|
|
id=uuid.uuid4(),
|
|
name=f"Group {i}",
|
|
sort_order=i,
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
for i in range(3)
|
|
]
|
|
print(groups)
|
|
db_session.add_all(groups)
|
|
db_session.commit()
|
|
|
|
# Bulk update sort orders (reverse order)
|
|
bulk_data = {
|
|
"groups": [
|
|
{"group_id": str(groups[0].id), "sort_order": 2},
|
|
{"group_id": str(groups[1].id), "sort_order": 1},
|
|
{"group_id": str(groups[2].id), "sort_order": 0},
|
|
]
|
|
}
|
|
response = admin_user_client.post("/groups/reorder", json=bulk_data)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
data = response.json()
|
|
assert len(data) == 3
|
|
|
|
# Create a dictionary for easy lookup of returned group data by ID
|
|
returned_groups_map = {item["id"]: item for item in data}
|
|
|
|
# Verify each group has its expected new sort_order
|
|
assert returned_groups_map[str(groups[0].id)]["sort_order"] == 2
|
|
assert returned_groups_map[str(groups[1].id)]["sort_order"] == 1
|
|
assert returned_groups_map[str(groups[2].id)]["sort_order"] == 0
|
|
|
|
# Verify in DB
|
|
db_groups = db_session.query(MockGroup).order_by(MockGroup.sort_order).all()
|
|
assert db_groups[0].sort_order == 2
|
|
assert db_groups[1].sort_order == 1
|
|
assert db_groups[2].sort_order == 0
|
|
|
|
|
|
def test_bulk_update_sort_orders_invalid_group(db_session: Session, admin_user_client):
|
|
# Create one group
|
|
group_id = uuid.uuid4()
|
|
test_group = MockGroup(
|
|
id=group_id,
|
|
name="Valid Group",
|
|
sort_order=1,
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
db_session.add(test_group)
|
|
db_session.commit()
|
|
|
|
# Try to update with invalid group
|
|
bulk_data = {
|
|
"groups": [
|
|
{"group_id": str(group_id), "sort_order": 2},
|
|
{"group_id": str(uuid.uuid4()), "sort_order": 1}, # Invalid group
|
|
]
|
|
}
|
|
response = admin_user_client.post("/groups/reorder", json=bulk_data)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert "not found" in response.json()["detail"]
|
|
|
|
# Verify original sort order unchanged
|
|
db_group = db_session.query(MockGroup).filter(MockGroup.id == group_id).first()
|
|
assert db_group.sort_order == 1
|