Centaur_Quant_Architecture/Python/tests/test_sdp_schema.py

# -*- coding: utf-8 -*-
"""
tests/test_sdp_schema.py — SDP Schema QA Harness (System Finalization)

Validates every SDP action envelope against the ratified wire schema and
exercises the router's strict parse path. Stdlib only (unittest).

Run:  python -m unittest tests.test_sdp_schema   (or)   python tests/test_sdp_schema.py
"""

import json
import os
import sys
import tempfile
import unittest

# --- make the Python/ package importable ---------------------------------
_PY = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, _PY)

# Route the router's TelemetryDB to a throwaway file before importing it.
_TMP_DB = os.path.join(tempfile.gettempdir(), "sdp_schema_test.db")
if os.path.exists(_TMP_DB):
    os.remove(_TMP_DB)
os.environ["CENTAUR_DB_PATH"] = _TMP_DB

REQUIRED_KEYS = ("sdp_version", "timestamp", "symbol", "timeframe",
                 "action_type", "algorithmic_confidence_score", "payload")
VALID_ACTIONS = {"Heartbeat", "Tick_Harvest", "Setup_Detected",
                 "Trade_Opened", "Trade_Closed"}


def validate_envelope(msg):
    """Return (ok: bool, errors: list[str]) against the ratified SDP schema."""
    errors = []
    if not isinstance(msg, dict):
        return False, ["not a JSON object"]
    for key in REQUIRED_KEYS:
        if key not in msg:
            errors.append("missing key '%s'" % key)
    if errors:
        return False, errors
    if not isinstance(msg["sdp_version"], str):
        errors.append("sdp_version must be str")
    if not isinstance(msg["timestamp"], str):
        errors.append("timestamp must be str")
    if not isinstance(msg["symbol"], str) or not msg["symbol"]:
        errors.append("symbol must be non-empty str")
    if not isinstance(msg["timeframe"], str):
        errors.append("timeframe must be str")
    if msg["action_type"] not in VALID_ACTIONS:
        errors.append("unknown action_type '%s'" % msg["action_type"])
    try:
        score = float(msg["algorithmic_confidence_score"])
        if not (0.0 <= score <= 100.0):
            errors.append("confidence out of [0,100]")
    except (TypeError, ValueError):
        errors.append("confidence must be numeric")
    if not isinstance(msg["payload"], dict):
        errors.append("payload must be an object")
    return (len(errors) == 0), errors


def envelope(action, payload, **overrides):
    msg = {
        "sdp_version": "1.0.0",
        "timestamp": "2026-08-12T18:00:00.000Z",
        "symbol": "XAUUSD",
        "timeframe": "M15",
        "action_type": action,
        "algorithmic_confidence_score": 62.5,
        "payload": payload,
    }
    msg.update(overrides)
    return msg


class TestSdpSchema(unittest.TestCase):
    """The 5 ratified action envelopes must all pass schema validation."""

    def test_all_actions_valid(self):
        cases = [
            envelope("Heartbeat", {"status": "alive"}),
            envelope("Tick_Harvest", {"bid": 2458.4, "ask": 2458.6,
                                      "spread_points": 20,
                                      "tick_time": "2026-08-12T18:00:00.000Z",
                                      "tick_volume": 12}),
            envelope("Setup_Detected", {"setup_type": "OB_FVG_BULLISH",
                                        "entry": 2458.5, "sl": 2450.0,
                                        "tp": 2475.0, "risk_reward": 2.0},
                     historical_context=[{"swing_high": 2460.0, "swing_low": 2450.0,
                                          "time": "2026-08-12T18:00:00.000Z"}]),
            envelope("Trade_Opened", {"ticket": 987654, "direction": "buy",
                                      "lot": 0.10, "entry_price": 2458.5,
                                      "sl": 2450.0, "tp": 2475.0}),
            envelope("Trade_Closed", {"ticket": 987654, "profit": 165.0,
                                      "r_multiple": 1.75,
                                      "initial_ai_score": 62.5}),
        ]
        for msg in cases:
            ok, errs = validate_envelope(msg)
            self.assertTrue(ok, "envelope %s invalid: %s" % (msg["action_type"], errs))

    def test_negative_cases(self):
        # empty object -> missing required keys
        ok, errs = validate_envelope({})
        self.assertFalse(ok)
        self.assertTrue(any("action_type" in e for e in errs))
        self.assertTrue(any("payload" in e for e in errs))

        # unknown action_type
        bad_action = envelope("Heartbeat", {})
        bad_action["action_type"] = "Something_Else"
        self.assertFalse(validate_envelope(bad_action)[0])

        # confidence out of range
        bad_score = envelope("Heartbeat", {})
        bad_score["algorithmic_confidence_score"] = 150.0
        self.assertFalse(validate_envelope(bad_score)[0])

        # payload must be an object
        bad_payload = envelope("Heartbeat", {})
        bad_payload["payload"] = [1, 2, 3]
        self.assertFalse(validate_envelope(bad_payload)[0])

    def test_roundtrip_via_router(self):
        """The router must accept real wire frames (CRLF and LF) and reject junk."""
        from router.main import FrameReader, parse_sdp_frame

        frame = json.dumps(envelope("Heartbeat", {"status": "alive"}))

        # CRLF-terminated frame
        reader = FrameReader()
        frames = reader.feed((frame + "\r\n").encode("utf-8"))
        self.assertEqual(len(frames), 1)
        msg = parse_sdp_frame(frames[0])
        self.assertIsNotNone(msg)
        self.assertEqual(msg["action_type"], "Heartbeat")

        # LF-only frame
        frames = reader.feed((frame + "\n").encode("utf-8"))
        self.assertEqual(len(frames), 1)

        # fragmented frame across reads
        reader2 = FrameReader()
        data = (frame + "\n").encode("utf-8")
        half = len(data) // 2
        self.assertEqual(len(reader2.feed(data[:half])), 0)
        self.assertEqual(len(reader2.feed(data[half:])), 1)

        # malformed -> rejected, never raised
        self.assertIsNone(parse_sdp_frame("{not json"))


if __name__ == "__main__":
    unittest.main()