John Harrison 7790d69cce
[lldb-dap] Refactoring IOStream into Transport handler. (#130026)
Instead of having two discrete InputStream and OutputStream helpers,
this merges the two into a unifed 'Transport' handler.

This handler is responsible for reading the DAP message headers, parsing
the resulting JSON and converting the messages into
`lldb_dap::protocol::Message`s for both input and output.

---------

Co-authored-by: Jonas Devlieghere <jonas@devlieghere.com>
2025-03-12 12:29:05 -07:00

85 lines
2.7 KiB
Python

"""
Test lldb-dap IO handling.
"""
import sys
from lldbsuite.test.decorators import *
import lldbdap_testcase
import dap_server
class TestDAP_io(lldbdap_testcase.DAPTestCaseBase):
def launch(self):
log_file_path = self.getBuildArtifact("dap.txt")
process, _ = dap_server.DebugAdapterServer.launch(
executable=self.lldbDAPExec, log_file=log_file_path
)
def cleanup():
# If the process is still alive, terminate it.
if process.poll() is None:
process.terminate()
process.wait()
stdout_data = process.stdout.read().decode()
stderr_data = process.stderr.read().decode()
print("========= STDOUT =========", file=sys.stderr)
print(stdout_data, file=sys.stderr)
print("========= END =========", file=sys.stderr)
print("========= STDERR =========", file=sys.stderr)
print(stderr_data, file=sys.stderr)
print("========= END =========", file=sys.stderr)
print("========= DEBUG ADAPTER PROTOCOL LOGS =========", file=sys.stderr)
with open(log_file_path, "r") as file:
print(file.read(), file=sys.stderr)
print("========= END =========", file=sys.stderr)
# Execute the cleanup function during test case tear down.
self.addTearDownHook(cleanup)
return process
def test_eof_immediately(self):
"""
lldb-dap handles EOF without any other input.
"""
process = self.launch()
process.stdin.close()
self.assertEqual(process.wait(timeout=5.0), 0)
def test_invalid_header(self):
"""
lldb-dap handles invalid message headers.
"""
process = self.launch()
process.stdin.write(b"not the corret message header")
process.stdin.close()
self.assertEqual(process.wait(timeout=5.0), 1)
def test_partial_header(self):
"""
lldb-dap handles parital message headers.
"""
process = self.launch()
process.stdin.write(b"Content-Length: ")
process.stdin.close()
self.assertEqual(process.wait(timeout=5.0), 1)
def test_incorrect_content_length(self):
"""
lldb-dap handles malformed content length headers.
"""
process = self.launch()
process.stdin.write(b"Content-Length: abc")
process.stdin.close()
self.assertEqual(process.wait(timeout=5.0), 1)
def test_partial_content_length(self):
"""
lldb-dap handles partial messages.
"""
process = self.launch()
process.stdin.write(b"Content-Length: 10\r\n\r\n{")
process.stdin.close()
self.assertEqual(process.wait(timeout=5.0), 1)