tests/test_models.py

from pydantic import ValidationError

from sysadmin_env.models import Action
from sysadmin_env.models import DifficultyTier
from sysadmin_env.models import Observation
from sysadmin_env.models import RewardSignal
from sysadmin_env.models import TaskMetadata


def test_action_valid():
    a = Action(command="systemctl restart nginx")
    assert a.command == "systemctl restart nginx"
    assert a.reasoning is None
    j = a.model_dump_json()
    assert "systemctl restart nginx" in j
    print("action valid ok")


def test_action_with_reasoning():
    a = Action(command="journalctl -u nginx", reasoning="checking logs for crash cause")
    assert a.reasoning == "checking logs for crash cause"
    roundtrip = Action.model_validate_json(a.model_dump_json())
    assert roundtrip == a
    print("action with reasoning ok")


def test_action_empty_command_rejected():
    try:
        Action(command="")
        raise AssertionError("should have raised")
    except ValidationError:
        pass
    print("action empty command rejected ok")


def test_action_missing_command_rejected():
    try:
        Action.model_validate({})
        raise AssertionError("should have raised")
    except ValidationError:
        pass
    print("action missing command rejected ok")


def test_observation_valid():
    o = Observation(
        stdout="active (running)",
        stderr="",
        exit_code=0,
        working_directory="/root",
        execution_time=0.042,
        reward=0.15,
        done=False,
        step_number=3,
        max_steps=50,
    )
    j = o.model_dump_json()
    roundtrip = Observation.model_validate_json(j)
    assert roundtrip == o
    print("observation valid ok")


def test_observation_negative_execution_time_rejected():
    try:
        Observation(
            stdout="",
            stderr="",
            exit_code=0,
            working_directory="/",
            execution_time=-1.0,
            reward=0.0,
            done=False,
            step_number=0,
            max_steps=10,
        )
        raise AssertionError("should have raised")
    except ValidationError:
        pass
    print("observation negative execution time rejected ok")


def test_observation_zero_max_steps_rejected():
    try:
        Observation(
            stdout="",
            stderr="",
            exit_code=0,
            working_directory="/",
            execution_time=0.0,
            reward=0.0,
            done=False,
            step_number=0,
            max_steps=0,
        )
        raise AssertionError("should have raised")
    except ValidationError:
        pass
    print("observation zero max steps rejected ok")


def test_task_metadata_valid():
    t = TaskMetadata(
        task_id="nginx_crash",
        difficulty=DifficultyTier.easy,
        description="nginx crashed with stale pid and config syntax error",
        max_steps=50,
        time_limit=300.0,
        base_filesystem_path="/assets/nginx_crash",
    )
    j = t.model_dump_json()
    roundtrip = TaskMetadata.model_validate_json(j)
    assert roundtrip == t
    assert roundtrip.difficulty == DifficultyTier.easy
    print("task metadata valid ok")


def test_task_metadata_invalid_difficulty_rejected():
    try:
        TaskMetadata(
            task_id="test",
            difficulty="legendary",
            description="invalid tier",
            max_steps=10,
            time_limit=60.0,
            base_filesystem_path="/tmp/test",
        )
        raise AssertionError("should have raised")
    except ValidationError:
        pass
    print("task metadata invalid difficulty rejected ok")


def test_task_metadata_empty_id_rejected():
    try:
        TaskMetadata(
            task_id="",
            difficulty="easy",
            description="empty id",
            max_steps=10,
            time_limit=60.0,
            base_filesystem_path="/tmp/test",
        )
        raise AssertionError("should have raised")
    except ValidationError:
        pass
    print("task metadata empty id rejected ok")


def test_reward_signal_valid():
    r = RewardSignal(
        health_delta=0.25,
        knowledge_delta=0.1,
        action_penalty=-0.01,
        total_reward=0.34,
    )
    j = r.model_dump_json()
    roundtrip = RewardSignal.model_validate_json(j)
    assert roundtrip == r
    print("reward signal valid ok")


def test_reward_signal_positive_penalty_rejected():
    try:
        RewardSignal(
            health_delta=0.0,
            knowledge_delta=0.0,
            action_penalty=0.5,
            total_reward=0.5,
        )
        raise AssertionError("should have raised")
    except ValidationError:
        pass
    print("reward signal positive penalty rejected ok")


def test_reward_signal_negative_knowledge_rejected():
    try:
        RewardSignal(
            health_delta=0.0,
            knowledge_delta=-0.1,
            action_penalty=-0.01,
            total_reward=-0.11,
        )
        raise AssertionError("should have raised")
    except ValidationError:
        pass
    print("reward signal negative knowledge rejected ok")


def test_all_models_serialization_roundtrip():
    """confirms all four models survive json roundtrip"""
    action = Action(command="ls -la", reasoning="listing files")
    obs = Observation(
        stdout="total 0",
        stderr="",
        exit_code=0,
        working_directory="/root",
        execution_time=0.001,
        reward=-0.01,
        done=False,
        step_number=1,
        max_steps=50,
    )
    task = TaskMetadata(
        task_id="disk_full",
        difficulty=DifficultyTier.medium,
        description="hidden sparse log file filling loopback mount",
        max_steps=75,
        time_limit=600.0,
        base_filesystem_path="/assets/disk_full",
    )
    reward = RewardSignal(
        health_delta=0.0,
        knowledge_delta=0.05,
        action_penalty=-0.01,
        total_reward=0.04,
    )

    for model_instance in [action, obs, task, reward]:
        json_str = model_instance.model_dump_json()
        restored = type(model_instance).model_validate_json(json_str)
        assert restored == model_instance

    print("all models serialization roundtrip ok")


if __name__ == "__main__":
    test_action_valid()
    test_action_with_reasoning()
    test_action_empty_command_rejected()
    test_action_missing_command_rejected()
    test_observation_valid()
    test_observation_negative_execution_time_rejected()
    test_observation_zero_max_steps_rejected()
    test_task_metadata_valid()
    test_task_metadata_invalid_difficulty_rejected()
    test_task_metadata_empty_id_rejected()
    test_reward_signal_valid()
    test_reward_signal_positive_penalty_rejected()
    test_reward_signal_negative_knowledge_rejected()
    test_all_models_serialization_roundtrip()
    print("all phase one tests passed")