Mocket

Neo: Whoa. Déjà vu.

Trinity: What did you just say?

Neo: Nothing. Just had a little déjà vu.

MocketSocket

The Oracle

SOCK_STREAM

socket()

bind()

listen()

socket()

accept()

connect()

def sendall(self, data, *args, **kwargs):

entry = Mocket.get_entry(self._host, self._port, data)

if not entry:

return self.true_sendall(data, *args, **kwargs)

entry.collect(data)

self.fd.seek(0)

self.fd.write(entry.get_response())

self.fd.seek(0)

def true_sendall(self, data, *args, **kwargs):

self.true_socket.connect(self._address)

self.true_socket.sendall(data, *args, **kwargs)

recv = True

while recv:

try:

recv = self.true_socket.recv(16)

self.true_socket.settimeout(0.0)

self.fd.write(recv)

except socket.error:

break

self.fd.seek(0)

self.true_socket.close()

write()

read()

write()

read()

pip install mockredispy

Mocket

@patch('redis.Redis', mock_redis_client)

The Keymaker

pip install httpretty

@httpretty.activate

@classmethod

def register(cls, *entries):

for entry in entries:

cls._entries[entry.location].append(entry)

@classmethod

def get_entry(cls, host, port, data):

entries = cls._entries.get((host, port), [])

for entry in entries:

if entry.can_handle(data):

return entry

@classmethod

def collect(cls, data):

cls._requests.append(data)

andrea de marco

giorgio salluzzo

@the_mindflayer

@z4erre

MOCK

backend devs @ buongiorno

appsfuel team

mip team

import unittest

class MyFuncTestCase(unittest.TestCase):

def testBasic(self):

a = ['larry', 'curly', 'moe']

self.assertEqual(my_func(a, 0), 'larry')

self.assertEqual(my_func(a, 1), 'curly')

them all

mocket

Mocketizer

agile

The Architect

"Code without tests is broken by design."

socket mock

framework

Jacob Kaplan-Moss

def __enter__(self):

Mocket.enable()

self.check_and_call('mocketize_setup')

def __exit__( self, type, value, tb ):

self.check_and_call('mocketize_teardown')

Mocket.disable()

Mocket.reset()

Neo: I know kung fu.

Morpheus: Show me.

'I know kung fu.\r\n'

Q&A

pip install mocket

Neo

Morpheus

thank you!

The Server

The Client

class Neo(object):

def iknow(self):

self._fp = self.sock.makefile('rb')

self.sock.sendall('I know kung fu.\r\n')

return self._fp.read().strip() == 'Show me.'

class Morpheus(asyncore.dispatcher_with_send):

def handle_read(self):

data = self.recv(8192).strip()

if data == 'I know kung fu.':

reply = 'Show me.'

else:

reply = 'Blue Pill.'

self.send(reply + '\r\n')

self.close()

github.com/mocketize/python-mocket

from unittest import TestCase

from mocket.mocket import Mocket, Mocketizer, MocketEntry

from neo import Neo

class RealTestCase(TestCase):

@Mocketizer.wrap

def test_ok(self):

neo = Neo()

self.assertTrue(neo.iknow())

self.assertEqual(len(Mocket._requests), 0)

class MockTestCase(TestCase):

@Mocketizer.wrap

def test_ok(self):

Mocket.register(MocketEntry(('localhost', 8080), ['Show me.\r\n']))

neo = Neo()

self.assertTrue(neo.iknow())

self.assertEqual(len(Mocket._requests), 1)

@Mocketizer.wrap

def test_ko(self):

Mocket.register(MocketEntry(('localhost', 8080), ['Blue Pill.\r\n']))

neo = Neo()

self.assertFalse(neo.iknow())

self.assertEqual(len(Mocket._requests), 1)

HTTP and REDIS

'Show me.\r\n'

are on GitHub

You take the red pill, you stay in Wonderland, and I show you how deep the rabbit hole goes.