Add Telnet regression test suite

2026-09-17 13:33:45 +02:00
parent 5e6d9138c4
commit da967f4d56
6 changed files with 258 additions and 11 deletions
@@ -0,0 +1,10 @@
+require_relative "test_helper"
+require_relative "telnet_test"
+require_relative "text_pipeline_test"
+
+begin
+  MeowTest.run!
+rescue MeowTest::Failure
+  exit(1) if __FILE__ == $PROGRAM_NAME
+  raise
+end
@@ -0,0 +1,97 @@
+# encoding: UTF-8
+
+require_relative "test_helper"
+require_relative "../lib/meow/telnet"
+
+module MeowTelnetTest
+  Telnet = MeowClient::Telnet
+
+  def self.bytes(*values)
+    values.pack("C*")
+  end
+
+  def self.collect(parser, chunks)
+    chunks.each_with_object({:data => +"".b, :replies => [], :events => []}) do |chunk, result|
+      parsed = parser.feed(chunk)
+      result[:data] << parsed.data
+      result[:replies].concat(parsed.replies)
+      result[:events].concat(parsed.events)
+    end
+  end
+
+  MeowTest.test("Telnet preserves data and unescapes IAC") do
+    parser = Telnet::Parser.new
+    result = parser.feed("look".b + bytes(Telnet::IAC, Telnet::IAC) + "!".b)
+    MeowTest.assert_equal("look".b + bytes(Telnet::IAC) + "!".b, result.data)
+    MeowTest.assert_empty(result.replies)
+    MeowTest.assert_empty(result.events)
+  end
+
+  MeowTest.test("Telnet negotiation survives every input boundary") do
+    packet = bytes(Telnet::IAC, Telnet::WILL, Telnet::ECHO)
+    whole = collect(Telnet::Parser.new, [packet])
+    split = collect(Telnet::Parser.new, packet.bytes.map { |byte| bytes(byte) })
+    MeowTest.assert_equal(whole, split)
+    MeowTest.assert_equal([bytes(Telnet::IAC, Telnet::DO, Telnet::ECHO)], split[:replies])
+    MeowTest.assert_equal([[:remote_echo, true]], split[:events])
+  end
+
+  MeowTest.test("Telnet tracks echo transitions without duplicate replies") do
+    parser = Telnet::Parser.new
+    enabled = parser.feed(bytes(Telnet::IAC, Telnet::WILL, Telnet::ECHO))
+    duplicate = parser.feed(bytes(Telnet::IAC, Telnet::WILL, Telnet::ECHO))
+    disabled = parser.feed(bytes(Telnet::IAC, Telnet::WONT, Telnet::ECHO))
+    disabled_duplicate = parser.feed(bytes(Telnet::IAC, Telnet::WONT, Telnet::ECHO))
+
+    MeowTest.assert_equal([bytes(Telnet::IAC, Telnet::DO, Telnet::ECHO)], enabled.replies)
+    MeowTest.assert_empty(duplicate.replies)
+    MeowTest.assert_equal([bytes(Telnet::IAC, Telnet::DONT, Telnet::ECHO)], disabled.replies)
+    MeowTest.assert_equal([[:remote_echo, false]], disabled.events)
+    MeowTest.assert_empty(disabled_duplicate.replies)
+    MeowTest.refute(parser.remote_echo)
+  end
+
+  MeowTest.test("Telnet accepts supported options and rejects unknown options") do
+    parser = Telnet::Parser.new
+    remote = parser.feed(bytes(Telnet::IAC, Telnet::WILL, Telnet::SUPPRESS_GO_AHEAD))
+    remote_unknown = parser.feed(bytes(Telnet::IAC, Telnet::WILL, 42))
+    local = parser.feed(bytes(Telnet::IAC, Telnet::DO, Telnet::TERMINAL_TYPE))
+    local_unknown = parser.feed(bytes(Telnet::IAC, Telnet::DO, 43))
+
+    MeowTest.assert_equal([bytes(Telnet::IAC, Telnet::DO, Telnet::SUPPRESS_GO_AHEAD)], remote.replies)
+    MeowTest.assert_equal([bytes(Telnet::IAC, Telnet::DONT, 42)], remote_unknown.replies)
+    MeowTest.assert_equal([bytes(Telnet::IAC, Telnet::WILL, Telnet::TERMINAL_TYPE)], local.replies)
+    MeowTest.assert_equal([bytes(Telnet::IAC, Telnet::WONT, 43)], local_unknown.replies)
+  end
+
+  MeowTest.test("Telnet sends terminal type only after accepting the option") do
+    request = bytes(Telnet::IAC, Telnet::SB, Telnet::TERMINAL_TYPE, Telnet::TTYPE_SEND, Telnet::IAC, Telnet::SE)
+    parser = Telnet::Parser.new
+    before = parser.feed(request)
+    parser.feed(bytes(Telnet::IAC, Telnet::DO, Telnet::TERMINAL_TYPE))
+    after = collect(parser, request.bytes.map { |byte| bytes(byte) })
+    expected = bytes(Telnet::IAC, Telnet::SB, Telnet::TERMINAL_TYPE, Telnet::TTYPE_IS) + "MEOW".b + bytes(Telnet::IAC, Telnet::SE)
+
+    MeowTest.assert_empty(before.replies)
+    MeowTest.assert_equal([[:subnegotiation, Telnet::TERMINAL_TYPE]], before.events)
+    MeowTest.assert_equal([expected], after[:replies])
+    MeowTest.assert_equal([[:terminal_type, "MEOW"]], after[:events])
+  end
+
+  MeowTest.test("Telnet handles escaped IAC inside subnegotiation") do
+    parser = Telnet::Parser.new
+    packet = bytes(Telnet::IAC, Telnet::SB, 77, 1, Telnet::IAC, Telnet::IAC, 2, Telnet::IAC, Telnet::SE)
+    result = collect(parser, packet.bytes.map { |byte| bytes(byte) })
+    MeowTest.assert_empty(result[:data])
+    MeowTest.assert_empty(result[:replies])
+    MeowTest.assert_equal([[:subnegotiation, 77]], result[:events])
+  end
+
+  MeowTest.test("Telnet ignores unknown commands and resumes ordinary data") do
+    parser = Telnet::Parser.new
+    result = parser.feed("a".b + bytes(Telnet::IAC, 241) + "b".b)
+    MeowTest.assert_equal("ab".b, result.data)
+    MeowTest.assert_empty(result.replies)
+    MeowTest.assert_empty(result.events)
+  end
+end
@@ -0,0 +1,54 @@
+module MeowTest
+  Failure = Class.new(StandardError)
+
+  @tests = []
+
+  class << self
+    attr_reader :tests, :last_result
+
+    def test(name, &block)
+      @tests << [name, block]
+    end
+
+    def assert(value, message = "Expected value to be truthy")
+      raise Failure, message unless value
+    end
+
+    def refute(value, message = "Expected value to be falsey")
+      raise Failure, message if value
+    end
+
+    def assert_equal(expected, actual, message = nil)
+      return if expected == actual
+
+      detail = "Expected #{expected.inspect}, got #{actual.inspect}"
+      raise Failure, [message, detail].compact.join(": ")
+    end
+
+    def assert_empty(value, message = nil)
+      assert(value.empty?, message || "Expected #{value.inspect} to be empty")
+    end
+
+    def run!
+      failures = []
+      @tests.each do |name, test|
+        test.call
+        puts "PASS #{name}"
+      rescue StandardError => error
+        failures << [name, error]
+        puts "FAIL #{name}: #{error.class}: #{error.message}"
+      end
+
+      @last_result = {
+        :tests => @tests.length,
+        :passed => @tests.length - failures.length,
+        :failures => failures.length,
+        :failure_details => failures.map { |name, error| "#{name}: #{error.class}: #{error.message}" }
+      }
+      puts "#{@last_result[:passed]}/#{@last_result[:tests]} tests passed"
+      raise Failure, @last_result[:failure_details].join("\n") unless failures.empty?
+
+      @last_result
+    end
+  end
+end
@@ -0,0 +1,77 @@
+# encoding: UTF-8
+
+require_relative "test_helper"
+require_relative "../lib/meow/telnet"
+require_relative "../lib/meow/text_pipeline"
+
+module MeowTextPipelineTest
+  Telnet = MeowClient::Telnet
+
+  def self.bytes(*values)
+    values.pack("C*")
+  end
+
+  MeowTest.test("ANSI CSI and OSC sequences are filtered across chunks") do
+    pipeline = MeowClient::TextPipeline.new("UTF-8")
+    records = []
+    records.concat(pipeline.feed("red\e[3".b).records)
+    records.concat(pipeline.feed("1m text\e]0;title\e".b).records)
+    records.concat(pipeline.feed("\\tail\r\n".b).records)
+    MeowTest.assert_equal(["red texttail"], records)
+  end
+
+  MeowTest.test("Line framing handles CRLF CR-NUL bare CR and LF across chunks") do
+    pipeline = MeowClient::TextPipeline.new("UTF-8")
+    MeowTest.assert_equal(["one"], pipeline.feed("one\r".b).records)
+    MeowTest.assert_equal(["two"], pipeline.feed("\ntwo\r".b).records)
+    MeowTest.assert_equal(["three"], pipeline.feed("\0three\n".b).records)
+    MeowTest.assert_equal(["four"], pipeline.feed("four\rfive".b).records)
+    MeowTest.assert_equal(["five"], pipeline.finish)
+  end
+
+  MeowTest.test("UTF-8 characters survive arbitrary byte boundaries") do
+    encoded = "Price: €\n".encode("UTF-8").b
+    pipeline = MeowClient::TextPipeline.new("UTF-8")
+    records = []
+    encoded.each_byte { |byte| records.concat(pipeline.feed(bytes(byte)).records) }
+    MeowTest.assert_equal(["Price: €"], records)
+  end
+
+  MeowTest.test("Invalid and incomplete UTF-8 are replaced safely") do
+    pipeline = MeowClient::TextPipeline.new("UTF-8")
+    invalid = pipeline.feed("bad\xC0\n".b)
+    MeowTest.assert_equal(["bad\uFFFD"], invalid.records)
+
+    incomplete = MeowClient::TextPipeline.new("UTF-8")
+    incomplete.feed("tail \xE2\x82".b)
+    MeowTest.assert_equal(["tail \uFFFD"], incomplete.finish)
+  end
+
+  MeowTest.test("Legacy encodings are converted to UTF-8") do
+    windows = MeowClient::TextPipeline.new("Windows-1252")
+    latin = MeowClient::TextPipeline.new("ISO-8859-1")
+    MeowTest.assert_equal(["€"], windows.feed("\x80\n".b).records)
+    MeowTest.assert_equal(["é"], latin.feed("\xE9\n".b).records)
+  end
+
+  MeowTest.test("Prompts can be detected flushed and finished") do
+    pipeline = MeowClient::TextPipeline.new("UTF-8")
+    MeowTest.assert_empty(pipeline.feed("Password: ".b).records)
+    MeowTest.assert(pipeline.prompt_due?(0))
+    MeowTest.assert_equal("Password: ", pipeline.flush_prompt)
+    MeowTest.refute(pipeline.prompt_due?(0))
+
+    pipeline.feed("unfinished".b)
+    MeowTest.assert_equal(["unfinished"], pipeline.finish)
+  end
+
+  MeowTest.test("Text pipeline propagates Telnet replies and echo state") do
+    pipeline = MeowClient::TextPipeline.new("UTF-8")
+    packet = "hello".b + bytes(Telnet::IAC, Telnet::WILL, Telnet::ECHO) + " world\r\n".b
+    result = pipeline.feed(packet)
+    MeowTest.assert_equal(["hello world"], result.records)
+    MeowTest.assert_equal([bytes(Telnet::IAC, Telnet::DO, Telnet::ECHO)], result.replies)
+    MeowTest.assert_equal([[:remote_echo, true]], result.events)
+    MeowTest.assert(pipeline.remote_echo?)
+  end
+end