diff --git a/tests/routers/test_priorities.py b/tests/routers/test_priorities.py new file mode 100644 index 0000000..1673bb8 --- /dev/null +++ b/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"] diff --git a/tests/utils/auth_test_fixtures.py b/tests/utils/auth_test_fixtures.py index cb983d7..e1ff564 100644 --- a/tests/utils/auth_test_fixtures.py +++ b/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: