demo.py

#!/usr/bin/python2.7
#
# Run tests with:
#
#    python -m unittest discover <directory> demo_test.py
#    python -m unittest discover . demo_test.py

# 1.3.3
import fabric
import fabric.api
import fabric.decorators
import fabric.state


ROLEDEFS = {
    'deploy': ['localhost'],
    'remote': ['www1.example.com', 'www2.example.com'],
}


@fabric.decorators.roles('deploy')
def prepare():
    print 'Preparing (locally)'

@fabric.decorators.roles('remote')
def update():
    print 'Updating (remotely)'

def deploy(targets=None):
    roledefs = ROLEDEFS

    if targets:
      roledefs['remote'] = targets                                                                            

    with fabric.context_managers.settings(roledefs=roledefs):
        fabric.api.execute(prepare)
        fabric.api.execute(update)

demo_test.py

import os
import tempfile
import unittest

import fabric
import fabric.api
import fabric.state
import lockfile
import mock

import demo


class TestFoo(unittest.TestCase):

    def setUp(self):
        self.prepare_hosts_log = tempfile.mkstemp()[1]
        self.update_hosts_log = tempfile.mkstemp()[1]
        self.addCleanup(os.remove, self.prepare_hosts_log)
        self.addCleanup(os.remove, self.update_hosts_log)

        self.prepare_hosts_lock = lockfile.FileLock(self.prepare_hosts_log)
        self.update_hosts_lock = lockfile.FileLock(self.update_hosts_log)

        def add_prepare_hosts(*args, **kwargs):
            self.prepare_hosts_lock.acquire(timeout=3)
            f = open(self.prepare_hosts_log, 'a')
            f.write('{}\n'.format(fabric.state.env.host))
            f.close()
            self.prepare_hosts_lock.release()

        def add_update_hosts(*args, **kwargs):
            self.update_hosts_lock.acquire(timeout=3)
            f = open(self.update_hosts_log, 'a')
            f.write('{}\n'.format(fabric.state.env.host))
            f.close()
            self.update_hosts_lock.release()

        # Save roles so they can be restored after mocking out the main methods.
        prepare_roles = demo.prepare.roles
        update_roles = demo.update.roles

        demo.prepare = mock.Mock(side_effect=add_prepare_hosts)
        demo.update = mock.Mock(side_effect=add_update_hosts)

        # Restore attributes that the mock overwrote.
        demo.prepare.roles = prepare_roles
        demo.update.roles = update_roles
        demo.prepare.hosts = []
        demo.update.hosts = []

    def test_deploy_with_targets(self):
        demo.deploy(targets=['dev1.example.com', 'dev2.example.com'])
        self.prepare_hosts_lock.acquire(timeout=3)
        self.update_hosts_lock.acquire(timeout=3)
        self.assertEqual(['localhost\n'], open(self.prepare_hosts_log).readlines())
        self.assertEqual(sorted(['dev1.example.com\n', 'dev2.example.com\n']), sorted(open(self.update_hosts_log).readlines()))

    def test_deploy_default(self):
        demo.deploy()
        self.prepare_hosts_lock.acquire(timeout=3)
        self.update_hosts_lock.acquire(timeout=3)
        self.assertEqual(['localhost\n'], open(self.prepare_hosts_log).readlines())
        self.assertEqual(sorted(['www1.example.com\n', 'www2.example.com\n']), sorted(open(self.update_hosts_log).readlines()))