Created unit tests for priorities.py router

tests/routers/test_priorities.py

@@ -0,0 +1,188 @@
+import pytest
+from fastapi import status
+from sqlalchemy.orm import Session
+
+from app.models.db import ChannelURL, Priority
+from app.routers.priorities import router as priorities_router
+
+# Import fixtures and mocks
+from tests.utils.auth_test_fixtures import (
+    admin_user_client,
+    db_session,
+    non_admin_user_client,
+)
+from tests.utils.db_mocks import MockChannelDB, MockChannelURL, MockPriority
+
+# --- Test Cases For Priority Creation ---
+
+
+def test_create_priority_success(db_session: Session, admin_user_client):
+    priority_data = {"id": 100, "description": "Test Priority"}
+    response = admin_user_client.post("/priorities/", json=priority_data)
+    assert response.status_code == status.HTTP_201_CREATED
+    data = response.json()
+    assert data["id"] == 100
+    assert data["description"] == "Test Priority"
+
+    # Verify in DB
+    db_priority = db_session.get(MockPriority, 100)
+    assert db_priority is not None
+    assert db_priority.description == "Test Priority"
+
+
+def test_create_priority_duplicate(db_session: Session, admin_user_client):
+    # Create initial priority
+    priority_data = {"id": 100, "description": "Original Priority"}
+    response1 = admin_user_client.post("/priorities/", json=priority_data)
+    assert response1.status_code == status.HTTP_201_CREATED
+
+    # Attempt to create with same ID
+    response2 = admin_user_client.post("/priorities/", json=priority_data)
+    assert response2.status_code == status.HTTP_409_CONFLICT
+    assert "already exists" in response2.json()["detail"]
+
+
+def test_create_priority_forbidden_for_non_admin(
+    db_session: Session, non_admin_user_client
+):
+    priority_data = {"id": 100, "description": "Test Priority"}
+    response = non_admin_user_client.post("/priorities/", json=priority_data)
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+    assert "required roles" in response.json()["detail"]
+
+
+# --- Test Cases For List Priorities ---
+
+
+def test_list_priorities_empty(db_session: Session, admin_user_client):
+    response = admin_user_client.get("/priorities/")
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json() == []
+
+
+def test_list_priorities_with_data(db_session: Session, admin_user_client):
+    # Create some test priorities
+    priorities = [
+        MockPriority(id=100, description="High"),
+        MockPriority(id=200, description="Medium"),
+        MockPriority(id=300, description="Low"),
+    ]
+    for priority in priorities:
+        db_session.add(priority)
+    db_session.commit()
+
+    response = admin_user_client.get("/priorities/")
+    assert response.status_code == status.HTTP_200_OK
+    data = response.json()
+    assert len(data) == 3
+    assert data[0]["id"] == 100
+    assert data[0]["description"] == "High"
+    assert data[1]["id"] == 200
+    assert data[1]["description"] == "Medium"
+    assert data[2]["id"] == 300
+    assert data[2]["description"] == "Low"
+
+
+def test_list_priorities_forbidden_for_non_admin(
+    db_session: Session, non_admin_user_client
+):
+    response = non_admin_user_client.get("/priorities/")
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+    assert "required roles" in response.json()["detail"]
+
+
+# --- Test Cases For Get Priority ---
+
+
+def test_get_priority_success(db_session: Session, admin_user_client):
+    # Create a test priority
+    priority = MockPriority(id=100, description="Test Priority")
+    db_session.add(priority)
+    db_session.commit()
+
+    response = admin_user_client.get("/priorities/100")
+    assert response.status_code == status.HTTP_200_OK
+    data = response.json()
+    assert data["id"] == 100
+    assert data["description"] == "Test Priority"
+
+
+def test_get_priority_not_found(db_session: Session, admin_user_client):
+    response = admin_user_client.get("/priorities/999")
+    assert response.status_code == status.HTTP_404_NOT_FOUND
+    assert "Priority not found" in response.json()["detail"]
+
+
+def test_get_priority_forbidden_for_non_admin(
+    db_session: Session, non_admin_user_client
+):
+    response = non_admin_user_client.get("/priorities/100")
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+    assert "required roles" in response.json()["detail"]
+
+
+# --- Test Cases For Delete Priority ---
+
+
+def test_delete_priority_success(db_session: Session, admin_user_client):
+    # Create a test priority
+    priority = MockPriority(id=100, description="To Delete")
+    db_session.add(priority)
+    db_session.commit()
+
+    response = admin_user_client.delete("/priorities/100")
+    assert response.status_code == status.HTTP_204_NO_CONTENT
+
+    # Verify priority is gone from DB
+    db_priority = db_session.get(MockPriority, 100)
+    assert db_priority is None
+
+
+def test_delete_priority_not_found(db_session: Session, admin_user_client):
+    response = admin_user_client.delete("/priorities/999")
+    assert response.status_code == status.HTTP_404_NOT_FOUND
+    assert "Priority not found" in response.json()["detail"]
+
+
+def test_delete_priority_in_use(db_session: Session, admin_user_client):
+    # Create a priority and a channel URL using it
+    priority = MockPriority(id=100, description="In Use")
+    db_session.add(priority)
+    db_session.commit()
+
+    # Create a channel first
+    channel = MockChannelDB(
+        name="Test Channel",
+        tvg_id="test.tv",
+        tvg_name="Test",
+        tvg_logo="test.png",
+        group_title="Test Group",
+    )
+    db_session.add(channel)
+    db_session.commit()
+
+    # Create URL associated with the channel and priority
+    channel_url = MockChannelURL(
+        url="http://test.com",
+        priority_id=100,
+        in_use=True,
+        channel_id=channel.id,  # Add the channel_id
+    )
+    db_session.add(channel_url)
+    db_session.commit()
+
+    response = admin_user_client.delete("/priorities/100")
+    assert response.status_code == status.HTTP_409_CONFLICT
+    assert "in use by channel URLs" in response.json()["detail"]
+
+    # Verify priority still exists
+    db_priority = db_session.get(MockPriority, 100)
+    assert db_priority is not None
+
+
+def test_delete_priority_forbidden_for_non_admin(
+    db_session: Session, non_admin_user_client
+):
+    response = non_admin_user_client.delete("/priorities/100")
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+    assert "required roles" in response.json()["detail"]

tests/utils/auth_test_fixtures.py

@@ -6,6 +6,7 @@ from sqlalchemy.orm import Session
 from app.auth.dependencies import get_current_user
 from app.models.auth import CognitoUser
 from app.routers.channels import router as channels_router
+from app.routers.priorities import router as priorities_router
 from app.utils.database import get_db
 from tests.utils.db_mocks import (
     MockBase,
@@ -56,6 +57,7 @@ def admin_user_client(db_session: Session):
     """Yields a TestClient configured with an admin user."""
     test_app = FastAPI()
     test_app.include_router(channels_router)
+    test_app.include_router(priorities_router)
     test_app.dependency_overrides[get_db] = mock_get_db
     test_app.dependency_overrides[get_current_user] = mock_get_current_user_admin
     with TestClient(test_app) as test_client:
@@ -67,6 +69,7 @@ def non_admin_user_client(db_session: Session):
     """Yields a TestClient configured with a non-admin user."""
     test_app = FastAPI()
     test_app.include_router(channels_router)
+    test_app.include_router(priorities_router)
     test_app.dependency_overrides[get_db] = mock_get_db
     test_app.dependency_overrides[get_current_user] = mock_get_current_user_non_admin
     with TestClient(test_app) as test_client: