Multi_Swarm_Coevolutive/tests/unit/test_protocol_validator.py

import json

import pytest

from multi_swarm.protocol.parser import parse_strategy
from multi_swarm.protocol.validator import ValidationError, validate_strategy


def _wrap(condition: dict, action: str = "entry-long") -> str:
    return json.dumps({"rules": [{"condition": condition, "action": action}]})


def test_valid_strategy_passes() -> None:
    src = _wrap(
        {
            "op": "gt",
            "args": [
                {"kind": "indicator", "name": "rsi", "params": [14]},
                {"kind": "literal", "value": 70.0},
            ],
        },
        action="entry-short",
    )
    ast = parse_strategy(src)
    validate_strategy(ast)  # no exception


def test_indicator_unknown_name_fails() -> None:
    src = _wrap(
        {
            "op": "gt",
            "args": [
                {"kind": "indicator", "name": "wibble", "params": [14]},
                {"kind": "literal", "value": 70.0},
            ],
        }
    )
    ast = parse_strategy(src)
    with pytest.raises(ValidationError, match="unknown indicator"):
        validate_strategy(ast)


def test_indicator_arity_too_few_fails() -> None:
    src = _wrap(
        {
            "op": "gt",
            "args": [
                {"kind": "indicator", "name": "rsi", "params": []},
                {"kind": "literal", "value": 70.0},
            ],
        }
    )
    ast = parse_strategy(src)
    with pytest.raises(ValidationError, match="arity"):
        validate_strategy(ast)


def test_indicator_arity_too_many_fails() -> None:
    src = _wrap(
        {
            "op": "gt",
            "args": [
                {"kind": "indicator", "name": "rsi", "params": [14, 28]},
                {"kind": "literal", "value": 70.0},
            ],
        }
    )
    ast = parse_strategy(src)
    with pytest.raises(ValidationError, match="arity"):
        validate_strategy(ast)


def test_macd_arity_zero_to_three_ok() -> None:
    for params in [[], [12], [12, 26], [12, 26, 9]]:
        src = _wrap(
            {
                "op": "gt",
                "args": [
                    {"kind": "indicator", "name": "macd", "params": params},
                    {"kind": "literal", "value": 0.0},
                ],
            }
        )
        ast = parse_strategy(src)
        validate_strategy(ast)


def test_macd_arity_four_fails() -> None:
    src = _wrap(
        {
            "op": "gt",
            "args": [
                {"kind": "indicator", "name": "macd", "params": [1, 2, 3, 4]},
                {"kind": "literal", "value": 0.0},
            ],
        }
    )
    ast = parse_strategy(src)
    with pytest.raises(ValidationError, match="arity"):
        validate_strategy(ast)


def test_comparator_wrong_arity_fails() -> None:
    src = _wrap({"op": "gt", "args": [{"kind": "literal", "value": 1.0}]})
    ast = parse_strategy(src)
    with pytest.raises(ValidationError, match="needs 2 args"):
        validate_strategy(ast)


def test_logical_not_arity_fails() -> None:
    src = _wrap(
        {
            "op": "not",
            "args": [
                {"kind": "literal", "value": 1.0},
                {"kind": "literal", "value": 2.0},
            ],
        }
    )
    ast = parse_strategy(src)
    with pytest.raises(ValidationError, match="'not' needs 1"):
        validate_strategy(ast)


def test_logical_and_arity_fails() -> None:
    src = _wrap({"op": "and", "args": [{"kind": "literal", "value": 1.0}]})
    ast = parse_strategy(src)
    with pytest.raises(ValidationError, match="and"):
        validate_strategy(ast)


def test_crossover_wrong_arity_fails() -> None:
    src = _wrap(
        {"op": "crossover", "args": [{"kind": "literal", "value": 1.0}]}
    )
    ast = parse_strategy(src)
    with pytest.raises(ValidationError, match="crossover"):
        validate_strategy(ast)


def test_feature_unknown_column_fails() -> None:
    src = _wrap(
        {
            "op": "gt",
            "args": [
                {"kind": "feature", "name": "wibble"},
                {"kind": "literal", "value": 100.0},
            ],
        }
    )
    ast = parse_strategy(src)
    with pytest.raises(ValidationError, match="unknown feature"):
        validate_strategy(ast)


@pytest.mark.parametrize("name", ["hour", "dow", "is_weekend", "minute_of_hour"])
def test_validator_accepts_temporal_feature(name: str) -> None:
    src = _wrap(
        {
            "op": "gt",
            "args": [
                {"kind": "feature", "name": name},
                {"kind": "literal", "value": 0.0},
            ],
        }
    )
    ast = parse_strategy(src)
    validate_strategy(ast)  # no exception


def test_validator_rejects_temporal_typo() -> None:
    src = _wrap(
        {
            "op": "gt",
            "args": [
                {"kind": "feature", "name": "weekday"},
                {"kind": "literal", "value": 0.0},
            ],
        }
    )
    ast = parse_strategy(src)
    with pytest.raises(ValidationError, match="unknown feature"):
        validate_strategy(ast)