1589 lines
55 KiB
Python
1589 lines
55 KiB
Python
import uuid
|
|
from unittest.mock import patch
|
|
|
|
from fastapi import status
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth.dependencies import get_current_user
|
|
from app.main import app
|
|
from app.utils.database import get_db
|
|
|
|
# Import mocks and fixtures
|
|
from tests.utils.auth_test_fixtures import (
|
|
admin_user_client,
|
|
db_session,
|
|
mock_get_current_user_admin,
|
|
mock_get_current_user_non_admin,
|
|
non_admin_user_client,
|
|
)
|
|
from tests.utils.db_mocks import (
|
|
MockBase,
|
|
MockChannelDB,
|
|
MockChannelURL,
|
|
MockGroup,
|
|
MockPriority,
|
|
create_mock_priorities_and_group,
|
|
engine_mock,
|
|
mock_get_db,
|
|
)
|
|
from tests.utils.db_mocks import session_mock as TestingSessionLocal
|
|
|
|
# Override dependencies for testing
|
|
app.dependency_overrides[get_db] = mock_get_db
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
# --- Test Cases For Channel Creation ---
|
|
|
|
|
|
def test_create_channel_success(db_session: Session, admin_user_client: TestClient):
|
|
# Create mock priority and group
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "Test Group"
|
|
)
|
|
|
|
channel_data = {
|
|
"tvg_id": "channel1.tv",
|
|
"name": "Test Channel 1",
|
|
"group_id": str(group_id),
|
|
"tvg_name": "TestChannel1",
|
|
"tvg_logo": "logo.png",
|
|
"urls": [{"url": "http://stream1.com/test", "priority_id": 100}],
|
|
}
|
|
response = admin_user_client.post("/channels/", json=channel_data)
|
|
assert response.status_code == status.HTTP_201_CREATED
|
|
data = response.json()
|
|
assert data["name"] == "Test Channel 1"
|
|
assert data["group_id"] == str(group_id)
|
|
assert data["tvg_id"] == "channel1.tv"
|
|
assert len(data["urls"]) == 1
|
|
assert data["urls"][0]["url"] == "http://stream1.com/test"
|
|
assert data["urls"][0]["priority_id"] == 100
|
|
|
|
# Verify in DB
|
|
db_channel = (
|
|
db_session.query(MockChannelDB)
|
|
.filter(MockChannelDB.name == "Test Channel 1")
|
|
.first()
|
|
)
|
|
assert db_channel is not None
|
|
assert db_channel.group_id == group_id
|
|
|
|
# Query URLs
|
|
db_urls = (
|
|
db_session.query(MockChannelURL)
|
|
.filter(MockChannelURL.channel_id == db_channel.id)
|
|
.all()
|
|
)
|
|
|
|
assert len(db_urls) == 1
|
|
assert db_urls[0].url == "http://stream1.com/test"
|
|
|
|
|
|
def test_create_channel_duplicate(db_session: Session, admin_user_client: TestClient):
|
|
# Create mock priority and group
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "Duplicate Group"
|
|
)
|
|
|
|
# Create initial channel
|
|
initial_channel_data = {
|
|
"tvg_id": "channel_dup.tv",
|
|
"name": "Duplicate Channel",
|
|
"group_id": str(group_id),
|
|
"tvg_name": "DuplicateChannelName",
|
|
"tvg_logo": "duplicate_logo.png",
|
|
"urls": [{"url": "http://stream_dup.com/test", "priority_id": 100}],
|
|
}
|
|
response1 = admin_user_client.post("/channels/", json=initial_channel_data)
|
|
assert response1.status_code == status.HTTP_201_CREATED
|
|
|
|
# Attempt to create the same channel again
|
|
response2 = admin_user_client.post("/channels/", json=initial_channel_data)
|
|
assert response2.status_code == status.HTTP_409_CONFLICT
|
|
assert "already exists" in response2.json()["detail"]
|
|
|
|
|
|
def test_create_channel_forbidden_for_non_admin(
|
|
db_session: Session, non_admin_user_client: TestClient
|
|
):
|
|
# Create mock priority and group
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "Forbidden Group"
|
|
)
|
|
|
|
channel_data = {
|
|
"tvg_id": "channel_forbidden.tv",
|
|
"name": "Forbidden Channel",
|
|
"group_id": str(group_id),
|
|
"tvg_name": "ForbiddenChannelName",
|
|
"tvg_logo": "forbidden_logo.png",
|
|
"urls": [{"url": "http://stream_forbidden.com/test", "priority_id": 100}],
|
|
}
|
|
response = non_admin_user_client.post("/channels/", json=channel_data)
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
assert "required roles" in response.json()["detail"]
|
|
|
|
|
|
def test_create_channel_group_not_found(
|
|
db_session: Session, admin_user_client: TestClient
|
|
):
|
|
"""Test creating channel with non-existent group returns 404"""
|
|
# No group created in DB
|
|
channel_data = {
|
|
"tvg_id": "no_group.tv",
|
|
"name": "No Group Channel",
|
|
"group_id": str(uuid.uuid4()), # Random non-existent group ID
|
|
"tvg_name": "NoGroupChannel",
|
|
"tvg_logo": "no_group_logo.png",
|
|
"urls": [{"url": "http://no_group.com/stream", "priority_id": 100}],
|
|
}
|
|
response = admin_user_client.post("/channels/", json=channel_data)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert "Group not found" in response.json()["detail"]
|
|
|
|
|
|
# --- Test Cases For Get Channel ---
|
|
|
|
|
|
def test_get_channel_success(db_session: Session, admin_user_client: TestClient):
|
|
# Create priority and group using utility function
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "Get Group"
|
|
)
|
|
|
|
# Create a channel first
|
|
|
|
channel_data_create = {
|
|
"tvg_id": "get_me.tv",
|
|
"name": "Get Me Channel",
|
|
"group_id": str(group_id),
|
|
"tvg_name": "GetMeChannelName",
|
|
"tvg_logo": "get_me_logo.png",
|
|
"urls": [{"url": "http://get_me.com/stream", "priority_id": 100}],
|
|
}
|
|
create_response = admin_user_client.post("/channels/", json=channel_data_create)
|
|
assert create_response.status_code == status.HTTP_201_CREATED
|
|
created_channel_id = create_response.json()["id"]
|
|
|
|
app.dependency_overrides[get_current_user] = (
|
|
mock_get_current_user_admin # Or a generic authenticated user
|
|
)
|
|
get_response = admin_user_client.get(f"/channels/{created_channel_id}")
|
|
assert get_response.status_code == status.HTTP_200_OK
|
|
data = get_response.json()
|
|
assert data["id"] == created_channel_id
|
|
assert data["name"] == "Get Me Channel"
|
|
assert data["group_id"] == str(group_id)
|
|
assert len(data["urls"]) == 1
|
|
app.dependency_overrides.pop(get_current_user, None)
|
|
|
|
|
|
def test_get_channel_not_found(db_session: Session, admin_user_client: TestClient):
|
|
app.dependency_overrides[get_current_user] = mock_get_current_user_admin
|
|
random_uuid = uuid.uuid4()
|
|
response = admin_user_client.get(f"/channels/{random_uuid}")
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert "Channel not found" in response.json()["detail"]
|
|
app.dependency_overrides.pop(get_current_user, None)
|
|
|
|
|
|
# --- Test Cases For Update Channel ---
|
|
|
|
|
|
def test_update_channel_success(db_session: Session, admin_user_client: TestClient):
|
|
# Create priority and group using utility function
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "Update Group"
|
|
)
|
|
|
|
initial_channel_data = {
|
|
"tvg_id": "update_me.tv",
|
|
"name": "Update Me Channel",
|
|
"group_id": str(group_id),
|
|
"tvg_name": "UpdateMeChannelName",
|
|
"tvg_logo": "update_me_logo.png",
|
|
"urls": [{"url": "http://update_me.com/stream", "priority_id": 100}],
|
|
}
|
|
create_response = admin_user_client.post("/channels/", json=initial_channel_data)
|
|
assert create_response.status_code == status.HTTP_201_CREATED
|
|
created_channel_id = create_response.json()["id"]
|
|
|
|
update_data = {"name": "Updated Channel Name", "tvg_logo": "new_logo.png"}
|
|
response = admin_user_client.put(
|
|
f"/channels/{created_channel_id}", json=update_data
|
|
)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
data = response.json()
|
|
assert data["id"] == created_channel_id
|
|
assert data["name"] == "Updated Channel Name"
|
|
assert data["group_id"] == str(group_id)
|
|
assert data["tvg_logo"] == "new_logo.png"
|
|
|
|
# Verify in DB
|
|
db_channel = (
|
|
db_session.query(MockChannelDB)
|
|
.filter(MockChannelDB.id == created_channel_id)
|
|
.first()
|
|
)
|
|
assert db_channel is not None
|
|
assert db_channel.name == "Updated Channel Name"
|
|
assert db_channel.tvg_logo == "new_logo.png"
|
|
|
|
|
|
def test_update_channel_conflict(db_session: Session, admin_user_client: TestClient):
|
|
# Create priorities and groups using utility function
|
|
group1_id = create_mock_priorities_and_group(db_session, [(100, "High")], "Group A")
|
|
group2_id = create_mock_priorities_and_group(db_session, [], "Group B")
|
|
|
|
# Create channel 1
|
|
channel1_data = {
|
|
"tvg_id": "c1.tv",
|
|
"name": "Channel One",
|
|
"group_id": str(group1_id),
|
|
"tvg_name": "C1Name",
|
|
"tvg_logo": "c1logo.png",
|
|
"urls": [{"url": "http://c1.com", "priority_id": 100}],
|
|
}
|
|
admin_user_client.post("/channels/", json=channel1_data)
|
|
|
|
# Create channel 2
|
|
channel2_data = {
|
|
"tvg_id": "c2.tv",
|
|
"name": "Channel Two",
|
|
"group_id": str(group2_id),
|
|
"tvg_name": "C2Name",
|
|
"tvg_logo": "c2logo.png",
|
|
"urls": [{"url": "http://c2.com", "priority_id": 100}],
|
|
}
|
|
response_c2 = admin_user_client.post("/channels/", json=channel2_data)
|
|
channel2_id = response_c2.json()["id"]
|
|
|
|
# Attempt to update channel 2 to conflict with channel 1
|
|
update_conflict_data = {"name": "Channel One", "group_id": str(group1_id)}
|
|
response = admin_user_client.put(
|
|
f"/channels/{channel2_id}", json=update_conflict_data
|
|
)
|
|
assert response.status_code == status.HTTP_409_CONFLICT
|
|
assert "already exists" in response.json()["detail"]
|
|
|
|
|
|
def test_update_channel_not_found(db_session: Session, admin_user_client: TestClient):
|
|
random_uuid = uuid.uuid4()
|
|
update_data = {"name": "Non Existent Update"}
|
|
response = admin_user_client.put(f"/channels/{random_uuid}", json=update_data)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert "Channel not found" in response.json()["detail"]
|
|
|
|
|
|
def test_update_channel_group_not_found(
|
|
db_session: Session, admin_user_client: TestClient
|
|
):
|
|
"""Test updating channel with non-existent group returns 404"""
|
|
# Create priority and group using utility function
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "Original Group"
|
|
)
|
|
|
|
initial_channel_data = {
|
|
"tvg_id": "original.tv",
|
|
"name": "Original Channel",
|
|
"group_id": str(group_id),
|
|
"tvg_name": "OriginalName",
|
|
"tvg_logo": "original_logo.png",
|
|
"urls": [{"url": "http://original.com", "priority_id": 100}],
|
|
}
|
|
create_response = admin_user_client.post("/channels/", json=initial_channel_data)
|
|
created_channel_id = create_response.json()["id"]
|
|
|
|
# Attempt to update with non-existent group
|
|
update_data = {"group_id": str(uuid.uuid4())}
|
|
response = admin_user_client.put(
|
|
f"/channels/{created_channel_id}", json=update_data
|
|
)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert "Group not found" in response.json()["detail"]
|
|
|
|
|
|
def test_update_channel_forbidden_for_non_admin(
|
|
db_session: Session,
|
|
non_admin_user_client: TestClient,
|
|
admin_user_client: TestClient,
|
|
):
|
|
# Create priority and group using utility function
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "Forbidden Update Group"
|
|
)
|
|
|
|
initial_channel_data = {
|
|
"tvg_id": "update_forbidden.tv",
|
|
"name": "Update Forbidden",
|
|
"group_id": str(group_id),
|
|
"tvg_name": "UFName",
|
|
"tvg_logo": "uflogo.png",
|
|
"urls": [{"url": "http://update_forbidden.com", "priority_id": 100}],
|
|
}
|
|
create_response = admin_user_client.post("/channels/", json=initial_channel_data)
|
|
created_channel_id = create_response.json()["id"]
|
|
|
|
update_data = {"name": "Attempted Update"}
|
|
response = non_admin_user_client.put(
|
|
f"/channels/{created_channel_id}", json=update_data
|
|
)
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
assert "required roles" in response.json()["detail"]
|
|
|
|
|
|
# --- Test Cases For Delete Channels ---
|
|
|
|
|
|
def test_delete_all_channels_success(db_session, admin_user_client):
|
|
"""Test reset channels endpoint"""
|
|
# Create test data
|
|
create_mock_priorities_and_group(db_session, [(100, "High")], "Test Group")
|
|
channel_data = [
|
|
{
|
|
"name": "Channel 1",
|
|
"group-title": "Group A",
|
|
"tvg-id": "channel1.tv",
|
|
"tvg-name": "Channel One",
|
|
"tvg-logo": "logo1.png",
|
|
"urls": ["http://stream1.com", "http://backup1.com"],
|
|
}
|
|
]
|
|
admin_user_client.post("/channels/bulk-upload", json=channel_data)
|
|
|
|
# Verify data exists
|
|
assert db_session.query(MockChannelDB).count() == 1
|
|
assert db_session.query(MockChannelURL).count() == 2
|
|
|
|
# Reset channels
|
|
response = admin_user_client.delete("/channels")
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert response.json()["deleted"] == 1
|
|
|
|
# Verify data is gone
|
|
assert db_session.query(MockChannelDB).count() == 0
|
|
assert db_session.query(MockChannelURL).count() == 0
|
|
|
|
|
|
def test_delete_all_channels_forbidden_for_non_admin(db_session, non_admin_user_client):
|
|
"""Test delete all channels requires admin role"""
|
|
response = non_admin_user_client.delete("/channels")
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
|
|
|
|
def test_delete_all_channels_error_handling(db_session, admin_user_client):
|
|
"""Test error handling when deleting channels fails"""
|
|
# Create some test data
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "Error Group"
|
|
)
|
|
channel = MockChannelDB(
|
|
name="Error Channel",
|
|
group_id=group_id,
|
|
tvg_id="error.tv",
|
|
urls=[MockChannelURL(url="http://error.com/stream", priority_id=100)],
|
|
)
|
|
db_session.add(channel)
|
|
db_session.commit()
|
|
|
|
# To test the error handling of the endpoint, we need to simulate a failure
|
|
# during the database operation. We patch the 'commit' method of the session
|
|
# class to raise an exception. This patch will be active during the API call.
|
|
with patch.object(
|
|
type(db_session), "commit", side_effect=Exception("Mock database commit error")
|
|
):
|
|
response = admin_user_client.delete("/channels/")
|
|
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
|
|
assert "Failed to delete channels" in response.json()["detail"]
|
|
|
|
|
|
def test_delete_channel_success(db_session: Session, admin_user_client: TestClient):
|
|
# Create priority and group
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "Delete Group"
|
|
)
|
|
|
|
initial_channel_data = {
|
|
"tvg_id": "delete_me.tv",
|
|
"name": "Delete Me Channel",
|
|
"group_id": str(group_id), # Use the ID of the created group
|
|
"tvg_name": "DMName",
|
|
"tvg_logo": "dmlogo.png",
|
|
"urls": [{"url": "http://delete_me.com/stream", "priority_id": 100}],
|
|
}
|
|
create_response = admin_user_client.post("/channels/", json=initial_channel_data)
|
|
assert create_response.status_code == status.HTTP_201_CREATED
|
|
created_channel_id = create_response.json()["id"]
|
|
|
|
# Verify it exists before delete
|
|
db_channel_before_delete = (
|
|
db_session.query(MockChannelDB)
|
|
.filter(MockChannelDB.id == uuid.UUID(created_channel_id))
|
|
.first()
|
|
)
|
|
assert db_channel_before_delete is not None
|
|
|
|
delete_response = admin_user_client.delete(f"/channels/{created_channel_id}")
|
|
assert delete_response.status_code == status.HTTP_204_NO_CONTENT
|
|
|
|
# Verify it's gone from DB
|
|
db_channel_after_delete = (
|
|
db_session.query(MockChannelDB)
|
|
.filter(MockChannelDB.id == uuid.UUID(created_channel_id))
|
|
.first()
|
|
)
|
|
assert db_channel_after_delete is None
|
|
|
|
# Also verify associated URLs are deleted (due to CASCADE in mock model)
|
|
db_urls_after_delete = (
|
|
db_session.query(MockChannelURL)
|
|
.filter(MockChannelURL.channel_id == uuid.UUID(created_channel_id))
|
|
.all()
|
|
)
|
|
assert len(db_urls_after_delete) == 0
|
|
|
|
|
|
def test_delete_channel_not_found(db_session: Session, admin_user_client: TestClient):
|
|
random_uuid = uuid.uuid4()
|
|
response = admin_user_client.delete(f"/channels/{random_uuid}")
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert "Channel not found" in response.json()["detail"]
|
|
|
|
|
|
def test_delete_channel_forbidden_for_non_admin(
|
|
db_session: Session,
|
|
non_admin_user_client: TestClient,
|
|
admin_user_client: TestClient,
|
|
):
|
|
# Create priority and group
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "Forbidden Delete Group"
|
|
)
|
|
|
|
initial_channel_data = {
|
|
"tvg_id": "delete_forbidden.tv",
|
|
"name": "Delete Forbidden",
|
|
"group_id": str(group_id), # Use the ID of the created group
|
|
"tvg_name": "DFName",
|
|
"tvg_logo": "dflogo.png",
|
|
"urls": [{"url": "http://delete_forbidden.com", "priority_id": 100}],
|
|
}
|
|
create_response = admin_user_client.post("/channels/", json=initial_channel_data)
|
|
created_channel_id = create_response.json()["id"]
|
|
|
|
response = non_admin_user_client.delete(f"/channels/{created_channel_id}")
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
assert "required roles" in response.json()["detail"]
|
|
|
|
# Ensure channel was not deleted
|
|
db_channel_not_deleted = (
|
|
db_session.query(MockChannelDB)
|
|
.filter(MockChannelDB.id == uuid.UUID(created_channel_id))
|
|
.first()
|
|
)
|
|
assert db_channel_not_deleted is not None
|
|
|
|
|
|
# --- Test Cases For List Channels ---
|
|
|
|
|
|
def test_list_channels_empty(db_session: Session, admin_user_client: TestClient):
|
|
response = admin_user_client.get("/channels/")
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert response.json() == []
|
|
|
|
|
|
def test_get_channels_by_group_success(
|
|
db_session: Session, admin_user_client: TestClient
|
|
):
|
|
"""Test getting channels for an existing group"""
|
|
# Create priority and groups
|
|
group1_id = create_mock_priorities_and_group(db_session, [(100, "High")], "Group 1")
|
|
|
|
group2_id = create_mock_priorities_and_group(db_session, [], "Group 2")
|
|
|
|
# Create 2 channels in group1 and 1 in group2
|
|
channels_group1 = [
|
|
{
|
|
"tvg_id": f"g1c{i}.tv",
|
|
"name": f"Group1 Channel {i}",
|
|
"group_id": str(group1_id),
|
|
"tvg_name": f"G1C{i}",
|
|
"tvg_logo": f"g1c{i}_logo.png",
|
|
"urls": [{"url": f"http://g1c{i}.com", "priority_id": 100}],
|
|
}
|
|
for i in range(2)
|
|
]
|
|
channel_group2 = {
|
|
"tvg_id": "g2c1.tv",
|
|
"name": "Group2 Channel 1",
|
|
"group_id": str(group2_id),
|
|
"tvg_name": "G2C1",
|
|
"tvg_logo": "g2c1_logo.png",
|
|
"urls": [{"url": "http://g2c1.com", "priority_id": 100}],
|
|
}
|
|
|
|
# Create all channels
|
|
for channel_data in channels_group1 + [channel_group2]:
|
|
admin_user_client.post("/channels/", json=channel_data)
|
|
|
|
# Get channels for group1
|
|
response = admin_user_client.get(f"/channels/groups/{group1_id}/channels")
|
|
assert response.status_code == status.HTTP_200_OK
|
|
data = response.json()
|
|
assert len(data) == 2
|
|
assert all(channel["group_id"] == str(group1_id) for channel in data)
|
|
|
|
|
|
def test_get_channels_by_group_not_found(
|
|
db_session: Session, admin_user_client: TestClient
|
|
):
|
|
"""Test getting channels for non-existent group returns 404"""
|
|
random_uuid = uuid.uuid4()
|
|
response = admin_user_client.get(f"/channels/groups/{random_uuid}/channels")
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert "Group not found" in response.json()["detail"]
|
|
|
|
|
|
def test_update_channel_group_success(
|
|
db_session: Session, admin_user_client: TestClient
|
|
):
|
|
"""Test successfully updating a channel's group"""
|
|
# Create priority and group
|
|
group1_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "Original Group"
|
|
)
|
|
|
|
channel_data = {
|
|
"tvg_id": "original.tv",
|
|
"name": "Original Channel",
|
|
"group_id": str(group1_id),
|
|
"tvg_name": "OriginalName",
|
|
"tvg_logo": "original_logo.png",
|
|
"urls": [{"url": "http://original.com", "priority_id": 100}],
|
|
}
|
|
create_response = admin_user_client.post("/channels/", json=channel_data)
|
|
channel_id = create_response.json()["id"]
|
|
|
|
# Create target group
|
|
group2_id = create_mock_priorities_and_group(db_session, [], "Target Group")
|
|
|
|
# Update channel's group
|
|
response = admin_user_client.put(
|
|
f"/channels/{channel_id}/group?group_id={group2_id}"
|
|
)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
data = response.json()
|
|
assert data["id"] == channel_id
|
|
assert data["group_id"] == str(group2_id)
|
|
|
|
# Verify in DB
|
|
db_channel = (
|
|
db_session.query(MockChannelDB)
|
|
.filter(MockChannelDB.id == uuid.UUID(channel_id))
|
|
.first()
|
|
)
|
|
assert db_channel.group_id == group2_id
|
|
|
|
|
|
def test_update_channel_group_channel_not_found(
|
|
db_session: Session, admin_user_client: TestClient
|
|
):
|
|
"""Test updating non-existent channel's group returns 404"""
|
|
# Create priority and group
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "Test Group"
|
|
)
|
|
|
|
# Attempt to update non-existent channel
|
|
random_uuid = uuid.uuid4()
|
|
response = admin_user_client.put(
|
|
f"/channels/{random_uuid}/group?group_id={group_id}"
|
|
)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert "Channel not found" in response.json()["detail"]
|
|
|
|
|
|
def test_update_channel_group_group_not_found(
|
|
db_session: Session, admin_user_client: TestClient
|
|
):
|
|
"""Test updating channel to non-existent group returns 404"""
|
|
# Create priority and group
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "Original Group"
|
|
)
|
|
|
|
# Create a channel in the original group
|
|
channel_data = {
|
|
"tvg_id": "original.tv",
|
|
"name": "Original Channel",
|
|
"group_id": str(group_id),
|
|
"tvg_name": "OriginalName",
|
|
"tvg_logo": "original_logo.png",
|
|
"urls": [{"url": "http://original.com", "priority_id": 100}],
|
|
}
|
|
create_response = admin_user_client.post("/channels/", json=channel_data)
|
|
channel_id = create_response.json()["id"]
|
|
|
|
# Attempt to update with non-existent group
|
|
response = admin_user_client.put(
|
|
f"/channels/{channel_id}/group?group_id={uuid.uuid4()}"
|
|
)
|
|
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert "Group not found" in response.json()["detail"]
|
|
|
|
|
|
def test_update_channel_group_duplicate_name(
|
|
db_session: Session, admin_user_client: TestClient
|
|
):
|
|
"""Test updating channel to group with duplicate name returns 409"""
|
|
# Create priority and groups
|
|
group1_id = create_mock_priorities_and_group(db_session, [(100, "High")], "Group 1")
|
|
group2_id = create_mock_priorities_and_group(db_session, [], "Group 2")
|
|
|
|
# Create channel in each group with same name
|
|
channel_name = "Duplicate Channel"
|
|
channel1_data = {
|
|
"tvg_id": "c1.tv",
|
|
"name": channel_name,
|
|
"group_id": str(group1_id),
|
|
"tvg_name": "C1",
|
|
"tvg_logo": "c1.png",
|
|
"urls": [{"url": "http://c1.com", "priority_id": 100}],
|
|
}
|
|
channel2_data = {
|
|
"tvg_id": "c2.tv",
|
|
"name": channel_name,
|
|
"group_id": str(group2_id),
|
|
"tvg_name": "C2",
|
|
"tvg_logo": "c2.png",
|
|
"urls": [{"url": "http://c2.com", "priority_id": 100}],
|
|
}
|
|
admin_user_client.post("/channels/", json=channel1_data)
|
|
create2_response = admin_user_client.post("/channels/", json=channel2_data)
|
|
channel2_id = create2_response.json()["id"]
|
|
|
|
# Attempt to move channel2 to group1 (would create duplicate name)
|
|
response = admin_user_client.put(
|
|
f"/channels/{channel2_id}/group?group_id={group1_id}"
|
|
)
|
|
assert response.status_code == status.HTTP_409_CONFLICT
|
|
assert "already exists" in response.json()["detail"]
|
|
|
|
|
|
def test_update_channel_group_forbidden_for_non_admin(
|
|
db_session: Session,
|
|
non_admin_user_client: TestClient,
|
|
admin_user_client: TestClient,
|
|
):
|
|
"""Test updating channel's group as non-admin returns 403"""
|
|
# Create priority and groups
|
|
group1_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "Original Group"
|
|
)
|
|
group2_id = create_mock_priorities_and_group(db_session, [], "Target Group")
|
|
|
|
channel_data = {
|
|
"tvg_id": "protected.tv",
|
|
"name": "Protected Channel",
|
|
"group_id": str(group1_id),
|
|
"tvg_name": "Protected",
|
|
"tvg_logo": "protected.png",
|
|
"urls": [{"url": "http://protected.com", "priority_id": 100}],
|
|
}
|
|
create_response = admin_user_client.post("/channels/", json=channel_data)
|
|
channel_id = create_response.json()["id"]
|
|
|
|
# Attempt to update group as non-admin
|
|
response = non_admin_user_client.put(
|
|
f"/channels/{channel_id}/group?group_id={group2_id}"
|
|
)
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
assert "required roles" in response.json()["detail"]
|
|
|
|
|
|
def test_list_channels_with_data_and_pagination(
|
|
db_session: Session, admin_user_client: TestClient
|
|
):
|
|
# Create priority and group
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "List Group"
|
|
)
|
|
|
|
# Create some channels
|
|
for i in range(5):
|
|
channel_data = {
|
|
"tvg_id": f"list_c{i}.tv",
|
|
"name": f"List Channel {i}",
|
|
"group_id": str(group_id),
|
|
"tvg_name": f"LCName{i}",
|
|
"tvg_logo": f"lclogo{i}.png",
|
|
"urls": [{"url": f"http://list_c{i}.com", "priority_id": 100}],
|
|
}
|
|
admin_user_client.post("/channels/", json=channel_data)
|
|
|
|
# Test default pagination (limit 100)
|
|
response_all = admin_user_client.get("/channels/")
|
|
assert response_all.status_code == status.HTTP_200_OK
|
|
data_all = response_all.json()
|
|
assert len(data_all) == 5
|
|
assert data_all[0]["name"] == "List Channel 0"
|
|
|
|
# Test limit
|
|
response_limit = admin_user_client.get("/channels/?limit=2")
|
|
assert response_limit.status_code == status.HTTP_200_OK
|
|
data_limit = response_limit.json()
|
|
assert len(data_limit) == 2
|
|
assert data_limit[0]["name"] == "List Channel 0"
|
|
assert data_limit[1]["name"] == "List Channel 1"
|
|
|
|
# Test skip and limit
|
|
response_skip_limit = admin_user_client.get("/channels/?skip=2&limit=2")
|
|
assert response_skip_limit.status_code == status.HTTP_200_OK
|
|
data_skip_limit = response_skip_limit.json()
|
|
assert len(data_skip_limit) == 2
|
|
assert data_skip_limit[0]["name"] == "List Channel 2"
|
|
assert data_skip_limit[1]["name"] == "List Channel 3"
|
|
|
|
# Test skip beyond data
|
|
response_skip_beyond = admin_user_client.get("/channels/?skip=10")
|
|
assert response_skip_beyond.status_code == status.HTTP_200_OK
|
|
assert response_skip_beyond.json() == []
|
|
|
|
|
|
def test_list_channels_forbidden_for_non_admin(
|
|
db_session: Session, non_admin_user_client: TestClient
|
|
):
|
|
response = non_admin_user_client.get("/channels/")
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
assert "required roles" in response.json()["detail"]
|
|
|
|
|
|
# --- Test Cases For Add Channel URL ---
|
|
|
|
|
|
def test_add_channel_url_success(db_session: Session, admin_user_client: TestClient):
|
|
# Setup priorities and create a group
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High"), (200, "Medium")], "URL Group"
|
|
)
|
|
|
|
channel_data = {
|
|
"tvg_id": "channel_for_url.tv",
|
|
"name": "Channel For URL",
|
|
"group_id": str(group_id),
|
|
"tvg_name": "CFUName",
|
|
"tvg_logo": "cfulogo.png",
|
|
"urls": [{"url": "http://initial.com/stream", "priority_id": 100}],
|
|
}
|
|
create_response = admin_user_client.post("/channels/", json=channel_data)
|
|
assert create_response.status_code == status.HTTP_201_CREATED
|
|
created_channel_id = create_response.json()["id"]
|
|
|
|
url_data = {"url": "http://new_stream.com/live", "priority_id": 200}
|
|
response = admin_user_client.post(
|
|
f"/channels/{created_channel_id}/urls", json=url_data
|
|
)
|
|
assert response.status_code == status.HTTP_201_CREATED
|
|
data = response.json()
|
|
assert data["url"] == "http://new_stream.com/live"
|
|
assert data["priority_id"] == 200
|
|
assert data["in_use"] is False # Default
|
|
|
|
# Verify in DB
|
|
db_url = (
|
|
db_session.query(MockChannelURL)
|
|
.filter(MockChannelURL.id == uuid.UUID(data["id"]))
|
|
.first()
|
|
)
|
|
assert db_url is not None
|
|
assert db_url.url == "http://new_stream.com/live"
|
|
assert db_url.priority_id == 200
|
|
assert db_url.channel_id == uuid.UUID(created_channel_id)
|
|
|
|
# Check the channel now has two URLs
|
|
# Re-fetch channel to get updated URLs list
|
|
|
|
# Expire to ensure fresh data from DB if ChannelResponse is not dynamic
|
|
db_session.expire_all()
|
|
|
|
# Let's verify by querying the database directly
|
|
# for the count of URLs for the channel
|
|
url_count = (
|
|
db_session.query(MockChannelURL)
|
|
.filter(MockChannelURL.channel_id == uuid.UUID(created_channel_id))
|
|
.count()
|
|
)
|
|
assert url_count == 2
|
|
|
|
# And also check the response from get_channel
|
|
channel_get_response = admin_user_client.get(f"/channels/{created_channel_id}")
|
|
assert channel_get_response.status_code == status.HTTP_200_OK
|
|
channel_details = channel_get_response.json()
|
|
assert len(channel_details["urls"]) == 2
|
|
|
|
|
|
def test_add_channel_url_channel_not_found(
|
|
db_session: Session, admin_user_client: TestClient
|
|
):
|
|
# Setup priority
|
|
create_mock_priorities_and_group(db_session, [(100, "High")], "My Group")
|
|
|
|
random_channel_uuid = uuid.uuid4()
|
|
url_data = {"url": "http://stream_no_channel.com", "priority_id": 100}
|
|
response = admin_user_client.post(
|
|
f"/channels/{random_channel_uuid}/urls", json=url_data
|
|
)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert "Channel not found" in response.json()["detail"]
|
|
|
|
|
|
def test_add_channel_url_forbidden_for_non_admin(
|
|
db_session: Session,
|
|
non_admin_user_client: TestClient,
|
|
admin_user_client: TestClient,
|
|
):
|
|
# Setup priority and create a channel with admin
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "URL Forbidden Group"
|
|
)
|
|
|
|
channel_data = {
|
|
"tvg_id": "url_forbidden.tv",
|
|
"name": "URL Forbidden",
|
|
"group_id": str(group_id), # Use the ID of the created group
|
|
"tvg_name": "UFName2",
|
|
"tvg_logo": "uflogo2.png",
|
|
"urls": [{"url": "http://url_forbidden.com", "priority_id": 100}],
|
|
}
|
|
create_response = admin_user_client.post("/channels/", json=channel_data)
|
|
created_channel_id = create_response.json()["id"]
|
|
|
|
url_data = {"url": "http://new_stream_forbidden.com", "priority_id": 100}
|
|
response = non_admin_user_client.post(
|
|
f"/channels/{created_channel_id}/urls", json=url_data
|
|
)
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
assert "required roles" in response.json()["detail"]
|
|
|
|
|
|
# --- Test Cases For Update Channel URL ---
|
|
|
|
|
|
def test_update_channel_url_success(db_session: Session, admin_user_client: TestClient):
|
|
# Setup priorities and create a channel with a URL
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High"), (200, "Medium"), (300, "Low")], "URL Update Group"
|
|
)
|
|
|
|
channel_data = {
|
|
"tvg_id": "ch_update_url.tv",
|
|
"name": "Channel Update URL",
|
|
"group_id": str(group_id), # Use the ID of the created group
|
|
"tvg_name": "CUUName",
|
|
"tvg_logo": "cuulogo.png",
|
|
"urls": [{"url": "http://original_url.com/stream", "priority_id": 100}],
|
|
}
|
|
create_ch_response = admin_user_client.post("/channels/", json=channel_data)
|
|
created_channel_id = create_ch_response.json()["id"]
|
|
# Get the ID of the initially created URL
|
|
initial_url_id = create_ch_response.json()["urls"][0]["id"]
|
|
|
|
update_url_data = {
|
|
"url": "http://updated_url.com/live",
|
|
"priority_id": 300,
|
|
"in_use": True,
|
|
}
|
|
response = admin_user_client.put(
|
|
f"/channels/{created_channel_id}/urls/{initial_url_id}", json=update_url_data
|
|
)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
data = response.json()
|
|
assert data["id"] == initial_url_id
|
|
assert data["url"] == "http://updated_url.com/live"
|
|
assert data["priority_id"] == 300
|
|
assert data["in_use"] is True
|
|
|
|
# Verify in DB
|
|
db_url = (
|
|
db_session.query(MockChannelURL)
|
|
.filter(MockChannelURL.id == uuid.UUID(initial_url_id))
|
|
.first()
|
|
)
|
|
assert db_url is not None
|
|
assert db_url.url == "http://updated_url.com/live"
|
|
assert db_url.priority_id == 300
|
|
assert db_url.in_use is True
|
|
|
|
|
|
def test_update_channel_url_partial_success(
|
|
db_session: Session, admin_user_client: TestClient
|
|
):
|
|
# Setup priorities and create a channel with a URL
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "URL Partial Update Group"
|
|
)
|
|
|
|
channel_data = {
|
|
"tvg_id": "ch_partial_update_url.tv",
|
|
"name": "Channel Partial Update URL",
|
|
"group_id": str(group_id),
|
|
"tvg_name": "CPUName",
|
|
"tvg_logo": "cpulogo.png",
|
|
"urls": [{"url": "http://partial_original.com/stream", "priority_id": 100}],
|
|
}
|
|
create_ch_response = admin_user_client.post("/channels/", json=channel_data)
|
|
created_channel_id = create_ch_response.json()["id"]
|
|
initial_url_id = create_ch_response.json()["urls"][0]["id"]
|
|
|
|
# Update only 'in_use'
|
|
update_url_data = {"in_use": True}
|
|
response = admin_user_client.put(
|
|
f"/channels/{created_channel_id}/urls/{initial_url_id}", json=update_url_data
|
|
)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
data = response.json()
|
|
assert data["id"] == initial_url_id
|
|
assert data["url"] == "http://partial_original.com/stream"
|
|
assert data["priority_id"] == 100
|
|
assert data["in_use"] is True
|
|
|
|
# Verify in DB
|
|
db_url = (
|
|
db_session.query(MockChannelURL)
|
|
.filter(MockChannelURL.id == uuid.UUID(initial_url_id))
|
|
.first()
|
|
)
|
|
assert db_url is not None
|
|
assert db_url.in_use is True
|
|
assert db_url.url == "http://partial_original.com/stream"
|
|
assert db_url.priority_id == 100
|
|
|
|
|
|
def test_update_channel_url_url_not_found(
|
|
db_session: Session, admin_user_client: TestClient
|
|
):
|
|
# Setup priority and create a group
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "URL Not Found Group"
|
|
)
|
|
|
|
channel_data = {
|
|
"tvg_id": "ch_url_not_found.tv",
|
|
"name": "Channel URL Not Found",
|
|
"group_id": str(group_id),
|
|
"tvg_name": "CUNFName",
|
|
"tvg_logo": "cunflogo.png",
|
|
"urls": [],
|
|
}
|
|
create_ch_response = admin_user_client.post("/channels/", json=channel_data)
|
|
created_channel_id = create_ch_response.json()["id"]
|
|
|
|
random_url_uuid = uuid.uuid4()
|
|
update_data = {"url": "http://does_not_matter.com"}
|
|
response = admin_user_client.put(
|
|
f"/channels/{created_channel_id}/urls/{random_url_uuid}", json=update_data
|
|
)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert "URL not found" in response.json()["detail"]
|
|
|
|
|
|
def test_update_channel_url_channel_id_mismatch_is_url_not_found(
|
|
db_session: Session, admin_user_client: TestClient
|
|
):
|
|
# This tests if a URL ID exists but is not associated
|
|
# with the given channel_id in the path
|
|
|
|
# Setup priority and create a group
|
|
group1_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "Ch1 Group"
|
|
)
|
|
|
|
# Create channel 1 with a URL
|
|
ch1_data = {
|
|
"tvg_id": "ch1_url_mismatch.tv",
|
|
"name": "CH1 URL Mismatch",
|
|
"group_id": str(group1_id),
|
|
"tvg_name": "C1UMName",
|
|
"tvg_logo": "c1umlogo.png",
|
|
"urls": [{"url": "http://ch1.url", "priority_id": 100}],
|
|
}
|
|
ch1_resp = admin_user_client.post("/channels/", json=ch1_data)
|
|
url_id_from_ch1 = ch1_resp.json()["urls"][0]["id"]
|
|
|
|
# Create another group
|
|
group2_id = create_mock_priorities_and_group(
|
|
db_session, [(200, "Medium")], "Ch2 Group"
|
|
)
|
|
|
|
# Create channel 2
|
|
ch2_data = {
|
|
"tvg_id": "ch2_url_mismatch.tv",
|
|
"name": "CH2 URL Mismatch",
|
|
"group_id": str(group2_id),
|
|
"tvg_name": "C2UMName",
|
|
"tvg_logo": "c2umlogo.png",
|
|
"urls": [],
|
|
} # priority_id not needed here
|
|
ch2_resp = admin_user_client.post("/channels/", json=ch2_data)
|
|
ch2_id = ch2_resp.json()["id"]
|
|
|
|
# Try to update URL from CH1 using CH2's ID in path
|
|
update_data = {"url": "http://mismatch_update.com"}
|
|
response = admin_user_client.put(
|
|
f"/channels/{ch2_id}/urls/{url_id_from_ch1}", json=update_data
|
|
)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert "URL not found" in response.json()["detail"]
|
|
|
|
|
|
def test_update_channel_url_forbidden_for_non_admin(
|
|
db_session: Session,
|
|
non_admin_user_client: TestClient,
|
|
admin_user_client: TestClient,
|
|
):
|
|
# Setup priority, group and create channel with URL using admin
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "URL Update Forbidden Group"
|
|
)
|
|
|
|
channel_data = {
|
|
"tvg_id": "ch_update_url_forbidden.tv",
|
|
"name": "Channel Update URL Forbidden",
|
|
"group_id": str(group_id),
|
|
"tvg_name": "CUFName",
|
|
"tvg_logo": "cuflgo.png",
|
|
"urls": [{"url": "http://original_forbidden.com/stream", "priority_id": 100}],
|
|
}
|
|
create_ch_response = admin_user_client.post("/channels/", json=channel_data)
|
|
created_channel_id = create_ch_response.json()["id"]
|
|
initial_url_id = create_ch_response.json()["urls"][0]["id"]
|
|
|
|
update_url_data = {"url": "http://attempted_update_forbidden.com"}
|
|
response = non_admin_user_client.put(
|
|
f"/channels/{created_channel_id}/urls/{initial_url_id}", json=update_url_data
|
|
)
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
assert "required roles" in response.json()["detail"]
|
|
|
|
|
|
# --- Test Cases For Delete Channel URL ---
|
|
|
|
|
|
def test_delete_channel_url_success(db_session: Session, admin_user_client: TestClient):
|
|
# Setup priority, group and create a channel with a URL
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "URL Delete Group"
|
|
)
|
|
|
|
channel_data = {
|
|
"tvg_id": "ch_delete_url.tv",
|
|
"name": "Channel Delete URL",
|
|
"group_id": str(group_id),
|
|
"tvg_name": "CDUName",
|
|
"tvg_logo": "cdulogo.png",
|
|
"urls": [{"url": "http://delete_this_url.com/stream", "priority_id": 100}],
|
|
}
|
|
create_ch_response = admin_user_client.post("/channels/", json=channel_data)
|
|
created_channel_id = create_ch_response.json()["id"]
|
|
url_to_delete_id = create_ch_response.json()["urls"][0]["id"]
|
|
|
|
# Verify URL exists before delete
|
|
db_url_before = (
|
|
db_session.query(MockChannelURL)
|
|
.filter(MockChannelURL.id == url_to_delete_id)
|
|
.first()
|
|
)
|
|
assert db_url_before is not None
|
|
|
|
delete_response = admin_user_client.delete(
|
|
f"/channels/{created_channel_id}/urls/{url_to_delete_id}"
|
|
)
|
|
assert delete_response.status_code == status.HTTP_204_NO_CONTENT
|
|
|
|
# Verify URL is gone from DB
|
|
db_url_after = (
|
|
db_session.query(MockChannelURL)
|
|
.filter(MockChannelURL.id == url_to_delete_id)
|
|
.first()
|
|
)
|
|
assert db_url_after is None
|
|
|
|
# Verify channel still exists and has no URLs
|
|
channel_response = admin_user_client.get(f"/channels/{created_channel_id}")
|
|
assert channel_response.status_code == status.HTTP_200_OK
|
|
assert len(channel_response.json()["urls"]) == 0
|
|
|
|
|
|
def test_delete_channel_url_url_not_found(
|
|
db_session: Session, admin_user_client: TestClient
|
|
):
|
|
# Setup priority, group and create a channel with a URL
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "URL Del Not Found Group"
|
|
)
|
|
|
|
channel_data = {
|
|
"tvg_id": "ch_del_url_not_found.tv",
|
|
"name": "Channel Del URL Not Found",
|
|
"group_id": str(group_id),
|
|
"tvg_name": "CDUNFName",
|
|
"tvg_logo": "cdunflogo.png",
|
|
"urls": [],
|
|
}
|
|
create_ch_response = admin_user_client.post("/channels/", json=channel_data)
|
|
created_channel_id = create_ch_response.json()["id"]
|
|
|
|
random_url_uuid = uuid.uuid4()
|
|
response = admin_user_client.delete(
|
|
f"/channels/{created_channel_id}/urls/{random_url_uuid}"
|
|
)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert "URL not found" in response.json()["detail"]
|
|
|
|
|
|
def test_delete_channel_url_channel_id_mismatch_is_url_not_found(
|
|
db_session: Session, admin_user_client: TestClient
|
|
):
|
|
# Setup priority, group 1 and create a channel 1 with a URL
|
|
group1_id = create_mock_priorities_and_group(db_session, [(100, "High")], "G1Del")
|
|
|
|
# Create channel 1 with a URL
|
|
ch1_data = {
|
|
"name": "CH1 Del URL Mismatch",
|
|
"tvg_id": "ch1_del_url_mismatch.tv",
|
|
"tvg_name": "CH1 Del URL Mismatch",
|
|
"tvg_logo": "ch1delogo.png",
|
|
"group_id": str(group1_id),
|
|
"urls": [{"url": "http://ch1del.url", "priority_id": 100}],
|
|
}
|
|
ch1_resp = admin_user_client.post("/channels/", json=ch1_data)
|
|
print(ch1_resp.json())
|
|
url_id_from_ch1 = ch1_resp.json()["urls"][0]["id"]
|
|
|
|
# Setup group 2 and create a channel 2
|
|
group2_id = create_mock_priorities_and_group(db_session, [], "G2Del")
|
|
|
|
# Create channel 2
|
|
ch2_data = {
|
|
"tvg_id": "ch2_del_url_mismatch.tv",
|
|
"name": "CH2 Del URL Mismatch",
|
|
"tvg_name": "CH2 Del URL Mismatch",
|
|
"tvg_logo": "ch2delogo.png",
|
|
"group_id": str(group2_id),
|
|
"urls": [],
|
|
}
|
|
ch2_resp = admin_user_client.post("/channels/", json=ch2_data)
|
|
ch2_id = ch2_resp.json()["id"]
|
|
|
|
# Try to delete URL from CH1 using CH2's ID in path
|
|
response = admin_user_client.delete(f"/channels/{ch2_id}/urls/{url_id_from_ch1}")
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert "URL not found" in response.json()["detail"]
|
|
|
|
# Ensure the original URL on CH1 was not deleted
|
|
db_url_ch1 = (
|
|
db_session.query(MockChannelURL)
|
|
.filter(MockChannelURL.id == url_id_from_ch1)
|
|
.first()
|
|
)
|
|
assert db_url_ch1 is not None
|
|
|
|
|
|
def test_delete_channel_url_forbidden_for_non_admin(
|
|
db_session: Session,
|
|
non_admin_user_client: TestClient,
|
|
admin_user_client: TestClient,
|
|
):
|
|
# Setup priority, group and create a channel with a URL
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "URL Del Forbidden Group"
|
|
)
|
|
|
|
channel_data = {
|
|
"tvg_id": "ch_del_url_forbidden.tv",
|
|
"name": "Channel Del URL Forbidden",
|
|
"group_id": str(group_id),
|
|
"tvg_name": "CDUFName",
|
|
"tvg_logo": "cduflogo.png",
|
|
"urls": [
|
|
{"url": "http://original_del_forbidden.com/stream", "priority_id": 100}
|
|
],
|
|
}
|
|
create_ch_response = admin_user_client.post("/channels/", json=channel_data)
|
|
created_channel_id = create_ch_response.json()["id"]
|
|
initial_url_id = create_ch_response.json()["urls"][0]["id"]
|
|
|
|
response = non_admin_user_client.delete(
|
|
f"/channels/{created_channel_id}/urls/{initial_url_id}"
|
|
)
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
assert "required roles" in response.json()["detail"]
|
|
|
|
# Ensure URL was not deleted
|
|
db_url_not_deleted = (
|
|
db_session.query(MockChannelURL)
|
|
.filter(MockChannelURL.id == initial_url_id)
|
|
.first()
|
|
)
|
|
assert db_url_not_deleted is not None
|
|
|
|
|
|
# --- Test Cases For List Channel URLs ---
|
|
|
|
|
|
def test_list_channel_urls_success(db_session: Session, admin_user_client: TestClient):
|
|
# Setup priorities, group and create a channel with a URL
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High"), (200, "Medium")], "URL List Group"
|
|
)
|
|
|
|
channel_data = {
|
|
"tvg_id": "ch_list_urls.tv",
|
|
"name": "Channel List URLs",
|
|
"group_id": str(group_id),
|
|
"tvg_name": "CLUName",
|
|
"tvg_logo": "clulogo.png",
|
|
"urls": [
|
|
{"url": "http://list_url1.com/stream", "priority_id": 100},
|
|
{"url": "http://list_url2.com/live", "priority_id": 200},
|
|
],
|
|
}
|
|
create_ch_response = admin_user_client.post("/channels/", json=channel_data)
|
|
created_channel_id = create_ch_response.json()["id"]
|
|
|
|
# URLs are added during channel creation,
|
|
# let's get their IDs for assertion if needed
|
|
# For now, we'll just check the count and content based on what was provided.
|
|
|
|
response = admin_user_client.get(f"/channels/{created_channel_id}/urls")
|
|
assert response.status_code == status.HTTP_200_OK
|
|
data = response.json()
|
|
assert len(data) == 2
|
|
|
|
# Check if the URLs returned match what we expect
|
|
# (order might not be guaranteed by default)
|
|
returned_urls_set = {(item["url"], item["priority_id"]) for item in data}
|
|
expected_urls_set = {
|
|
("http://list_url1.com/stream", 100),
|
|
("http://list_url2.com/live", 200),
|
|
}
|
|
assert returned_urls_set == expected_urls_set
|
|
|
|
|
|
def test_list_channel_urls_empty(db_session: Session, admin_user_client: TestClient):
|
|
# Create a channel with no URLs initially
|
|
# No need to set up MockPriority if no URLs with priority_id are being created.
|
|
|
|
# Setup group
|
|
group_id = create_mock_priorities_and_group(db_session, [], "URL List Empty Group")
|
|
|
|
channel_data = {
|
|
"tvg_id": "ch_list_empty_urls.tv",
|
|
"name": "Channel List Empty URLs",
|
|
"group_id": str(group_id),
|
|
"tvg_name": "CLEUName",
|
|
"tvg_logo": "cleulogo.png",
|
|
"urls": [],
|
|
}
|
|
create_ch_response = admin_user_client.post("/channels/", json=channel_data)
|
|
created_channel_id = create_ch_response.json()["id"]
|
|
|
|
response = admin_user_client.get(f"/channels/{created_channel_id}/urls")
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert response.json() == []
|
|
|
|
|
|
def test_list_channel_urls_channel_not_found(
|
|
db_session: Session, admin_user_client: TestClient
|
|
):
|
|
random_channel_uuid = uuid.uuid4()
|
|
response = admin_user_client.get(f"/channels/{random_channel_uuid}/urls")
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
assert "Channel not found" in response.json()["detail"]
|
|
|
|
|
|
def test_list_channel_urls_forbidden_for_non_admin(
|
|
db_session: Session,
|
|
non_admin_user_client: TestClient,
|
|
admin_user_client: TestClient,
|
|
):
|
|
# Setup priority, group and create a channel admin
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "URL Del Forbidden Group"
|
|
)
|
|
|
|
channel_data = {
|
|
"tvg_id": "ch_list_url_forbidden.tv",
|
|
"name": "Channel List URL Forbidden",
|
|
"group_id": str(group_id),
|
|
"tvg_name": "CLUFName",
|
|
"tvg_logo": "cluflogo.png",
|
|
"urls": [{"url": "http://list_url_forbidden.com", "priority_id": 100}],
|
|
}
|
|
create_ch_response = admin_user_client.post("/channels/", json=channel_data)
|
|
created_channel_id = create_ch_response.json()["id"]
|
|
|
|
response = non_admin_user_client.get(f"/channels/{created_channel_id}/urls")
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
assert "required roles" in response.json()["detail"]
|
|
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
assert "required roles" in response.json()["detail"]
|
|
|
|
|
|
# --- Test Cases For Bulk Upload Channels ---
|
|
|
|
|
|
def test_bulk_upload_channels_success(
|
|
db_session: Session, admin_user_client: TestClient
|
|
):
|
|
"""Test successful bulk upload of channels"""
|
|
# Create priorities
|
|
priorities = [(100, "High"), (200, "Medium"), (300, "Low")]
|
|
for id, desc in priorities:
|
|
db_session.add(MockPriority(id=id, description=desc))
|
|
db_session.commit()
|
|
|
|
# Test data matching channels.json format
|
|
channels = [
|
|
{
|
|
"name": "Channel 1",
|
|
"group-title": "Group A",
|
|
"tvg-id": "channel1.tv",
|
|
"tvg-name": "Channel One",
|
|
"tvg-logo": "logo1.png",
|
|
"urls": ["http://stream1.com", "http://backup1.com"],
|
|
},
|
|
{
|
|
"name": "Channel 2",
|
|
"group-title": "Group B",
|
|
"tvg-id": "channel2.tv",
|
|
"tvg-name": "Channel Two",
|
|
"tvg-logo": "logo2.png",
|
|
"urls": ["http://stream2.com"],
|
|
},
|
|
]
|
|
|
|
response = admin_user_client.post("/channels/bulk-upload", json=channels)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert response.json()["processed"] == 2
|
|
|
|
# Verify groups created
|
|
groups = db_session.query(MockGroup).all()
|
|
assert len(groups) == 2
|
|
assert {g.name for g in groups} == {"Group A", "Group B"}
|
|
|
|
# Verify channels and URLs
|
|
channel1 = db_session.query(MockChannelDB).filter_by(name="Channel 1").first()
|
|
assert channel1 is not None
|
|
assert len(channel1.urls) == 2
|
|
assert channel1.urls[0].url == "http://stream1.com"
|
|
assert channel1.urls[0].priority_id == 100
|
|
assert channel1.urls[1].url == "http://backup1.com"
|
|
assert channel1.urls[1].priority_id == 200
|
|
|
|
channel2 = db_session.query(MockChannelDB).filter_by(name="Channel 2").first()
|
|
assert channel2 is not None
|
|
assert len(channel2.urls) == 1
|
|
assert channel2.urls[0].url == "http://stream2.com"
|
|
assert channel2.urls[0].priority_id == 100
|
|
|
|
|
|
def test_bulk_upload_empty(db_session, admin_user_client):
|
|
"""Test bulk upload with empty list"""
|
|
response = admin_user_client.post("/channels/bulk-upload", json=[])
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert response.json()["processed"] == 0
|
|
|
|
|
|
def test_bulk_upload_forbidden_for_non_admin(db_session, non_admin_user_client):
|
|
"""Test bulk upload requires admin role"""
|
|
response = non_admin_user_client.post("/channels/bulk-upload", json=[{}])
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
|
|
|
|
def test_bulk_upload_skip_missing_group_name(db_session, admin_user_client):
|
|
"""Test bulk upload skips channels with missing group name"""
|
|
channel_data = [{"name": "No Group Channel", "urls": ["http://no-group.com"]}]
|
|
|
|
response = admin_user_client.post("/channels/bulk-upload", json=channel_data)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert response.json()["processed"] == 0
|
|
assert db_session.query(MockChannelDB).count() == 0
|
|
|
|
|
|
def test_bulk_upload_single_url_conversion(db_session, admin_user_client):
|
|
"""Test bulk upload converts single URL string to list"""
|
|
# Create priorities
|
|
priorities = [(100, "High"), (200, "Medium"), (300, "Low")]
|
|
for id, desc in priorities:
|
|
db_session.add(MockPriority(id=id, description=desc))
|
|
db_session.commit()
|
|
|
|
channel_data = [
|
|
{
|
|
"name": "Single URL Channel",
|
|
"group-title": "Test Group",
|
|
"urls": ["http://single-url.com"],
|
|
}
|
|
]
|
|
|
|
response = admin_user_client.post("/channels/bulk-upload", json=channel_data)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert response.json()["processed"] == 1
|
|
|
|
channel = db_session.query(MockChannelDB).first()
|
|
assert channel is not None
|
|
urls = db_session.query(MockChannelURL).filter_by(channel_id=channel.id).all()
|
|
assert len(urls) == 1
|
|
assert urls[0].url == "http://single-url.com"
|
|
|
|
|
|
def test_bulk_upload_update_existing_channel(db_session, admin_user_client):
|
|
"""Test bulk upload updates existing channel and recreates URLs"""
|
|
# First create a group and channel
|
|
group_id = create_mock_priorities_and_group(
|
|
db_session, [(100, "High")], "Existing Group"
|
|
)
|
|
channel = MockChannelDB(
|
|
name="Existing Channel",
|
|
group_id=group_id,
|
|
tvg_id="existing.tv",
|
|
tvg_name="Existing",
|
|
tvg_logo="existing.png",
|
|
)
|
|
db_session.add(channel)
|
|
db_session.commit()
|
|
db_session.refresh(channel)
|
|
|
|
# Add some URLs
|
|
db_url1 = MockChannelURL(
|
|
channel_id=channel.id, url="http://old1.com", priority_id=100, in_use=False
|
|
)
|
|
db_url2 = MockChannelURL(
|
|
channel_id=channel.id, url="http://old2.com", priority_id=200, in_use=False
|
|
)
|
|
db_session.add_all([db_url1, db_url2])
|
|
db_session.commit()
|
|
|
|
# Bulk upload with same group and channel name but different URLs
|
|
channel_data = [
|
|
{
|
|
"name": "Existing Channel",
|
|
"group-title": "Existing Group",
|
|
"tvg-id": "updated.tv",
|
|
"tvg-name": "Updated",
|
|
"tvg-logo": "updated.png",
|
|
"urls": ["http://new1.com", "http://new2.com"],
|
|
}
|
|
]
|
|
|
|
response = admin_user_client.post("/channels/bulk-upload", json=channel_data)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert response.json()["processed"] == 1
|
|
|
|
# Verify channel was updated
|
|
print(
|
|
"Channel after update:",
|
|
db_session.query(MockChannelDB).filter().first().id,
|
|
channel.tvg_id,
|
|
)
|
|
updated_channel = db_session.query(MockChannelDB).filter_by(id=channel.id).first()
|
|
assert updated_channel.tvg_id == "updated.tv"
|
|
assert updated_channel.tvg_name == "Updated"
|
|
assert updated_channel.tvg_logo == "updated.png"
|
|
|
|
# Verify old URLs were deleted and new ones created
|
|
urls = db_session.query(MockChannelURL).filter_by(channel_id=channel.id).all()
|
|
assert len(urls) == 2
|
|
assert {url.url for url in urls} == {"http://new1.com", "http://new2.com"}
|
|
|
|
|
|
def test_bulk_upload_error_handling(db_session, admin_user_client):
|
|
"""Test error handling in bulk upload continues processing"""
|
|
# Create a group first
|
|
create_mock_priorities_and_group(db_session, [(100, "High")], "Error Group")
|
|
|
|
channel_data = [
|
|
{
|
|
"name": "Good Channel",
|
|
"group-title": "Error Group",
|
|
"urls": ["http://good.com"],
|
|
},
|
|
{
|
|
"name": "Bad Channel",
|
|
"group-title": "Error Group",
|
|
"urls": None, # This will cause an error
|
|
},
|
|
{
|
|
"name": "Another Good Channel",
|
|
"group-title": "Error Group",
|
|
"urls": ["http://another-good.com"],
|
|
},
|
|
]
|
|
|
|
response = admin_user_client.post("/channels/bulk-upload", json=channel_data)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert response.json()["processed"] == 2 # Only 2 successful
|
|
|
|
# Verify the good channels were processed
|
|
channels = db_session.query(MockChannelDB).all()
|
|
assert len(channels) == 2
|
|
assert {c.name for c in channels} == {"Good Channel", "Another Good Channel"}
|
|
|
|
|
|
def test_bulk_upload_channels_no_priorities(
|
|
db_session: Session, admin_user_client: TestClient
|
|
):
|
|
"""Test bulk upload when no priorities are defined in the database.
|
|
The channel should be processed, but its URLs should be skipped.
|
|
"""
|
|
# Ensure no priorities exist in the database
|
|
db_session.query(MockPriority).delete()
|
|
db_session.commit()
|
|
|
|
# Create a group for the channel
|
|
group = MockGroup(name="No Priority Group")
|
|
db_session.add(group)
|
|
db_session.commit()
|
|
db_session.refresh(group)
|
|
|
|
channel_data = [
|
|
{
|
|
"name": "Channel With No Priority URL",
|
|
"group-title": "No Priority Group",
|
|
"tvg-id": "nopriority.tv",
|
|
"tvg-name": "NoPriorityChannel",
|
|
"tvg-logo": "nopriority.png",
|
|
"urls": ["http://nopriority.com/stream"],
|
|
}
|
|
]
|
|
|
|
# Call the bulk upload endpoint
|
|
response = admin_user_client.post("/channels/bulk-upload", json=channel_data)
|
|
|
|
# Assert the response indicates 1 processed channel
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert response.json()["processed"] == 1
|
|
|
|
# Verify the channel was added to the database
|
|
db_channel = (
|
|
db_session.query(MockChannelDB)
|
|
.filter_by(name="Channel With No Priority URL")
|
|
.first()
|
|
)
|
|
assert db_channel is not None
|
|
assert db_channel.group_id == group.id
|
|
|
|
# Verify no URLs were added for this channel
|
|
assert (
|
|
db_session.query(MockChannelURL).filter_by(channel_id=db_channel.id).count()
|
|
== 0
|
|
)
|