iptv-manager-service/tests/routers/test_priorities.py

import uuid
from datetime import datetime, timezone

from fastapi import status
from sqlalchemy.orm import Session

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,
    MockGroup,
    MockPriority,
    create_mock_priorities_and_group,
)

# --- 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_all_priorities_success(db_session, admin_user_client):
    """Test reset priorities endpoint"""
    # Create test data
    priorities = [(100, "High"), (200, "Medium"), (300, "Low")]
    for id, desc in priorities:
        db_session.add(MockPriority(id=id, description=desc))
    db_session.commit()

    # Create channel using priority 100
    create_mock_priorities_and_group(db_session, [], "Test Group")
    channel_data = [
        {
            "group-title": "Test Group",
            "tvg_id": "test.tv",
            "name": "Test Channel",
            "urls": ["http://test.com"],
        }
    ]
    admin_user_client.post("/channels/bulk-upload", json=channel_data)

    # Delete all priorities
    response = admin_user_client.delete("/priorities")
    assert response.status_code == status.HTTP_200_OK
    assert response.json()["deleted"] == 2  # Medium and Low priorities
    assert response.json()["skipped"] == 1  # High priority is in use

    # Verify only priority 100 remains
    priorities = db_session.query(MockPriority).all()
    assert len(priorities) == 1
    assert priorities[0].id == 100


def test_reset_priorities_forbidden_for_non_admin(db_session, non_admin_user_client):
    """Test reset priorities requires admin role"""
    response = non_admin_user_client.delete("/priorities")
    assert response.status_code == status.HTTP_403_FORBIDDEN


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")
    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_all([priority, test_group])
    db_session.commit()

    # Create a channel first
    channel = MockChannelDB(
        name="Test Channel",
        tvg_id="test.tv",
        tvg_name="Test",
        tvg_logo="test.png",
        group_id=group_id,
    )
    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"]