Twisted ПАиПП — Техника разработки
Введение
Сообщество разработчиков Twisted открыто ведет процесс разработки и документирует его. Это позволяет взглянуть на работу профессионалов и позаимствовать базу знаний.
Разработка через тестирование (Test driven development/TDD)
Twisted имеет модуль для тестирования кода(unit tests). Он называется Trial.
Посмотрев примеры кода самого twisted, мы наверняка найдем схожую с нашей задачей проблему, которая была решена ранее. Так мы можем найти примеры тестов для протоколов. Такие тесты используются в проекте FATS.
Немного о технике TDD:
- Мы создаем код теста и пишем в нем пустые методы для нашего кода.
- Нашего кода не существует! Мы ставим себе задачу для его реализации.
- Запускаем Trial. Видим что у нас все горит, — ничего не работает.
- Начинаем выполнять эти пункты! Таким образом мы сразу тестируем свой код и планируем работы.
Для изучения техники тестирования советую прочитать книгу “Разработка через тестирование в примерах”, автор Кент Бек.
В проектах FATS и DBsync использовались шаблоны тестирования, например Подделка Объекта(Mock Object).
FATS включает в себя MockAsterisk и MockADBAPI.
Подделка сервера Астериск(MockAsterisk):
FATS сервис работает за счет того, что управляет сервером при помощи команд протокола FastAGI и получает соответствующие ответы. Чтобы не издеваться над своим терпением, проще запускать тесты так, чтобы приложение думало, что общается с сервером, а мы тем временем подменяли ответы Астериска на свои. Такая ситуация может привести к путанице, мы же не можем помнить все команды и их результаты. Другая проблема в том, что команды не имеет общего правила при ответе т.к. проект Астериск писали разные люди. Для этого реализована карта команд, которая используется в реализации сервера FATS и используется наизнанку в обе стороны.
Пример кода: тест для сервиса “hello FATS!”
from twisted.fats.test.asterisk import AGITestCase, ENV, COMMANDS, SUCCESS, FAILURE
from twisted.fats.examples.hello_agi import HelloCallHandler
#from twisted.fats.test.tool import MockADBAPI
class HelloFastAGIExampleTest(AGITestCase):
def _baseCall(self, expectedResponse):
def callResult(result):
self.assertEqual(expectedResponse, result)
call = HelloCallHandler()
call.agi = self.agi
df = call.startCall()
df.addCallback(callResult)
return df
def test_callScriptAndAsteriskCollaboration(self):
# Call script execute commands in the sequence:
# --> answer : * response SUCCESS
# --> say_number : * response SUCCESS
asterisk_responses = [
COMMANDS['ANSWER'][SUCCESS],
COMMANDS['SAY NUMBER'][SUCCESS]]
req_result = 'magic_result'
call = self._baseCall(req_result)
self.asterisk.setResponseList(asterisk_responses)
self.asterisk.start()
return call
Пример кода кода: сервис “hello FATS!”
from zope.interface import implements, Interface
from twisted.python import log, components
from twisted.application import internet, service
from twisted.internet.defer import inlineCallbacks, returnValue
from twisted.fats.service import FastAGIFactory, IFastAGIFactory, CallHandler
class HelloCallHandler(CallHandler):
"""My first call handler.
"""
def startCall(self):
log.msg('Hello FastAGI logging system.')
# Answer the call.
df = self.agi.answer()
# Say number and stop call session
df.addCallback(lambda _: self.agi.sayNumber(666))
return df
class IHelloFastAGIService(Interface):
"""Example service interface
"""
def my_method(param):
"""Example method
@param param: parameter
"""
class ExampleService:
implements(IHelloFastAGIService)
class HelloFastAGIFactoryFromService(FastAGIFactory):
"""My factory from service.
Implement service method and use them in the factory if it's
required.
"""
implements(IFastAGIFactory)
handler = HelloCallHandler
def __init__(self, service):
self.service = service
def my_method(self, param):
"""Adapt method from the service.
"""
return self.service.my_method(param)
components.registerAdapter(HelloFastAGIFactoryFromService,
IHelloFastAGIService,
IFastAGIFactory)
# create application and TCP server with factory.
PORT = 9000
factory = HelloFastAGIFactoryFromService(ExampleService)
application = service.Application("HelloFastAGI")
fastagi_service = internet.TCPServer(PORT, factory)
fastagi_service.setServiceParent(application)
Пример запуска тестов при помощи trial
$ trial twisted.fats.examples
twisted.fats.examples.test.test_hello_agi
HelloFastAGIExampleTest
test_callScriptAndAsteriskCollaboration ... [OK]
-------------------------------------------------------------------------------
Ran 1 tests in 0.007s
PASSED (successes=1)
В FATS удобно использовать MockADBAPI т.к. часто сервисы используют базу данных. И требуется аналогичная MockAsterisk проверка фиктивных данных.
DBSync использует Поддельную библиотеку ADBAPI(MockADBAPI).
Мы заменяем ADBAPI twisted своим объектом, который отвечает так же как и реальная база данных. Таким образом нам не надо ждать ответа от реальной базы и не надо создавать\обновлять тестовые данные. Это особенно важно с учетом того, что наш сервис должен работать с большим количеством данных и уметь их обрабатывать.
Тест журнала master базы DBSync
from twisted.trial import unittest
from sportbox.dbsync.journal import MasterJournal
from sportbox.dbsync.test.mock import MockADBAPI
from sportbox.dbsync.model.test.test_slave import testSchema
class MasterJournalTest(unittest.TestCase):
def test_getLastUpdateTime(self):
db = MockADBAPI(range(3))
journal = MasterJournal(db)
def checkResult(mod):
self.assertEqual(mod.last_update('baz'), 0)
self.assertEqual(mod.last_update('bar'), 2)
self.assertEqual(mod.last_update('foo'), 1)
df = journal.getLastUpdateTime(testSchema.tables())
df.addCallback(checkResult)
Запуск теста
$ trial sportbox.dbsync.test.test_journal
sportbox.dbsync.test.test_journal
MasterJournalTest
test_getFoulRecords ... [OK]
test_getLastUpdateTime ... [OK]
test_getRegisterRecords ... [OK]
MasterStorageModificationTest
test_init ... [OK]
test_last_update ... [OK]
test_len_exception ... [OK]
test_wrongTable ... [OK]
SlaveJournalTest
test_getLastModTime ... [OK]
test_getRecordsIntersection ... [OK]
SlaveStorageModificationTest
test_init ... [OK]
test_last_update ... [OK]
test_wrongTable ... [OK]
StorageModificationTest
test_init ... [OK]
test_last_update ... [OK]
test_wrongTable ... [OK]
-------------------------------------------------------------------------------
Ran 15 tests in 0.031s
PASSED (successes=15)
Для эффективной работы с тестированием и модификацией кода рекомендую прочитать книгу “Рефакторинг: улучшение существующего кода”, автор Мартин Фаулер.
Это поможет лучше структурировать код и понять принцип декомпозиции. Разделить логику внутри системы и получить уверенность в том, что код покрытый тестами делает только то, о чем его просят внешние вызовы.
Компонентная архитектура(zope.interface)
В синхронизаторе баз данных активно используется шаблон адаптер (Adapter) и стратегия (Strategy). Кроме того позаимствована идея реализации виртуальной файловой системы(vfs) в ядре ОС linux.
У нас есть два хранилища данных:
- Master storage
- Slave storage
После запроса к системе журналирования мы получаем данные о измененных сущностях в Slave storage относительно Master storage. Сервер скачивает данные и формирует в памяти абстрактную базу данных, после чего используя описанные ранее правила для данной сессии синхронизации, запускает механизм преобразования полученных данных к необходимому для записи в Master storage виду.
Пример кода: slave модуль
class Student(SlaveElement):
query = Query(
"""exec get_student @id=$id;""")
table = 'student'
def mod_gender(integer)
int_map = {
1: 't',
2: 'f',
None: None, # special description
0: None} # description too
if integer:
return int_map[int(integer)]
return None
class SchoolStudentAdapter(MasterAtom):
def __init__(self, slave_atom):
if slave_atom[0].event == 'delete':
MasterAtom.__init__(
self, slave=ISchoolSlaveInfo(slave_atom[0]))
else:
slave_info, name, age, gender = slave_atom
MasterAtom.__init__(
self, slave=ISchoolSlaveInfo(slave_info), name=name,
age=age, gender=mod_gender(gender))
components.registerAdapter(SchoolStudentAdapter,
SlaveAtom, ISchoolStudent)
Пример кода: master модуль
class Students(MasterElement):
implements(IStudents)
atomSchema = ISchoolStudent
query = MultiQuery(
insert=Query("""select school.update_student('school_xxx',
$slave, '$name', $age, $gender);"""),
update=Query("""select school.update_student('school_xxx',
$slave, '$name', $age, $gender);"""),
delete=Query("select dbsync.delete('school_xxx', $slave);"))
Пример кода: тест для конвертера данных slave->master
from twisted.trial import unittest
from sportbox.dbsync.model.slave import SlaveAtom, SlaveInfo
from sportbox.plugins.school.master import ISchoolStudent
class SchoolStudentAdapterTest(unittest.TestCase):
def test_adapter(self):
slave = SlaveAtom((SlaveInfo(1, 'foo', 'insert', 12345),
'name', 14, 2))
master = ISchoolStudent(slave)
# check result data
- Конвертер(Converter) — реализован на базе шаблона стратегия Strategy, адаптирует сущности под необходимую схему данных.
Существуют механизмы для управления и контролирования итерациями синхронизации, чтобы не превысить расход системных ресурсов за один цикл синхронизации. После модернизации данных, система формирует взаимоисключающие запросы на запись в Master storage. Для того, чтобы не возиться с написанием логики БД для каждого экземпляра синхронизатора, логика вынесена в хранимые процедуры, но может быть описана и в коде модулей DBSync на языке python.
Таким образом, технологии Twisted дали нам возможность быстро создавать и модифицировать существующие синхронизаторы данных. Работа превратилась в написание хранимых процедур и правил адаптации данных. Все остальное делает DBsync. Наличие тестов дает нам преимущество в отладке кода, ускоряет процесс разработки в несколько раз. Все не выполненные задачи легко видны при запуске тестов системы, формируется TODO. Сложные задачи превращаются в счетное количество простых задач.
Все украдено до нас! (Так часто говорит очень дорогой мне человек)
