Mox
install
sudo apt-get install python-mox
notes:
- self.mox.CreateMockAnything() or mox.MockAnything() :
- creates a mock object and returns it. however, nothing has not been stubbed out. you need to explicitly stub it out with self.mox.Stubs.Set(…)
- CreateMock(…) vs MockAnything() is that CreateMock(…) can create a mock object instance of a class. thus CreateMock(…) has input argument class_to_mock, while MockAnything() has no input. thus, CreateMock(…) is recommended to mock class instance, while MockAnything() is recommended for anything else.
- self.mox.StubOutWithMock(…) :
- marks something is stubbed out, and the stubbed out thing is a new created mock object. notice, the function doesn't return anything. the mock obj is not returned but it has been saved internally. when you refer the attribute, you are actually refer the mock object. so it is just like a combination of self.mox.CreateMockAnything(…) + self.mox.Stubs.Set(…). read the mox source code, you will understand.
- self.mox.StubOutClassWithMocks (…):
- this is similar with to self.mox.StubOutWithMock(…). it stubs out a class factory.
illustration
- suppose in cfg.py model is like:
#!/usr/bin/python TO_ADDRESS = 'test@yahoo.com' class Tag(object): def getName(self): return "hello"
- the main.py module is like:
#!/usr/bin/python import cfg def GetAddress(): return cfg.TO_ADDRESS def GetName(): tag = cfg.Tag() return tag.getName() def main(): print('address: %s' % GetAddress()) if __name__ == '__main__': main()
- the main_test.py is like:
#!/usr/bin/python import main import mox import unittest class MainTest(unittest.TestCase): def setUp(self): self.mox = mox.Mox() def tearDown(self): self.mox.VerifyAll() self.mox.UnsetStubs() print('after tear now, the address is: %s' % main.cfg.TO_ADDRESS) def testMain(self): self.mox.StubOutWithMock(main, 'cfg') # this stubs out the 'cfg' with a mock object. notice, notice the mock obj is not returned but it has been saved internally. main.cfg.TO_ADDRESS = 'test@gmail.com' # here main.cfg is already referred to the internal mock object. self.mox.ReplayAll() result = main.GetAddress() print('the result is: %s' % result) self.assertEqual('test@gmail.com', result) def testMain2(self): mock = self.mox.CreateMockAnything() # this just create a mock object mock.TO_ADDRESS = 'test@gmail.com' self.mox.stubs.Set(main, 'cfg', mock) # you need to explicitly stubs something with this mock object self.mox.ReplayAll() result = main.GetAddress() print('the result2 is: %s' % result) self.assertEqual('test@gmail.com', result) def testName(self): self.mox.StubOutClassWithMocks(main.cfg, 'Tag') # stub out the Tag's class Factory mock = main.cfg.Tag() mock.getName().AndReturn('world') self.mox.ReplayAll() result = main.GetName() print('the get name result is: %s' % result) self.assertEqual('world', result) def testName2(self): mock = self.mox.CreateMock(main.cfg.Tag) # create a mock object instance of class main.cfg.Tag mock.getName().AndReturn('world2') self.mox.StubOutWithMock(main.cfg, 'Tag') # stub out the Tag's class Factory main.cfg.Tag().AndReturn(mock) self.mox.ReplayAll() result = main.GetName() print('the get name result2 is: %s' % result) self.assertEqual('world2', result) def testName3(self): self.mox.StubOutWithMock(main.cfg.Tag, 'getName') main.cfg.Tag.getName().AndReturn('world3') self.mox.ReplayAll() result = main.GetName() print('the get name result3 is: %s' % result) self.assertEqual('world3', result) if __name__ == '__main__': unittest.main()
- run a single test case
python ./main_test.py MainTest.testMain2
or
./main_test.py MainTest.testMain2
if your main_test.py has no 'name== main' statement, you can run it as :
python -m unittest main_test.MainTest.testMain2
misc
- stub out instance's method or property
# Note the difference: we instantiate the class *before* Replaying. request = module_under_test.Request() self.mox.StubOutWithMock(request, 'Nomalize') # stub out method Nomalize foo_instance.Nomalize().AndReturn('....') ... self.mox.ReplayAll() handler.process(request) self.mox.VerifyAll()
- the subbed out method has no return value.
handler = MyRequestHandler() m = mox.Mox() m.StubOutWithMock(handler, "Process") handler.Process(IsA(Request)) m.ReplayAll() handler.Process(request) m.UnsetStubs() m.VerifyAll()
- stub out call back
m = mox.Mox() mock_callback = m.CreateMockAnything() # MockAnything is callable test_object.SetCallback(mock_callback) mock_callback(42) # Expect this to be called. m.ReplayAll() test_object.DoStuff() # Which in turn calls mock_callback...
- mock exception
def test_feed(self): zookeeper = self.m.CreateMock(zoo.ZooKeeper) zookeeper.feed(mox.StrContains("penguin")) zookeeper.feed(mox.StrContains("swan")).AndRaise(zoo.ZooException) z = zoo.Zoo(zookeeper) self.m.ReplayAll() with self.assertRaises(zoo.ZooException): z.feed() self.m.VerifyAll()
- good tutorial
mock usage1
mock usage2
mock usage3
- others
Mocking out a class in a different module
the mox doc here and there is extremely poor.
here is my case:
# com.xyz.authorize.py import ... oauth class Authorize(): def __init__(self, scope): self.scope = scope def verify(self, credentials): oauth_client = oauth.OAuth((self.scope) return oauth_client.Authorize(credentials)
- the good test approach 1:
# com.xyz.authorize_test.py import com.xyz.authorize as authorize class Test(unittest.TestCase): def setUp(self): self.mox = mox.Mox() def tearDown(self): self.mox.UnsetStubs() def testVerify(self): self.mox.StubOutClassWithMocks(authorize.oauth, 'OAuth') # This creates a internal class factory mock. mock1 = authorize.oauth.OAuth('scope123') # when you create a new class, you will be returned a mock object as the class factory is stubbed out, mock1.Authorize('password123').AndReturn(True) self.mox.ReplayAll() t = authorize.Authorize('scope123') self.assertTrue(t.verify('password123')) self.mox.VerifyAll()
- a clunky approach 2
# com.xyz.authorize_test.py import com.xyz.authorize as authorize class Test(unittest.TestCase): def setUp(self): self.mox = mox.Mox() def tearDown(self): self.mox.UnsetStubs() def testVerify(self): mock1 = self.mox.CreateMock(authorize.oauth.OAuth) mock1.Authorize('password123').AndReturn(True) self.mox.StubOutWithMock(authorize.oauth, 'OAuth') # this actually stubs out the class factory. authorize.oauth.OAuth('scope123').AndReturn(mock1) self.mox.ReplayAll() t = authorize.Authorize('scope123') self.assertTrue(t.verify('password123')) self.mox.VerifyAll()
- A less attractive approach 3 by mocking out a class method.
# com.xyz.authorize_test.py import com.xyz.authorize as authorize class Test(unittest.TestCase): def setUp(self): self.mox = mox.Mox() def tearDown(self): self.mox.UnsetStubs() def testVerify(self): # stub out a class's method self.mox.StubOutWithMock(authorize.oauth.OAuth, 'Authorize') authorize.oauth.OAuth.Authorize('password123').AndReturn(True) self.mox.ReplayAll() t = authorize.Authorize('scope123') self.assertTrue(t.verify('password123')) self.mox.VerifyAll()
that this might cause issues if you wish to concurrently mock multiple instances of OAuth since we stub out class method
http://kathrynbrisbin.blogspot.com/
tips
anyTrue
import itertools
def anyTrue(predicate, sequence):
return True in itertools.imap(predicate, sequence)
if anyTrue(filename.endswith, (".jpg", ".gif", ".png")):
arguparser
import argparse
parser = argparse.ArgumentParser()
parser_subparsers = parser.add_subparsers()
sub1 = parser_subparsers.add_parser('test')
sub1.add_argument('--a')
sub2 = parser_subparsers.add_parser('list')
sub2.add_argument('--b', type=str)
sub2.add_argument('--c', type=str)
sub2.add_argument('-t', action='store_true', default=False,
help='Set a switch to true')
def main():
x = parser.parse_args()
print x.b, x.c, x.t
restart program
import sys
import os
def restart_program():
"""Restarts the current program.
Note: this function does not return. Any cleanup action (like
saving data) must be done before calling this function."""
print 'restarting'
python = sys.executable
os.execl(python, python, * sys.argv)
if __name__ == "__main__":
answer = raw_input("Do you want to restart this program ? ")
if answer.strip() in "y Y yes Yes YES".split():
restart_program()
log
import logging
import sys
logger = logging.getLogger(__name__)
# http://www.lightbird.net/py-by-example/logging.html
# http://stackoverflow.com/questions/7016056/python-logging-not-outputting-anything
# http://stackoverflow.com/questions/14058453/making-python-loggers-output-all-messages-to-stdout-in-addition-to-log
if __name__ == '__main__':
# approch 1
# logging.basicConfig(level=logging.DEBUG)
# approach 2
ch = logging.StreamHandler(sys.stdout)
logger.addHandler(ch)
logger.setLevel(logging.DEBUG)
logger.info('I told you so')
execute multiple ssh commands
import subprocess
cmd = ('sshpass -p "password" ssh master@localhost "'
' ls ~;'
' echo "I am echoed";'
' pwd;'
' date;'
'"')
# os.system(cmd)
# result = subprocess.check_output(cmd, shell=True, stderr=STDOUT)
proc = subprocess.Popen(
cmd,
shell=True,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE) # stderr=subprocess.STDOUT to direct stderr to stdout,
result, error = proc.communicate()
logger.info('error=%s', error)
logger.info('result=%s', result)
you can use pexpect or another fashion:
http://stackoverflow.com/questions/17118239/how-to-give-subprocess-a-password-and-get-stdout-at-the-same-time
def ssh(cmd, pw):
pid, fd = pty.fork()
if pid == 0: # Child
os.system(cmd)
os._exit(1) # fail to execv
# read '..... password:', write password
os.read(fd, 1024)
os.write(fd, pw + '\n')
result = []
while True:
try:
data = os.read(fd, 1024)
except OSError:
break
if not data:
break
result.append(data)
pid, status = os.waitpid(pid, 0)
return status, ''.join(result)
http://helpful.knobs-dials.com/index.php/Python_usage_notes/pty_and_pexpect
http://stackoverflow.com/questions/4412238/whats-the-cleanest-way-to-ssh-and-run-multiple-commands-in-bash
ssh tunel explained
http://chamibuddhika.wordpress.com/2012/03/21/ssh-tunnelling-explained/
timeout
import signal
class TimeoutException(Exception):
pass
def signal_handler(signum, frame):
raise TimeoutException('timeout')
def run2(cmd, pw, timeout=2):
signal.signal(signal.SIGALRM, signal_handler)
signal.alarm(timeout) # seconds
try:
result = ssh(cmd, pw)
signal.alarm(0) # reset the alarm
return result
except TimeoutException:
return 'time out'
thread approach:
def run(cmd, timeout=2):
result = None
process = None
def target():
print 'Thread started'
process = subprocess.Popen(self.cmd, shell=True)
process.communicate()
print 'Thread finished'
thread = threading.Thread(target=target)
thread.start()
thread.join(timeout) # seconds
if thread.is_alive():
print 'Terminating process'
self.process.terminate()
thread.join()
if process:
return process.returncode
select approach
def read(fd, size, timeout): # seconds
"""Read size bytes from the serial fd. If a timeout is set it may
return less characters as requested. With no timeout it will block
until the requested number of bytes is read."""
read = ''
if size > 0:
while len(read) < size:
print "\tread(): size", size, "have", len(read), read #debug
ready, _, _ = select.select([fd], [], [], timeout)
if not ready:
break # timeout
print 'read from fd...'
buf = os.read(fd, size - len(read))
read = read + buf
if timeout >= 0 and not buf:
break #early abort on timeout
return read
http://snipplr.com/view/59069/runwithtimeout/
http://sqizit.bartletts.id.au/2011/02/14/pseudo-terminals-in-python/
import subprocess, shlex
from threading import Timer
def run(cmd, timeout_sec):
proc = subprocess.Popen(shlex.split(cmd), stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
kill_proc = lambda p: p.kill()
timer = Timer(timeout_sec, kill_proc, [proc])
timer.start()
stdout,stderr = proc.communicate()
timer.cancel()
http://stackoverflow.com/questions/1191374/subprocess-with-timeout
http://helpful.knobs-dials.com/index.php/Python_usage_notes/pty_and_pexpect
pipe vs pty
http://codeidol.com/python/python3/Parallel-System-Tools/Pipes/
daemon
http://www.jejik.com/articles/2007/02/a_simple_unix_linux_daemon_in_python/
http://stackoverflow.com/questions/3575554/python-subprocess-with-timeout-and-large-output-64k
organize project
+ XyzProject
|---xyz
| |--- __init__.py
| |--- common
| | |--- __init__.py
| | |--- util.py
| |--- client
| |--- __init__.py
| |---client_lib.py
|--- run_client.py
xyz/common/util.py
class Util(object):
def __init__(self):
print 'Util'
xyz/client/client_lib.py
import xyz.common.util
class Client(object):
def __init__(self):
print xyz.common
u = xyz.common.util.Util()
print 'client'
if __name__ == '__main__':
c = Client()
we can run the client_lib.py in XyzProject directory as:
python -m xyz.client.client_lib
run_client.py
import xyz.client.client_lib
if __name__ == '__main__':
c = xyz.client.client_lib.Client()
print 'Hello'
we can run the run_client.py in XyzProject directory as:
python -m run_client
or python run_client.py
another organize
+ XyzProject
|---xyz
| |--- __init__.py
| |--- common
| | |--- __init__.py
| | |--- util.py
| |--- server
| | |--- __init__.py
| | |--- server_lib.py
| |--- run_server.py
xyz/common/util.py
class Util(object):
def __init__(self):
print 'Util'
xyz/server/server_lib.py
import xyz.common.util
class Server(object):
def __init__(self):
print xyz.common
u = xyz.common.util.Util()
print 'server'
if __name__ == '__main__':
c = Server()
we can run the server_lib.py in XyzProject directory as:
python -m xyz.server.server_lib
xyz/run_server.py
import xyz.server.server_lib
if __name__ == '__main__':
c = xyz.server.server_lib.Server()
print 'World'
we can run the xyz/run_server.py in XyzProject directory as:
python -m xyz.run_server
notice python xyz/run_server.py won't work
https://ncoghlan_devs-python-notes.readthedocs.org/en/latest/python_concepts/import_traps.html
http://pythonadventures.wordpress.com/2012/03/31/import-a-file-from-another-directory-of-the-project/
http://stackoverflow.com/questions/1642975/folder-and-file-organization-for-python-development