wrolpi/modules/videos/test/test_common.py

import pathlib
import subprocess
import tempfile
from copy import copy
from unittest import mock

import pytest
from PIL import Image

from modules.videos.models import Channel, Video
from wrolpi.collections import Collection
from wrolpi.conftest import config_was_imported, probe_write_lock_is_held, production_like_sessions
from wrolpi.common import get_absolute_media_path, get_wrolpi_config, get_relative_to_media_directory
from wrolpi.downloader import Download, DownloadFrequency
from wrolpi.vars import PROJECT_DIR
from .. import common
from ..common import convert_image, update_view_counts_and_censored, extract_video_duration, generate_video_poster, \
    is_valid_poster
from ..lib import save_channels_config, get_channels_config, import_channels_config, ChannelsConfig


def test_get_absolute_media_path(test_directory):
    path = get_absolute_media_path('videos')
    assert str(path) == f'{test_directory}/videos'

    # An absolute path is not changed.
    path = get_absolute_media_path(path)
    assert str(path) == f'{test_directory}/videos'


def test_extract_video_duration(test_directory, corrupted_video_file):
    """
    Video duration can be retrieved from the video file.
    """
    video_path = PROJECT_DIR / 'test/big_buck_bunny_720p_1mb.mp4'
    assert extract_video_duration(video_path) == 5

    with pytest.raises(FileNotFoundError):
        extract_video_duration(PROJECT_DIR / 'test/does not exist.mp4')

    with pytest.raises(subprocess.CalledProcessError):
        empty_file = test_directory / 'empty.mp4'
        empty_file.touch()
        extract_video_duration(empty_file)

    with pytest.raises(subprocess.CalledProcessError):
        extract_video_duration(corrupted_video_file)


def test_convert_image():
    """
    An image's format can be changed.  This tests that convert_image() converts from a WEBP format, to a JPEG format.
    """
    foo = Image.new('RGB', (25, 25), color='grey')

    with tempfile.TemporaryDirectory() as tempdir:
        tempdir = pathlib.Path(tempdir)

        # Save the new image to "foo.webp".
        existing_path = tempdir / 'foo.webp'
        foo.save(existing_path)
        assert Image.open(existing_path).format == 'WEBP'

        destination_path = tempdir / 'foo.jpg'
        assert not destination_path.is_file()

        # Convert the WEBP to a JPEG.  The WEBP image should be removed.
        convert_image(existing_path, destination_path)
        assert not existing_path.is_file()
        assert destination_path.is_file()
        assert Image.open(destination_path).format == 'JPEG'


@pytest.mark.parametrize(
    'ext,expected',
    [
        ('jpg', True),
        ('jpeg', True),
        ('png', False),
        ('webp', False),
        ('tif', False),
    ]
)
def test_is_valid_poster(ext, expected, test_directory):
    foo = Image.new('RGB', (25, 25), color='blue')
    image_path = test_directory / f'image.{ext}'
    foo.save(image_path)
    assert is_valid_poster(image_path) == expected, f'is_valid_poster({image_path}) should be {expected}'


@pytest.mark.asyncio
async def test_update_view_count(test_session, channel_factory, video_factory):
    def check_view_counts(view_counts):
        for source_id, view_count in view_counts.items():
            video = test_session.query(Video).filter_by(source_id=source_id).one()
            assert video.view_count == view_count

    channel1, channel2, channel3 = channel_factory(), channel_factory(), channel_factory()
    video_factory(channel_id=channel1.id, with_poster_ext='jpg', title='vid1')
    video_factory(channel_id=channel2.id, title='vid2')
    video_factory(channel_id=channel3.id, with_poster_ext='jpg', title='vid3')
    video_factory(channel_id=channel3.id, title='vid4')
    channel1.info_json = {'entries': [{'id': 'vid1', 'view_count': 10}]}
    channel2.info_json = {'entries': [{'id': 'vid2', 'view_count': 11}, {'id': 'bad_id', 'view_count': 12}]}
    channel3.info_json = {'entries': [{'id': 'vid3', 'view_count': 13}, {'id': 'vid4', 'view_count': 14}]}
    test_session.commit()

    # Check all videos are empty.
    check_view_counts({'vid1': None, 'vid2': None, 'vid3': None, 'vid4': None})

    # Channel 1 is updated, the other channels are left alone.
    await update_view_counts_and_censored(channel1.id)
    check_view_counts({'vid1': 10, 'vid2': None, 'vid3': None, 'vid4': None})

    # Channel 2 is updated, the other channels are left alone.  The 'bad_id' video is ignored.
    await update_view_counts_and_censored(channel2.id)
    check_view_counts({'vid1': 10, 'vid2': 11, 'vid3': None, 'vid4': None})

    # All videos are updated.
    await update_view_counts_and_censored(channel3.id)
    check_view_counts({'vid1': 10, 'vid2': 11, 'vid3': 13, 'vid4': 14})

    # An outdated view count will be overwritten.
    vid = test_session.query(Video).filter_by(id=1).one()
    vid.view_count = 8
    test_session.commit()  # Commit so raw SQL in update_view_counts_and_censored doesn't deadlock
    check_view_counts({'vid1': 8, 'vid2': 11, 'vid3': 13, 'vid4': 14})
    await update_view_counts_and_censored(channel1.id)
    test_session.expire_all()  # Expire cached values to see raw SQL changes
    check_view_counts({'vid1': 10, 'vid2': 11, 'vid3': 13, 'vid4': 14})


@pytest.mark.asyncio
async def test_update_view_count_missing_view_count(test_session, channel_factory, video_factory):
    """Entries without a view_count key (e.g. private/unavailable videos) are skipped without error."""
    channel = channel_factory()
    video_factory(channel_id=channel.id, with_poster_ext='jpg', title='vid1')
    channel.info_json = {'entries': [
        {'id': 'vid1', 'view_count': 42},
        {'id': 'vid_private'},  # No view_count key.
        {'id': 'vid_null', 'view_count': None},
    ]}
    test_session.commit()

    # Should not raise KeyError.
    await update_view_counts_and_censored(channel.id)

    test_session.expire_all()
    video = test_session.query(Video).filter_by(source_id='vid1').one()
    assert video.view_count == 42


def test_generate_video_poster(video_file):
    """
    A poster can be generated from a video file.
    """
    poster_path = video_file.with_suffix('.jpg')
    _, duration = generate_video_poster(video_file)
    assert poster_path.is_file(), f'{poster_path} was not created!'
    assert poster_path.stat().st_size > 0
    assert duration == 5


@pytest.mark.asyncio
async def test_import_channel_downloads(await_switches, test_session, channel_factory, test_channels_config,
                                        tag_factory):
    """Importing the Channels' config should create any missing download records"""
    channel1 = channel_factory(source_id='foo', url='https://example.com/channel1')
    channel2 = channel_factory(source_id='bar', url='https://example.com/channel2')
    tag = await tag_factory()
    assert len(test_session.query(Channel).all()) == 2
    test_session.commit()

    def update_channel_config(conf: ChannelsConfig, source_id, d):
        # Creat a copy of the old config, replace the data of the provided Channel.
        old_config = conf._config.copy()
        for c in old_config['channels']:
            if c['source_id'] == source_id:
                c.update(d)
        conf.channels = old_config['channels']
        conf.save()

    # Two Channels, but neither have Downloads.
    save_channels_config()
    await await_switches()
    # Import after config is saved.
    import_channels_config()
    assert not channel1.downloads
    assert not channel2.downloads
    assert len(test_session.query(Channel).all()) == 2
    assert test_session.query(Download).count() == 0

    # Add a frequency to `channel1`.
    channels_config = get_channels_config()
    update_channel_config(channels_config, 'foo',
                          {'downloads': [
                              {'url': 'https://example.com/channel1', 'frequency': DownloadFrequency.weekly.value}
                          ]})
    await await_switches()

    # Download record is created on import.
    import_channels_config()
    assert channel1.downloads
    assert len(test_session.query(Channel).all()) == 2, 'Both Channels should still exist'
    assert test_session.query(Download).count() == 1, 'One Channel has a Download'
    download: Download = test_session.query(Download).one()
    assert download.url == channel1.url
    assert download.frequency
    assert download.downloader == 'video_channel'

    # Download.next_download was not deleted.
    next_download = str(download.next_download)
    import_channels_config()
    download: Download = test_session.query(Download).one()
    assert next_download == str(download.next_download)

    # Creating Download that matches Channel2's URL means they are related.  Delete it and it should be re-created.
    channel2.get_or_create_download(test_session, channel2.url, 60)
    save_channels_config()
    Download.find_by_url(test_session, channel2.url).delete(add_to_skip_list=False)
    test_session.commit()
    await await_switches()

    # Missing Download is re-created on import.
    import_channels_config()
    channel2 = Channel.get_by_id(test_session, channel2.id)
    downloads = test_session.query(Download).all()
    assert len(downloads) == 1, downloads
    assert len(channel1.downloads) == 1
    assert len(channel2.downloads) == 0
    assert downloads[0].downloader == 'video_channel'

    # Add a Download to Channel2 which does not match Channel.url.
    channel2.get_or_create_download(test_session, 'https://example.org', 60)
    channel2.get_or_create_download(test_session, 'https://example.com/channel2', 60)
    test_session.commit()
    save_channels_config()
    await await_switches()
    # Check config is written to match new URLs.
    config = get_channels_config()
    assert len(config.channels) == 2
    for channel_config in config.channels:
        if channel_config['source_id'] == channel2.source_id:
            assert channel_config['source_id'] == 'bar'
            assert len(channel_config['downloads']) == 2
            assert {i['url'] for i in channel_config['downloads']} \
                   == {'https://example.com/channel2', 'https://example.org'}
        else:
            assert channel_config['source_id'] == 'foo'

    # Get Channels with their Downloads.
    channels_backup = copy(get_channels_config().channels)
    # Reset Downloads.  Downloads should be recreated from the channels config file.
    [i.delete(add_to_skip_list=False) for i in test_session.query(Download).all()]
    # Write config with Channel Downloads now that the delete() has removed them from the config.
    get_channels_config().channels = channels_backup
    # Deleting the Downloads made the database newer than the file, so restoring from the file is the user's choice.
    get_channels_config().import_forced()
    channel1, channel2 = test_session.query(Channel).order_by(Channel.url).all()
    assert test_session.query(Download).count() == 3
    assert channel1.url == 'https://example.com/channel1'
    assert channel2.url == 'https://example.com/channel2'
    assert {i.url for i in channel1.downloads} == {'https://example.com/channel1'}
    assert {i.url for i in channel2.downloads} == {'https://example.com/channel2', 'https://example.org'}

    # A Channel with a `tag_name` is tagged.
    update_channel_config(get_channels_config(),
                          channel1.source_id,
                          {'tag_name': tag.name})
    import_channels_config()
    channel1, channel2 = test_session.query(Channel).order_by(Channel.url).all()
    assert channel1.tag == tag and channel1.tag_id == tag.id
    assert channel2.tag is None and channel2.tag_id is None


@pytest.mark.asyncio
async def test_import_channel_delete_missing_channels(await_switches, test_session, channel_factory,
                                                      test_channels_config):
    """The Channel import function deletes any Channels that are not in the config."""
    # Create a DB and config with two Channels.
    channel1 = channel_factory(source_id='foo')
    channel2 = channel_factory(source_id='bar')
    test_session.commit()

    # Capture directory values before deletion (channels will be detached after deletion)
    # Config stores relative paths for portability
    channel1_dir = str(get_relative_to_media_directory(channel1.directory))
    channel2_dir = str(get_relative_to_media_directory(channel2.directory))

    # Write Channels to the config file.
    save_channels_config()
    test_session.delete(channel1)
    test_session.delete(channel2)
    test_session.commit()
    await await_switches()

    # Importing the config creates two Channels.
    import_channels_config()
    assert len(test_session.query(Channel).all()) == 2
    assert channel1_dir in test_channels_config.read_text()
    assert channel2_dir in test_channels_config.read_text()

    # Delete channel2 from the config file.
    config = get_channels_config()
    config_dict = config.dict()
    config_dict['channels'] = [i for i in config.channels if i['directory'] != channel2_dir]
    config.update(config_dict)
    await await_switches()
    assert channel1_dir in test_channels_config.read_text()
    assert channel2_dir not in test_channels_config.read_text()

    # Importing the config deletes the Channel record.
    import_channels_config()
    assert len(test_session.query(Channel).all()) == 1
    assert channel1_dir in test_channels_config.read_text()
    assert channel2_dir not in test_channels_config.read_text()

    # Saving and importing does not change anything.
    save_channels_config()
    import_channels_config()
    assert channel1_dir in test_channels_config.read_text()
    assert channel2_dir not in test_channels_config.read_text()


@pytest.mark.asyncio
async def test_import_channel_deletes_orphaned_collections(await_switches, test_session, channel_factory,
                                                           test_channels_config):
    """When a Channel is deleted from config, its associated Collection should also be deleted.

    This prevents orphaned Collections (kind='channel') from appearing in the UI with null channel_id.
    """
    # Create two channels (each creates a Collection with kind='channel')
    channel1 = channel_factory(source_id='foo')
    channel2 = channel_factory(source_id='bar')
    test_session.commit()

    # Verify both Collections exist
    assert test_session.query(Collection).filter_by(kind='channel').count() == 2

    # Capture directory values before deletion (config stores relative paths)
    channel1_dir = str(get_relative_to_media_directory(channel1.directory))
    channel2_dir = str(get_relative_to_media_directory(channel2.directory))
    collection2_id = channel2.collection_id

    # Write Channels to the config file
    save_channels_config()
    await await_switches()

    # Remove channel2 from the config file
    config = get_channels_config()
    config_dict = config.dict()
    config_dict['channels'] = [i for i in config.channels if i['directory'] != channel2_dir]
    config.update(config_dict)
    await await_switches()

    # Import config - this should delete channel2 AND its Collection
    config_was_imported(test_session, 'channels.yaml')
    import_channels_config()

    # Verify only one Channel remains
    assert test_session.query(Channel).count() == 1

    # Verify the orphaned Collection was also deleted (this is the bug we're fixing)
    assert test_session.query(Collection).filter_by(kind='channel').count() == 1
    assert test_session.query(Collection).filter_by(id=collection2_id).count() == 0


@pytest.mark.asyncio
async def test_import_channel_cleans_up_preexisting_orphaned_collections(await_switches, test_session, channel_factory,
                                                                          test_channels_config):
    """Pre-existing orphaned Collections (kind='channel') with no Channel should be deleted during import.

    This handles the case where a Collection was orphaned before the cleanup fix was deployed.
    """
    # Create a channel (which creates a Collection)
    channel1 = channel_factory(source_id='foo')
    test_session.commit()

    # Create an orphaned Collection directly (simulating pre-existing orphan)
    orphan_collection = Collection(
        name='Orphaned Channel',
        kind='channel',
        directory=None,
    )
    test_session.add(orphan_collection)
    test_session.commit()
    orphan_id = orphan_collection.id

    # Verify we have 2 Collections with kind='channel'
    assert test_session.query(Collection).filter_by(kind='channel').count() == 2

    # Write the valid channel to config
    save_channels_config()
    await await_switches()

    # Import config - should clean up the pre-existing orphan
    import_channels_config()

    # Verify only one Collection remains (the one with a Channel)
    assert test_session.query(Collection).filter_by(kind='channel').count() == 1
    assert test_session.query(Collection).filter_by(id=orphan_id).count() == 0


@pytest.mark.asyncio
async def test_once_download_excluded_from_channel_config(await_switches, test_session, channel_factory,
                                                          test_channels_config):
    """A once-download (no frequency) associated with a Channel's Collection should not be saved to the channels
    config.  This prevents errors during import when a once-download has no frequency."""
    channel = channel_factory(download_frequency=DownloadFrequency.weekly)

    # Create a once-download associated with the channel's collection (simulates a video download that got
    # associated with the channel via apply_yt_channel).
    once_download = Download(url='https://www.youtube.com/watch?v=once_video', downloader='video',
                             collection_id=channel.collection_id)
    test_session.add(once_download)
    test_session.commit()
    # The once-download is in the Collection, but Channel.downloads filters it out.
    assert len(channel.collection.downloads) == 2
    assert len(channel.downloads) == 1

    save_channels_config()
    await await_switches()

    # The once-download should not appear in the config.
    config_text = test_channels_config.read_text()
    assert 'once_video' not in config_text
    # The recurring download should still be there.
    assert channel.url in config_text

    # Importing should succeed without the "Refusing to create Download" error.
    test_session.delete(once_download)
    test_session.commit()
    import_channels_config()
    assert len(channel.downloads) == 1


@pytest.mark.asyncio
async def test_import_channel_download_comments(await_switches, test_session, channel_factory,
                                                test_channels_config):
    channel = channel_factory()
    assert channel.download_missing_data is True

    # download_missing_data is saved to the config.
    save_channels_config()
    await await_switches()
    assert 'download_missing_data: true' in test_channels_config.read_text()

    # change download_missing_data to False, channel should be updated on import.
    contents = test_channels_config.read_text()
    contents = contents.replace('download_missing_data: true', 'download_missing_data: false')
    test_channels_config.write_text(contents)
    import_channels_config()
    assert channel.download_missing_data is False


@pytest.mark.asyncio
async def test_save_channels_config_does_not_wipe_config(await_switches, test_session, channel_factory,
                                                          test_channels_config):
    """Saving channels config when DB is empty must not overwrite an existing config file that has channels.

    Regression test: a fresh DB (e.g. after repair) triggering save_channels_config would write channels: []
    to the config file, permanently wiping the channel definitions.
    """
    # Create a channel and save it to the config file.
    channel = channel_factory()
    test_session.commit()
    save_channels_config()
    await await_switches()
    assert test_channels_config.is_file()
    config_text = test_channels_config.read_text()
    assert 'channels: []' not in config_text, 'Config should have a channel'

    # Simulate a fresh DB by deleting all channels.
    test_session.query(Channel).delete()
    test_session.query(Collection).filter_by(kind='channel').delete()
    test_session.commit()
    assert test_session.query(Channel).count() == 0

    # Saving with an empty DB should NOT wipe the config file.
    save_channels_config()
    await await_switches()
    config_text = test_channels_config.read_text()
    assert 'channels: []' not in config_text, 'Config was wiped by saving with empty DB!'


@pytest.mark.asyncio
async def test_ffprobe_json(async_client, video_file, corrupted_video_file):
    content = await common.ffprobe_json(video_file)
    assert not content['chapters']
    assert content['format']['duration'] == '5.312000'
    assert content['format']['size'] == '1056318'
    assert content['streams']
    assert content['streams'][0]['codec_name'] == 'h264'

    with pytest.raises(RuntimeError):
        await common.ffprobe_json(corrupted_video_file)


def test_read_write_ffprobe_json_file(test_directory):
    """Test reading and writing .ffprobe.json files."""
    ffprobe_path = test_directory / 'video.ffprobe.json'
    ffprobe_data = {'streams': [{'codec_name': 'h264'}], 'format': {'duration': '5.0'}}

    # File doesn't exist, should return None
    assert common.read_ffprobe_json_file(ffprobe_path) is None
    assert common.read_ffprobe_json_file(None) is None

    # Write the file
    common.write_ffprobe_json_file(ffprobe_path, ffprobe_data)
    assert ffprobe_path.is_file()

    # File should be human-readable (indented)
    contents = ffprobe_path.read_text()
    assert '\n' in contents, 'ffprobe json should be formatted with newlines'

    # Read it back
    loaded = common.read_ffprobe_json_file(ffprobe_path)
    assert loaded == ffprobe_data

    # Invalid JSON should return None
    ffprobe_path.write_text('not valid json')
    assert common.read_ffprobe_json_file(ffprobe_path) is None


@pytest.mark.asyncio
async def test_video_ffprobe_json(async_client, test_session, video_file, refresh_files):
    """ffprobe data is extracted when a video is modeled."""
    with mock.patch('modules.videos.lib.extract_video_duration') as mock_extract_video_duration:
        mock_extract_video_duration.side_effect = Exception('duration should be from ffprobe json')
        await refresh_files()

    video = test_session.query(Video).one()
    assert video.ffprobe_json
    assert video.file_group.length


@pytest.mark.asyncio
async def test_video_ffprobe_json_caches_to_file(async_client, test_session, video_file, refresh_files):
    """ffprobe results are cached to .ffprobe.json file."""
    await refresh_files()

    video = test_session.query(Video).one()

    # .ffprobe.json file should have been created during validation
    ffprobe_path = video.ffprobe_json_path
    assert ffprobe_path.is_file(), '.ffprobe.json file should have been created'

    # File should contain the ffprobe data
    cached_data = common.read_ffprobe_json_file(ffprobe_path)
    assert cached_data is not None
    assert 'streams' in cached_data
    assert 'format' in cached_data


@pytest.mark.asyncio
async def test_video_ffprobe_json_reads_from_cache(async_client, test_session, video_file, refresh_files):
    """get_ffprobe_json() uses cached .ffprobe.json file instead of running ffprobe."""
    await refresh_files()

    video = test_session.query(Video).one()
    ffprobe_path = video.ffprobe_json_path

    # Clear the DB cache to simulate a fresh reindex
    video.ffprobe_json = None
    test_session.commit()

    # Write a fake ffprobe cache file with a marker
    fake_data = {'streams': [{'codec_name': 'cached_marker'}], 'format': {'duration': '999'}}
    common.write_ffprobe_json_file(ffprobe_path, fake_data)

    # get_ffprobe_json should read from the cached file, not call ffprobe
    result = await video.get_ffprobe_json()
    assert result['streams'][0]['codec_name'] == 'cached_marker', \
        'Should have read cached file instead of running ffprobe'
    assert video.ffprobe_json == fake_data, 'DB should be updated with cached data'


@pytest.mark.asyncio
async def test_video_ffprobe_json_returns_db_first(async_client, test_session, video_file, refresh_files):
    """get_ffprobe_json() returns DB data if already populated, without reading file."""
    await refresh_files()

    video = test_session.query(Video).one()
    original_data = video.ffprobe_json.copy()

    # Write different data to the file (simulating stale cache)
    fake_data = {'streams': [{'codec_name': 'stale_data'}], 'format': {}}
    common.write_ffprobe_json_file(video.ffprobe_json_path, fake_data)

    # Should return DB data, not file data
    result = await video.get_ffprobe_json()
    assert result == original_data
    assert result['streams'][0]['codec_name'] != 'stale_data'


def test_get_videos_directory(test_directory):
    """Default videos directory path"""
    # Directory does not yet exist.
    assert not (test_directory / 'videos').exists()

    # Directory is created when first gotten.
    assert common.get_videos_directory() == (test_directory / 'videos')
    assert (test_directory / 'videos').is_dir()


def test_get_custom_videos_directory(test_directory, test_wrolpi_config):
    """Custom directory can be used for videos directory."""
    config = get_wrolpi_config()
    config.videos_destination = 'custom/directory/videos'

    assert common.get_videos_directory() == (test_directory / 'custom/directory/videos')
    assert (test_directory / 'custom/directory/videos').is_dir()
    assert not (test_directory / 'videos').is_dir()


@pytest.mark.asyncio
async def test_video_modeler_does_not_hold_write_lock_during_ffprobe(async_client, test_session, video_factory):
    """Modeling must not hold the SQLite write lock while ffprobe runs.

    `video_modeler` opened one write session per batch and `await`ed `get_ffprobe_json` -- an ffprobe
    subprocess -- for up to 20 videos inside it.  Every other writer on the box waits on that lock
    and gives up after `busy_timeout` (30s).  ffprobe belongs outside the transaction: read the
    batch, probe the files, then write what was learned.
    """
    import modules.videos as videos_module
    from modules.videos import video_modeler

    video = video_factory(with_video_file=True)
    video.file_group.indexed = False
    video.file_group.model = None
    test_session.delete(video)
    test_session.commit()

    db_file = test_session.get_bind().url.database
    lock_held_during_ffprobe = []
    real_get_or_create_ffprobe_json = videos_module.get_or_create_ffprobe_json

    async def probing_ffprobe(video_path):
        lock_held_during_ffprobe.append(probe_write_lock_is_held(db_file))
        return await real_get_or_create_ffprobe_json(video_path)

    with production_like_sessions(test_session):
        with mock.patch.object(videos_module, 'get_or_create_ffprobe_json', probing_ffprobe):
            await video_modeler()

    assert lock_held_during_ffprobe, 'ffprobe never ran, so the test proves nothing'
    assert not any(lock_held_during_ffprobe), \
        'the write lock was held while ffprobe ran; every other writer waits out busy_timeout'


def test_ffmpeg_video_complete_real_files(test_directory, video_file):
    """A whole video passes; a truncated copy fails.  Uses the real ffmpeg on the example video."""
    from modules.videos.common import ffmpeg_video_complete, tail_decode_command
    assert ffmpeg_video_complete(video_file) is True

    truncated = test_directory / 'truncated.mp4'
    data = video_file.read_bytes()
    truncated.write_bytes(data[:int(len(data) * 0.6)])
    assert ffmpeg_video_complete(truncated) is False

    empty = test_directory / 'empty.mp4'
    empty.write_bytes(b'')
    with pytest.raises(RuntimeError, match='empty'):
        ffmpeg_video_complete(empty)
    with pytest.raises(FileNotFoundError):
        ffmpeg_video_complete(test_directory / 'missing.mp4')

    cmd = tail_decode_command(video_file)
    assert cmd[cmd.index('-sseof') + 1] == '-30' and cmd[cmd.index('-f') + 1] == 'null' and cmd[-1] == '-'


NULL_MUXER_DTS = b'[null @ 0x5555faf7fa70] Application provided invalid, non monotonically increasing dts to muxer' \
                 b' in stream 0: 460 >= 460\n'


@pytest.mark.parametrize('return_code,stderr,expected_ok,expected_errors', [
    (0, b'', True, ''),
    # Duplicate timestamps are reported by the null muxer, not a decoder; the stream decoded.
    (0, NULL_MUXER_DTS, True, ''),
    (0, NULL_MUXER_DTS * 3, True, ''),
    # ffmpeg collapses identical messages into a summary line when stderr is not a terminal.
    (0, NULL_MUXER_DTS + b'    Last message repeated 118 times\n', True, ''),
    (0, NULL_MUXER_DTS + b'    Last message repeated 118 times\n[h264 @ 0x1] Invalid NAL unit size\n', False,
     '[h264 @ 0x1] Invalid NAL unit size'),
    (0, NULL_MUXER_DTS + b'[h264 @ 0x1] Invalid NAL unit size\n', False, '[h264 @ 0x1] Invalid NAL unit size'),
    (0, b'[h264 @ 0x1] error while decoding MB 3 4\n', False, '[h264 @ 0x1] error while decoding MB 3 4'),
    (1, NULL_MUXER_DTS, False, ''),
])
def test_tail_decode_verdict(return_code, stderr, expected_ok, expected_errors):
    """Only decode errors, or a failed exit, mark a video incomplete."""
    from modules.videos.common import _tail_decode_verdict
    assert _tail_decode_verdict(return_code, stderr) == (expected_ok, expected_errors)
    assert _tail_decode_verdict(return_code, stderr.decode()) == (expected_ok, expected_errors)


@pytest.mark.asyncio
async def test_ffmpeg_video_complete_async_real_files(test_directory, video_file):
    """The async variant reaches the same verdicts and returns ffmpeg's error text."""
    from modules.videos.common import ffmpeg_video_complete_async
    ok, errors = await ffmpeg_video_complete_async(video_file)
    assert ok is True and errors == ''

    truncated = test_directory / 'truncated.mp4'
    data = video_file.read_bytes()
    truncated.write_bytes(data[:int(len(data) * 0.6)])
    ok, errors = await ffmpeg_video_complete_async(truncated)
    assert ok is False and errors