momentry_core/test_subprocess_asr.py

#!/opt/homebrew/bin/python3.11
"""Test ASR processor via subprocess (like test_all_videos.py does)."""

import subprocess
import tempfile
import os
import time
import sys


def test_subprocess_small():
    """Test subprocess call with small video."""
    test_video = "/Users/accusys/test_video/BigBuckBunny_320x180.mp4"
    if not os.path.exists(test_video):
        print(f"Test video not found: {test_video}")
        return

    # Create temp output
    with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
        output_path = f.name

    try:
        cmd = [
            "/opt/homebrew/bin/python3.11",
            "scripts/asr_processor_debug.py",
            test_video,
            output_path,
            "--uuid",
            "test_subprocess",
        ]

        print(f"Running command: {' '.join(cmd[:3])} ...")
        print(f"Output will be written to: {output_path}")

        start = time.time()
        proc = subprocess.Popen(
            cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
        )

        # Wait with timeout
        timeout = 60  # 1 minute
        try:
            stdout, stderr = proc.communicate(timeout=timeout)
            elapsed = time.time() - start
            success = proc.returncode == 0

            print(f"Process completed in {elapsed:.1f}s")
            print(f"Return code: {proc.returncode}")

            if stderr:
                print("Stderr output (last 10 lines):")
                for line in stderr.strip().split("\n")[-10:]:
                    print(f"  {line}")

            if success:
                print("✓ Subprocess succeeded")
                if os.path.exists(output_path):
                    import json

                    with open(output_path, "r") as f:
                        data = json.load(f)
                    print(f"  Segments: {len(data.get('segments', []))}")
            else:
                print("✗ Subprocess failed")

        except subprocess.TimeoutExpired:
            print(f"✗ Subprocess timed out after {timeout}s")
            proc.kill()
            stdout, stderr = proc.communicate()
            print(f"Stderr (last 20 lines):")
            for line in stderr.strip().split("\n")[-20:]:
                print(f"  {line}")

    except Exception as e:
        print(f"✗ Test failed: {e}")
        import traceback

        traceback.print_exc()
    finally:
        if os.path.exists(output_path):
            os.unlink(output_path)
            print("✓ Cleaned up output file")


def test_subprocess_large():
    """Test subprocess call with large video (should hang)."""
    test_video = "/Users/accusys/test_video/1636719d-c31f-78ac-f1dd-8ab0b0b36c66.mov"
    if not os.path.exists(test_video):
        print(f"Large test video not found: {test_video}")
        return

    # Create temp output
    with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
        output_path = f.name

    try:
        cmd = [
            "/opt/homebrew/bin/python3.11",
            "scripts/asr_processor_debug.py",
            test_video,
            output_path,
            "--uuid",
            "test_large_subprocess",
            "--chunk-duration",
            "60",  # 1 minute chunks for debugging
        ]

        print(f"\nTesting large video via subprocess...")
        print(f"Video: {os.path.basename(test_video)}")
        print(f"Size: {os.path.getsize(test_video) / (1024**3):.2f} GB")

        start = time.time()
        proc = subprocess.Popen(
            cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
        )

        # Wait with shorter timeout to catch hang quickly
        timeout = 30  # 30 seconds
        try:
            stdout, stderr = proc.communicate(timeout=timeout)
            elapsed = time.time() - start
            success = proc.returncode == 0

            print(f"Process completed in {elapsed:.1f}s")
            print(f"Return code: {proc.returncode}")

            if stderr:
                print("Stderr output (last 20 lines):")
                for line in stderr.strip().split("\n")[-20:]:
                    print(f"  {line}")

            if success:
                print("✓ Subprocess succeeded (unexpected!)")
            else:
                print("✗ Subprocess failed")

        except subprocess.TimeoutExpired:
            elapsed = time.time() - start
            print(f"✗ Subprocess timed out after {elapsed:.1f}s (hang detected)")
            print("Killing process...")
            proc.kill()
            stdout, stderr = proc.communicate()
            print(f"Stderr (last 30 lines):")
            for line in stderr.strip().split("\n")[-30:]:
                print(f"  {line}")

    except Exception as e:
        print(f"✗ Test failed: {e}")
        import traceback

        traceback.print_exc()
    finally:
        if os.path.exists(output_path):
            os.unlink(output_path)
            print("✓ Cleaned up output file")


if __name__ == "__main__":
    test_subprocess_small()
    test_subprocess_large()