pure3270.session Module

Session management for pure3270, handling synchronous and asynchronous 3270 connections.

exception pure3270.session.SessionError(message: str, context: Dict[str, Any] | None = None)

Bases: Exception

Base exception for session-related errors.

When context is provided, include key details in the string representation so tests can assert on them without reaching into attributes.

__init__(message: str, context: Dict[str, Any] | None = None)

exception pure3270.session.ConnectionError(message: str, context: Dict[str, Any] | None = None)

Bases: SessionError

Raised when connection fails.

class pure3270.session.Session(host: str | None = None, port: int = 23, ssl_context: Any | None = None, force_mode: str | None = None, allow_fallback: bool = True, enable_trace: bool = False, terminal_type: str = 'IBM-3278-2')

Bases: object

Synchronous wrapper for AsyncSession.

This class provides a synchronous interface to the asynchronous 3270 session. All methods use asyncio.run() to execute async operations.

__init__(host: str | None = None, port: int = 23, ssl_context: Any | None = None, force_mode: str | None = None, allow_fallback: bool = True, enable_trace: bool = False, terminal_type: str = 'IBM-3278-2') → None

Initialize a synchronous session with a dedicated thread and event loop.

connect(host: str | None = None, port: int | None = None, ssl_context: Any | None = None) → None

Connect to the 3270 host synchronously.

Parameters:

• host – Optional host override.

• port – Optional port override (default 23).

• ssl_context – Optional SSL context override.

Raises:

ConnectionError – If connection fails.

send(data: bytes) → None

Send data to the session.

Parameters:

data – Bytes to send.

Raises:

SessionError – If send fails.

read(timeout: float = 5.0) → bytes

Read data from the session.

Parameters:

timeout – Read timeout in seconds.

Returns:

Received bytes.

Raises:

SessionError – If read fails.

get_aid() → int | None

Get AID synchronously (last known AID value).

close() → None

Close the session synchronously.

property connected: bool

Check if session is connected.

get_trace_events() → List[Any]

open(host: str, port: int = 23) → None

Open connection synchronously (s3270 Open() action).

property screen_buffer: ScreenBuffer

Expose screen buffer for sync Session tests.

property input_inhibited: bool

Get Input Inhibited state (keyboard locked).

property tn3270e_mode: bool

Check if session is in TN3270E mode.

Preference order: 1) Async session’s tn3270 mode flag when True 2) Handler’s negotiated_tn3270e flag as a fallback

close_script() → None

Close script synchronously (s3270 CloseScript() action).

ascii(data: bytes) → str

Convert EBCDIC data to ASCII text (s3270 Ascii() action).

Parameters:

data – EBCDIC bytes to convert.

Returns:

ASCII string representation.

ebcdic(text: str) → bytes

Convert ASCII text to EBCDIC data (s3270 Ebcdic() action).

Parameters:

text – ASCII text to convert.

Returns:

EBCDIC bytes representation.

ascii1(byte_val: int) → str

Convert a single EBCDIC byte to ASCII character.

This is a pure conversion utility and does not require a live session.

ebcdic1(char: str) → int

Convert a single ASCII character to EBCDIC byte.

This is a pure conversion utility and does not require a live session.

ascii_field(field_index: int) → str

Convert field content to ASCII text (s3270 AsciiField() action).

Parameters:

field_index – Index of field to convert.

Returns:

ASCII string representation of field content.

cursor_select() → None

Select field at cursor (s3270 CursorSelect() action).

Raises:

SessionError – If not connected.

delete_field() → None

Delete field at cursor (s3270 DeleteField() action).

Raises:

SessionError – If not connected.

string(text: str) → None

Send string to the session (s3270 String() action).

circum_not() → None

Toggle circumvention of field protection (s3270 CircumNot() action).

Raises:

SessionError – If not connected.

script(commands: str) → None

Execute script (s3270 Script() action).

Parameters:

commands – Script commands.

Raises:

SessionError – If not connected.

execute(command: str) → str

Execute external command synchronously (s3270 Execute() action).

info() → str

Get session information synchronously.

query(query_type: str = 'All') → str

Query session information synchronously.

Supported query types (s3270 protocol parity):

• All: Summary information

• CodePage: Host code page

• ConnectionState: Connection state

• Cursor: Cursor position

• Model: 3270 model information

• ScreenCurSize: Current screen dimensions

• Tn3270eOptions: Active TN3270E functions

• TelnetHostOptions: Host TELNET options

• TelnetMyOptions: Client TELNET options

• And 30+ more query types…

move_cursor1(row: int, col: int) → None

Move cursor to specified position (s3270 MoveCursor1 action).

hex_string(hex_data: str) → None

Send raw hex bytes as input (s3270 HexString() action).

set(option: str, value: str) → None

Set option synchronously.

print_text(text: str) → None

Print text synchronously.

snap() → None

Take snapshot synchronously.

show() → None

Show screen synchronously.

trace(on: bool = True) → None

Enable/disable tracing synchronously.

wait(seconds: float = 1.0) → None

Wait synchronously.

sleep(seconds: float = 1.0) → None

Sleep synchronously.

transfer(file: str) → None

Transfer file synchronously.

source(file: str) → None

Source file synchronously.

expect(pattern: str, timeout: float = 10.0) → bool

Expect pattern synchronously.

fail(message: str) → None

Fail with message synchronously.

compose(text: str) → None

Compose text synchronously.

set_field_attribute(field_index: int, attr: str, value: int) → None

Set a screen field attribute via the underlying async session.

This is a synchronous passthrough used by compatibility tests; the underlying implementation is synchronous as well.

cookie(cookie_string: str) → None

Set cookie synchronously.

interrupt() → None

Send interrupt synchronously (s3270 Interrupt() action).

key(keyname: str) → None

Send key synchronously (s3270 Key() action).

submit(aid: int) → None

Submit with AID synchronously.

home() → None

Move cursor to home position.

up() → None

Move cursor up.

down() → None

Move cursor down.

tab() → None

Move cursor to next tab stop.

backtab() → None

Move cursor to previous tab stop.

backspace() → None

Send backspace key.

enter() → None

Send Enter key.

pf(n: str) → None

Send PF key.

pa(n: str) → None

Send PA key.

erase() → None

Erase entire screen.

clear() → None

Clear entire screen (alias for erase).

newline() → None

Move cursor to start of next line.

erase_eof() → None

Erase to end of field.

erase_input() → None

Erase all input fields.

field_end() → None

Move cursor to end of field.

field_mark() → None

Set field mark.

dup() → None

Duplicate character.

pause(seconds: float = 1.0) → None

Pause session.

bell() → None

Ring bell.

left2() → None

Move cursor left by 2.

right() → None

Move cursor right.

right2() → None

Move cursor right by 2.

reset() → None

Reset session.

field_exit() → None

Exit field.

sysreq() → None

Send SysReq key.

attn() → None

Send Attention key.

test() → None

Send Test key.

left() → None

Move cursor left.

class pure3270.session.AsyncSession(host: str | None = None, port: int = 23, ssl_context: Any | None = None, force_mode: str | None = None, allow_fallback: bool = True, enable_trace: bool = False, terminal_type: str = 'IBM-3278-2', is_printer_session: bool = False)

Bases: object

Asynchronous 3270 session handler.

This class provides an asynchronous interface to the 3270 session. All operations are async and use asyncio for non-blocking I/O.

AID_MAP: Dict[str, int] = {'Attn': 241, 'BackSpace': 248, 'CLEAR': 109, 'Dup': 245, 'Enter': 125, 'PA(1)': 108, 'PA(2)': 110, 'PA(3)': 107, 'PF(1)': 241, 'PF(10)': 122, 'PF(11)': 123, 'PF(12)': 124, 'PF(13)': 193, 'PF(14)': 194, 'PF(15)': 195, 'PF(16)': 196, 'PF(17)': 197, 'PF(18)': 198, 'PF(19)': 199, 'PF(2)': 242, 'PF(20)': 200, 'PF(21)': 201, 'PF(22)': 74, 'PF(23)': 75, 'PF(24)': 76, 'PF(3)': 243, 'PF(4)': 244, 'PF(5)': 245, 'PF(6)': 246, 'PF(7)': 247, 'PF(8)': 248, 'PF(9)': 249, 'RESET': 106, 'SysReq': 240, 'TEST': 17, 'Test': 17}

__init__(host: str | None = None, port: int = 23, ssl_context: Any | None = None, force_mode: str | None = None, allow_fallback: bool = True, enable_trace: bool = False, terminal_type: str = 'IBM-3278-2', is_printer_session: bool = False) → None

Initialize the AsyncSession.

Parameters:

• host – Host to connect to.

• port – Port to connect to.

• ssl_context – SSL context for secure connections.

• force_mode – Force specific TN3270 mode.

• allow_fallback – Allow fallback to TN3270 if TN3270E fails.

• enable_trace – Enable tracing.

• terminal_type – Terminal model type.

• is_printer_session – True if this is a printer session.

resources: Dict[str, str]

property host: str | None

Public host accessor for tests.

property port: int

Public port accessor for tests.

property model: str

property screen_buffer: ScreenBuffer

Get the screen buffer.

property screen: ScreenBuffer

Get the screen buffer (alias for screen_buffer).

property input_inhibited: bool

Get Input Inhibited state (keyboard locked).

property handler: TN3270Handler | None

Get the underlying TN3270 handler (may be None if not connected).

property tn3270e_mode: bool

Check if session is in TN3270E mode.

property connected: bool

Check if session is connected.

async connect(host: str | None = None, port: int | None = None, ssl_context: Any | None = None) → None

Connect to the 3270 host asynchronously.

async send(data: bytes) → None

Send data to the session with retry logic.

async send_data(data: bytes) → None

Send data to the session (alias for send).

async read(timeout: float = 5.0) → bytes

Read data from the session with retry logic.

async receive_data(timeout: float = 5.0) → bytes

Receive data from the session (alias for read).

get_aid() → int | None

Get the last AID value.

async close() → None

Close the session.

async close_script() → None

Close script (s3270 CloseScript() action).

get_trace_events() → List[Any]

Get trace events.

managed() → AsyncIterator[AsyncSession]

Provide a managed async context that guarantees close() on exit.

ascii(data: bytes) → str

Convert EBCDIC to ASCII.

ebcdic(text: str) → bytes

Convert ASCII to EBCDIC.

ascii1(byte_val: int) → str

Convert single EBCDIC byte to ASCII.

ebcdic1(char: str) → int

Convert single ASCII char to EBCDIC.

ascii_field(field_index: int) → str

Convert field to ASCII.

async cursor_select() → None

Select field at cursor.

async delete_field() → None

Delete field at cursor.

async insert_text(text: str) → None

Insert text at cursor.

async string(text: str) → None

Send string to the session.

async circum_not() → None

Toggle circumvention.

async script(commands: str) → None

Execute script command(s).

Supports both simple method names and function-style commands like: - Simple: “connect”, “disconnect” - With args: “String(test)”, “MoveCursor(5,10)”

async execute(command: str) → str

Execute external command.

async interrupt() → None

Send interrupt.

async key(keyname: str) → None

Send key synchronously (s3270 Key() action).

capabilities() → str

Get capabilities.

async set_option(option: str, value: str) → None

Set option (placeholder for future functionality).

async query(query_type: str = 'All') → str

Query screen.

async set(option: str, value: str) → None

Set option (alias).

async exit() → None

Exit session.

async bell() → None

Ring bell.

async pause(seconds: float = 1.0) → None

Pause.

async ansi_text(data: bytes) → str

Convert EBCDIC to ANSI.

async hex_string(hex_str: str) → bytes

Convert hex to bytes.

async show() → None

Show screen.

async snap() → None

Save snapshot.

async newline() → None

Move cursor to start of next line.

async page_down() → None

Page down: wrap to top row for tests.

async page_up() → None

Page up: move to top row for tests.

async paste_string(text: str) → None

async end() → None

Move cursor to end of current line.

async move_cursor(row: int, col: int) → None

async move_cursor1(row1: int, col1: int) → None

async next_word() → None

async previous_word() → None

async toggle_insert() → None

async flip() → None

async insert() → None

async delete() → None

Delete character at cursor by shifting remainder left and clearing last.

async disconnect() → None

async left() → None

Move cursor left.

async right() → None

Move cursor right.

async left2() → None

Move cursor left by 2.

async right2() → None

Move cursor right by 2.

async mono_case() → None

Toggle monocase.

async nvt_text(text: str) → None

Send NVT text.

async print_text(text: str) → None

Print text.

async prompt(message: str) → str

Prompt for input.

async read_buffer() → bytes

Read buffer.

async reconnect() → None

Reconnect.

async info() → str

Get session information.

async quit() → None

Quit the session.

async home() → None

Move cursor to home position.

async up() → None

Move cursor up.

async down() → None

Move cursor down.

async tab() → None

Send Tab key.

async backtab() → None

Move cursor to previous tab stop.

async backspace() → None

Send backspace key.

async enter() → None

Send Enter key.

async clear() → None

Clear screen locally.

async pf(n: str) → None

Send PF key.

async pa(n: str) → None

Send PA key.

async macro(commands: List[str]) → None

Execute a sequence of commands.

async erase() → None

Erase entire screen (local).

async erase_eof() → None

Erase to end of field.

async erase_input() → None

Erase all input fields.

async field_end() → None

Move cursor to end of field.

async field_mark() → None

Set field mark.

async dup() → None

Duplicate character.

async field_exit() → None

Exit field.

async sysreq() → None

Send SysReq key.

async attn() → None

Send Attention key.

async test() → None

Send Test key.

async screen_trace() → None

Screen trace.

async source(file: str) → None

Source script.

async subject_names() → None

Subject names.

async sys_req(command: str | None = None) → None

Send SysReq.

async send_break() → None

Send Telnet BREAK.

async send_soh_message(status_code: int) → None

Send SOH message.

async toggle_option(option: str) → None

Toggle option.

async trace(on: bool) → None

Enable/disable tracing.

async transfer(file: str) → None

Transfer file.

async send_file(local_path: str, remote_name: str) → None

Send a file to the host using IND$FILE protocol.

async receive_file(remote_name: str, local_path: str) → None

Receive a file from the host using IND$FILE protocol.

async wait_condition(condition: str) → None

Wait for condition.

async compose(text: str) → None

Compose text.

async cookie(cookie_string: str) → None

Set a cookie.

This sets the cookie immediately and returns None. Tests can call this without awaiting since the side-effect happens immediately.

async expect(pattern: str, timeout: float = 10.0) → bool

Wait for pattern.

async fail(message: str) → None

Fail with message.

async select_light_pen(row: int, col: int) → None

Select light pen.

async start_lu_lu_session(lu_name: str) → None

Start LU-LU session.

property lu_lu_session: Any

Get the current LU-LU session if active.

async load_resource_definitions(file_path: str) → None

Load resources.

async apply_resources() → None

Apply resources.

set_field_attribute(field_index: int, attr: str, value: int) → None

Set field attribute.

async submit(aid: int) → None

Submit with AID.