#!/usr/bin/env python3 """ Tests for Story 8.01: OpenClaw Bridge — Inbound Message Schema AIVA RLM Nexus PRD v2 — Track A, Module 8 Black box tests (BB1-BB3): verify public contract — instantiation, enum values, optional field behaviour — without referencing internal implementation details. White box tests (WB1-WB3): verify internal implementation properties — dataclass equality semantics, deliberate absence of priority validation in the schema layer, and clean import with zero side effects. All tests are pure in-process — no I/O, no Redis, no network. """ from __future__ import annotations import importlib import sys from datetime import datetime, timezone import pytest sys.path.insert(0, "/mnt/e/genesis-system") from core.bridge.openclaw_bridge import MessageDirection, OpenClawMessage # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_msg(**overrides) -> OpenClawMessage: """Return a fully-populated OpenClawMessage with sensible defaults.""" defaults = dict( message_id="550e8400-e29b-41d4-a716-446655440000", session_id="session-test-001", direction=MessageDirection.AIVA_TO_GENESIS, payload={"intent": "task_request", "body": "test payload"}, priority=2, created_at=datetime(2026, 2, 25, 10, 0, 0, tzinfo=timezone.utc), ) defaults.update(overrides) return OpenClawMessage(**defaults) # --------------------------------------------------------------------------- # BB1: OpenClawMessage instantiates with all required fields # --------------------------------------------------------------------------- class TestBB1_Instantiation: """BB1: OpenClawMessage can be created with all six required fields.""" def test_instantiation_with_required_fields_only(self): """Creating a message without expires_at must succeed.""" msg = _make_msg() assert isinstance(msg, OpenClawMessage) def test_message_id_stored_correctly(self): msg = _make_msg(message_id="test-uuid-1234") assert msg.message_id == "test-uuid-1234" def test_session_id_stored_correctly(self): msg = _make_msg(session_id="my-session") assert msg.session_id == "my-session" def test_direction_stored_correctly(self): msg = _make_msg(direction=MessageDirection.GENESIS_TO_AIVA) assert msg.direction == MessageDirection.GENESIS_TO_AIVA def test_payload_stored_correctly(self): payload = {"intent": "inject_context", "data": [1, 2, 3]} msg = _make_msg(payload=payload) assert msg.payload == payload def test_priority_stored_correctly(self): msg = _make_msg(priority=1) assert msg.priority == 1 def test_created_at_stored_correctly(self): ts = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc) msg = _make_msg(created_at=ts) assert msg.created_at == ts # --------------------------------------------------------------------------- # BB2: MessageDirection enum has exactly 2 values with correct string values # --------------------------------------------------------------------------- class TestBB2_MessageDirectionEnum: """BB2: MessageDirection has exactly 2 members with the correct .value strings.""" def test_enum_has_exactly_two_members(self): members = list(MessageDirection) assert len(members) == 2, ( f"Expected 2 MessageDirection members, got {len(members)}: {members}" ) def test_aiva_to_genesis_value(self): assert MessageDirection.AIVA_TO_GENESIS.value == "aiva_to_genesis" def test_genesis_to_aiva_value(self): assert MessageDirection.GENESIS_TO_AIVA.value == "genesis_to_aiva" def test_both_members_accessible_by_name(self): assert MessageDirection["AIVA_TO_GENESIS"] is MessageDirection.AIVA_TO_GENESIS assert MessageDirection["GENESIS_TO_AIVA"] is MessageDirection.GENESIS_TO_AIVA def test_both_members_accessible_by_value(self): assert MessageDirection("aiva_to_genesis") is MessageDirection.AIVA_TO_GENESIS assert MessageDirection("genesis_to_aiva") is MessageDirection.GENESIS_TO_AIVA # --------------------------------------------------------------------------- # BB3: expires_at=None is valid (no expiry) # --------------------------------------------------------------------------- class TestBB3_ExpiresAtOptional: """BB3: expires_at defaults to None and can be set to a datetime.""" def test_expires_at_defaults_to_none(self): """Omitting expires_at must yield None (dataclass default).""" msg = _make_msg() assert msg.expires_at is None def test_expires_at_none_explicit(self): msg = _make_msg(expires_at=None) assert msg.expires_at is None def test_expires_at_accepts_datetime(self): expiry = datetime(2026, 12, 31, 23, 59, 59, tzinfo=timezone.utc) msg = _make_msg(expires_at=expiry) assert msg.expires_at == expiry def test_seven_fields_total(self): """OpenClawMessage must have exactly 7 fields (6 required + 1 optional).""" import dataclasses fields = dataclasses.fields(OpenClawMessage) assert len(fields) == 7, ( f"Expected 7 dataclass fields, found {len(fields)}: " f"{[f.name for f in fields]}" ) def test_six_required_fields(self): """Exactly 6 fields must have no default (i.e., are required).""" import dataclasses fields = dataclasses.fields(OpenClawMessage) required = [f for f in fields if f.default is dataclasses.MISSING and f.default_factory is dataclasses.MISSING] # type: ignore[misc] assert len(required) == 6, ( f"Expected 6 required fields, got {len(required)}: " f"{[f.name for f in required]}" ) # --------------------------------------------------------------------------- # WB1: dataclass __eq__ works correctly # --------------------------------------------------------------------------- class TestWB1_DataclassEquality: """WB1: Two OpenClawMessage instances with identical field values compare equal.""" def test_equal_instances_are_equal(self): ts = datetime(2026, 2, 25, 12, 0, 0, tzinfo=timezone.utc) a = OpenClawMessage( message_id="abc", session_id="s1", direction=MessageDirection.AIVA_TO_GENESIS, payload={"k": "v"}, priority=3, created_at=ts, ) b = OpenClawMessage( message_id="abc", session_id="s1", direction=MessageDirection.AIVA_TO_GENESIS, payload={"k": "v"}, priority=3, created_at=ts, ) assert a == b def test_different_message_ids_not_equal(self): ts = datetime(2026, 2, 25, 12, 0, 0, tzinfo=timezone.utc) a = _make_msg(message_id="id-001", created_at=ts) b = _make_msg(message_id="id-002", created_at=ts) assert a != b def test_different_directions_not_equal(self): ts = datetime(2026, 2, 25, 12, 0, 0, tzinfo=timezone.utc) a = _make_msg(direction=MessageDirection.AIVA_TO_GENESIS, created_at=ts) b = _make_msg(direction=MessageDirection.GENESIS_TO_AIVA, created_at=ts) assert a != b def test_different_priorities_not_equal(self): ts = datetime(2026, 2, 25, 12, 0, 0, tzinfo=timezone.utc) a = _make_msg(priority=1, created_at=ts) b = _make_msg(priority=3, created_at=ts) assert a != b # --------------------------------------------------------------------------- # WB2: priority=4 does not raise (validation belongs to routing layer) # --------------------------------------------------------------------------- class TestWB2_PriorityNoValidation: """WB2: Schema layer accepts any int for priority — routing layer validates range.""" def test_priority_4_does_not_raise(self): """Priority 4 (out of 1-3 spec range) must NOT raise at schema level.""" try: msg = _make_msg(priority=4) except Exception as exc: # pragma: no cover pytest.fail( f"OpenClawMessage raised on priority=4, but schema must not validate: {exc}" ) assert msg.priority == 4 def test_priority_0_does_not_raise(self): msg = _make_msg(priority=0) assert msg.priority == 0 def test_priority_negative_does_not_raise(self): msg = _make_msg(priority=-1) assert msg.priority == -1 def test_priority_is_int_field(self): import dataclasses fields = {f.name: f for f in dataclasses.fields(OpenClawMessage)} assert fields["priority"].type is int or str(fields["priority"].type) in ( "int", "" ), "priority field type annotation must be int" # --------------------------------------------------------------------------- # WB3: Module imports standalone with no side effects # --------------------------------------------------------------------------- class TestWB3_CleanImport: """WB3: Importing openclaw_bridge has zero side effects (no I/O, no DB connections).""" def test_module_importable(self): """Module must be importable without raising.""" import core.bridge.openclaw_bridge as mod # noqa: F401 assert mod is not None def test_package_init_importable(self): """Package __init__ must re-export both names without error.""" import core.bridge as pkg assert hasattr(pkg, "MessageDirection") assert hasattr(pkg, "OpenClawMessage") def test_no_sqlite3_import(self): """sqlite3 is forbidden (Rule 7 — no local SQLite).""" import inspect import core.bridge.openclaw_bridge as mod source = inspect.getsource(mod) assert "import sqlite3" not in source, ( "sqlite3 is FORBIDDEN in openclaw_bridge.py (Rule 7)" ) def test_no_network_calls_on_import(self): """Re-importing the module must not open any sockets.""" import socket original_socket = socket.socket calls: list[str] = [] class _TrackingSocket(socket.socket): # type: ignore[misc] def __init__(self, *a, **kw): calls.append("socket_created") super().__init__(*a, **kw) socket.socket = _TrackingSocket # type: ignore[misc] try: importlib.reload(sys.modules["core.bridge.openclaw_bridge"]) finally: socket.socket = original_socket # type: ignore[misc] assert calls == [], ( f"Import opened {len(calls)} socket(s) — module must have zero side effects" ) def test_all_names_in_all(self): """Package __all__ must export MessageDirection and OpenClawMessage (at minimum). NOTE (Story 8.02): __init__.py was extended per PRD to also export BridgeWriter and the two queue-key constants. This test now asserts that the original 8.01 symbols are present as a *subset*, rather than asserting exact equality, so the regression suite stays green as the package evolves through later stories. """ import core.bridge as pkg required_8_01 = {"MessageDirection", "OpenClawMessage"} assert required_8_01.issubset(set(pkg.__all__)), ( f"__all__ is missing 8.01 symbols: {required_8_01 - set(pkg.__all__)}" ) # --------------------------------------------------------------------------- # Run summary # --------------------------------------------------------------------------- if __name__ == "__main__": result = pytest.main([__file__, "-v", "--tb=short"]) sys.exit(result)