Merge pull request #10604 from vzakaznikov/fix_timeouts_in_new_live_view_tests

Fixing hard coded timeouts in new live view tests.

tests/queries/0_stateless/01246_insert_into_watch_live_view.py

+#!/usr/bin/env python
+import os
+import sys
+import time
+import signal
+
+CURDIR = os.path.dirname(os.path.realpath(__file__))
+sys.path.insert(0, os.path.join(CURDIR, 'helpers'))
+
+from client import client, prompt, end_of_block
+
+log = None
+# uncomment the line below for debugging
+#log=sys.stdout
+
+with client(name='client1>', log=log) as client1, client(name='client2>', log=log) as client2, client(name='client3>', log=log) as client3:
+    client1.expect(prompt)
+    client2.expect(prompt)
+    client3.expect(prompt)
+
+    client1.send('SET allow_experimental_live_view = 1')
+    client1.expect(prompt)
+    client3.send('SET allow_experimental_live_view = 1')
+    client3.expect(prompt)
+
+    client1.send('DROP TABLE IF EXISTS test.lv')
+    client1.expect(prompt)
+    client1.send('DROP TABLE IF EXISTS test.mt')
+    client1.expect(prompt)
+    client1.send('DROP TABLE IF EXISTS test.sums')
+    client1.expect(prompt)
+    client1.send('CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple()')
+    client1.expect(prompt)
+    client1.send('CREATE LIVE VIEW test.lv AS SELECT sum(a) AS s FROM test.mt')
+    client1.expect(prompt)
+    client1.send('CREATE TABLE test.sums (s Int32, version Int32) Engine=MergeTree ORDER BY tuple()')
+    client1.expect(prompt)
+    client3.send('CREATE LIVE VIEW test.lv_sums AS SELECT * FROM test.sums ORDER BY version')
+    client3.expect(prompt)
+
+    client3.send("WATCH test.lv_sums FORMAT CSV")
+    client3.expect(r'0.*1' + end_of_block)
+
+    client1.send('INSERT INTO test.sums WATCH test.lv')
+    client1.expect(end_of_block)
+    
+    client3.expect('0,1.*\r\n')
+
+    client2.send('INSERT INTO test.mt VALUES (1),(2),(3)')
+    client2.expect(prompt)
+    client3.expect('6,2.*\r\n')
+
+    client2.send('INSERT INTO test.mt VALUES (4),(5),(6)')
+    client2.expect(prompt)
+    client3.expect('21,3.*\r\n')
+
+    # send Ctrl-C
+    client3.send('\x03', eol='')
+    match = client3.expect('(%s)|([#\$] )' % prompt)
+    if match.groups()[1]:
+        client3.send(client3.command)
+        client3.expect(prompt)
+
+    # send Ctrl-C
+    client1.send('\x03', eol='')
+    match = client1.expect('(%s)|([#\$] )' % prompt)
+    if match.groups()[1]:
+        client1.send(client1.command)
+        client1.expect(prompt)    
+
+    client2.send('DROP TABLE test.lv')
+    client2.expect(prompt)
+    client2.send('DROP TABLE test.lv_sums')
+    client2.expect(prompt)
+    client2.send('DROP TABLE test.sums')
+    client2.expect(prompt)
+    client2.send('DROP TABLE test.mt')
+    client2.expect(prompt)

tests/queries/0_stateless/01247_insert_into_watch_live_view_url.py

+#!/usr/bin/env python
+import os
+import sys
+import time
+import signal
+import requests
+
+CURDIR = os.path.dirname(os.path.realpath(__file__))
+sys.path.insert(0, os.path.join(CURDIR, 'helpers'))
+
+from httpechoserver import HTTP_SERVER_URL_STR
+from client import client, prompt, end_of_block
+from shell import shell
+
+log = None
+# uncomment the line below for debugging
+#log=sys.stdout
+
+for output_format in ['CSV', 'JSONEachRow', 'JSONEachRowWithProgress']:
+    with shell(log=log) as bash:
+        try:
+            bash.send("python \"%s\" 3" % os.path.join(CURDIR, 'helpers', 'httpechoserver.py'))
+            bash.expect("\n+")
+            with client(name='client1>', log=log) as client1, client(name='client2>', log=log) as client2:
+                client1.expect(prompt)
+                client2.expect(prompt)
+
+                client1.send('SET allow_experimental_live_view = 1')
+                client1.expect(prompt)
+
+                client1.send('DROP TABLE IF EXISTS test.lv')
+                client1.expect(prompt)
+                client1.send('DROP TABLE IF EXISTS test.mt')
+                client1.expect(prompt)
+                client1.send('CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple()')
+                client1.expect(prompt)
+                client1.send('CREATE LIVE VIEW test.lv AS SELECT sum(a) AS s FROM test.mt')
+                client1.expect(prompt)
+
+                client1.send("INSERT INTO FUNCTION url('%s', %s, 's Int32, version Int32') WATCH test.lv" % (HTTP_SERVER_URL_STR, output_format))
+                client1.expect(end_of_block)
+                bash.expect("0.*1.*\r\n")
+
+                client2.send('INSERT INTO test.mt VALUES (1),(2),(3)')
+                client2.expect(prompt)
+                bash.expect("6.*2.*\r\n")
+
+                client2.send('INSERT INTO test.mt VALUES (4),(5),(6)')
+                client2.expect(prompt)
+                bash.expect("21.*3.*\r\n")
+
+                # send Ctrl-C
+                client1.send('\x03', eol='')
+                match = client1.expect('(%s)|([#\$] )' % prompt)
+                if match.groups()[1]:
+                    client1.send(client1.command)
+                    client1.expect(prompt)
+
+                client1.send('DROP TABLE test.lv')
+                client1.expect(prompt)
+                client1.send('DROP TABLE test.mt')
+                client1.expect(prompt)
+        finally:
+            try:
+                for i in range(3):
+                    requests.post(HTTP_SERVER_URL_STR, data=b"0\r\n", timeout=1)
+            except Exception:
+                pass

tests/queries/0_stateless/helpers/httpechoserver.py

@@ -20,7 +20,7 @@ HTTP_SERVER_ADDRESS = (HTTP_SERVER_HOST, HTTP_SERVER_PORT)
 HTTP_SERVER_URL_STR = 'http://' + ':'.join(str(s) for s in HTTP_SERVER_ADDRESS) + "/"
 
 ostream = StringIO()
-istream = StringIO()
+istream = sys.stdout
 
 class EchoCSVHTTPServer(BaseHTTPRequestHandler):
     def _set_headers(self):
@@ -57,10 +57,8 @@ class EchoCSVHTTPServer(BaseHTTPRequestHandler):
             chunk = self.read_chunk()
             if not chunk:
                 break
-            pos = istream.tell()
-            istream.seek(0, SEEK_END)
             istream.write(chunk)
-            istream.seek(pos)
+            istream.flush()
         text = ""
         self._set_headers()
         self.wfile.write("ok")
@@ -78,11 +76,10 @@ def start_server(requests_amount, test_data="Hello,2,-2,7.7\nWorld,2,-5,8.8"):
             httpd.handle_request()
 
     t = threading.Thread(target=real_func)
-    t.out = istream
     return t
 
-def run():
-    t = start_server(1)
+def run(requests_amount=1):
+    t = start_server(requests_amount)
     t.start()
     t.join()
 
@@ -90,11 +87,11 @@ if __name__ == "__main__":
     exception_text = ''
     for i in range(1, 5):
         try:
-            run()
+            run(int(sys.argv[1]) if len(sys.argv) > 1 else 1)
             break
         except Exception as ex:
             exception_text = str(ex)
-            time.sleep(0.1)
+            time.sleep(1)
 
     if exception_text:
         print("Exception: {}".format(exception_text), file=sys.stderr)

tests/queries/0_stateless/helpers/shell.py

+import os
+import sys
+import time
+
+CURDIR = os.path.dirname(os.path.realpath(__file__))
+
+sys.path.insert(0, os.path.join(CURDIR))
+
+import uexpect
+
+class shell(object):
+    def __init__(self, command=None, name='', log=None, prompt='[#\$] '):
+        if command is None:
+            command = ['/bin/bash', '--noediting']
+        self.prompt = prompt
+        self.client = uexpect.spawn(command)
+        self.client.eol('\r')
+        self.client.logger(log, prefix=name)
+        self.client.timeout(20)
+        self.client.expect(prompt, timeout=60)
+
+    def __enter__(self):
+        io = self.client.__enter__()
+        io.prompt = self.prompt
+        return io
+
+    def __exit__(self, type, value, traceback):
+        self.client.reader['kill_event'].set()
+        # send Ctrl-C
+        self.client.send('\x03', eol='')
+        time.sleep(0.3)
+        self.client.send('exit', eol='\r')
+        self.client.send('\x03', eol='')
+        return self.client.__exit__(type, value, traceback)