Tips

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

  1. suppose in cfg.py model is like:
#!/usr/bin/python
 
TO_ADDRESS = 'test@yahoo.com'
 
class Tag(object):
  def getName(self):
    return "hello"
  1. 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()
  1. 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()
  1. 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

python archive file

Unless otherwise stated, the content of this page is licensed under Creative Commons Attribution-ShareAlike 3.0 License