"""Test module for erlang.reactor functionality.

This module provides Python-side tests for the reactor API.

These can be called from Erlang tests or run standalone.

"""

import socket

def test_protocol_creation():

"""Test Protocol class can be instantiated and subclassed."""

import sys

sys.path.insert(0, 'priv')

from _erlang_impl import reactor

# Test base protocol

proto = reactor.Protocol()

assert proto.fd == -1

assert proto.client_info == {}

assert proto.write_buffer == bytearray()

assert proto.closed is False

# Test connection_made

proto.connection_made(42, {'addr': '127.0.0.1', 'port': 8080})

assert proto.fd == 42

assert proto.client_info == {'addr': '127.0.0.1', 'port': 8080}

return True

def test_echo_protocol():

"""Test a simple echo protocol with socketpair."""

import sys

sys.path.insert(0, 'priv')

from _erlang_impl import reactor

class EchoProtocol(reactor.Protocol):

def data_received(self, data):

self.write_buffer.extend(data)

return "write_pending"

def write_ready(self):

if not self.write_buffer:

return "read_pending"

written = self.write(bytes(self.write_buffer))

del self.write_buffer[:written]

return "continue" if self.write_buffer else "read_pending"

# Create socketpair for testing

s1, s2 = socket.socketpair()

s1.setblocking(False)

s2.setblocking(False)

try:

# Set up protocol

reactor.set_protocol_factory(EchoProtocol)

reactor.init_connection(s1.fileno(), {'type': 'test'})

# Send data through s2

s2.send(b'hello world')

# Trigger read on protocol side

action = reactor.on_read_ready(s1.fileno())

assert action == "write_pending"

# Check write buffer

proto = reactor.get_protocol(s1.fileno())

assert bytes(proto.write_buffer) == b'hello world'

# Trigger write

action = reactor.on_write_ready(s1.fileno())

# Should be read_pending since buffer was flushed

assert action == "read_pending"

# Read the echoed data

echoed = s2.recv(1024)

assert echoed == b'hello world'

# Clean up

reactor.close_connection(s1.fileno())

assert proto.closed is True

finally:

s1.close()

s2.close()

return True

def test_multiple_protocols():

"""Test multiple protocols can be registered simultaneously."""

import sys

sys.path.insert(0, 'priv')

from _erlang_impl import reactor

class SimpleProtocol(reactor.Protocol):

instances = []

def __init__(self):

super().__init__()

SimpleProtocol.instances.append(self)

def data_received(self, data):

return "continue"

def write_ready(self):

return "close"

# Reset instances

SimpleProtocol.instances = []

# Create multiple socketpairs

pairs = []

for _ in range(5):

s1, s2 = socket.socketpair()

s1.setblocking(False)

pairs.append((s1, s2))

try:

# Set factory and init connections

reactor.set_protocol_factory(SimpleProtocol)

for s1, s2 in pairs:

reactor.init_connection(s1.fileno(), {})

# Verify all instances created

assert len(SimpleProtocol.instances) == 5

# Verify each has correct fd

for (s1, s2), proto in zip(pairs, SimpleProtocol.instances):

assert proto.fd == s1.fileno()

# Clean up

for s1, s2 in pairs:

reactor.close_connection(s1.fileno())

finally:

for s1, s2 in pairs:

s1.close()

s2.close()

return True

def test_reactor_buffer():

"""Test ReactorBuffer behaves like bytes."""

import erlang

# Get ReactorBuffer type from erlang module

ReactorBuffer = erlang.ReactorBuffer

# Create test buffer using the _test_create class method

test_data = b'GET / HTTP/1.1\r\n'

buf = ReactorBuffer._test_create(test_data)

# Test len()

assert len(buf) == len(test_data), f"len mismatch: {len(buf)} != {len(test_data)}"

# Test startswith()

assert buf.startswith(b'GET'), "startswith(b'GET') failed"

assert buf.startswith(b'GET /'), "startswith(b'GET /') failed"

assert not buf.startswith(b'POST'), "startswith(b'POST') should be False"

# Test endswith()

assert buf.endswith(b'\r\n'), "endswith(b'\\r\\n') failed"

assert not buf.endswith(b'END'), "endswith(b'END') should be False"

# Test indexing

assert buf[0] == ord('G'), f"buf[0] should be ord('G'), got {buf[0]}"

assert buf[-1] == ord('\n'), f"buf[-1] should be ord('\\n'), got {buf[-1]}"

# Test slicing

assert buf[0:3] == b'GET', f"buf[0:3] should be b'GET', got {buf[0:3]}"

assert buf[-2:] == b'\r\n', f"buf[-2:] should be b'\\r\\n', got {buf[-2:]}"

# Test bytes() conversion

assert bytes(buf) == test_data, f"bytes(buf) mismatch"

# Test memoryview

mv = memoryview(buf)

assert mv[0] == ord('G'), f"memoryview[0] should be ord('G')"

assert bytes(mv) == test_data, "memoryview bytes mismatch"

# Test find()

assert buf.find(b'HTTP') == 6, f"find(b'HTTP') should be 6, got {buf.find(b'HTTP')}"

assert buf.find(b'HTTPS') == -1, "find(b'HTTPS') should be -1"

# Test rfind()

assert buf.rfind(b'/') == 4, f"rfind(b'/') should be 4, got {buf.rfind(b'/')}"

# Test count()

assert buf.count(b'/') == 1, f"count(b'/') should be 1, got {buf.count(b'/')}"

# Test 'in' operator

assert b'HTTP' in buf, "b'HTTP' in buf should be True"

assert b'HTTPS' not in buf, "b'HTTPS' not in buf should be True"

# Test decode()

decoded = buf.decode('utf-8')

assert decoded == 'GET / HTTP/1.1\r\n', f"decode mismatch: {decoded}"

# Test comparison with bytes

assert buf == test_data, "buf == test_data should be True"

assert buf != b'other', "buf != b'other' should be True"

return True

def test_reactor_buffer_with_protocol():

"""Test ReactorBuffer works transparently in protocol data_received."""

import sys

sys.path.insert(0, 'priv')

from _erlang_impl import reactor

import erlang

# Track what was received

received_data = []

received_types = []

class TestProtocol(reactor.Protocol):

def data_received(self, data):

received_data.append(bytes(data)) # Convert to bytes for comparison

received_types.append(type(data).__name__)

# Test that bytes-like operations work

if data.startswith(b'PING'):

self.write_buffer.extend(b'PONG')

return "write_pending"

return "continue"

def write_ready(self):

if not self.write_buffer:

return "read_pending"

written = self.write(bytes(self.write_buffer))

del self.write_buffer[:written]

return "continue" if self.write_buffer else "read_pending"

import socket

s1, s2 = socket.socketpair()

s1.setblocking(False)

s2.setblocking(False)

try:

reactor.set_protocol_factory(TestProtocol)

reactor.init_connection(s1.fileno(), {'type': 'test'})

# Send test data

s2.send(b'PING')

# Call on_read_ready - this will use ReactorBuffer internally

action = reactor.on_read_ready(s1.fileno())

# Verify the protocol received the data

assert len(received_data) == 1, f"Expected 1 received, got {len(received_data)}"

assert received_data[0] == b'PING', f"Expected b'PING', got {received_data[0]}"

assert action == "write_pending", f"Expected 'write_pending', got {action}"

# The type should be ReactorBuffer (not bytes) when using NIF

# But if running without NIF initialization, it might be bytes

# We accept both for compatibility

assert received_types[0] in ('ReactorBuffer', 'bytes'), \

f"Expected ReactorBuffer or bytes, got {received_types[0]}"

reactor.close_connection(s1.fileno())

finally:

s1.close()

s2.close()

return True

def run_all_tests():

"""Run all reactor tests."""

tests = [

test_protocol_creation,

test_echo_protocol,

test_multiple_protocols,

test_reactor_buffer,

test_reactor_buffer_with_protocol,

]

results = []

for test in tests:

try:

result = test()

results.append((test.__name__, 'PASS' if result else 'FAIL'))

except Exception as e:

results.append((test.__name__, f'ERROR: {e}'))

return results

if __name__ == '__main__':

results = run_all_tests()

for name, status in results:

print(f'{name}: {status}')