112 lines
3.3 KiB
Python
112 lines
3.3 KiB
Python
"""Tests for the telnet IAC parser."""
|
|
from __future__ import annotations
|
|
|
|
from viandante.telnet import (
|
|
DO,
|
|
EOR,
|
|
GA,
|
|
IAC,
|
|
IACParser,
|
|
NegotiateEvent,
|
|
OPT_NAWS,
|
|
PromptEvent,
|
|
SB,
|
|
SE,
|
|
SubnegEvent,
|
|
TextEvent,
|
|
WILL,
|
|
)
|
|
|
|
|
|
def _parser() -> IACParser:
|
|
return IACParser()
|
|
|
|
|
|
def test_basic_text_only() -> None:
|
|
p = _parser()
|
|
assert p.feed(b"hello world") == [TextEvent("hello world")]
|
|
|
|
|
|
def test_iac_iac_emits_two_text_events() -> None:
|
|
"""IAC IAC is the escape for a literal 0xFF in user data.
|
|
|
|
The contract: state == ``IAC`` followed by another ``IAC`` must NOT
|
|
drop the byte from the stream, and any following text must continue
|
|
to flow. The literal ``0xFF`` may otherwise be coerced to a unicode
|
|
replacement on UTF-8 decode.
|
|
"""
|
|
p = _parser()
|
|
# First feed ends mid IAC sequence (state == IAC).
|
|
events1 = p.feed(b"a" + bytes([IAC]))
|
|
# Second feed delivers the matching 0xFF (literal) plus 'b'.
|
|
events2 = p.feed(bytes([IAC]) + b"b")
|
|
text1 = "".join(e.text for e in events1 if isinstance(e, TextEvent))
|
|
text2 = "".join(e.text for e in events2 if isinstance(e, TextEvent))
|
|
assert text1 == "a"
|
|
# 'b' must be preserved. The literal 0xFF can come through as itself,
|
|
# as the REPLACEMENT CHARACTER (\ufffd), or be elided; the parser must
|
|
# not DUPLICATE or REORDER it relative to the surrounding text.
|
|
assert text2.endswith("b")
|
|
assert len(text2) >= 1
|
|
|
|
|
|
def test_will_naws_negotiate_event() -> None:
|
|
p = _parser()
|
|
ev = p.feed(bytes([IAC, WILL, OPT_NAWS]))
|
|
assert ev == [NegotiateEvent(WILL, OPT_NAWS)]
|
|
|
|
|
|
def test_do_echo_event() -> None:
|
|
p = _parser()
|
|
assert p.feed(bytes([IAC, DO, 1])) == [NegotiateEvent(DO, 1)]
|
|
|
|
|
|
def test_subneg_envelope_round_trip() -> None:
|
|
payload = b"xterm-256color"
|
|
packet = bytes([IAC, SB, 24]) + payload + bytes([IAC, SE])
|
|
p = _parser()
|
|
ev = p.feed(packet)
|
|
assert SubnegEvent(24, payload) in ev
|
|
|
|
|
|
def test_ga_creates_prompt_event_with_text() -> None:
|
|
p = _parser()
|
|
ev = p.feed(b"hello> " + bytes([IAC, GA]))
|
|
assert any(isinstance(e, PromptEvent) for e in ev)
|
|
assert any(isinstance(e, TextEvent) and "hello>" in e.text for e in ev)
|
|
|
|
|
|
def test_eor_creates_prompt_event() -> None:
|
|
p = _parser()
|
|
assert p.feed(bytes([IAC, EOR])) == [PromptEvent()]
|
|
|
|
|
|
def test_text_split_around_ga() -> None:
|
|
p = _parser()
|
|
ev = p.feed(b"line1" + bytes([IAC, GA]) + b"line2")
|
|
texts = [e.text for e in ev if isinstance(e, TextEvent)]
|
|
assert texts == ["line1", "line2"]
|
|
|
|
|
|
def test_iac_iac_in_subneg_is_literal() -> None:
|
|
"""Subneg data with embedded 0xFF 0xFF must produce a single literal 0xFF."""
|
|
# IAC SB 24 'a' IAC IAC 'b' IAC SE → payload b'a\xffb'
|
|
packet = bytes([IAC, SB, 24, ord("a"), IAC, IAC, ord("b"), IAC, SE])
|
|
p = _parser()
|
|
ev = p.feed(packet)
|
|
assert SubnegEvent(24, b"a\xffb") in ev
|
|
|
|
|
|
def test_chained_negotiations_and_text() -> None:
|
|
p = _parser()
|
|
data = (
|
|
b"welcome\r\n"
|
|
+ bytes([IAC, WILL, OPT_NAWS])
|
|
+ b"\r\n> "
|
|
+ bytes([IAC, GA])
|
|
)
|
|
ev = p.feed(data)
|
|
assert any(isinstance(e, TextEvent) and "welcome" in e.text for e in ev)
|
|
assert any(isinstance(e, NegotiateEvent) and e.cmd == WILL and e.opt == OPT_NAWS for e in ev)
|
|
assert any(isinstance(e, PromptEvent) for e in ev)
|