i2p-python / tests / test_client / test_session.py

"""Tests for I2CP session state machine."""

import os

import pytest


class TestSessionState:
    def test_enum_values(self):
        from i2p_client.session import SessionState
        assert SessionState.INIT.value == 0
        assert SessionState.OPENING.value == 1
        assert SessionState.OPEN.value == 2
        assert SessionState.CLOSING.value == 3
        assert SessionState.CLOSED.value == 4
        assert SessionState.ERROR.value == 5


class TestSessionId:
    def test_construct(self):
        from i2p_client.session import SessionId
        sid = SessionId(42)
        assert int(sid) == 42

    def test_equality(self):
        from i2p_client.session import SessionId
        assert SessionId(1) == SessionId(1)
        assert SessionId(1) != SessionId(2)

    def test_hash(self):
        from i2p_client.session import SessionId
        s = {SessionId(1), SessionId(1), SessionId(2)}
        assert len(s) == 2

    def test_range_validation(self):
        from i2p_client.session import SessionId
        SessionId(0)
        SessionId(0xFFFF)
        with pytest.raises(ValueError):
            SessionId(-1)
        with pytest.raises(ValueError):
            SessionId(0x10000)


class TestI2CPSession:
    def test_initial_state(self):
        from i2p_client.session import I2CPSession, SessionState
        session = I2CPSession()
        assert session.state == SessionState.INIT

    def test_open_transitions_to_opening(self):
        from i2p_client.session import I2CPSession, SessionState
        from i2p_client.session_config import SessionConfig
        session = I2CPSession()
        dest_data = os.urandom(387)
        session.open(dest_data, SessionConfig())
        assert session.state == SessionState.OPENING

    def test_confirm_open(self):
        from i2p_client.session import I2CPSession, SessionState, SessionId
        from i2p_client.session_config import SessionConfig
        session = I2CPSession()
        session.open(os.urandom(387), SessionConfig())
        session.confirm_open(SessionId(5))
        assert session.state == SessionState.OPEN
        assert session.session_id == SessionId(5)

    def test_close(self):
        from i2p_client.session import I2CPSession, SessionState, SessionId
        from i2p_client.session_config import SessionConfig
        session = I2CPSession()
        session.open(os.urandom(387), SessionConfig())
        session.confirm_open(SessionId(1))
        session.close()
        assert session.state == SessionState.CLOSING

    def test_confirm_close(self):
        from i2p_client.session import I2CPSession, SessionState, SessionId
        from i2p_client.session_config import SessionConfig
        session = I2CPSession()
        session.open(os.urandom(387), SessionConfig())
        session.confirm_open(SessionId(1))
        session.close()
        session.confirm_close()
        assert session.state == SessionState.CLOSED

    def test_error_from_opening(self):
        from i2p_client.session import I2CPSession, SessionState
        from i2p_client.session_config import SessionConfig
        session = I2CPSession()
        session.open(os.urandom(387), SessionConfig())
        session.set_error("refused")
        assert session.state == SessionState.ERROR
        assert session.error_message == "refused"

    def test_cannot_open_twice(self):
        from i2p_client.session import I2CPSession
        from i2p_client.session_config import SessionConfig
        session = I2CPSession()
        session.open(os.urandom(387), SessionConfig())
        with pytest.raises(RuntimeError):
            session.open(os.urandom(387), SessionConfig())

    def test_cannot_confirm_from_init(self):
        from i2p_client.session import I2CPSession, SessionId
        session = I2CPSession()
        with pytest.raises(RuntimeError):
            session.confirm_open(SessionId(1))

    def test_cannot_close_from_init(self):
        from i2p_client.session import I2CPSession
        session = I2CPSession()
        with pytest.raises(RuntimeError):
            session.close()

    def test_message_callback(self):
        from i2p_client.session import I2CPSession, SessionState, SessionId
        from i2p_client.session_config import SessionConfig
        received = []
        session = I2CPSession()
        session.on_message(lambda msg: received.append(msg))
        session.open(os.urandom(387), SessionConfig())
        session.confirm_open(SessionId(1))
        session.deliver_message(b"hello i2p")
        assert len(received) == 1
        assert received[0] == b"hello i2p"

    def test_destination_data_stored(self):
        from i2p_client.session import I2CPSession
        from i2p_client.session_config import SessionConfig
        dest = os.urandom(387)
        session = I2CPSession()
        session.open(dest, SessionConfig())
        assert session.destination_data == dest

    def test_config_stored(self):
        from i2p_client.session import I2CPSession
        from i2p_client.session_config import SessionConfig
        cfg = SessionConfig({"inbound.quantity": "5"})
        session = I2CPSession()
        session.open(os.urandom(387), cfg)
        assert session.config.get("inbound.quantity") == "5"