or1-design / tests / test_repl.py

"""Unit tests for monitor REPL functionality.

Tests cover all acceptance criteria for AC4.1-AC4.9, verifying REPL
commands work correctly with a synchronized backend and proper
state management.
"""

from __future__ import annotations

import io
import sys
from pathlib import Path
from tempfile import NamedTemporaryFile

import pytest

from cm_inst import MemOp
from monitor.backend import SimulationBackend
from monitor.repl import MonitorREPL
from tokens import MonadToken, SMToken


@pytest.fixture
def temp_dfasm_file():
    """Create a temporary dfasm file for testing.

    A minimal valid dfasm program that loads without errors.
    """
    content = """@system pe=1, sm=0
&c1 <| const, 3
&c2 <| const, 7
&result <| add
&c1 |> &result:L
&c2 |> &result:R
"""
    with NamedTemporaryFile(mode="w", suffix=".dfasm", delete=False) as f:
        f.write(content)
        f.flush()
        path = Path(f.name)

    yield path

    # Cleanup
    path.unlink()


@pytest.fixture
def backend():
    """Create and start a simulation backend for testing."""
    b = SimulationBackend()
    b.start()
    yield b
    b.stop()


@pytest.fixture
def repl(backend):
    """Create a REPL instance with a test backend."""
    return MonitorREPL(backend)


class TestREPLLoad:
    """Tests for AC4.1: load command assembles and loads dfasm."""

    def test_load_valid_file(self, repl, temp_dfasm_file):
        """load <path> should load a valid dfasm file."""
        # Capture output
        output = io.StringIO()
        repl.stdout = output

        # Call do_load
        repl.do_load(str(temp_dfasm_file))

        # Check state
        assert repl._loaded is True
        assert repl._last_snapshot is not None
        assert "Loaded" in output.getvalue()

    def test_load_nonexistent_file(self, repl):
        """load with nonexistent path should fail gracefully."""
        output = io.StringIO()
        repl.stdout = output

        repl.do_load("/nonexistent/path.dfasm")

        assert repl._loaded is False
        assert "not found" in output.getvalue()

    def test_load_invalid_syntax(self, repl):
        """load with invalid dfasm syntax should report error."""
        # Create invalid file
        with NamedTemporaryFile(mode="w", suffix=".dfasm", delete=False) as f:
            f.write("@@@ invalid syntax @@@")
            f.flush()
            path = Path(f.name)

        try:
            output = io.StringIO()
            repl.stdout = output

            repl.do_load(str(path))

            assert repl._loaded is False
            out = output.getvalue()
            assert "Error" in out
        finally:
            path.unlink()

    def test_load_without_args(self, repl):
        """load without args should print usage."""
        output = io.StringIO()
        repl.stdout = output

        repl.do_load("")

        assert "Usage" in output.getvalue()


class TestREPLStep:
    """Tests for AC4.2: step and stepe commands advance simulation."""

    def test_step_without_load(self, repl):
        """step without loaded simulation should error (AC4.8)."""
        output = io.StringIO()
        repl.stdout = output

        repl.do_step("")

        assert "No simulation loaded" in output.getvalue()

    def test_step_after_load(self, repl, temp_dfasm_file):
        """step after loading should advance simulation."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))
        assert repl._loaded

        # Clear and step
        output = io.StringIO()
        repl.stdout = output

        repl.do_step("")

        # Should produce output (snapshot and/or events)
        assert len(output.getvalue()) > 0

    def test_stepe_after_load(self, repl, temp_dfasm_file):
        """stepe after loading should advance by one event."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))
        assert repl._loaded

        # Clear and step
        output = io.StringIO()
        repl.stdout = output

        repl.do_stepe("")

        # Should produce output
        assert len(output.getvalue()) > 0

    def test_stepe_without_load(self, repl):
        """stepe without loaded simulation should error (AC4.8)."""
        output = io.StringIO()
        repl.stdout = output

        repl.do_stepe("")

        assert "No simulation loaded" in output.getvalue()


class TestREPLRun:
    """Tests for AC4.3: run command runs until target time."""

    def test_run_without_load(self, repl):
        """run without loaded simulation should error (AC4.8)."""
        output = io.StringIO()
        repl.stdout = output

        repl.do_run("10")

        assert "No simulation loaded" in output.getvalue()

    def test_run_after_load(self, repl, temp_dfasm_file):
        """run <until> should run until target time."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))
        assert repl._loaded

        # Clear and run
        output = io.StringIO()
        repl.stdout = output

        repl.do_run("10")

        # Should produce output
        assert len(output.getvalue()) > 0

    def test_run_invalid_time(self, repl, temp_dfasm_file):
        """run with invalid time should error."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))

        # Try invalid time
        output = io.StringIO()
        repl.stdout = output

        repl.do_run("not_a_number")

        assert "Invalid time" in output.getvalue()

    def test_run_without_args(self, repl, temp_dfasm_file):
        """run without args should print usage."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))

        # Try without args
        output = io.StringIO()
        repl.stdout = output

        repl.do_run("")

        assert "Usage" in output.getvalue()


class TestREPLSend:
    """Tests for AC4.4: send command injects token via FIFO."""

    def test_send_monad_without_load(self, repl):
        """send without loaded simulation should error (AC4.8)."""
        output = io.StringIO()
        repl.stdout = output

        repl.do_send("0 0 0 42")

        assert "No simulation loaded" in output.getvalue()

    def test_send_monad_token(self, repl, temp_dfasm_file):
        """send <target> <offset> <ctx> <data> should send MonadToken."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))
        assert repl._loaded

        # Clear and send
        output = io.StringIO()
        repl.stdout = output

        repl.do_send("0 0 0 42")

        # Should produce output (no exception)
        assert "Sent" in output.getvalue()

    def test_send_sm_token(self, repl, temp_dfasm_file):
        """send <target> --sm <addr> <op> should send SMToken."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))
        assert repl._loaded

        # Clear and send SM token
        output = io.StringIO()
        repl.stdout = output

        repl.do_send("0 --sm 5 READ")

        # Should produce output (no exception)
        assert "Sent" in output.getvalue()

    def test_send_sm_token_with_data(self, repl, temp_dfasm_file):
        """send <target> --sm <addr> <op> <data> should work."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))
        assert repl._loaded

        # Send SM token with data
        output = io.StringIO()
        repl.stdout = output

        repl.do_send("0 --sm 5 WRITE 123")

        assert "Sent" in output.getvalue()

    def test_send_invalid_args(self, repl, temp_dfasm_file):
        """send with invalid args should error gracefully."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))

        # Try invalid args
        output = io.StringIO()
        repl.stdout = output

        repl.do_send("not_an_int")

        assert "Error" in output.getvalue()

    def test_send_unknown_memop(self, repl, temp_dfasm_file):
        """send with unknown MemOp should error."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))

        # Try unknown MemOp
        output = io.StringIO()
        repl.stdout = output

        repl.do_send("0 --sm 5 UNKNOWN_OP")

        assert "Unknown MemOp" in output.getvalue()

    def test_send_without_args(self, repl, temp_dfasm_file):
        """send without args should print usage."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))

        # Try without args
        output = io.StringIO()
        repl.stdout = output

        repl.do_send("")

        assert "Usage" in output.getvalue()


class TestREPLInject:
    """Tests for AC4.5: inject command does direct injection."""

    def test_inject_monad_without_load(self, repl):
        """inject without loaded simulation should error (AC4.8)."""
        output = io.StringIO()
        repl.stdout = output

        repl.do_inject("0 0 0 42")

        assert "No simulation loaded" in output.getvalue()

    def test_inject_monad_token(self, repl, temp_dfasm_file):
        """inject <target> <offset> <ctx> <data> should inject MonadToken."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))
        assert repl._loaded

        # Clear and inject
        output = io.StringIO()
        repl.stdout = output

        repl.do_inject("0 0 0 42")

        # Should produce output (no exception)
        assert "Injected" in output.getvalue()

    def test_inject_sm_token(self, repl):
        """inject <target> --sm <addr> <op> should inject SMToken without throwing."""
        # For this test, we just verify token parsing works
        # Actual injection will fail without SM, but parsing should succeed
        output = io.StringIO()
        repl.stdout = output

        # Set loaded manually just to bypass the check
        repl._loaded = True

        # Just verify the token parsing happens without exception
        token = repl._parse_token_args("0 --sm 5 WRITE")
        assert token.addr == 5

    def test_inject_invalid_args(self, repl, temp_dfasm_file):
        """inject with invalid args should error gracefully."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))

        # Try invalid args
        output = io.StringIO()
        repl.stdout = output

        repl.do_inject("invalid")

        assert "Error" in output.getvalue()


class TestREPLState:
    """Tests for AC4.6: state, pe, sm commands display readable state."""

    def test_state_without_load(self, repl):
        """state without loaded simulation should error (AC4.8)."""
        output = io.StringIO()
        repl.stdout = output

        repl.do_state("")

        assert "No simulation loaded" in output.getvalue()

    def test_state_after_load(self, repl, temp_dfasm_file):
        """state should display system state summary."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))
        assert repl._loaded

        # Clear and call state
        output = io.StringIO()
        repl.stdout = output

        repl.do_state("")

        out = output.getvalue()
        # Should show PE/SM counts and time info
        assert len(out) > 0
        assert ("PE" in out or "SM" in out or "Time" in out)

    def test_pe_without_load(self, repl):
        """pe without loaded simulation should error (AC4.8)."""
        output = io.StringIO()
        repl.stdout = output

        repl.do_pe("0")

        assert "No simulation loaded" in output.getvalue()

    def test_pe_after_load(self, repl, temp_dfasm_file):
        """pe <id> should display PE state."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))
        assert repl._loaded

        # Clear and call pe
        output = io.StringIO()
        repl.stdout = output

        repl.do_pe("0")

        out = output.getvalue()
        # Should show PE state or not found message
        assert len(out) > 0
        # Verify formatting includes lane information or empty tag store marker (case-insensitive)
        assert "lane" in out.lower() or "tag store: (empty)" in out.lower()

    def test_pe_invalid_id(self, repl, temp_dfasm_file):
        """pe with non-integer ID should error."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))

        # Try invalid ID
        output = io.StringIO()
        repl.stdout = output

        repl.do_pe("invalid_id")

        assert "Invalid PE ID" in output.getvalue()

    def test_sm_without_load(self, repl):
        """sm without loaded simulation should error (AC4.8)."""
        output = io.StringIO()
        repl.stdout = output

        repl.do_sm("0")

        assert "No simulation loaded" in output.getvalue()

    def test_sm_after_load(self, repl, temp_dfasm_file):
        """sm <id> should display SM state."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))
        assert repl._loaded

        # Clear and call sm
        output = io.StringIO()
        repl.stdout = output

        repl.do_sm("0")

        out = output.getvalue()
        # Should show SM state or not found message
        assert len(out) > 0

    def test_sm_invalid_id(self, repl, temp_dfasm_file):
        """sm with non-integer ID should error."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))

        # Try invalid ID
        output = io.StringIO()
        repl.stdout = output

        repl.do_sm("invalid_id")

        assert "Invalid SM ID" in output.getvalue()


class TestREPLLog:
    """Tests for AC4.7: log command shows recent events."""

    def test_log_without_events(self, repl):
        """log with no events should indicate no events."""
        output = io.StringIO()
        repl.stdout = output

        repl.do_log("")

        assert "No events" in output.getvalue()

    def test_log_after_step(self, repl, temp_dfasm_file):
        """log after stepping should show events."""
        # Load and step
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))
        repl.do_step("")

        # Now log
        output = io.StringIO()
        repl.stdout = output

        repl.do_log("")

        # Output depends on whether there were events
        out = output.getvalue()
        assert len(out) > 0

    def test_log_filter_pe(self, repl, temp_dfasm_file):
        """log pe:<id> should filter by PE component."""
        # Load and step a program with multiple components
        output = io.StringIO()
        repl.stdout = output

        # Create a dfasm with multiple PEs to test filtering
        import tempfile
        from pathlib import Path
        with tempfile.NamedTemporaryFile(mode='w', suffix='.dfasm', delete=False) as f:
            f.write("""\
@system pe=2, sm=1
&c0|pe0 <| const, 1
&c1|pe1 <| const, 2
&pass0|pe0 <| pass
&pass1|pe1 <| pass
&c0|pe0 |> &pass0|pe0:L
&c1|pe1 |> &pass1|pe1:L
""")
            temp_file = f.name

        try:
            repl.do_load(temp_file)
            # Run simulation to capture events (cycle-accurate timing starts events at time 1+)
            repl.do_run("100")

            # Filter by PE 0
            output = io.StringIO()
            repl.stdout = output
            repl.do_log("pe:0")

            # Should complete without exception and contain pe:0 events
            output_str = output.getvalue()
            assert len(output_str) > 0, "Expected output from filtering"
            # The filtered output should contain references to pe:0
            assert "pe:0" in output_str or "PE 0" in output_str, \
                "Expected filtered output to contain pe:0"
            # Verify that filtering actually excludes pe:1
            assert "pe:1" not in output_str and "PE 1" not in output_str, \
                "Expected filtered output to exclude pe:1"
        finally:
            Path(temp_file).unlink(missing_ok=True)

    def test_log_filter_sm(self, repl, temp_dfasm_file):
        """log sm:<id> should filter by SM component."""
        # Load and step a program with SM operations
        output = io.StringIO()
        repl.stdout = output

        # Create a dfasm with SM operations to generate SM events
        import tempfile
        from pathlib import Path
        with tempfile.NamedTemporaryFile(mode='w', suffix='.dfasm', delete=False) as f:
            f.write("""\
@system pe=2, sm=2
&c0|pe0 <| const, 100
&write_op|pe0 <| write
&c1|pe1 <| const, 200
&read_op|pe1 <| read
&c0|pe0 |> &write_op|pe0:L
&c1|pe1 |> &read_op|pe1:L
""")
            temp_file = f.name

        try:
            repl.do_load(temp_file)
            # Step multiple times to ensure events are generated
            for _ in range(5):
                repl.do_step("")

            # Filter by SM 0
            output = io.StringIO()
            repl.stdout = output
            repl.do_log("sm:0")

            # Should complete without exception
            output_str = output.getvalue()
            assert len(output_str) > 0, "Expected output from SM filtering"
            # Verify that sm:1 is NOT in the filtered output (exclusion check)
            assert "sm:1" not in output_str, \
                "Expected filtered output (sm:0) to exclude sm:1 events"
        finally:
            Path(temp_file).unlink(missing_ok=True)

    def test_log_filter_by_type(self, repl, temp_dfasm_file):
        """log <type> should filter by event type name."""
        # Load and step
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))
        repl.do_step("")

        # Filter by type
        output = io.StringIO()
        repl.stdout = output

        repl.do_log("TokenReceived")

        # Should complete without exception
        assert len(output.getvalue()) > 0


class TestREPLTime:
    """Tests for time command."""

    def test_time_shows_current_time(self, repl, temp_dfasm_file):
        """time should show current simulation time."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))

        # Check time
        output = io.StringIO()
        repl.stdout = output

        repl.do_time("")

        assert "Simulation time" in output.getvalue()


class TestREPLReset:
    """Tests for AC4.9: reset command clears simulation state."""

    def test_reset_clears_state(self, repl, temp_dfasm_file):
        """reset should clear loaded state and event history."""
        # Load first
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))
        assert repl._loaded

        # Reset
        output = io.StringIO()
        repl.stdout = output

        repl.do_reset("")

        assert repl._loaded is False
        assert len(repl._event_history) == 0
        assert "reset" in output.getvalue()

    def test_step_after_reset_errors(self, repl, temp_dfasm_file):
        """step after reset should fail (AC4.8)."""
        # Load and reset
        output = io.StringIO()
        repl.stdout = output

        repl.do_load(str(temp_dfasm_file))
        repl.do_reset("")

        # Try step
        output = io.StringIO()
        repl.stdout = output

        repl.do_step("")

        assert "No simulation loaded" in output.getvalue()


class TestREPLQuit:
    """Tests for AC4.9: quit command exits cleanly."""

    def test_quit_returns_true(self, repl):
        """do_quit should return True to exit cmdloop."""
        result = repl.do_quit("")
        assert result is True

    def test_eof_returns_true(self, repl):
        """do_EOF should return True for Ctrl-D."""
        result = repl.do_EOF("")
        assert result is True


class TestTokenParsing:
    """Tests for token parsing in send/inject."""

    def test_parse_monad_token(self, repl):
        """Parse MonadToken from args."""
        token = repl._parse_token_args("0 10 5 42")
        assert isinstance(token, MonadToken)
        assert token.target == 0
        assert token.offset == 10
        assert token.act_id == 5
        assert token.data == 42
        assert token.inline is False

    def test_parse_sm_token_read(self, repl):
        """Parse SMToken with READ op."""
        token = repl._parse_token_args("1 --sm 20 READ")
        assert isinstance(token, SMToken)
        assert token.target == 1
        assert token.addr == 20
        assert token.op == MemOp.READ
        assert token.data is None

    def test_parse_sm_token_with_data(self, repl):
        """Parse SMToken with data."""
        token = repl._parse_token_args("1 --sm 20 WRITE 999")
        assert isinstance(token, SMToken)
        assert token.target == 1
        assert token.addr == 20
        assert token.op == MemOp.WRITE
        assert token.data == 999

    def test_parse_invalid_target(self, repl):
        """Parse should fail on invalid target."""
        with pytest.raises(ValueError):
            repl._parse_token_args("not_a_number 0 0 0")

    def test_parse_monad_missing_args(self, repl):
        """Parse should fail if MonadToken args are incomplete."""
        with pytest.raises(ValueError):
            repl._parse_token_args("0 10 5")  # Missing data

    def test_parse_sm_missing_args(self, repl):
        """Parse should fail if SMToken args are incomplete."""
        with pytest.raises(ValueError):
            repl._parse_token_args("0 --sm 20")  # Missing op

    def test_parse_sm_unknown_op(self, repl):
        """Parse should fail on unknown MemOp."""
        with pytest.raises(ValueError, match="Unknown MemOp"):
            repl._parse_token_args("0 --sm 20 INVALID_OP")