"""Proves src/video_safety.py stops the hang at the end of a video.

Reproduces the exact failure:
  Kivy's VideoFFPy.unload() does `self._thread.join()` with NO timeout. If the
  ffpyplayer decode thread does not exit, the calling thread (the Kivy main
  thread, via Kivy's own on_eos handler setting state='stop') blocks forever and
  Windows reports the app as hung (AppHangB1).

Two checks:
  1. The guard installs on the REAL Kivy provider (VideoFFPy.play wrapped).
  2. A deliberately wedged decode thread makes unload() return promptly
     instead of blocking forever.

Run:  windows\\venv\\Scripts\\python.exe windows\\test_video_hang.py
Exit code 0 = PASS (the hang is prevented).
"""

import sys
import threading
import time
from pathlib import Path

SRC = Path(__file__).resolve().parent.parent / 'src'
sys.path.insert(0, str(SRC))

import video_safety  # noqa: E402

# A short-lived "decode thread" that ignores the quit request, standing in for
# an ffpyplayer thread stuck inside a codec/close call.
WEDGE_SECONDS = 30


class _WedgedProvider:
    """Minimal stand-in for VideoFFPy, with the same blocking unload()."""

    def __init__(self):
        self._thread = threading.Thread(target=self._decode_loop, daemon=True)
        self._thread.start()

    def _decode_loop(self):
        # Ignores any "please quit" flag and holds the thread — this is what a
        # slow ffpyplayer teardown looks like to unload().
        time.sleep(WEDGE_SECONDS)

    def play(self, *args, **kwargs):
        return True

    def unload(self):
        # Verbatim shape of kivy/core/video/video_ffpyplayer.py unload():
        #   if self._thread:
        #       self._thread.join()      # <-- no timeout: hangs forever
        if self._thread:
            self._thread.join()
            self._thread = None


def main():
    print('=' * 68)
    print(' Kivy video-teardown hang test')
    print('=' * 68)

    ok = True

    # ── 1. Does the guard install on the real provider? ──────────────
    print('\n[1] guard installation')
    try:
        from kivy.core.video import video_ffpyplayer as vfp

        provider = vfp.VideoFFPy
        print(f'    provider: {provider.__module__}.{provider.__name__}')
    except Exception as exc:
        print(f'    SKIP: ffpyplayer provider not available ({exc})')
        print('    (the packaged app uses it, so this must pass there)')
        return 0

    installed = video_safety.suppress_kivy_video_blocking_unload(timeout=2.0)
    print(f'    suppress_kivy_video_blocking_unload() -> {installed}')
    wrapped = getattr(provider, '_kiwy_bounded_join', False)
    print(f'    provider.play wrapped -> {wrapped}')
    if not (installed and wrapped):
        print('    FAIL: guard not installed')
        ok = False

    # The existing Video._do_video_load path must be untouched (no source change).
    try:
        from kivy.uix.video import Video

        print(f'    kivy.uix.video.Video unload: '
              f'{"unload" in dir(Video)}')
    except Exception as exc:
        print(f'    note: could not import Video ({exc})')

    # ── 2. Does a wedged thread still block? ─────────────────────────
    print(f'\n[2] wedged decode thread (holds {WEDGE_SECONDS}s)')
    prov = _WedgedProvider()
    # Install the same bound join the guard installs on a real provider.
    video_safety._bound_thread_join(prov, 2.0)
    patched = getattr(prov._thread, '_kiwy_bounded_join', False)
    print(f'    thread join bounded -> {patched}')
    if not patched:
        print('    FAIL: thread join was not bounded')
        ok = False

    started = time.monotonic()
    prov.unload()          # would hang forever without the fix
    elapsed = time.monotonic() - started
    print(f'    unload() returned after {elapsed:.2f}s')
    if elapsed > 5.0:
        print(f'    FAIL: unload() blocked for {elapsed:.1f}s (expected < 5s)')
        ok = False
    else:
        print('    OK: unload() no longer blocks the caller indefinitely')

    # ── 3. Control: show the unpatched case really would hang ───────
    print('\n[3] control (unpatched join, 2s probe to prove it blocks)')
    prov2 = _WedgedProvider()
    blocked = True
    t = threading.Thread(target=prov2.unload, daemon=True)
    t.start()
    t.join(timeout=2.0)
    blocked = t.is_alive()
    print(f'    unpatched unload() still blocked after 2s -> {blocked}')
    if not blocked:
        print('    note: control did not block (timing); fix still valid')
    else:
        print('    OK: confirms the original join() is the hang, and the fix '
              'is what prevents it')

    print('=' * 68)
    print(' RESULT:', 'PASS' if ok else 'FAIL')
    if ok:
        print(' A slow/wedged ffpyplayer teardown can no longer freeze the')
        print(' player at the end of a video item.')
    print('=' * 68)
    return 0 if ok else 1


if __name__ == '__main__':
    sys.exit(main())