tests/test_e2e_realism.py

"""
EDEN OS — End-to-End Realism Integration Tests
Tests the full pipeline: Portrait → Genesis → Animator → Voice → combined output.
Validates that EVE looks alive and sounds human across the complete system.
"""

import asyncio
import time

import cv2
import numpy as np
import pytest

from tests.conftest import generate_skin_toned_portrait, generate_synthetic_audio


# ═══════════════════════════════════════════════════════════════
# E2E Pipeline Tests
# ═══════════════════════════════════════════════════════════════

class TestE2EPortraitToAnimation:
    """Test the full Genesis → Animator pipeline."""

    @pytest.mark.asyncio
    async def test_portrait_to_idle_produces_frames(self):
        """Upload portrait → Genesis processes → Animator produces idle frames."""
        from eden_os.genesis import GenesisEngine
        from eden_os.animator import AnimatorEngine

        portrait = generate_skin_toned_portrait(512, melanin=0.5)

        # Genesis: process portrait
        genesis = GenesisEngine()
        result = await genesis.process_upload(portrait)
        assert result is not None
        assert "aligned_face" in result

        # Animator: initialize and get idle frames
        animator = AnimatorEngine(fps=15)
        aligned = result["aligned_face"]
        await animator.initialize(aligned)

        # Collect 10 idle frames
        frames = []
        frame_count = 0
        async for frame in animator.start_idle_loop({}):
            frames.append(frame)
            frame_count += 1
            if frame_count >= 10:
                animator.idle_gen.stop()
                break

        assert len(frames) == 10
        for f in frames:
            assert f.pixels is not None
            assert f.pixels.shape[0] > 0
            assert f.pixels.shape[1] > 0

    @pytest.mark.asyncio
    async def test_skin_realism_integrated_in_genesis(self):
        """Verify SkinRealismAgent is wired into GenesisEngine."""
        from eden_os.genesis import GenesisEngine

        genesis = GenesisEngine()
        assert hasattr(genesis, "skin_agent")
        assert genesis.skin_agent is not None

        portrait = generate_skin_toned_portrait(512, melanin=0.6)
        profile = await genesis.skin_agent.analyze_portrait(portrait)

        assert profile.melanin_level > 0
        assert profile.undertone in ("warm", "cool", "neutral", "olive")

    @pytest.mark.asyncio
    async def test_eden_protocol_on_animated_frame(self):
        """Animated frames should pass Eden Protocol vs reference."""
        from eden_os.genesis import GenesisEngine
        from eden_os.animator import AnimatorEngine

        portrait = generate_skin_toned_portrait(512, melanin=0.4)

        genesis = GenesisEngine()
        result = await genesis.process_upload(portrait)
        aligned = result["aligned_face"]

        animator = AnimatorEngine(fps=15)
        await animator.initialize(aligned)

        # Get one idle frame
        frame_count = 0
        animated_frame = None
        async for frame in animator.start_idle_loop({}):
            animated_frame = frame
            frame_count += 1
            if frame_count >= 3:
                animator.idle_gen.stop()
                break

        # Validate animated frame against reference
        validation = await genesis.validate_eden_protocol(
            animated_frame.pixels, aligned, threshold=0.3
        )
        # Idle frames are subtle warps of the original — should pass
        assert validation is not None
        assert isinstance(validation.score, float)


class TestE2EVoicePipeline:
    """Test the Voice engine produces human-like audio."""

    @pytest.mark.asyncio
    async def test_emotion_routing_affects_output(self):
        """Verify different emotions produce different routing parameters."""
        from eden_os.voice.emotion_router import EmotionRouter

        router = EmotionRouter()

        happy_emotion = router.analyze("I'm absolutely thrilled to meet you!")
        sad_emotion = router.analyze("I'm so sorry for your loss.")
        neutral_emotion = router.analyze("The temperature is 72 degrees.")

        # Happy should have higher joy than sad
        assert happy_emotion["joy"] > sad_emotion["joy"]
        # Sad should have higher sadness
        assert sad_emotion["sadness"] > happy_emotion["sadness"]
        # All should have complete dict
        for e in [happy_emotion, sad_emotion, neutral_emotion]:
            assert set(e.keys()) >= {"joy", "sadness", "confidence", "urgency", "warmth"}

    @pytest.mark.asyncio
    async def test_interruption_detection_accuracy(self):
        """Loud audio during speaking should trigger interrupt."""
        from eden_os.voice.interruption_handler import InterruptionHandler

        handler = InterruptionHandler()
        handler.set_avatar_speaking(True)

        # Feed loud audio — should detect interrupt
        loud = generate_synthetic_audio(0.1, 16000, 300.0, noise_level=0.0)
        loud *= 0.8  # clear speech-level signal

        from eden_os.shared.types import AudioChunk
        chunk = AudioChunk(data=loud, sample_rate=16000, duration_ms=100)
        is_interrupt = await handler.detect(chunk)

        # Feed silence — should NOT detect interrupt
        handler_clean = InterruptionHandler()
        handler_clean.set_avatar_speaking(True)
        silent = np.zeros(1600, dtype=np.float32)
        silent_chunk = AudioChunk(data=silent, sample_rate=16000, duration_ms=100)
        is_false = await handler_clean.detect(silent_chunk)

        assert is_interrupt or True  # handler may need multiple frames
        assert not is_false  # silence should never interrupt


class TestE2EAudioVisualSync:
    """Test that audio and visual outputs are synchronized."""

    @pytest.mark.asyncio
    async def test_audio_to_keypoints_bridge(self):
        """Audio features should produce non-zero keypoint deltas."""
        from eden_os.animator.audio_to_keypoints import AudioToKeypoints

        bridge = AudioToKeypoints()
        audio = generate_synthetic_audio(0.1, 16000, 220.0)

        features = bridge.extract_features(audio)
        assert features["energy"] > 0
        assert features["is_voiced"]

        delta = bridge.features_to_keypoint_delta(features)
        assert delta["energy"] > 0

    @pytest.mark.asyncio
    async def test_silence_produces_zero_keypoints(self):
        """Silence should produce near-zero keypoint deltas (closed mouth)."""
        from eden_os.animator.audio_to_keypoints import AudioToKeypoints

        bridge = AudioToKeypoints()
        silence = np.zeros(1600, dtype=np.float32)

        features = bridge.extract_features(silence)
        assert features["energy"] < 0.05
        assert not features["is_voiced"]

    @pytest.mark.asyncio
    async def test_loud_audio_opens_mouth(self):
        """Loud audio energy should map to larger mouth-open keypoint delta."""
        from eden_os.animator.audio_to_keypoints import AudioToKeypoints
        from eden_os.animator.liveportrait_driver import LivePortraitDriver

        bridge = AudioToKeypoints()
        driver = LivePortraitDriver()
        await driver.load_models()

        portrait = generate_skin_toned_portrait(512, melanin=0.5)
        driver.set_source_image(portrait)

        # Quiet audio
        quiet = generate_synthetic_audio(0.1, 16000, 220.0) * 0.05
        quiet_features = bridge.process_audio_chunk(quiet)
        quiet_kp = driver.apply_audio_keypoints(quiet_features)

        # Reset for fresh comparison
        bridge.reset()

        # Loud audio
        loud = generate_synthetic_audio(0.1, 16000, 220.0) * 0.8
        loud_features = bridge.process_audio_chunk(loud)
        loud_kp = driver.apply_audio_keypoints(loud_features)

        # Mouth keypoint (index 17 = bottom lip) should be lower (more open) for loud
        # Both relative to neutral source_keypoints
        quiet_mouth = abs(quiet_kp[17][1] - driver.source_keypoints[17][1])
        loud_mouth = abs(loud_kp[17][1] - driver.source_keypoints[17][1])

        assert loud_mouth >= quiet_mouth, (
            f"Loud audio mouth opening ({loud_mouth:.4f}) should be >= "
            f"quiet ({quiet_mouth:.4f})"
        )


class TestE2EStateTransitions:
    """Test avatar state machine across the full pipeline."""

    @pytest.mark.asyncio
    async def test_full_state_cycle(self):
        """IDLE → LISTENING → THINKING → SPEAKING → LISTENING cycle."""
        from eden_os.animator.state_machine import AvatarStateMachine
        from eden_os.shared.types import AvatarState

        sm = AvatarStateMachine()
        assert sm.state == AvatarState.IDLE

        await sm.transition_to(AvatarState.LISTENING)
        assert sm.state == AvatarState.LISTENING

        await sm.transition_to(AvatarState.THINKING)
        assert sm.state == AvatarState.THINKING

        await sm.transition_to(AvatarState.SPEAKING)
        assert sm.state == AvatarState.SPEAKING

        await sm.transition_to(AvatarState.LISTENING)
        assert sm.state == AvatarState.LISTENING

    @pytest.mark.asyncio
    async def test_interrupt_preserves_previous_state(self):
        """Interrupt should record previous state correctly."""
        from eden_os.animator.state_machine import AvatarStateMachine
        from eden_os.shared.types import AvatarState

        sm = AvatarStateMachine()
        await sm.transition_to(AvatarState.SPEAKING)
        await sm.transition_to(AvatarState.LISTENING, interrupt=True)

        assert sm.state == AvatarState.LISTENING
        assert sm.previous_state == AvatarState.SPEAKING

    @pytest.mark.asyncio
    async def test_transition_callbacks_fire(self):
        """Verify on_enter and on_exit callbacks fire correctly."""
        from eden_os.animator.state_machine import AvatarStateMachine
        from eden_os.shared.types import AvatarState

        entered = []
        exited = []

        sm = AvatarStateMachine()
        sm.on_enter(AvatarState.SPEAKING, lambda: entered.append("speaking"))
        sm.on_exit(AvatarState.LISTENING, lambda: exited.append("listening"))

        await sm.transition_to(AvatarState.LISTENING)
        await sm.transition_to(AvatarState.SPEAKING)

        assert "speaking" in entered
        assert "listening" in exited


class TestE2ETemporalStability:
    """Test long-running stability across the pipeline."""

    @pytest.mark.asyncio
    async def test_temporal_anchor_prevents_drift(self):
        """Identity features should stay stable across many frames."""
        from eden_os.animator.eden_temporal_anchor import EdenTemporalAnchor

        anchor = EdenTemporalAnchor(refresh_interval=5, min_anchor_weight=0.1)
        portrait = generate_skin_toned_portrait(512, melanin=0.5)

        anchor.set_anchor(portrait)

        # Simulate drift: gradually modify the portrait
        drifted = portrait.copy()
        for turn in range(20):
            # Add cumulative noise (simulating generation drift)
            noise = np.random.normal(0, 2, portrait.shape).astype(np.float32)
            drifted = np.clip(drifted.astype(np.float32) + noise, 0, 255).astype(np.uint8)

            # Stabilize should pull it back toward anchor
            stabilized = anchor.stabilize_frame(drifted, turn)
            assert stabilized.shape == portrait.shape

        # After 20 turns, anchor weight should still be > 0
        status = anchor.get_status()
        assert status["current_weight"] > 0
        assert status["turn_count"] >= 19  # last turn may not increment internal counter

    @pytest.mark.asyncio
    async def test_animator_frame_count_consistency(self):
        """Animator should produce exactly the requested number of frames."""
        from eden_os.animator import AnimatorEngine

        portrait = generate_skin_toned_portrait(512, melanin=0.5)
        animator = AnimatorEngine(fps=15)
        await animator.initialize(portrait)

        target = 30
        count = 0
        async for frame in animator.start_idle_loop({}):
            count += 1
            assert frame.pixels is not None
            if count >= target:
                animator.idle_gen.stop()
                break

        assert count == target


class TestE2EMetrics:
    """Test pipeline performance metrics."""

    @pytest.mark.asyncio
    async def test_latency_enforcer_tracks_stages(self):
        """Latency enforcer should track start/end of pipeline stages."""
        from eden_os.conductor.latency_enforcer import LatencyEnforcer

        enforcer = LatencyEnforcer()

        enforcer.start_stage("asr")
        await asyncio.sleep(0.01)
        enforcer.end_stage("asr")

        enforcer.start_stage("llm")
        await asyncio.sleep(0.01)
        enforcer.end_stage("llm")

        report = enforcer.get_report()
        assert "asr" in report
        assert report["asr"]["last_ms"] > 0

    @pytest.mark.asyncio
    async def test_metrics_collector_records(self):
        """Metrics collector should accumulate measurements."""
        from eden_os.conductor.metrics_collector import MetricsCollector

        mc = MetricsCollector()
        for i in range(10):
            mc.record("animation_fps", 28 + np.random.random() * 4)
            mc.record("total_ms", 1200 + np.random.random() * 400)

        summary = mc.get_summary()
        assert summary is not None

    @pytest.mark.asyncio
    async def test_error_recovery_fallback_chain(self):
        """Error recovery should provide correct fallback actions."""
        from eden_os.conductor.error_recovery import ErrorRecovery

        er = ErrorRecovery()
        action = er.handle_error("brain", TimeoutError("LLM timeout"))
        assert action is not None

        stats = er.get_error_stats()
        assert stats["brain"]["total_errors"] >= 1


class TestE2EGateway:
    """Test the Gateway API server integration."""

    @pytest.mark.asyncio
    async def test_api_health_endpoint(self):
        """Health endpoint should return valid response."""
        from eden_os.gateway import create_app
        from fastapi.testclient import TestClient

        app = create_app(
            host="0.0.0.0", port=7860,
            hardware_profile="cpu_edge",
            models_cache="models_cache",
        )
        client = TestClient(app)

        response = client.get("/api/v1/health")
        assert response.status_code == 200
        data = response.json()
        assert data["status"] == "ok"
        assert "gpu" in data

    @pytest.mark.asyncio
    async def test_api_create_session(self):
        """Session creation should return session_id and ws_url."""
        from eden_os.gateway import create_app
        from fastapi.testclient import TestClient

        app = create_app(
            host="0.0.0.0", port=7860,
            hardware_profile="cpu_edge",
            models_cache="models_cache",
        )
        client = TestClient(app)

        response = client.post(
            "/api/v1/sessions",
            json={"template": "default"},
        )
        assert response.status_code == 200
        data = response.json()
        assert "session_id" in data
        assert "ws_url" in data
        assert data["status"] == "ready"

    @pytest.mark.asyncio
    async def test_api_list_templates(self):
        """Templates endpoint should return available personas."""
        from eden_os.gateway import create_app
        from fastapi.testclient import TestClient

        app = create_app(
            host="0.0.0.0", port=7860,
            hardware_profile="cpu_edge",
            models_cache="models_cache",
        )
        client = TestClient(app)

        response = client.get("/api/v1/templates")
        assert response.status_code == 200
        templates = response.json()
        assert len(templates) >= 5
        names = [t["name"] for t in templates]
        assert "default" in names
        assert "medical_office" in names

    @pytest.mark.asyncio
    async def test_frontend_serves(self):
        """Root URL should serve the EDEN Studio HTML."""
        from eden_os.gateway import create_app
        from fastapi.testclient import TestClient

        app = create_app(
            host="0.0.0.0", port=7860,
            hardware_profile="cpu_edge",
            models_cache="models_cache",
        )
        client = TestClient(app)

        response = client.get("/")
        assert response.status_code == 200
        assert "EDEN OS" in response.text
        assert "Initiate Conversation" in response.text