В предыдущей части мы сделали инстурменты для тестирования серверного кода без участия сокетов. Но это получился самый тривиальный из видов тестов ­— 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 33

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+2:])     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)

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

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

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, '')

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


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

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

Сервер сворачивается в один вызов 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

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, чтобы он его использовал и передавал дальше уже всё готовенькое:


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

1 2 3

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

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

1 2	>>> 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 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/


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