allow arbitrary python streams to be converted to SBFile

Summary:
This patch adds SWIG typemaps that can convert arbitrary python
file objects into lldb_private::File.

A SBFile may be initialized from a python file using the
constructor.   There are also alternate, tagged constructors
that allow python files to be borrowed, and for the caller
to control whether or not the python I/O methods will be
called even when a file descriptor is available.I

Reviewers: JDevlieghere, jasonmolenda, labath

Reviewed By: labath

Subscribers: zturner, amccarth, lldb-commits

Tags: #lldb

Differential Revision: https://reviews.llvm.org/D68188

llvm-svn: 374225

1 files changed, 301 insertions, 17 deletions

diff --git a/lldb/packages/Python/lldbsuite/test/python_api/file_handle/TestFileHandle.py b/lldb/packages/Python/lldbsuite/test/python_api/file_handle/TestFileHandle.py
index 3231fdfe688..c6bfc35f0ac 100644
--- a/lldb/packages/Python/lldbsuite/test/python_api/file_handle/TestFileHandle.py
+++ b/lldb/packages/Python/lldbsuite/test/python_api/file_handle/TestFileHandle.py
@@ -13,8 +13,53 @@ from contextlib import contextmanager
 import lldb
 from lldbsuite.test import  lldbtest
 from lldbsuite.test.decorators import (
-    add_test_categories, no_debug_info_test, skipIf)
-
+    add_test_categories, skipIf, skipIfWindows)
+
+class OhNoe(Exception):
+    pass
+
+class BadIO(io.TextIOBase):
+    @property
+    def closed(self):
+        return False
+    def writable(self):
+        return True
+    def readable(self):
+        return True
+    def write(self, s):
+        raise OhNoe('OH NOE')
+    def read(self, n):
+        raise OhNoe("OH NOE")
+    def flush(self):
+        raise OhNoe('OH NOE')
+
+# This class will raise an exception while it's being
+# converted into a C++ object by swig
+class ReallyBadIO(io.TextIOBase):
+    def fileno(self):
+        return 999
+    def writable(self):
+        raise OhNoe("OH NOE!!!")
+
+class MutableBool():
+    def __init__(self, value):
+        self.value = value
+    def set(self, value):
+        self.value = bool(value)
+    def __bool__(self):
+        return self.value
+
+class FlushTestIO(io.StringIO):
+    def __init__(self, mutable_flushed, mutable_closed):
+        super(FlushTestIO, self).__init__()
+        self.mut_flushed = mutable_flushed
+        self.mut_closed = mutable_closed
+    def close(self):
+        self.mut_closed.set(True)
+        return super(FlushTestIO, self).close()
+    def flush(self):
+        self.mut_flushed.set(True)
+        return super(FlushTestIO, self).flush()
 
 @contextmanager
 def replace_stdout(new):
@@ -36,6 +81,7 @@ def readStrippedLines(f):
 
 class FileHandleTestCase(lldbtest.TestBase):
 
+    NO_DEBUG_INFO_TESTCASE = True
     mydir = lldbtest.Base.compute_mydir(__file__)
 
     # The way this class interacts with the debugger is different
@@ -84,7 +130,8 @@ class FileHandleTestCase(lldbtest.TestBase):
 
 
     @add_test_categories(['pyapi'])
-    @no_debug_info_test
+    @skipIfWindows # FIXME pre-existing bug, should be fixed
+                   # when we delete the FILE* typemaps.
     def test_legacy_file_out_script(self):
         with open(self.out_filename, 'w') as f:
             self.debugger.SetOutputFileHandle(f, False)
@@ -100,7 +147,6 @@ class FileHandleTestCase(lldbtest.TestBase):
 
 
     @add_test_categories(['pyapi'])
-    @no_debug_info_test
     def test_legacy_file_out(self):
         with open(self.out_filename, 'w') as f:
             self.debugger.SetOutputFileHandle(f, False)
@@ -110,7 +156,8 @@ class FileHandleTestCase(lldbtest.TestBase):
             self.assertIn('deadbeef', f.read())
 
     @add_test_categories(['pyapi'])
-    @no_debug_info_test
+    @skipIfWindows # FIXME pre-existing bug, should be fixed
+                   # when we delete the FILE* typemaps.
     def test_legacy_file_err_with_get(self):
         with open(self.out_filename, 'w') as f:
             self.debugger.SetErrorFileHandle(f, False)
@@ -124,7 +171,6 @@ class FileHandleTestCase(lldbtest.TestBase):
 
 
     @add_test_categories(['pyapi'])
-    @no_debug_info_test
     def test_legacy_file_err(self):
         with open(self.out_filename, 'w') as f:
             self.debugger.SetErrorFileHandle(f, False)
@@ -135,7 +181,6 @@ class FileHandleTestCase(lldbtest.TestBase):
 
 
     @add_test_categories(['pyapi'])
-    @no_debug_info_test
     def test_sbfile_type_errors(self):
         sbf = lldb.SBFile()
         self.assertRaises(TypeError, sbf.Write, None)
@@ -146,8 +191,7 @@ class FileHandleTestCase(lldbtest.TestBase):
 
 
     @add_test_categories(['pyapi'])
-    @no_debug_info_test
-    def test_sbfile_write(self):
+    def test_sbfile_write_fileno(self):
         with open(self.out_filename, 'w') as f:
             sbf = lldb.SBFile(f.fileno(), "w", False)
             self.assertTrue(sbf.IsValid())
@@ -161,8 +205,20 @@ class FileHandleTestCase(lldbtest.TestBase):
 
 
     @add_test_categories(['pyapi'])
-    @no_debug_info_test
-    def test_sbfile_read(self):
+    def test_sbfile_write(self):
+        with open(self.out_filename, 'w') as f:
+            sbf = lldb.SBFile(f)
+            e, n = sbf.Write(b'FOO\n')
+            self.assertTrue(e.Success())
+            self.assertEqual(n, 4)
+            sbf.Close()
+            self.assertTrue(f.closed)
+        with open(self.out_filename, 'r') as f:
+            self.assertEqual(f.read().strip(), 'FOO')
+
+
+    @add_test_categories(['pyapi'])
+    def test_sbfile_read_fileno(self):
         with open(self.out_filename, 'w') as f:
             f.write('FOO')
         with open(self.out_filename, 'r') as f:
@@ -175,7 +231,21 @@ class FileHandleTestCase(lldbtest.TestBase):
 
 
     @add_test_categories(['pyapi'])
-    @no_debug_info_test
+    def test_sbfile_read(self):
+        with open(self.out_filename, 'w') as f:
+            f.write('foo')
+        with open(self.out_filename, 'r') as f:
+            sbf = lldb.SBFile(f)
+            buf = bytearray(100)
+            e, n = sbf.Read(buf)
+            self.assertTrue(e.Success())
+            self.assertEqual(n, 3)
+            self.assertEqual(buf[:n], b'foo')
+            sbf.Close()
+            self.assertTrue(f.closed)
+
+
+    @add_test_categories(['pyapi'])
     def test_fileno_out(self):
         with open(self.out_filename, 'w') as f:
             sbf = lldb.SBFile(f.fileno(), "w", False)
@@ -189,7 +259,6 @@ class FileHandleTestCase(lldbtest.TestBase):
 
 
     @add_test_categories(['pyapi'])
-    @no_debug_info_test
     def test_fileno_help(self):
         with open(self.out_filename, 'w') as f:
             sbf = lldb.SBFile(f.fileno(), "w", False)
@@ -201,7 +270,6 @@ class FileHandleTestCase(lldbtest.TestBase):
 
 
     @add_test_categories(['pyapi'])
-    @no_debug_info_test
     def test_immediate(self):
         with open(self.out_filename, 'w') as f:
             ret = lldb.SBCommandReturnObject()
@@ -220,7 +288,6 @@ class FileHandleTestCase(lldbtest.TestBase):
 
 
     @add_test_categories(['pyapi'])
-    @no_debug_info_test
     def test_fileno_inout(self):
         with open(self.in_filename, 'w') as f:
             f.write("help help\n")
@@ -244,7 +311,6 @@ class FileHandleTestCase(lldbtest.TestBase):
 
 
     @add_test_categories(['pyapi'])
-    @no_debug_info_test
     def test_fileno_error(self):
         with open(self.out_filename, 'w') as f:
 
@@ -263,7 +329,6 @@ class FileHandleTestCase(lldbtest.TestBase):
 
     #FIXME This shouldn't fail for python2 either.
     @add_test_categories(['pyapi'])
-    @no_debug_info_test
     @skipIf(py_version=['<', (3,)])
     def test_replace_stdout(self):
         f = io.StringIO()
@@ -272,3 +337,222 @@ class FileHandleTestCase(lldbtest.TestBase):
             self.handleCmd('script sys.stdout.write("lol")',
                 collect_result=False, check=False)
             self.assertEqual(sys.stdout, f)
+
+
+    @add_test_categories(['pyapi'])
+    def test_sbfile_write_borrowed(self):
+        with open(self.out_filename, 'w') as f:
+            sbf = lldb.SBFile.Create(f, borrow=True)
+            e, n = sbf.Write(b'FOO')
+            self.assertTrue(e.Success())
+            self.assertEqual(n, 3)
+            sbf.Close()
+            self.assertFalse(f.closed)
+            f.write('BAR\n')
+        with open(self.out_filename, 'r') as f:
+            self.assertEqual(f.read().strip(), 'FOOBAR')
+
+
+
+    @add_test_categories(['pyapi'])
+    @skipIf(py_version=['<', (3,)])
+    def test_sbfile_write_forced(self):
+        with open(self.out_filename, 'w') as f:
+            written = MutableBool(False)
+            orig_write = f.write
+            def mywrite(x):
+                written.set(True)
+                return orig_write(x)
+            f.write = mywrite
+            sbf = lldb.SBFile.Create(f, force_io_methods=True)
+            e, n = sbf.Write(b'FOO')
+            self.assertTrue(written)
+            self.assertTrue(e.Success())
+            self.assertEqual(n, 3)
+            sbf.Close()
+            self.assertTrue(f.closed)
+        with open(self.out_filename, 'r') as f:
+            self.assertEqual(f.read().strip(), 'FOO')
+
+
+    @add_test_categories(['pyapi'])
+    @skipIf(py_version=['<', (3,)])
+    def test_sbfile_write_forced_borrowed(self):
+        with open(self.out_filename, 'w') as f:
+            written = MutableBool(False)
+            orig_write = f.write
+            def mywrite(x):
+                written.set(True)
+                return orig_write(x)
+            f.write = mywrite
+            sbf = lldb.SBFile.Create(f, borrow=True, force_io_methods=True)
+            e, n = sbf.Write(b'FOO')
+            self.assertTrue(written)
+            self.assertTrue(e.Success())
+            self.assertEqual(n, 3)
+            sbf.Close()
+            self.assertFalse(f.closed)
+        with open(self.out_filename, 'r') as f:
+            self.assertEqual(f.read().strip(), 'FOO')
+
+
+    @add_test_categories(['pyapi'])
+    @skipIf(py_version=['<', (3,)])
+    def test_sbfile_write_string(self):
+        f = io.StringIO()
+        sbf = lldb.SBFile(f)
+        e, n = sbf.Write(b'FOO')
+        self.assertEqual(f.getvalue().strip(), "FOO")
+        self.assertTrue(e.Success())
+        self.assertEqual(n, 3)
+        sbf.Close()
+        self.assertTrue(f.closed)
+
+    @add_test_categories(['pyapi'])
+    @skipIf(py_version=['<', (3,)])
+    def test_sbfile_write_bytes(self):
+        f = io.BytesIO()
+        sbf = lldb.SBFile(f)
+        e, n = sbf.Write(b'FOO')
+        self.assertEqual(f.getvalue().strip(), b"FOO")
+        self.assertTrue(e.Success())
+        self.assertEqual(n, 3)
+        sbf.Close()
+        self.assertTrue(f.closed)
+
+    @add_test_categories(['pyapi'])
+    @skipIf(py_version=['<', (3,)])
+    def test_sbfile_read_string(self):
+        f = io.StringIO('zork')
+        sbf = lldb.SBFile(f)
+        buf = bytearray(100)
+        e, n = sbf.Read(buf)
+        self.assertTrue(e.Success())
+        self.assertEqual(buf[:n], b'zork')
+
+
+    @add_test_categories(['pyapi'])
+    @skipIf(py_version=['<', (3,)])
+    def test_sbfile_read_string_one_byte(self):
+        f = io.StringIO('z')
+        sbf = lldb.SBFile(f)
+        buf = bytearray(1)
+        e, n = sbf.Read(buf)
+        self.assertTrue(e.Fail())
+        self.assertEqual(n, 0)
+        self.assertEqual(e.GetCString(), "can't read less than 6 bytes from a utf8 text stream")
+
+
+    @add_test_categories(['pyapi'])
+    @skipIf(py_version=['<', (3,)])
+    def test_sbfile_read_bytes(self):
+        f = io.BytesIO(b'zork')
+        sbf = lldb.SBFile(f)
+        buf = bytearray(100)
+        e, n = sbf.Read(buf)
+        self.assertTrue(e.Success())
+        self.assertEqual(buf[:n], b'zork')
+
+
+    @add_test_categories(['pyapi'])
+    @skipIf(py_version=['<', (3,)])
+    def test_sbfile_out(self):
+        with open(self.out_filename, 'w') as f:
+            sbf = lldb.SBFile(f)
+            status = self.debugger.SetOutputFile(sbf)
+            self.assertTrue(status.Success())
+            self.handleCmd('script 2+2')
+        with open(self.out_filename, 'r') as f:
+            self.assertEqual(f.read().strip(), '4')
+
+
+    @add_test_categories(['pyapi'])
+    def test_sbfile_error(self):
+        with open(self.out_filename, 'w') as f:
+            sbf = lldb.SBFile(f)
+            status = self.debugger.SetErrorFile(sbf)
+            self.assertTrue(status.Success())
+            self.handleCmd('lolwut', check=False, collect_result=False)
+        with open(self.out_filename, 'r') as f:
+            errors = f.read()
+            self.assertTrue(re.search(r'error:.*lolwut', errors))
+
+
+    @add_test_categories(['pyapi'])
+    def test_exceptions(self):
+        self.assertRaises(TypeError, lldb.SBFile, None)
+        self.assertRaises(TypeError, lldb.SBFile, "ham sandwich")
+        if sys.version_info[0] < 3:
+            self.assertRaises(TypeError, lldb.SBFile, ReallyBadIO())
+        else:
+            self.assertRaises(OhNoe, lldb.SBFile, ReallyBadIO())
+            error, n = lldb.SBFile(BadIO()).Write(b"FOO")
+            self.assertEqual(n, 0)
+            self.assertTrue(error.Fail())
+            self.assertEqual(error.GetCString(), "OhNoe('OH NOE')")
+            error, n = lldb.SBFile(BadIO()).Read(bytearray(100))
+            self.assertEqual(n, 0)
+            self.assertTrue(error.Fail())
+            self.assertEqual(error.GetCString(), "OhNoe('OH NOE')")
+
+
+    @add_test_categories(['pyapi'])
+    @skipIf(py_version=['<', (3,)])
+    def test_exceptions_logged(self):
+        messages = list()
+        self.debugger.SetLoggingCallback(messages.append)
+        self.handleCmd('log enable lldb script')
+        self.debugger.SetOutputFile(lldb.SBFile(BadIO()))
+        self.handleCmd('script 1+1')
+        self.assertTrue(any('OH NOE' in msg for msg in messages))
+
+
+    @add_test_categories(['pyapi'])
+    @skipIf(py_version=['<', (3,)])
+    def test_flush(self):
+        flushed = MutableBool(False)
+        closed = MutableBool(False)
+        f = FlushTestIO(flushed, closed)
+        self.assertFalse(flushed)
+        self.assertFalse(closed)
+        sbf = lldb.SBFile(f)
+        self.assertFalse(flushed)
+        self.assertFalse(closed)
+        sbf = None
+        self.assertFalse(flushed)
+        self.assertTrue(closed)
+        self.assertTrue(f.closed)
+
+        flushed = MutableBool(False)
+        closed = MutableBool(False)
+        f = FlushTestIO(flushed, closed)
+        self.assertFalse(flushed)
+        self.assertFalse(closed)
+        sbf = lldb.SBFile.Create(f, borrow=True)
+        self.assertFalse(flushed)
+        self.assertFalse(closed)
+        sbf = None
+        self.assertTrue(flushed)
+        self.assertFalse(closed)
+        self.assertFalse(f.closed)
+
+
+    @add_test_categories(['pyapi'])
+    def test_fileno_flush(self):
+        with open(self.out_filename, 'w') as f:
+            f.write("foo")
+            sbf = lldb.SBFile(f)
+            sbf.Write(b'bar')
+            sbf = None
+            self.assertTrue(f.closed)
+        with open(self.out_filename, 'r') as f:
+            self.assertEqual(f.read(), 'foobar')
+
+        with open(self.out_filename, 'w+') as f:
+            f.write("foo")
+            sbf = lldb.SBFile.Create(f, borrow=True)
+            sbf.Write(b'bar')
+            sbf = None
+            self.assertFalse(f.closed)
+            f.seek(0)
+            self.assertEqual(f.read(), 'foobar')