wrolpi/modules/videos/channel/test/test_models.py

import json
from http import HTTPStatus

import pytest

from modules.videos.models import Channel
from modules.videos.models import Video
from wrolpi.downloader import Download, DownloadFrequency


def test_delete_channel_no_url(test_session, channel_factory):
    """
    A Channel can be deleted even if it has no URL.
    """
    channel = channel_factory()
    channel.url = None
    test_session.commit()

    channel.delete_with_videos()


@pytest.mark.asyncio
async def test_delete_channel_with_videos(test_session, async_client, channel_factory, video_factory):
    """Videos are disowned when their Channel is deleted. The Collection should also be deleted."""
    from wrolpi.collections import Collection

    channel = channel_factory()
    video = video_factory(channel_id=channel.id)
    video_path, video_id = video.video_path, video.id
    collection_id = channel.collection_id
    test_session.commit()

    # Delete the Channel.
    request, response = await async_client.delete(f'/api/videos/channels/{channel.id}')
    assert response.status_code == HTTPStatus.NO_CONTENT

    # Video still exists, but has no channel.
    video: Video = test_session.query(Video).one()
    assert video.video_path == video_path and video.id == video_id, 'Video entry should not be changed'
    assert video.video_path.is_file(), 'Video should not be deleted'
    assert not video.channel_id and not video.channel, 'Video should not have a Channel'

    # Collection should also be deleted (this is the bug we're fixing)
    assert test_session.query(Collection).filter_by(id=collection_id).count() == 0, \
        'Collection should be deleted when Channel is deleted'


@pytest.mark.asyncio
async def test_nested_channel_directories(test_session, async_client, test_directory, channel_factory,
                                          video_factory):
    """Channel directories cannot contain another Channel's directory."""
    (test_directory / 'foo').mkdir()
    channel1_directory = test_directory / 'foo/one'
    channel_factory(directory=channel1_directory)
    test_session.commit()

    # Channel 2 cannot be in Channel 1's directory.
    channel2_directory = channel1_directory / 'two'
    channel2_directory.mkdir()
    content = dict(
        name='channel 2',
        directory=str(channel2_directory),
    )
    request, response = await async_client.post('/api/videos/channels', content=json.dumps(content))
    assert response.status_code == HTTPStatus.BAD_REQUEST
    assert response.json['cause'] and \
           response.json['cause']['message'] == 'The directory is already used by another channel.'

    # Channel 3 cannot be above Channel 1's directory.
    content = dict(
        name='channel 3',
        directory=str(test_directory),
    )
    request, response = await async_client.post('/api/videos/channels', content=json.dumps(content))
    assert response.status_code == HTTPStatus.BAD_REQUEST
    assert response.json['cause'] and \
           response.json['cause']['message'] == 'The directory is already used by another channel.'


@pytest.mark.asyncio()
async def test_channel_download_relationships(test_session, download_channel):
    """Test relationships of Channel and Download."""
    test_session.flush()

    download = test_session.query(Download).one()
    channel = test_session.query(Channel).one()
    assert len(channel.downloads) == 1
    assert channel.downloads
    assert download in channel.downloads

    test_session.commit()

    download: Download = test_session.query(Download).one()
    channel: Channel = test_session.query(Channel).one()
    assert len(channel.downloads) == 1
    assert channel.downloads[0] == download and channel.downloads[0].url == 'https://example.com/channel1'
    assert download.frequency == DownloadFrequency.weekly
    assert download.destination == channel.directory

    # Deleting Download deletes the Download, but not the Channel.
    download.delete()
    assert not test_session.query(Download).all()
    assert test_session.query(Channel).one(), 'Channel should not have been deleted.'

    # Create a Download again.
    download_channel.get_or_create_download(test_session, 'https://example.com/1', 60, reset_attempts=True)
    assert test_session.query(Download).count() == 1
    # Delete the Channel, and any relationships.
    download_channel.delete_with_videos()
    assert test_session.query(Download).count() == 0


@pytest.mark.asyncio
async def test_channel_statistics_with_tagged_videos(async_client, test_session, channel_factory, video_factory,
                                                     tag_factory):
    """Tagging a Video must not inflate its Channel's statistics.

    A Video with multiple Tags was counted once per Tag because of the tag_file join."""
    channel = channel_factory()
    video1 = video_factory(channel_id=channel.id)
    video2 = video_factory(channel_id=channel.id)
    video3 = video_factory(channel_id=channel.id)
    test_session.commit()

    expected_size = video1.file_group.size + video2.file_group.size + video3.file_group.size

    # Statistics are correct before any Videos are tagged.
    statistics = channel.get_statistics()
    assert statistics['video_count'] == 3
    assert statistics['size'] == expected_size
    assert statistics['video_tags'] == 0

    # video1 has two Tags, video2 has one Tag, video3 has none.
    tag1, tag2 = await tag_factory(), await tag_factory()
    video1.add_tag(test_session, tag1.name)
    video1.add_tag(test_session, tag2.name)
    video2.add_tag(test_session, tag1.name)
    test_session.commit()

    # Tagging Videos must not change the number of Videos in the Channel.
    statistics = channel.get_statistics()
    assert statistics['video_count'] == 3
    assert statistics['size'] == expected_size
    assert statistics['video_tags'] == 2


def test_channel_directory_video_counts(test_session, channel_factory, video_factory):
    """A Channel can count the video FileGroups in its directory, and which Channel they are assigned to.

    These counts are logged after a Channel refresh to help debug Videos missing from their Channel."""
    channel1 = channel_factory()
    channel2 = channel_factory()
    video_factory(channel_id=channel1.id)
    video_factory(channel_id=channel1.id)
    # These Videos' files are in channel1's directory, but they are not assigned to channel1.
    other_channel_video = video_factory(with_video_file=channel1.directory / 'other channel.mp4')
    unassigned_video = video_factory(with_video_file=channel1.directory / 'unassigned.mp4')
    other_channel_video.channel_id = channel2.id
    unassigned_video.channel_id = None
    test_session.commit()

    assert channel1.get_directory_video_counts() == dict(
        video_file_groups=4,
        assigned_to_channel=2,
        assigned_to_other_channels=1,
        unassigned=1,
    )

    # channel2's directory contains no files.
    assert channel2.get_directory_video_counts() == dict(
        video_file_groups=0,
        assigned_to_channel=0,
        assigned_to_other_channels=0,
        unassigned=0,
    )


def test_channel_info_json(test_session, test_directory):
    from wrolpi.collections import Collection

    # Create Collection first without directory
    collection = Collection(name='new channel', kind='channel')
    test_session.add(collection)
    test_session.flush()

    channel = Channel(collection_id=collection.id)
    test_session.add(channel)
    test_session.flush()

    with pytest.raises(FileNotFoundError) as e:
        # Cannot get info json because directory is not defined.
        channel.info_json_path
    assert 'directory' in str(e)

    # Channel's directory is created if it does not exist.
    channel.directory = test_directory / 'new channel'
    assert channel.info_json_path == test_directory / 'new channel/new channel.info.json'
    assert channel.directory.is_dir()
    # info json file is not yet created.
    assert not channel.info_json_path.exists()