Long body in responses and requests

tempesta_fw/t/functional/health_monitoring/test_health_monitor.py

@@ -29,7 +29,7 @@ def prepare(self):
         self.tester.current_chain = copy.copy(self.chain)
         self.tester.recieved_chain = deproxy.MessageChain.empty()
         self.client.clear()
-        self.client.set_request(self.tester.current_chain.request)
+        self.client.set_request(self.tester.current_chain)
 
     def check_transition(self, messages):
         expected = None

tempesta_fw/t/functional/helpers/__init__.py

@@ -1,4 +1,4 @@
 __all__ = ['tf_cfg', 'deproxy', 'nginx', 'tempesta', 'error', 'flacky',
-           'analyzer', 'stateful', 'dmesg']
+           'analyzer', 'stateful', 'dmesg', 'wrk']
 
 # vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4

tempesta_fw/t/functional/helpers/chains.py

@@ -33,6 +33,37 @@ def make_502_expected():
     )
     return response
 
+def response_500():
+    date = deproxy.HttpMessage.date_time_string()
+    headers = [
+        'Date: %s' % date,
+        'Content-Length: 0',
+        'Connection: keep-alive'
+    ]
+    response = deproxy.Response.create(status=500, headers=headers)
+    return response
+
+def response_403(date=None, connection=None):
+    if date is None:
+        date = deproxy.HttpMessage.date_time_string()
+    headers = ['Content-Length: 0']
+    if connection != None:
+        headers.append('Connection: %s' % connection)
+
+    return deproxy.Response.create(status=403, headers=headers,
+                                   date=date, body='')
+
+def response_400(date=None, connection=None):
+    if date is None:
+        date = deproxy.HttpMessage.date_time_string()
+    headers = ['Content-Length: 0']
+    if connection != None:
+        headers.append('Connection: %s' % connection)
+
+    resp = deproxy.Response.create(status=400, headers=headers,
+                                   date=date, body='')
+    return resp
+
 def base(uri='/', method='GET', forward=True, date=None):
     """Base message chain. Looks like simple Curl request to Tempesta and
     response for it.
@@ -85,7 +116,6 @@ def base(uri='/', method='GET', forward=True, date=None):
     common_resp_date = date
     # response body
     common_resp_body = ''
-    common_resp_body_void = False
     # common part of response headers 
     common_resp_headers = [
         'Connection: keep-alive',
@@ -122,8 +152,6 @@ def base(uri='/', method='GET', forward=True, date=None):
         ]
         if method == "GET":
             common_resp_body = sample_body
-        else:
-            common_resp_body_void = True
 
     elif method == "POST":
         common_req_headers += [
@@ -165,12 +193,13 @@ def base(uri='/', method='GET', forward=True, date=None):
         uri=uri,
         body=common_req_body
     )
+
     tempesta_resp = deproxy.Response.create(
         status=common_resp_code,
         headers=common_resp_headers + tempesta_resp_headers_addn,
         date=common_resp_date,
         body=common_resp_body,
-        body_void=common_resp_body_void
+        method=method,
     )
 
     if forward:
@@ -180,12 +209,13 @@ def base(uri='/', method='GET', forward=True, date=None):
             uri=uri,
             body=common_req_body
         )
+
         backend_resp = deproxy.Response.create(
             status=common_resp_code,
             headers=common_resp_headers + backend_resp_headers_addn,
             date=common_resp_date,
             body=common_resp_body,
-            body_void=common_resp_body_void
+            method=method,
         )
     else:
         tempesta_req = None
@@ -197,7 +227,6 @@ def base(uri='/', method='GET', forward=True, date=None):
                                       server_response=backend_resp)
     return copy.copy(base_chain)
 
-
 def base_chunked(uri='/'):
     """Same as chains.base(), but returns a copy of message chain with
     chunked body.

tempesta_fw/t/functional/helpers/control.py

@@ -127,18 +127,27 @@ def __init__(self, threads=-1, uri='/', ssl=False):
         Client.__init__(self, binary='wrk', uri=uri, ssl=ssl)
         self.threads = threads
         self.script = ''
-
-    def set_script(self, script):
-        self.script = script
+        self.local_scriptdir = ''.join([
+                            os.path.dirname(os.path.realpath(__file__)),
+                            '/../wrk/'])
+        self.copy_script = True
+
+    def set_script(self, script, content=None):
+        self.script = script + ".lua"
+        if content == None:
+            local_path = ''.join([self.local_scriptdir, self.script])
+            local_script_path = os.path.abspath(local_path)
+            assert os.path.isfile(local_script_path), \
+                   'No script found: %s !' % local_script_path
+            f = open(local_script_path, 'r')
+            self.files.append((self.script, f.read()))
+        else:
+            self.node.copy_file(self.script, content)
 
     def append_script_option(self):
         if not self.script:
             return
-        path = ''.join([os.path.dirname(os.path.realpath(__file__)),
-                        '/../wrk/', self.script, '.lua'])
-        script_path = os.path.abspath(path)
-        assert os.path.isfile(script_path), \
-               'No script found: %s !' % script_path
+        script_path = self.workdir + "/" + self.script
         self.options.append('-s %s' % script_path)
 
     def form_command(self):

tempesta_fw/t/functional/helpers/deproxy.py

@@ -25,7 +25,6 @@ class ParseError(Exception):
 class IncompliteMessage(ParseError):
     pass
 
-
 class HeaderCollection(object):
     """
     A collection class for HTTP Headers. This class combines aspects of a list
@@ -207,10 +206,10 @@ def __repr__(self):
 class HttpMessage(object):
     __metaclass__ = abc.ABCMeta
 
-    def __init__(self, message_text=None, body_parsing=True, body_void=False):
+    def __init__(self, message_text=None, body_parsing=True, method="GET"):
         self.msg = ''
+        self.method = method
         self.body_parsing = True
-        self.body_void = body_void # For responses to HEAD requests
         self.headers = HeaderCollection()
         self.trailer = HeaderCollection()
         self.body = ''
@@ -222,41 +221,44 @@ def parse_text(self, message_text, body_parsing=True):
         self.body_parsing = body_parsing
         stream = StringIO(message_text)
         self.__parse(stream)
-        self.__set_str_msg()
+        self.build_message()
 
     def __parse(self, stream):
         self.parse_firstline(stream)
         self.parse_headers(stream)
         self.body = ''
         self.parse_body(stream)
 
-    def __set_str_msg(self):
+    def build_message(self):
         self.msg = str(self)
 
     @abc.abstractmethod
     def parse_firstline(self, stream):
         pass
 
+    @abc.abstractmethod
+    def parse_body(self, stream):
+        pass
+
     def get_firstline(self):
         return ''
 
     def parse_headers(self, stream):
         self.headers = HeaderCollection.from_stream(stream)
 
-    def parse_body(self, stream):
-        if self.body_parsing and 'Transfer-Encoding' in self.headers:
-            enc = self.headers['Transfer-Encoding']
-            option = enc.split(',')[-1] # take the last option
+    def read_encoded_body(self, stream):
+        """ RFC 7230. 3.3.3 #3 """
+        enc = self.headers['Transfer-Encoding']
+        option = enc.split(',')[-1] # take the last option
 
-            if option.strip().lower() == 'chunked':
-                self.read_chunked_body(stream)
-            else:
-                error.bug('Not implemented!')
-        elif self.body_parsing and 'Content-Length' in self.headers:
-            length = int(self.headers['Content-Length'])
-            self.read_sized_body(stream, length)
+        if option.strip().lower() == 'chunked':
+            self.read_chunked_body(stream)
         else:
-            self.body = stream.read()
+            error.bug('Not implemented!')
+
+    def read_rest_body(self, stream):
+        """ RFC 7230. 3.3.3 #7 """
+        self.body = stream.read()
 
     def read_chunked_body(self, stream):
         while True:
@@ -278,11 +280,10 @@ def read_chunked_body(self, stream):
         # Parsing trailer will eat last CRLF
         self.parse_trailer(stream)
 
-    def read_sized_body(self, stream, size):
-        if self.body_void:
-            return
-        if size == 0:
-            return
+    def read_sized_body(self, stream):
+        """ RFC 7230. 3.3.3 #5 """
+        size = int(self.headers['Content-Length'])
+
         self.body = stream.read(size)
         if len(self.body) != size:
             raise ParseError(("Wrong body size: expect %d but got %d!"
@@ -379,6 +380,19 @@ def parse_firstline(self, stream):
     def get_firstline(self):
         return ' '.join([self.method, self.uri, self.version])
 
+    def parse_body(self, stream):
+        """ RFC 7230 3.3.3 """
+        # 3.3.3 3
+        if 'Transfer-Encoding' in self.headers:
+            self.read_encoded_body(stream)
+            return
+        # 3.3.3 5
+        if 'Content-Length' in self.headers:
+            self.read_sized_body(stream)
+            return
+        # 3.3.3 6
+        self.body = ''
+
     def __eq__(self, other):
         return ((self.method == other.method)
                 and (self.version == other.version)
@@ -422,6 +436,31 @@ def parse_firstline(self, stream):
         except:
             raise ParseError('Invalid Status code!')
 
+    def parse_body(self, stream):
+        """ RFC 7230 3.3.3 """
+        # 3.3.3 1
+        if self.method == "HEAD":
+            return
+        code = int(self.status)
+        if code >= 100 and code <= 199 or \
+            code == 204 or code == 304:
+            return
+        # 3.3.3 2
+        if self.method == "CONNECT" and code >= 200 and code <= 299:
+            error.bug('Not implemented!')
+            return
+        # 3.3.3 3
+        if 'Transfer-Encoding' in self.headers:
+            self.read_encoded_body(stream)
+            return
+        # TODO: check 3.3.3 4
+        # 3.3.3 5
+        if 'Content-Length' in self.headers:
+            self.read_sized_body(stream)
+            return
+        # 3.3.3 7
+        self.read_rest_body(stream)
+
     def get_firstline(self):
         status = int(self.status)
         reason = BaseHTTPRequestHandler.responses[status][0]
@@ -438,12 +477,12 @@ def __ne__(self, other):
 
     @staticmethod
     def create(status, headers, version='HTTP/1.1', date=False,
-               srv_version=None, body=None, body_void=False):
+               srv_version=None, body=None, method='GET'):
         reason = BaseHTTPRequestHandler.responses
         first_line = ' '.join([version, str(status), reason[status][0]])
         msg = HttpMessage.create(first_line, headers, date=date,
                                  srv_version=srv_version, body=body)
-        return Response(msg, body_void=body_void)
+        return Response(msg, method=method)
 
 #-------------------------------------------------------------------------------
 # HTTP Client/Server
@@ -481,10 +520,10 @@ def run_start(self):
     def clear(self):
         self.request_buffer = ''
 
-    def set_request(self, request):
-        if request:
-            self.request = request
-            self.request_buffer = request.msg
+    def set_request(self, message_chain):
+        if message_chain:
+            self.request = message_chain.request
+            self.request_buffer = message_chain.request.msg
 
     def set_tester(self, tester):
         self.tester = tester
@@ -503,7 +542,7 @@ def handle_read(self):
         tf_cfg.dbg(5, self.response_buffer)
         try:
             response = Response(self.response_buffer,
-                                body_void=(self.request.method == 'HEAD'))
+                                method=self.request.method)
             self.response_buffer = self.response_buffer[len(response.msg):]
         except IncompliteMessage:
             return
@@ -512,6 +551,10 @@ def handle_read(self):
                            '<<<<<\n%s>>>>>'
                            % self.response_buffer))
             raise
+        if len(self.response_buffer) > 0:
+            # TODO: take care about pipelined case
+            raise ParseError('Garbage after response end:\n```\n%s\n```\n' % \
+                             self.response_buffer)
         if self.tester:
             self.tester.recieved_response(response)
         self.response_buffer = ''
@@ -529,7 +572,11 @@ def handle_write(self):
 
     def handle_error(self):
         _, v, _ = sys.exc_info()
-        error.bug('\tDeproxy: Client: %s' % v)
+        if type(v) == ParseError or type(v) == AssertionError:
+            raise v
+        else:
+            error.bug('\tDeproxy: Client: %s' % v)
+
 
 
 class ServerConnection(asyncore.dispatcher_with_send):
@@ -640,17 +687,24 @@ def handle_accept(self):
             self.connections.append(handler)
             assert len(self.connections) <= self.conns_n, \
                 ('Too lot connections, expect %d, got %d'
-                 & (self.conns_n, len(self.connections)))
+                 % (self.conns_n, len(self.connections)))
+
+    def handle_read_event(self):
+        asyncore.dispatcher.handle_read_event(self)
 
     def active_conns_n(self):
         return len(self.connections)
 
     def handle_error(self):
         _, v, _ = sys.exc_info()
-        raise  Exception('\tDeproxy: Server %s:%d: %s' % (self.ip, self.port, v))
+        if type(v) == AssertionError:
+            raise v
+        else:
+            raise  Exception('\tDeproxy: Server %s:%d: %s' % \
+             (self.ip, self.port, type(v)))
 
     def handle_close(self):
-        self.stop()
+        self.close()
 
 
 #-------------------------------------------------------------------------------
@@ -732,7 +786,7 @@ def run(self):
         for self.current_chain in self.message_chains:
             self.recieved_chain = MessageChain.empty()
             self.client.clear()
-            self.client.set_request(self.current_chain.request)
+            self.client.set_request(self.current_chain)
             self.loop()
             self.check_expectations()