Python — Веб-сервер своими руками. Часть 3 — фиксим фичи, добавляем баги

В предыдущей части мы сделали инстурменты для тестирования серверного кода без участия сокетов. Но это получился самый тривиальный из видов тестов ­— Smoke Test. Сервер запрос обработал, но что именно произошло остаётся загадкой.

Как мы помним из кода, липовое соединение содержит в себе буфер отправленного, в котором оказывается ответ сервера. Можно было бы его сравнить с эталонной строкой, но каждый раз её составлять неудобно и муторно. Поэтому неплохо было бы его распарсить.

Но один раз у нас уже кто-то что-то парсит, а именно — сервер, при получении запроса от клиента. Внимательно посмотрев на траффик можно обнаружить, что протокол практически симметричен. И клиент и сервер обмениваются «сообщениями», состоящими из одних и тех же элементов: строка запроса или ответа (формат одинаковый, немного отличается содержимое), заголовки (формат одинаковый) и тело (необязательное для клиента при GET и для сервера при всяких хитрых статусах).

В то же время, наш тестовый клиент уже содержит генератор запросов, преобразующий аргументы функции согласно протоколу.

Вынесем эти две части как из клиента, так и из сервера:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32

def parse_http(data): lines = data.split('\r\n') query = lines<0>.split(' ') headers = {} for pos, line in enumerate(lines<1:>): if not line.strip(): break key, value = line.split(': ', 1) headers<key.upper> = value body = '\r\n'.join(lines<pos>) return query, headers, body def encode_http(query, body='', **headers): data = <" ".join(query)> headers = "\r\n".join("%s: %s" % ("-".join(part.title() for part in key.split('_')), value) for key, value in sorted(headers.iteritems())) if headers: data.append(headers) data.append('') if body: data.append(body) return "\r\n".join(data)</pos></key.upper>

Прежде чем менять актуальный и уже рабочий код, сделаем простенькие тесты конкретно для этого элемента. Они будут лежать в своём отдельном контейнере:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

class TestHTTP(object): def test_request(self): """тестирование в режиме запроса: клиент сериализует, сервер разбирает""" eq_('', encode_http(('GET', '/', 'HTTP/1.0'), user_agent="test/shmest")) eq_('', encode_http(('POST', '/', 'HTTP/1.0'), 'post=body', user_agent="test/shmest")) eq_((), parse_http('POST / HTTP/1.0\r\nUser-Agent: test/shmest\r\n\r\npost=body')) def test_response(self): """тестирование в режиме ответа: сервер сериализует, клиент разбирает""" data = 'HTTP/1.0 200 OK\r\nSpam: eggs\r\nTest-Me: please\r\n\r\nHellow, orld!\n' eq_(data, encode_http(('HTTP/1.0', '200', 'OK'), 'Hellow, orld!\n', test_me='please', spam="eggs")) reply, headers, body = parse_http(data) eq_(reply, ) eq_(headers, {}) eq_(body, '')

Это обычные заготовки тестов, которые при запуске будут фейлиться и сообщать, что получено не то, что ожидалось:

1	AssertionError: 'GET / HTTP/1.0\r\nUser-Agent: test/shmest\r\n' != ''

Убедившись «на глаз», что на выходе получается ровно то, что там должно быть согласно входным данным, копипастим (вдумчиво и внимательно!) значения в тесты.

Теперь, когда мы уверены, что всё работает, можно убрать дублирующийся код из клиента и сервера, а заодно добавить функционала.

Сервер сворачивается в один вызов parse и один encode и теперь готов к дальнейшему расширению без лишних усилий на ручное де/кодирование ответов:

1 2 3 4 5 6 7 8 9 10

def on_connect(self, conn, addr): """Соединение установлено, вычитываем запрос""" (method, url, proto), headers, body = parse_http(conn.recv(1024)) self.on_request(method, url, headers, body, conn) def on_request(self, method, url, headers, body, conn): """Обработка запроса""" print method, url, repr(body) conn.send(encode_http(("HTTP/1.0", "200", "OK"), "Hi there!\n", server="OwnHands/0.1")) conn.close()

Тест-клиент делает тоже самое, только в обратном порядке:

1 2 3 4

def __call__(self, url, method="GET", body='', **headers): conn = MockConnection(encode_http((method, url, "HTTP/1.0"), body, **headers)) self.server.on_connect(conn, None) return parse_http(conn.sent)

Теперь, пнув сервер в лабораторных условиях, мы можем точно узнать его реакцию, дописав соответствующий тест:

1 2 3 4 5 6 7 8

server = HTTPServer() client = MockClient(server) reply, headers, body = client('/hellow/orld/') eq_(reply, ) eq_(headers<'SERVER'>, '') eq_(body, '')

Уже заполненые тесты лежат на гуглькоде, но я считаю, что намного полезней и интересней поиграться и изучить всё это самим.

Заодно обратите внимание, что в отчёте nosetests функции, имеющие строки документации теперь отображаются в человечьем виде вместо «ехал гитлер^W тест через тест…»:

1 2 3

Тестирование в режиме запроса: клиент сериализует, сервер разбирает ... ok Тестирование в режиме ответа: сервер сериализует, клиент разбирает ... ok tests.TestServer.test_serve ... ok

Вот теперь мы готовы запилить что-нибудь полезное. Сервер выдающий один и тот же ответ мало интересен, поэтому надо сделать возможность расширения функционала. Тоесть потребуются какие-то обработчики и возможность их встраивать в код без наследования и перезаписи кусков кода сервера.

Можно было бы вешать код просто на URL, но это малоинтересно и не позволяет сделать какие-нибудь более продвинутые схемы. Сразу разделим обработчики на две фазы: pattern и handler. Первый занимается определением, надо ли вообще вызывать обработчик - получает всё, что сервер знает о запросе и выдаёт своё веское решение. Второй собственно знает, что его вызывают не просто так и пора заняться своей непосредственной работой - ответом.

Но сервер знает много чего, и это много передавать в виде аргументов очень неудобно. Поэтому завернём всё наше хозяйство в объект Request:

1 2 3 4 5 6 7 8 9

class Request(object): """Контейнер с данными текущего запроса и средством ответа на него""" def __init__(self, method, url, headers, body, conn): self.method = method self.url = url self.headers = headers self.body = body self.conn = conn

Пропишем сразу серверу в on_connect, чтобы он его использовал и передавал дальше уже всё готовенькое:

1	self.on_request(Request(method, url, headers, body, conn))

Сам же on_request теряет всю свою кучу аргументов и получает один (два, если вместе с self):

1 2 3

def on_request(self, request): """Обработка запроса""" print request

Хм.. При запросе сервер выводит какую-то нечитабельную лабуду в консоль. Это легко исправить. print пытается все свои аргументы привести сначала строковому виду, тоесть к типу str. Посмотреть что будет выводиться можно в терминале, сделав это вручную:

1 2	&gt;&gt;&gt; str(Request()) '<__main__.Request instance at 0x7f5a8a564488>'

В питоне всё-это-объект™ и у всех объектов может быть определён «волшебный» метод __str__ который будет в таких случаях вызываться. Там есть ещё много других интересных и странных методов, позволяющих сделать объект функцией или словарём или чёрти чем ещё. Пока что ограничимся просто читабельностью нашего контейнера и покажем пользователю немного содержимого:

1 2	def __str__(self): return "%s %s %r" % (self.method, self.url, self.headers)

Время разработчика очень ценно, а дублирование кода очень вредно. Поэтому, чтобы поймать сразу двух зайцев, скроем работу с соединением за функцией-помощником reply:

1 2 3 4 5 6 7 8 9

def reply(self, code='200', status='OK', body='', **headers): headers.setdefault('server', 'OwnHands/0.1') headers.setdefault('content_type', 'text/plain') headers.setdefault('content_length', len(body)) headers.setdefault('connection', 'close') headers.setdefault('date', datetime.now().ctime()) self.conn.send(encode_http(('HTTP/1.0', code, status), body, **headers)) self.conn.close()

Она сразу выставит дефолтные заголовки, которые при желании можно передать самому, но они практически обязательны и совершенно нет смысла их формировать каждый раз вручную. При очень большом желании, обработчик может взять request.conn и ответить так, как ему надо. Но такое требуется редко.

Посылка готова, можно отправлять. Но ещё надо составить список возможных получателей. Добавим в конструктор сервера инициализацию списка обработчиков:

1	self.handlers =

И метод их регистрации, в котором просто добавляем пару шаблон-обработчик в этот список:

1 2	def register(self, pattern, handler): self.handlers.append((pattern, handler))

Теперь on_request может стать диспетчером:

1 2 3 4 5 6 7

for pattern, handler in self.handlers: if pattern(request): # aim! handler(request) # fire! return True # работа по запросу завершена, откидываемся # никто не взялся ответить request.reply('404', 'Not found', 'Письмо самурай получил\nТают следы на песке\nСтраница не найдена')

Обновим тесты, с учётом всех нововведений. Класс, содержащий сценарии тестирования будет иметь несколько методов, каждый из которых будет создавать сервер, клиент для него и дальше делать свои дела. Дублирование кода детектед! К счастью, методика модульного тестирования уже давно решила эту задачу. Собственно для этого мы и используем тут классы, а не просто функции test_something. Специальный метод setup позволяет делать одинаковую настройку для каждого последующего запуска серии тестов:

1 2 3 4

class TestServer(object): def setup(self): self.server = HTTPServer() self.client = MockClient(self.server)

Попробуем теперь протестировать поведение пустого сервера без обработчиков. Клиент уже создан и настроен, поэтому сразу выстреливаем запрос:

1 2 3 4

def test_404(self): reply, headers, body = self.client('/you/cant/find/me/?yet') eq_(reply, <'HTTP/1.0', '404', 'Not found'>) eq_(headers<'SERVER'>, 'OwnHands/0.1')

Всё в порядке, можем продолжать. Зарегистрируем пару обработчиков и попробуем наш API на вкус:

1 2 3 4 5 6 7

def test_handlers(self): self.server.register(lambda r: r.url.startswith('/hello/'), # pattern lambda r: r.reply(body='hi')) # handler reply, headers, body = self.client('/hello/world/') eq_(reply<1>, '200') eq_(body, 'hi')

Одна из самых удобных возможностей питона — передавать функции в качестве аргументов, укладывать их в списки и назначать в переменные. Безо всяких if/case/goto и подобной чертовщины. lambda это выражение для создания анонимной функции; сжатый аналог def, которую можно создавать на ходу и передавать дальше не отвлекаясь от структуры кода.

Как и обещалось, проверять можно не только урл, но и всё, что доступно в запросе:

1 2 3 4 5 6

self.server.register(lambda r: r.method == 'POST', # отлавливать все посты lambda r: r.reply(body=r.body)) # зеркалим тело запроса reply, headers, body = self.client('/looking/for/a/POST/', 'POST', 'any url') # отправляем eq_(reply<1>, '200') # всё в порядке eq_(body, 'any url') # ловим то, что отправили

Тесты работают и можно приступать к реализации модулей, описаных в первой части.

PS: Специальный бонус для осиливших весь пост целиком \o/


(Полный код тестов и сервера)