Kiwy-Signage/linux/test_media_state.py

"""test_media_state.py — the conversion-flag state machine.

The rules here are easy to get subtly wrong, and getting them wrong is
user-visible in two opposite ways: playing a file that is mid-conversion
(truncated video), or skipping forever an item that is actually fine.

Covered:

1. A normal-size video resolves to itself and is playable.
2. An oversized video with no completed conversion reports ``pending``.
3. While the ``.kiwy-converting`` marker exists it reports ``converting``.
4. Once the output + metadata exist it resolves to the **converted** file.
5. A stale marker (from a crash) does not park the item forever.
6. Metadata that does not match the current source is ignored — otherwise a
   different video reusing the same filename would play the previous one.
7. The MP4 header parser agrees with ffprobe (the player must not spawn a
   subprocess on the playback path, so it reads the container directly).

Run:
    .venv/bin/python linux/test_media_state.py
"""

from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
import tempfile
import time
from pathlib import Path

HERE = Path(__file__).resolve().parent
ROOT = HERE.parent
SRC = ROOT / 'src'
for path in (str(HERE), str(SRC)):
    if path not in sys.path:
        sys.path.insert(0, path)

import media_state as ms  # noqa: E402

failures: list[str] = []
checks = 0


def check(label, condition, detail=''):
    global checks
    checks += 1
    print(f'  {"PASS" if condition else "FAIL"}  {label}'
          + (f' — {detail}' if not condition and detail else ''))
    if not condition:
        failures.append(label)


def make_video(path, width, height, seconds=1):
    """Create a tiny real video of the given size (or None if ffmpeg is absent)."""
    if not shutil.which('ffmpeg'):
        return False
    cmd = [
        'ffmpeg', '-hide_banner', '-loglevel', 'error',
        '-f', 'lavfi', '-i', f'testsrc=size={width}x{height}:rate=10:duration={seconds}',
        '-c:v', 'libx264', '-preset', 'ultrafast', '-pix_fmt', 'yuv420p',
        '-y', path,
    ]
    return subprocess.run(cmd, capture_output=True, check=False).returncode == 0


workdir = Path(tempfile.mkdtemp(prefix='kiwy-mediastate-'))
print(f'workdir: {workdir}')

try:
    big = workdir / 'big.mp4'
    small = workdir / 'small.mp4'

    if not make_video(big, 2560, 1440) or not make_video(small, 1280, 720):
        print('SKIP: ffmpeg not available to build fixtures')
        raise SystemExit(0)

    # ── 1. Within-limit video is playable as itself ──────────────────
    print('\n[1] A within-limit video resolves to itself')
    chosen, state = ms.resolve_playable(str(small))
    check('state is ready', state == 'ready', f'got {state}')
    check('chosen path is the original', chosen == str(small), f'got {chosen}')

    # ── 2. Oversized + no conversion -> pending ──────────────────────
    print('\n[2] Oversized video with no conversion reports pending')
    chosen, state = ms.resolve_playable(str(big))
    check('state is pending', state == 'pending', f'got {state}')
    oversized, _w, _h = ms.is_oversized(str(big))
    check('is_oversized is True', oversized is True)
    check('a within-limit file is not oversized',
          ms.is_oversized(str(small))[0] is False)

    # ── 3. Converting marker -> converting ───────────────────────────
    print('\n[3] The converting marker suppresses playback')
    ms.begin_conversion(str(big))
    chosen, state = ms.resolve_playable(str(big))
    check('state is converting', state == 'converting', f'got {state}')
    check('is_converting is True', ms.is_converting(str(big)) is True)

    # ── 4. Completed conversion resolves to the output ───────────────
    print('\n[4] A finished conversion resolves to the converted file')
    output = ms.normalized_output(str(big))
    shutil.copy2(str(small), output)          # stand-in for the 1080p result
    with open(output + ms.MARKER_SUFFIX, 'w') as fh:
        json.dump({'source_size': os.path.getsize(big),
                   'width': 2560, 'height': 1440}, fh)
    ms.end_conversion(str(big))

    chosen, state = ms.resolve_playable(str(big))
    check('state is ready', state == 'ready', f'got {state}')
    check('chosen path is the converted file', chosen == output, f'got {chosen}')

    # ── 5. Stale marker is ignored ───────────────────────────────────
    print('\n[5] A stale (crashed) marker does not block the item forever')
    os.remove(output)
    os.remove(output + ms.MARKER_SUFFIX)
    ms.begin_conversion(str(big))
    old = time.time() - (ms.STALE_CONVERSION_SECONDS + 60)
    os.utime(ms.converting_marker(str(big)), (old, old))
    check('a stale marker is not treated as converting',
          ms.is_converting(str(big)) is False)
    _chosen, state = ms.resolve_playable(str(big))
    check('so the item is pending rather than converting',
          state == 'pending', f'got {state}')
    ms.end_conversion(str(big))

    # ── 6. Mismatched metadata is ignored ────────────────────────────
    print('\n[6] Metadata for a different source is rejected')
    shutil.copy2(str(small), output)
    with open(output + ms.MARKER_SUFFIX, 'w') as fh:
        json.dump({'source_size': 12345,   # does not match big.mp4
                   'width': 2560, 'height': 1440}, fh)
    check('a stale output is not accepted', ms.normalized_file(str(big)) is None)
    _chosen, state = ms.resolve_playable(str(big))
    check('the item is treated as pending', state == 'pending', f'got {state}')

    # ── 7. Header parser agrees with ffprobe ─────────────────────────
    print('\n[7] The dependency-free MP4 parser matches ffprobe')
    for label, path, expect in (('2560x1440', str(big), (2560, 1440)),
                                ('1280x720', str(small), (1280, 720))):
        parsed = ms.read_video_size(path)
        check(f'{label} parsed correctly', parsed == expect, f'got {parsed}')

    if shutil.which('ffprobe'):
        out = subprocess.run(
            ['ffprobe', '-v', 'error', '-select_streams', 'v:0',
             '-show_entries', 'stream=width,height', '-of', 'csv=p=0', str(big)],
            capture_output=True, text=True, check=False).stdout.strip()
        fw, fh = (int(x) for x in out.split(',')[:2])
        check('parser matches ffprobe for the oversized file',
              ms.read_video_size(str(big)) == (fw, fh),
              f'parser={ms.read_video_size(str(big))} ffprobe={(fw, fh)}')

    check('a non-video file yields (None, None)',
          ms.read_video_size(str(workdir / 'missing.mp4')) == (None, None))
    check('a corrupt file yields (None, None)', (lambda p: (
        p.write_bytes(b'not a video'), ms.read_video_size(str(p)))[1]
    )(workdir / 'corrupt.mp4') == (None, None))

finally:
    shutil.rmtree(workdir, ignore_errors=True)

print(f'\n{checks - len(failures)}/{checks} checks passed')
if failures:
    print('\nFailed:')
    for name in failures:
        print(f'  - {name}')
    raise SystemExit(1)
print('All checks passed.')