test: Improved testing around flush method and exception safety.

Author: amunra

TODO.rst

@@ -2,38 +2,24 @@
 TODO
 ====
 
-Stuff to get done before 1.0.0
-
 
 Build Tooling
 =============
-* **[HIGH]** Integrate tooling to build binaries for a matrix of operating systems,
-  architectures and Python versions. This will also be our CI.
+* **[HIGH]** Integrate tooling to build binaries for a matrix of operating
+  systems, architectures and Python versions. This will also be our CI.
   This should help: https://github.com/pypa/cibuildwheel
 
-* **[MEDIUM]** Integrate TOX https://tox.wiki/en/latest/ to trigger our testcases.
-  This makes it easier to eventually run tests vs. different dependency
-  versions.
-
-* **[MEDIUM]** Figure out how ``bumpversion`` works, how it cuts tags, etc.
-  Does ``bumpversion`` help us with ``CHANGELOG.rst``?
-
-* **[LOW]** Consider converting `setup.py` to `build.py` and transitioning to
-  `poetry`. Whilst we don't have other python dependencies for now
-  (though we will have numpy and pandas eventually), it would standardise the
-  way we build other python packages.
-  *This can probably wait for a future release.*
 
 Docs
 ====
-* **[MEDIUM]** Document on a per-version basis.
-
 * **[HIGH]** Author a few examples of how to use the client.
   This will help people get started. The examples should be presented in Sphinx
   using ``.. literalinclude::``.
   See: https://www.sphinx-doc.org/en/master/usage/restructuredtext/directives.html#directive-literalinclude
   The examples should be in the ``examples/`` directory in the repo.
 
+* **[MEDIUM]** Document on a per-version basis.
+
 * **[MEDIUM]** These examples should be tested as part of the unit tests (as they
   are in the C client). This is to ensure they don't "bit rot" as the code
   changes.
@@ -43,14 +29,12 @@ Development
 ===========
 * **[HIGH]** Review API naming!
 
-* **[HIGH]** Implement ``tabular()`` API in the buffer.
-
-* **[HIGH]** Test the flush API carefully with exceptions before / after flushing.
-
 * **[HIGH]** Implement the auto-commit logic based on a watermark.
 
 * **[MEDIUM]** Once we're done with them, merge in changes in the ``py_client_tweaks`` branch
   of the C client.
 
+* **[LOW]** Implement ``tabular()`` API in the buffer.
+
 * **[LOW]** Implement ``pandas()`` API in the buffer.
   *This can probably wait for a future release.*

proj.py

@@ -48,6 +48,8 @@ def _rmtree(path: pathlib.Path):
 
 
 def _arg2bool(arg):
+    if isinstance(arg, bool):
+        return arg
     return arg.lower() in ('true', 'yes', '1')
 
 
@@ -81,11 +83,11 @@ def doc(http_serve=False, port=None):
 
 
 @command
-def test(all=False, patch_path='1'):
+def test(all=False, patch_path='1', *args):
     env = {'TEST_QUESTDB_PATCH_PATH': patch_path}
     if _arg2bool(all):
         env['TEST_QUESTDB_INTEGRATION'] = '1'
-    _run('python3', 'test/test.py', '-v',
+    _run('python3', 'test/test.py', '-v', *args,
         env=env)
 
 

test/mock_server.py

@@ -33,8 +33,11 @@ def recv(self, wait_timeout_sec=0.1):
         buf = b''
         while True:
             # Block for *some* data.
-            select.select([self._client_sock], [], [])
-            buf += self._client_sock.recv(1024)
+            res =  select.select([self._client_sock], [], [], 1.0)
+            new_data = self._client_sock.recv(1024)
+            if not new_data:
+                return []
+            buf += new_data
             if len(buf) < 2:
                 continue
             if (buf[-1] == ord('\n')) and (buf[-2] != ord('\\')):
@@ -44,10 +47,13 @@ def recv(self, wait_timeout_sec=0.1):
         self.msgs.extend(new_msgs)
         return new_msgs
 
-    def __exit__(self, _ex_type, _ex_value, _ex_tb):
+    def close(self):
         if self._client_sock:
             self._client_sock.close()
             self._client_sock = None
         if self._sock:
             self._sock.close()
             self._sock = None
+
+    def __exit__(self, _ex_type, _ex_value, _ex_tb):
+        self.close()

test/test.py

@@ -5,6 +5,7 @@
 import os
 import unittest
 import datetime
+import time
 
 import patch_path
 from mock_server import Server
@@ -129,6 +130,82 @@ def test_row_before_connect(self):
         finally:
             sender.close()
 
+    def test_flush_1(self):
+        with Server() as server:
+            with ilp.Sender('localhost', server.port) as sender:
+                server.accept()
+                with self.assertRaisesRegex(ilp.IlpError, 'Column names'):
+                    sender.row('tbl1', symbols={'...bad name..': 'val1'})
+                self.assertEqual(str(sender), '')
+                sender.flush()
+                self.assertEqual(str(sender), '')
+            msgs = server.recv()
+            self.assertEqual(msgs, [])
+
+    def test_flush_2(self):
+        with Server() as server:
+            with ilp.Sender('localhost', server.port) as sender:
+                server.accept()
+                server.close()
+
+                # We enter a bad state where we can't flush again.
+                with self.assertRaises(ilp.IlpError):
+                    for _ in range(1000):
+                        time.sleep(0.01)
+                        sender.row('tbl1', symbols={'a': 'b'})
+                        sender.flush()
+
+                # We should still be in a bad state.
+                with self.assertRaises(ilp.IlpError):
+                    sender.row('tbl1', symbols={'a': 'b'})
+                    sender.flush()
+
+            # Leaving the `with` scope will call __exit__ and here we test
+            # that a prior exception will not cause subsequent problems.
+
+    def test_flush_3(self):
+        # Same as test_flush_2, but we catch the exception _outside_ the
+        # sender's `with` block, to ensure no exceptions get trapped.
+        with Server() as server:
+            with self.assertRaises(ilp.IlpError):
+                with ilp.Sender('localhost', server.port) as sender:
+                    server.accept()
+                    server.close()
+                    for _ in range(1000):
+                        time.sleep(0.01)
+                        sender.row('tbl1', symbols={'a': 'b'})
+                        sender.flush()
+
+    def test_independent_buffer(self):
+        buf = ilp.Buffer()
+        buf.row('tbl1', symbols={'sym1': 'val1'})
+        exp = 'tbl1,sym1=val1\n'
+        bexp = exp[:-1].encode('utf-8')
+        self.assertEqual(str(buf), exp)
+
+        with Server() as server1, Server() as server2:
+            with ilp.Sender('localhost', server1.port) as sender1, \
+                 ilp.Sender('localhost', server2.port) as sender2:
+                    server1.accept()
+                    server2.accept()
+
+                    sender1.flush(buf, clear=False)
+                    self.assertEqual(str(buf), exp)
+
+                    sender2.flush(buf, clear=False)
+                    self.assertEqual(str(buf), exp)
+
+                    msgs1 = server1.recv()
+                    msgs2 = server2.recv()
+                    self.assertEqual(msgs1, [bexp])
+                    self.assertEqual(msgs2, [bexp])
+
+                    sender1.flush(buf)
+                    self.assertEqual(server1.recv(), [bexp])
+
+                    # The buffer is now auto-cleared.
+                    self.assertEqual(str(buf), '')
+
 
 if __name__ == '__main__':
     unittest.main()