Skip to content

Instantly share code, notes, and snippets.

@jcockhren
Created November 18, 2013 17:23
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save jcockhren/e807b011d61e22d7686c to your computer and use it in GitHub Desktop.
Save jcockhren/e807b011d61e22d7686c to your computer and use it in GitHub Desktop.
This is an example of patching/mocking 3rd party libs. Some libs touch the underlying system and thus we need a way to isolate their usage in the code we're testing.
from flask import Flask
from flask import request
import os
'''
Notice we're not using the 'from' statement here and notice
the calls using these modules. Ex. git.Repo.
This will allow the mocking to work within the write context.
'''
import git
import subprocess
app = Flask(__name__)
app.config['DEBUG'] = True
BASE_DIRS = ['/var/www', '/apps']
def _find_site(site_slug):
location = None
for base in BASE_DIRS:
l = os.path.join(base, site_slug)
if os.path.isdir(l):
location = l
if location is None:
location = os.path.join(BASE_DIRS[0], site_slug)
return location
@app.route("/deploy", methods=['POST'])
def deploy():
data = request.json
real_repo = data['repository']['absolute_url']
real_repo = real_repo[1:-1]+".git"
dir_name = _find_site(data['repository']['slug'])
if not os.path.isdir(dir_name):
'''
Do something smart here like check folder permissions and such
'''
try:
subprocess.check_call(["git", "clone", "git@bitbucket.org:"+real_repo, dir_name])
except subprocess.CalledProcessError as e:
return "Error Occured"+e.message+"\n"
repo = git.Repo(dir_name)
else:
repo = git.Repo(dir_name)
origin = repo.remotes.origin
origin.fetch()
origin.pull()
return "Deployed "+repo.head.commit.hexsha+"\n"
import hook
import json
from mock import patch
import git
import subprocess
import unittest
class DeployTest(unittest.TestCase):
def setUp(self):
hook.app.config['TESTING'] = True
self.app = hook.app.test_client()
def test_client_is_active(self):
res = self.app.get('/')
self.assertEqual("Hello World!", res.data)
@unittest.skip
def test_post_deploy(self):
headers = [('Content-Type', 'application/json')]
j = json.dumps(dict(repository=dict(absolute_url='s', slug='blah')))
with patch('git.Repo', spec=True, side_effect=None) as git.Repo:
git.Repo.return_value.head.commit.hexsha = 'abc123'
res = self.app.post('/deploy', headers=headers, data=j)
self.assertEqual("Deployed abc123\n", res.data)
def test_deploy_failure_clone(self):
headers = [('Content-Type', 'application/json')]
j = json.dumps(dict(repository=dict(absolute_url='s', slug='blah')))
with patch('subprocess.check_call', side_effect=subprocess.CalledProcessError) as subprocess.check_call:
subprocess.check_call.return_value = ''
try:
res = self.app.post('/deploy', headers=headers, data=j)
except subprocess.CalledProcessError:
assert False # fail if exception isn't caught
self.assertEqual("Cannot Create Directory\n", res.data)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment