#!/usr/bin/env python3
"""
Tests for Story 8.03: OpenClaw Bridge — Redis Queue Reader
AIVA RLM Nexus PRD v2 — Track A, Module 8

Black box tests (BB1-BB3): verify observable contract from the outside —
correct deserialization of a queued message, None returned on timeout, and
ValueError raised on malformed JSON — without inspecting internals.

White box tests (WB1-WB3): verify implementation properties —
blpop is used (not lpop in a loop), timeout_s is forwarded to blpop
(not hardcoded), and the return value is an OpenClawMessage instance
(not a raw dict).

All Redis I/O is mocked via unittest.mock.MagicMock.  No live Redis
connection is required or expected.
"""

from __future__ import annotations

import asyncio
import json
import sys
from datetime import datetime, timezone
from unittest.mock import MagicMock, call

import pytest

sys.path.insert(0, "/mnt/e/genesis-system")

from core.bridge.openclaw_bridge import (
    BridgeReader,
    BridgeWriter,
    BRIDGE_QUEUE_AIVA_TO_GENESIS,
    BRIDGE_QUEUE_GENESIS_TO_AIVA,
    MessageDirection,
    OpenClawMessage,
)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _run(coro):
    """Execute a coroutine synchronously (works without a running event loop)."""
    return asyncio.get_event_loop().run_until_complete(coro)


def _make_msg(**overrides) -> OpenClawMessage:
    """Return a fully-populated OpenClawMessage with sensible defaults."""
    defaults = dict(
        message_id="803e8400-e29b-41d4-a716-446655440000",
        session_id="session-test-803",
        direction=MessageDirection.AIVA_TO_GENESIS,
        payload={"intent": "task_request", "body": "test payload 8.03"},
        priority=2,
        created_at=datetime(2026, 2, 25, 10, 0, 0, tzinfo=timezone.utc),
    )
    defaults.update(overrides)
    return OpenClawMessage(**defaults)


def _serialise_msg(msg: OpenClawMessage) -> str:
    """Serialise a message to the same JSON format BridgeWriter uses."""
    data = {
        "message_id": msg.message_id,
        "session_id": msg.session_id,
        "direction": msg.direction.value,
        "payload": msg.payload,
        "priority": msg.priority,
        "created_at": msg.created_at.isoformat(),
        "expires_at": msg.expires_at.isoformat() if msg.expires_at is not None else None,
    }
    return json.dumps(data)


def _make_reader_with_message(msg: OpenClawMessage) -> tuple[BridgeReader, MagicMock]:
    """Return a (BridgeReader, mock_redis) pair pre-seeded with *msg* in the queue."""
    mock_redis = MagicMock()
    raw = _serialise_msg(msg).encode()
    # blpop returns (key_bytes, value_bytes) tuple
    if msg.direction == MessageDirection.AIVA_TO_GENESIS:
        key = BRIDGE_QUEUE_AIVA_TO_GENESIS.encode()
    else:
        key = BRIDGE_QUEUE_GENESIS_TO_AIVA.encode()
    mock_redis.blpop.return_value = (key, raw)
    reader = BridgeReader(redis_client=mock_redis)
    return reader, mock_redis


def _make_empty_reader() -> tuple[BridgeReader, MagicMock]:
    """Return a (BridgeReader, mock_redis) pair simulating an empty/timeout queue."""
    mock_redis = MagicMock()
    mock_redis.blpop.return_value = None  # Redis blpop returns None on timeout
    reader = BridgeReader(redis_client=mock_redis)
    return reader, mock_redis


def _make_malformed_reader(raw_bytes: bytes) -> tuple[BridgeReader, MagicMock]:
    """Return a (BridgeReader, mock_redis) pair that returns malformed data."""
    mock_redis = MagicMock()
    mock_redis.blpop.return_value = (
        BRIDGE_QUEUE_AIVA_TO_GENESIS.encode(),
        raw_bytes,
    )
    reader = BridgeReader(redis_client=mock_redis)
    return reader, mock_redis


# ---------------------------------------------------------------------------
# BB1: Message in queue → poll() returns OpenClawMessage with all fields correct
# ---------------------------------------------------------------------------


class TestBB1_MessageReturned:
    """BB1: When a message is in the queue, poll() returns a correct OpenClawMessage."""

    def test_returns_openclaw_message_instance(self):
        """poll() must return an OpenClawMessage, not a raw dict or None."""
        msg = _make_msg()
        reader, _ = _make_reader_with_message(msg)

        result = _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

        assert isinstance(result, OpenClawMessage)

    def test_message_id_roundtrip(self):
        """message_id must survive the JSON round-trip unchanged."""
        msg = _make_msg(message_id="roundtrip-id-xyz")
        reader, _ = _make_reader_with_message(msg)

        result = _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

        assert result.message_id == "roundtrip-id-xyz"

    def test_session_id_roundtrip(self):
        """session_id must survive the JSON round-trip unchanged."""
        msg = _make_msg(session_id="session-roundtrip")
        reader, _ = _make_reader_with_message(msg)

        result = _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

        assert result.session_id == "session-roundtrip"

    def test_direction_roundtrip_aiva_to_genesis(self):
        """direction AIVA_TO_GENESIS must survive the JSON round-trip."""
        msg = _make_msg(direction=MessageDirection.AIVA_TO_GENESIS)
        reader, _ = _make_reader_with_message(msg)

        result = _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

        assert result.direction == MessageDirection.AIVA_TO_GENESIS

    def test_direction_roundtrip_genesis_to_aiva(self):
        """direction GENESIS_TO_AIVA must survive the JSON round-trip."""
        msg = _make_msg(direction=MessageDirection.GENESIS_TO_AIVA)
        reader, _ = _make_reader_with_message(msg)

        result = _run(reader.poll(MessageDirection.GENESIS_TO_AIVA))

        assert result.direction == MessageDirection.GENESIS_TO_AIVA

    def test_payload_roundtrip(self):
        """Nested payload dict must survive the JSON round-trip unchanged."""
        payload = {"intent": "inject_context", "data": [1, 2, 3], "nested": {"k": "v"}}
        msg = _make_msg(payload=payload)
        reader, _ = _make_reader_with_message(msg)

        result = _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

        assert result.payload == payload

    def test_priority_roundtrip(self):
        """priority must survive the JSON round-trip unchanged."""
        msg = _make_msg(priority=1)
        reader, _ = _make_reader_with_message(msg)

        result = _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

        assert result.priority == 1

    def test_created_at_roundtrip(self):
        """created_at (UTC datetime) must survive the ISO 8601 round-trip."""
        ts = datetime(2026, 2, 25, 10, 0, 0, tzinfo=timezone.utc)
        msg = _make_msg(created_at=ts)
        reader, _ = _make_reader_with_message(msg)

        result = _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

        # Compare as UTC-aware datetimes
        assert result.created_at.replace(tzinfo=timezone.utc) == ts

    def test_expires_at_none_roundtrip(self):
        """expires_at=None must survive the JSON round-trip as None."""
        msg = _make_msg(expires_at=None)
        reader, _ = _make_reader_with_message(msg)

        result = _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

        assert result.expires_at is None

    def test_expires_at_datetime_roundtrip(self):
        """A non-None expires_at must survive the ISO 8601 round-trip."""
        expiry = datetime(2026, 12, 31, 23, 59, 59, tzinfo=timezone.utc)
        msg = _make_msg(expires_at=expiry)
        reader, _ = _make_reader_with_message(msg)

        result = _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

        assert result.expires_at is not None
        assert result.expires_at.replace(tzinfo=timezone.utc) == expiry

    def test_aiva_to_genesis_reads_correct_queue(self):
        """poll(AIVA_TO_GENESIS) must call blpop on bridge:queue:aiva_to_genesis."""
        msg = _make_msg(direction=MessageDirection.AIVA_TO_GENESIS)
        reader, mock_redis = _make_reader_with_message(msg)

        _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

        actual_key = mock_redis.blpop.call_args[0][0]
        assert actual_key == BRIDGE_QUEUE_AIVA_TO_GENESIS

    def test_genesis_to_aiva_reads_correct_queue(self):
        """poll(GENESIS_TO_AIVA) must call blpop on bridge:queue:genesis_to_aiva."""
        msg = _make_msg(direction=MessageDirection.GENESIS_TO_AIVA)
        reader, mock_redis = _make_reader_with_message(msg)

        _run(reader.poll(MessageDirection.GENESIS_TO_AIVA))

        actual_key = mock_redis.blpop.call_args[0][0]
        assert actual_key == BRIDGE_QUEUE_GENESIS_TO_AIVA


# ---------------------------------------------------------------------------
# BB2: Empty queue / timeout → returns None (not exception)
# ---------------------------------------------------------------------------


class TestBB2_TimeoutReturnsNone:
    """BB2: When the queue is empty and timeout elapses, poll() returns None."""

    def test_none_returned_on_timeout(self):
        """poll() must return None when blpop returns None (timeout reached)."""
        reader, _ = _make_empty_reader()

        result = _run(reader.poll(MessageDirection.AIVA_TO_GENESIS, timeout_s=0))

        assert result is None

    def test_no_exception_on_timeout(self):
        """poll() must not raise any exception when the queue is empty."""
        reader, _ = _make_empty_reader()

        try:
            _run(reader.poll(MessageDirection.AIVA_TO_GENESIS, timeout_s=0))
        except Exception as exc:
            pytest.fail(f"poll() raised on empty queue: {exc}")

    def test_genesis_to_aiva_also_returns_none_on_timeout(self):
        """Timeout must return None for any direction."""
        reader, _ = _make_empty_reader()

        result = _run(reader.poll(MessageDirection.GENESIS_TO_AIVA, timeout_s=0))

        assert result is None

    def test_return_type_is_none_not_false_or_zero(self):
        """None — not False or 0 — is the sentinel for 'no message'."""
        reader, _ = _make_empty_reader()

        result = _run(reader.poll(MessageDirection.AIVA_TO_GENESIS, timeout_s=0))

        assert result is None  # strict identity check


# ---------------------------------------------------------------------------
# BB3: Malformed JSON in queue → ValueError raised
# ---------------------------------------------------------------------------


class TestBB3_MalformedJsonRaisesValueError:
    """BB3: If the raw bytes in the queue cannot be parsed as JSON, ValueError is raised."""

    def test_invalid_json_raises_value_error(self):
        """Completely invalid JSON bytes must raise ValueError."""
        reader, _ = _make_malformed_reader(b"not json at all {{{{")

        with pytest.raises(ValueError):
            _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

    def test_empty_bytes_raises_value_error(self):
        """Empty bytes are not valid JSON and must raise ValueError."""
        reader, _ = _make_malformed_reader(b"")

        with pytest.raises(ValueError):
            _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

    def test_json_array_raises_value_error(self):
        """A JSON array (not object) is missing required fields → ValueError."""
        reader, _ = _make_malformed_reader(b"[1, 2, 3]")

        with pytest.raises(ValueError):
            _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

    def test_missing_required_field_raises_value_error(self):
        """JSON object missing 'message_id' must raise ValueError."""
        incomplete = json.dumps({
            "session_id": "s1",
            "direction": "aiva_to_genesis",
            "payload": {},
            "priority": 1,
            "created_at": "2026-02-25T10:00:00+00:00",
            "expires_at": None,
            # 'message_id' deliberately omitted
        }).encode()
        reader, _ = _make_malformed_reader(incomplete)

        with pytest.raises(ValueError):
            _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

    def test_invalid_direction_string_raises_value_error(self):
        """An unknown direction string must raise ValueError during deserialization."""
        bad_direction = json.dumps({
            "message_id": "id-001",
            "session_id": "s1",
            "direction": "totally_unknown_direction",  # invalid enum value
            "payload": {},
            "priority": 1,
            "created_at": "2026-02-25T10:00:00+00:00",
            "expires_at": None,
        }).encode()
        reader, _ = _make_malformed_reader(bad_direction)

        with pytest.raises(ValueError):
            _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

    def test_valid_json_but_invalid_created_at_raises_value_error(self):
        """A non-ISO-8601 created_at string must raise ValueError."""
        bad_ts = json.dumps({
            "message_id": "id-001",
            "session_id": "s1",
            "direction": "aiva_to_genesis",
            "payload": {},
            "priority": 1,
            "created_at": "not-a-date",  # invalid ISO 8601
            "expires_at": None,
        }).encode()
        reader, _ = _make_malformed_reader(bad_ts)

        with pytest.raises(ValueError):
            _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))


# ---------------------------------------------------------------------------
# WB1: BLPOP is used (not LPOP in a loop)
# ---------------------------------------------------------------------------


class TestWB1_BlpopUsed:
    """WB1: poll() must call redis.blpop(), never redis.lpop()."""

    def test_blpop_called_exactly_once(self):
        """A single blpop call must be made per poll() invocation."""
        msg = _make_msg()
        reader, mock_redis = _make_reader_with_message(msg)

        _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

        mock_redis.blpop.assert_called_once()

    def test_lpop_never_called(self):
        """lpop must NEVER be called — that would be a busy-poll loop."""
        msg = _make_msg()
        reader, mock_redis = _make_reader_with_message(msg)

        _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

        mock_redis.lpop.assert_not_called()

    def test_blpop_called_on_timeout_too(self):
        """blpop must be called even when the result is None (timeout path)."""
        reader, mock_redis = _make_empty_reader()

        _run(reader.poll(MessageDirection.AIVA_TO_GENESIS, timeout_s=0))

        mock_redis.blpop.assert_called_once()

    def test_no_loop_calls_multiple_blpop(self):
        """Exactly 1 blpop call — not 2, not 0 — per single poll() invocation."""
        msg = _make_msg()
        reader, mock_redis = _make_reader_with_message(msg)

        _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

        assert mock_redis.blpop.call_count == 1, (
            f"Expected exactly 1 blpop call, got {mock_redis.blpop.call_count}"
        )


# ---------------------------------------------------------------------------
# WB2: timeout_s is passed through to BLPOP (not hardcoded)
# ---------------------------------------------------------------------------


class TestWB2_TimeoutPassthrough:
    """WB2: poll(timeout_s=N) must pass N to blpop, not a hardcoded value."""

    def test_default_timeout_1_passed_to_blpop(self):
        """Default timeout_s=1 must be forwarded to blpop."""
        reader, mock_redis = _make_empty_reader()

        _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))  # default timeout_s=1

        _, kwargs = mock_redis.blpop.call_args
        actual_timeout = kwargs.get("timeout")
        assert actual_timeout == 1, (
            f"Expected timeout=1 passed to blpop, got timeout={actual_timeout}"
        )

    def test_explicit_timeout_0_passed_to_blpop(self):
        """timeout_s=0 (non-blocking) must be forwarded to blpop as timeout=0."""
        reader, mock_redis = _make_empty_reader()

        _run(reader.poll(MessageDirection.AIVA_TO_GENESIS, timeout_s=0))

        _, kwargs = mock_redis.blpop.call_args
        actual_timeout = kwargs.get("timeout")
        assert actual_timeout == 0, (
            f"Expected timeout=0 passed to blpop, got timeout={actual_timeout}"
        )

    def test_explicit_timeout_5_passed_to_blpop(self):
        """timeout_s=5 must be forwarded to blpop as timeout=5."""
        reader, mock_redis = _make_empty_reader()

        _run(reader.poll(MessageDirection.AIVA_TO_GENESIS, timeout_s=5))

        _, kwargs = mock_redis.blpop.call_args
        actual_timeout = kwargs.get("timeout")
        assert actual_timeout == 5, (
            f"Expected timeout=5 passed to blpop, got timeout={actual_timeout}"
        )

    def test_explicit_timeout_30_passed_to_blpop(self):
        """Large timeout_s=30 must also be forwarded verbatim."""
        reader, mock_redis = _make_empty_reader()

        _run(reader.poll(MessageDirection.AIVA_TO_GENESIS, timeout_s=30))

        _, kwargs = mock_redis.blpop.call_args
        actual_timeout = kwargs.get("timeout")
        assert actual_timeout == 30, (
            f"Expected timeout=30 passed to blpop, got timeout={actual_timeout}"
        )


# ---------------------------------------------------------------------------
# WB3: Return type is OpenClawMessage (not raw dict)
# ---------------------------------------------------------------------------


class TestWB3_ReturnTypeIsOpenClawMessage:
    """WB3: poll() must return an OpenClawMessage dataclass, not a dict or string."""

    def test_return_type_is_openclaw_message(self):
        """result must be an instance of OpenClawMessage."""
        msg = _make_msg()
        reader, _ = _make_reader_with_message(msg)

        result = _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

        assert isinstance(result, OpenClawMessage), (
            f"Expected OpenClawMessage, got {type(result).__name__}"
        )

    def test_return_type_is_not_dict(self):
        """result must NOT be a plain dict."""
        msg = _make_msg()
        reader, _ = _make_reader_with_message(msg)

        result = _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

        assert not isinstance(result, dict), "poll() must return OpenClawMessage, not dict"

    def test_return_type_is_not_string(self):
        """result must NOT be a raw JSON string."""
        msg = _make_msg()
        reader, _ = _make_reader_with_message(msg)

        result = _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

        assert not isinstance(result, (str, bytes)), (
            "poll() must return OpenClawMessage, not str/bytes"
        )

    def test_return_has_direction_as_enum(self):
        """direction field in the returned message must be a MessageDirection enum."""
        msg = _make_msg(direction=MessageDirection.AIVA_TO_GENESIS)
        reader, _ = _make_reader_with_message(msg)

        result = _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

        assert isinstance(result.direction, MessageDirection), (
            "direction must be deserialized back to MessageDirection enum"
        )

    def test_return_has_created_at_as_datetime(self):
        """created_at field must be a datetime, not a raw ISO string."""
        msg = _make_msg()
        reader, _ = _make_reader_with_message(msg)

        result = _run(reader.poll(MessageDirection.AIVA_TO_GENESIS))

        assert isinstance(result.created_at, datetime), (
            "created_at must be deserialized back to datetime, not left as string"
        )

    def test_bridgereader_exported_from_package(self):
        """BridgeReader must be accessible directly from the core.bridge package."""
        import core.bridge as pkg

        assert hasattr(pkg, "BridgeReader"), (
            "BridgeReader must be exported from core/bridge/__init__.py"
        )
        assert pkg.BridgeReader is BridgeReader

    def test_bridgereader_in_all(self):
        """BridgeReader must appear in core.bridge.__all__."""
        import core.bridge as pkg

        assert "BridgeReader" in pkg.__all__, (
            f"'BridgeReader' missing from __all__: {pkg.__all__}"
        )


# ---------------------------------------------------------------------------
# Run summary
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    result = pytest.main([__file__, "-v", "--tb=short"])
    sys.exit(result)