httpbin: Improve request parsing/logging.
This commit is contained in:
parent
fd264631c4
commit
c530312aca
@ -3,12 +3,16 @@ import sys
|
|||||||
import socket
|
import socket
|
||||||
import functools
|
import functools
|
||||||
import os.path
|
import os.path
|
||||||
|
import collections
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
import pytestqt.plugin
|
import pytestqt.plugin
|
||||||
from PyQt5.QtCore import pyqtSlot, pyqtSignal, QProcess, QObject
|
from PyQt5.QtCore import pyqtSlot, pyqtSignal, QProcess, QObject
|
||||||
|
|
||||||
|
|
||||||
|
Request = collections.namedtuple('Request', 'verb, url')
|
||||||
|
|
||||||
|
|
||||||
class InvalidLine(Exception):
|
class InvalidLine(Exception):
|
||||||
|
|
||||||
"""Exception raised when HTTPBin prints a line which is not parsable."""
|
"""Exception raised when HTTPBin prints a line which is not parsable."""
|
||||||
@ -19,12 +23,26 @@ class InvalidLine(Exception):
|
|||||||
class HTTPBin(QObject):
|
class HTTPBin(QObject):
|
||||||
|
|
||||||
ready = pyqtSignal()
|
ready = pyqtSignal()
|
||||||
got_new_url = pyqtSignal(str)
|
new_request = pyqtSignal(Request)
|
||||||
|
|
||||||
|
LOG_RE = re.compile(r"""
|
||||||
|
(?P<host>[^ ]*)
|
||||||
|
\ ([^ ]*) # ignored
|
||||||
|
\ (?P<user>[^ ]*)
|
||||||
|
\ \[(?P<date>[^]]*)\]
|
||||||
|
\ "(?P<request>
|
||||||
|
(?P<verb>[^ ]*)
|
||||||
|
\ (?P<url>[^ ]*)
|
||||||
|
\ (?P<protocol>[^ ]*)
|
||||||
|
)"
|
||||||
|
\ (?P<status>[^ ]*)
|
||||||
|
\ (?P<size>[^ ]*)
|
||||||
|
""", re.VERBOSE)
|
||||||
|
|
||||||
def __init__(self, parent=None):
|
def __init__(self, parent=None):
|
||||||
super().__init__(parent)
|
super().__init__(parent)
|
||||||
self._invalid = False
|
self._invalid = False
|
||||||
self._visited_urls = []
|
self._requests = []
|
||||||
self.port = self._get_port()
|
self.port = self._get_port()
|
||||||
self.proc = QProcess()
|
self.proc = QProcess()
|
||||||
self.proc.setReadChannel(QProcess.StandardError)
|
self.proc.setReadChannel(QProcess.StandardError)
|
||||||
@ -36,13 +54,13 @@ class HTTPBin(QObject):
|
|||||||
sock.close()
|
sock.close()
|
||||||
return port
|
return port
|
||||||
|
|
||||||
def get_visited(self):
|
def get_requests(self):
|
||||||
self.proc.waitForReadyRead(500)
|
self.proc.waitForReadyRead(500)
|
||||||
self.read_urls()
|
self.read_log()
|
||||||
return self._visited_urls
|
return self._requests
|
||||||
|
|
||||||
@pyqtSlot()
|
@pyqtSlot()
|
||||||
def read_urls(self):
|
def read_log(self):
|
||||||
while self.proc.canReadLine():
|
while self.proc.canReadLine():
|
||||||
line = self.proc.readLine()
|
line = self.proc.readLine()
|
||||||
line = bytes(line).decode('utf-8').rstrip('\n')
|
line = bytes(line).decode('utf-8').rstrip('\n')
|
||||||
@ -53,25 +71,29 @@ class HTTPBin(QObject):
|
|||||||
self.ready.emit()
|
self.ready.emit()
|
||||||
continue
|
continue
|
||||||
|
|
||||||
match = re.match(r'.*"(GET [^ ]*) .*', line) # FIXME
|
match = self.LOG_RE.match(line)
|
||||||
if match is None:
|
if match is None:
|
||||||
self._invalid = True
|
self._invalid = True
|
||||||
print("INVALID: {}".format(line))
|
print("INVALID: {}".format(line))
|
||||||
continue
|
continue
|
||||||
url = match.group(1)
|
|
||||||
print(url)
|
# FIXME do we need to allow other options?
|
||||||
self._visited_urls.append(url)
|
assert match.group('protocol') == 'HTTP/1.1'
|
||||||
self.got_new_url.emit(url)
|
|
||||||
|
request = Request(verb=match.group('verb'), url=match.group('url'))
|
||||||
|
print(request)
|
||||||
|
self._requests.append(request)
|
||||||
|
self.new_request.emit(request)
|
||||||
|
|
||||||
def start(self):
|
def start(self):
|
||||||
filename = os.path.join(os.path.dirname(__file__), 'webserver.py')
|
filename = os.path.join(os.path.dirname(__file__), 'webserver.py')
|
||||||
self.proc.start(sys.executable, [filename, str(self.port)])
|
self.proc.start(sys.executable, [filename, str(self.port)])
|
||||||
ok = self.proc.waitForStarted()
|
ok = self.proc.waitForStarted()
|
||||||
assert ok
|
assert ok
|
||||||
self.proc.readyRead.connect(self.read_urls)
|
self.proc.readyRead.connect(self.read_log)
|
||||||
|
|
||||||
def after_test(self):
|
def after_test(self):
|
||||||
self._visited_urls.clear()
|
self._requests.clear()
|
||||||
if self._invalid:
|
if self._invalid:
|
||||||
raise InvalidLine
|
raise InvalidLine
|
||||||
|
|
||||||
|
@ -11,7 +11,7 @@ import pytest
|
|||||||
('/data/hello.txt', 'Hello World!', True),
|
('/data/hello.txt', 'Hello World!', True),
|
||||||
])
|
])
|
||||||
def test_httpbin(httpbin, qtbot, path, content, expected):
|
def test_httpbin(httpbin, qtbot, path, content, expected):
|
||||||
with qtbot.waitSignal(httpbin.got_new_url, raising=True, timeout=100):
|
with qtbot.waitSignal(httpbin.new_request, raising=True, timeout=100):
|
||||||
url = 'http://localhost:{}{}'.format(httpbin.port, path)
|
url = 'http://localhost:{}{}'.format(httpbin.port, path)
|
||||||
try:
|
try:
|
||||||
response = urllib.request.urlopen(url)
|
response = urllib.request.urlopen(url)
|
||||||
@ -25,5 +25,5 @@ def test_httpbin(httpbin, qtbot, path, content, expected):
|
|||||||
|
|
||||||
data = response.read().decode('utf-8')
|
data = response.read().decode('utf-8')
|
||||||
|
|
||||||
assert httpbin.get_visited() == ['GET {}'.format(path)]
|
assert httpbin.get_requests() == [('GET', path)]
|
||||||
assert (content in data) == expected
|
assert (content in data) == expected
|
||||||
|
Loading…
Reference in New Issue
Block a user