def test_get_jira_user_multi(self):
"""
Tests the _get_jira_user method when JIRA returns more than one
matching user.
"""
self.mock_auth.search_users = Mock(return_value=[self.mock_jira_user_1,
self.mock_jira_user_2])
with patch('ambassador.transport.Transport.get_key_cert',
return_value=self.mock_cert):
with patch('platforms.jira.handlers.jira.JIRA',
return_value=self.mock_auth):
with patch.dict('platforms.jira.handlers.settings.JIRA',
self.mock_settings):
handler_w_user = IssueAPI(endpoint=self.endpoint,
user=self.user)
actual = handler_w_user._get_jira_user()
expected = None
self.assertEqual(actual, expected)
python类dict()的实例源码
def test_trims_path(self, virtual_env):
# pylint: disable=no-self-use
venv_path = '/Y1w4sD/DELETE/THIS/ISUy0r'
values = {
'VIRTUAL_ENV': venv_path,
'PATH': ':'.join(('/0ho8Ke/hL9DsW/ymH81W',
'/L51pua',
'/Vngp3V/G2m7Ih/05m7qW/LyrYkK/l5NuwA/oq1DPp',
venv_path,
'/Zpdhu4/bjvuqt',
'/STGtcb/FhAnWH/HwTvOr/gngGiB',
'/Zizj4D/szncsv/O5wO6X/joFHVT'))
}
with patch.dict('os.environ', values=values, clear=True):
env = virtual_env.sanitized_env()
assert venv_path not in env['PATH']
assert env['PATH'] == (
'/0ho8Ke/hL9DsW/ymH81W:/L51pua:/Vngp3V/G2m7Ih/05m7qW/LyrYkK/'
'l5NuwA/oq1DPp:/Zpdhu4/bjvuqt:/STGtcb/FhAnWH/HwTvOr/gngGiB:'
'/Zizj4D/szncsv/O5wO6X/joFHVT')
def test_disqus_sso_payload_auth_user(self, mock_b64encode, mock_hmac):
"""Test Disqus SSO payload auth works."""
user = UserFactory.create()
DISQUS_PUBLIC_KEY = 'public'
DISQUS_SECRET_KEY = 'secret'
patch_dict = {'DISQUS_PUBLIC_KEY': DISQUS_PUBLIC_KEY,
'DISQUS_SECRET_KEY': DISQUS_SECRET_KEY}
data = json.dumps({'id': user.id,
'username': user.name,
'email': user.email_addr})
mock_b64encode.return_value = data
with patch.dict(self.flask_app.config, patch_dict):
message, timestamp, sig, pub_key = util.get_disqus_sso_payload(user)
mock_b64encode.assert_called_with(data)
mock_hmac.assert_called_with(DISQUS_SECRET_KEY, '%s %s' % (data, timestamp),
hashlib.sha1)
assert timestamp
assert sig
assert pub_key == DISQUS_PUBLIC_KEY
def test_disqus_sso_payload_anon_user(self, mock_b64encode, mock_hmac):
"""Test Disqus SSO payload anon works."""
DISQUS_PUBLIC_KEY = 'public'
DISQUS_SECRET_KEY = 'secret'
patch_dict = {'DISQUS_PUBLIC_KEY': DISQUS_PUBLIC_KEY,
'DISQUS_SECRET_KEY': DISQUS_SECRET_KEY}
data = json.dumps({})
mock_b64encode.return_value = data
with patch.dict(self.flask_app.config, patch_dict):
message, timestamp, sig, pub_key = util.get_disqus_sso_payload(None)
mock_b64encode.assert_called_with(data)
mock_hmac.assert_called_with(DISQUS_SECRET_KEY, '%s %s' % (data, timestamp),
hashlib.sha1)
assert timestamp
assert sig
assert pub_key == DISQUS_PUBLIC_KEY
def test_handle_content_type_json_error(self, mocklast, mockjsonify,
mockrender, mockrequest):
fake_d = {'Content-Type': 'application/json'}
mockrequest.headers.__getitem__.side_effect = fake_d.__getitem__
mockrequest.headers.get.side_effect = fake_d.get
mockrequest.headers.__iter__.side_effect = fake_d.__iter__
mockjsonify.side_effect = myjsonify
res, code = util.handle_content_type(
dict(
template='example.html',
code=404,
description="Not found"))
err_msg = "template key should exist"
assert res.get('template') == 'example.html', err_msg
err_msg = "jsonify should be called"
assert mockjsonify.called, err_msg
err_msg = "Error code should exist"
assert res.get('code') == 404, err_msg
assert code == 404, err_msg
err_msg = "Error description should exist"
assert res.get('description') is not None, err_msg
def test_handle_content_type_json_form(self, mocklast, mockcsrf,
mockjsonify, mockrender,
mockrequest):
fake_d = {'Content-Type': 'application/json'}
mockrequest.headers.__getitem__.side_effect = fake_d.__getitem__
mockrequest.headers.get.side_effect = fake_d.get
mockrequest.headers.__iter__.side_effect = fake_d.__iter__
mockjsonify.side_effect = myjsonify
mockcsrf.return_value = "yourcsrf"
form = MagicMock(spec=Form, data=dict(foo=1), errors=None)
res = util.handle_content_type(dict(template='example.html',
form=form))
err_msg = "template key should exist"
assert res.get('template') == 'example.html', err_msg
err_msg = "jsonify should be called"
assert mockjsonify.called, err_msg
err_msg = "Form should exist"
assert res.get('form'), err_msg
err_msg = "Form should have a csrf key/value"
assert res.get('form').get('csrf') == 'yourcsrf', err_msg
err_msg = "There should be the keys of the form"
keys = ['foo', 'errors', 'csrf']
assert res.get('form').keys().sort() == keys.sort(), err_msg
def test_handle_content_type_html(self, mockjsonify,
mockrender, mockrequest):
fake_d = {'Content-Type': 'text/html'}
mockrequest.headers.__getitem__.side_effect = fake_d.__getitem__
mockrequest.headers.get.side_effect = fake_d.get
mockrequest.headers.__iter__.side_effect = fake_d.__iter__
mockjsonify.side_effect = myjsonify
mockrender.side_effect = myrender
pagination = util.Pagination(page=1, per_page=5, total_count=10)
template, data = util.handle_content_type(dict(template='example.html',
pagination=pagination))
err_msg = "Template should be rendered"
assert template == 'example.html', err_msg
err_msg = "Template key should not exist"
assert data.get('template') is None, err_msg
err_msg = "jsonify should not be called"
assert mockjsonify.called is False, err_msg
err_msg = "render_template should be called"
assert mockrender.called is True, err_msg
def test_handle_content_type_html_error(self, mockjsonify,
mockrender, mockrequest):
fake_d = {'Content-Type': 'text/html'}
mockrequest.headers.__getitem__.side_effect = fake_d.__getitem__
mockrequest.headers.get.side_effect = fake_d.get
mockrequest.headers.__iter__.side_effect = fake_d.__iter__
mockjsonify.side_effect = myjsonify
mockrender.side_effect = myrender
template, code = util.handle_content_type(dict(template='example.html',
code=404))
data = template[1]
template = template[0]
err_msg = "Template should be rendered"
assert template == 'example.html', err_msg
err_msg = "Template key should not exist"
assert data.get('template') is None, err_msg
err_msg = "jsonify should not be called"
assert mockjsonify.called is False, err_msg
err_msg = "render_template should be called"
assert mockrender.called is True, err_msg
err_msg = "There should be an error"
assert code == 404, err_msg
err_msg = "There should not be code key"
assert data.get('code') is None, err_msg
def test_UnicodeWriter(self):
"""Test UnicodeWriter class works."""
tmp = tempfile.NamedTemporaryFile()
uw = util.UnicodeWriter(tmp)
fake_csv = ['one, two, three, {"i": 1}']
for row in csv.reader(fake_csv):
# change it for a dict
row[3] = dict(i=1)
uw.writerow(row)
tmp.seek(0)
err_msg = "It should be the same CSV content"
with open(tmp.name, 'rb') as f:
reader = csv.reader(f)
for row in reader:
for item in row:
assert item in fake_csv[0], err_msg
def test_register_json_errors_get(self):
"""Test WEB register errors JSON works"""
with patch.dict(self.flask_app.config, {'WTF_CSRF_ENABLED': True}):
csrf = self.get_csrf('/account/register')
userdict = {'fullname': 'a', 'name': 'name',
'email_addr': None, 'password': 'p'}
res = self.app.post('/account/register', data=json.dumps(userdict),
content_type='application/json',
headers={'X-CSRFToken': csrf})
# The output should have a mime-type: application/json
errors = json.loads(res.data).get('form').get('errors')
assert res.mimetype == 'application/json', res.data
err_msg = "There should be an error with the email"
assert errors.get('email_addr'), err_msg
err_msg = "There should be an error with fullname"
assert errors.get('fullname'), err_msg
err_msg = "There should be an error with password"
assert errors.get('password'), err_msg
err_msg = "There should NOT be an error with name"
assert errors.get('name') is None, err_msg
def test_register_confirmation_validates_email(self, fake_signer):
"""Test WEB validates email"""
self.register()
user = db.session.query(User).get(1)
user.valid_email = False
user.confirmation_email_sent = True
db.session.commit()
fake_signer.loads.return_value = dict(fullname=user.fullname,
name=user.name,
email_addr=user.email_addr)
self.app.get('/account/register/confirmation?key=valid-key')
user = db.session.query(User).get(1)
assert user is not None
msg = "Email has not been validated"
assert user.valid_email, msg
msg = "Confirmation email flag has not been restored"
assert user.confirmation_email_sent is False, msg
def test_confirm_account_newsletter(self, fake_signer, url_for, newsletter):
"""Test WEB confirm email shows newsletter or home."""
newsletter.ask_user_to_subscribe.return_value = True
self.register()
user = db.session.query(User).get(1)
user.valid_email = False
db.session.commit()
fake_signer.loads.return_value = dict(fullname=user.fullname,
name=user.name,
email_addr=user.email_addr)
self.app.get('/account/register/confirmation?key=valid-key')
url_for.assert_called_with('account.newsletter_subscribe', next=None)
newsletter.ask_user_to_subscribe.return_value = False
self.app.get('/account/register/confirmation?key=valid-key')
url_for.assert_called_with('home.home')
def test_account_upload_avatar(self):
"""Test WEB Account upload avatar."""
import io
owner = UserFactory.create()
url = '/account/%s/update?api_key=%s' % (owner.name,
owner.api_key)
avatar = (io.BytesIO(b'test'), 'test_file.jpg')
payload = dict(btn='Upload', avatar=avatar,
id=owner.id, x1=0, y1=0,
x2=100, y2=100)
res = self.app.post(url, follow_redirects=True,
content_type="multipart/form-data", data=payload)
assert res.status_code == 200
u = user_repo.get(owner.id)
assert u.info['avatar'] is not None
assert u.info['container'] is not None
avatar_url = '/uploads/%s/%s' % (u.info['container'], u.info['avatar'])
assert u.info['avatar_url'] == avatar_url
def test_update_project_json_as_user(self):
"""Test WEB JSON update project as user."""
admin = UserFactory.create()
owner = UserFactory.create()
user = UserFactory.create()
project = ProjectFactory.create(owner=owner)
url = '/project/%s/update?api_key=%s' % (project.short_name, user.api_key)
res = self.app_get_json(url)
data = json.loads(res.data)
assert data['code'] == 403, data
old_data = dict()
old_data['description'] = 'foobar'
res = self.app_post_json(url, data=old_data)
data = json.loads(res.data)
assert data['code'] == 403, data
def test_26_tutorial_signed_user(self):
"""Test WEB tutorials work as signed in user"""
self.create()
project1 = db.session.query(Project).get(1)
project1.info = dict(tutorial="some help", task_presenter="presenter")
db.session.commit()
self.register()
# First time accessing the project should redirect me to the tutorial
res = self.app.get('/project/test-app/newtask', follow_redirects=True)
err_msg = "There should be some tutorial for the project"
assert "some help" in res.data, err_msg
# Second time should give me a task, and not the tutorial
res = self.app.get('/project/test-app/newtask', follow_redirects=True)
assert "some help" not in res.data
# Check if the tutorial can be accessed directly
res = self.app.get('/project/test-app/tutorial', follow_redirects=True)
err_msg = "There should be some tutorial for the project"
assert "some help" in res.data, err_msg
def test_26_tutorial_signed_user_json(self):
"""Test WEB tutorials work as signed in user"""
self.create()
project1 = db.session.query(Project).get(1)
project1.info = dict(tutorial="some help", task_presenter="presenter")
db.session.commit()
self.register()
# First time accessing the project should redirect me to the tutorial
res = self.app.get('/project/test-app/newtask', follow_redirects=True)
err_msg = "There should be some tutorial for the project"
assert "some help" in res.data, err_msg
# Second time should give me a task, and not the tutorial
res = self.app.get('/project/test-app/newtask', follow_redirects=True)
assert "some help" not in res.data
# Check if the tutorial can be accessed directly
res = self.app_get_json('/project/test-app/tutorial')
data = json.loads(res.data)
err_msg = 'key missing'
assert 'owner' in data, err_msg
assert 'project' in data, err_msg
assert 'template' in data, err_msg
assert 'title' in data, err_msg
err_msg = 'project tutorial missing'
assert 'My New Project' in data['title'], err_msg
def test_27_tutorial_anonymous_user_json(self):
"""Test WEB tutorials work as an anonymous user"""
self.create()
project = db.session.query(Project).get(1)
project.info = dict(tutorial="some help", task_presenter="presenter")
db.session.commit()
# First time accessing the project should redirect me to the tutorial
res = self.app.get('/project/test-app/newtask', follow_redirects=True)
err_msg = "There should be some tutorial for the project"
assert "some help" in res.data, err_msg
# Second time should give me a task, and not the tutorial
res = self.app.get('/project/test-app/newtask', follow_redirects=True)
assert "some help" not in res.data
# Check if the tutorial can be accessed directly
res = self.app_get_json('/project/test-app/tutorial')
data = json.loads(res.data)
err_msg = 'key missing'
assert 'owner' in data, err_msg
assert 'project' in data, err_msg
assert 'template' in data, err_msg
assert 'title' in data, err_msg
err_msg = 'project tutorial missing'
assert 'My New Project' in data['title'], err_msg
def test_anon(self):
"""Test with user authenticated."""
url = 'api/disqus/sso'
DISQUS_PUBLIC_KEY = 'public'
DISQUS_SECRET_KEY = 'secret'
patch_dict = {'DISQUS_PUBLIC_KEY': DISQUS_PUBLIC_KEY,
'DISQUS_SECRET_KEY': DISQUS_SECRET_KEY}
with patch.dict(self.flask_app.config, patch_dict):
res = self.app.get(url)
data = json.loads(res.data)
assert res.status_code == 200, res.status_code
assert data['remote_auth_s3'] is not None, data
assert data['api_key'] is not None, data
def test_get_jobs_only_on_sunday_variant(self, mock_datetime):
"""Test JOB get jobs for weekly stats works only on Sunday variant."""
user = UserFactory.create(pro=True)
pr = ProjectFactory(owner=user)
task = TaskFactory.create(project=pr)
TaskRunFactory.create(project=pr, task=task)
mock_date = MagicMock()
mock_date.strftime.return_value = 'Sunday'
mock_datetime.today.return_value = mock_date
jobs = get_weekly_stats_update_projects()
for job in jobs:
assert type(job) == dict, type(job)
assert job['name'] == send_weekly_stats_project
assert job['args'] == [pr.id]
assert job['kwargs'] == {}
assert job['timeout'] == self.flask_app.config.get('TIMEOUT')
assert job['queue'] == 'low'
def test_get_jobs_no_pro_feature_for_everyone(self, mock_datetime):
"""Test JOB get jobs for weekly stats works for non pros if feature is
only for everyone."""
user = UserFactory.create(pro=False)
pr = ProjectFactory(owner=user)
task = TaskFactory.create(project=pr)
TaskRunFactory.create(project=pr, task=task)
mock_date = MagicMock()
mock_date.strftime.return_value = 'Sunday'
mock_datetime.today.return_value = mock_date
jobs = [job for job in get_weekly_stats_update_projects()]
assert len(jobs) == 1
for job in jobs:
assert type(job) == dict, type(job)
assert job['name'] == send_weekly_stats_project
assert job['args'] == [pr.id]
assert job['kwargs'] == {}
assert job['timeout'] == self.flask_app.config.get('TIMEOUT')
assert job['queue'] == 'low'
def test_get_jobs_only_on_featured_variant(self, mock_datetime):
"""Test JOB get jobs for weekly stats works for featured."""
user = UserFactory.create(pro=False)
pr = ProjectFactory(owner=user, featured=True)
task = TaskFactory.create(project=pr)
TaskRunFactory.create(project=pr, task=task)
mock_date = MagicMock()
mock_date.strftime.return_value = 'Sunday'
mock_datetime.today.return_value = mock_date
jobs = get_weekly_stats_update_projects()
for job in jobs:
assert type(job) == dict, type(job)
assert job['name'] == send_weekly_stats_project
assert job['args'] == [pr.id]
assert job['kwargs'] == {}
assert job['timeout'] == self.flask_app.config.get('TIMEOUT')
assert job['queue'] == 'low'
def test_autoimport_jobs_with_autoimporter(self):
"""Test JOB autoimport jobs returns projects with autoimporter."""
user = UserFactory.create(pro=True)
project = ProjectFactory.create(owner=user,info=dict(autoimporter='foobar'))
jobs_generator = get_autoimport_jobs()
jobs = []
for job in jobs_generator:
jobs.append(job)
msg = "There should be 1 job."
assert len(jobs) == 1, msg
job = jobs[0]
msg = "It sould be the same project."
assert job['args'][0] == project.id, msg
msg = "It sould be created as an auto import job."
assert job['args'][1] == True, msg
msg = "There sould be the kwargs."
assert job['kwargs'] == 'foobar', msg
def test_autoimport_jobs_without_pro_when_for_everyone(self):
"""Test JOB autoimport jobs returns normal user owned projects
if autoimporter is enabled for everyone."""
project = ProjectFactory.create(info=dict(autoimporter='foobar'))
jobs_generator = get_autoimport_jobs()
jobs = []
for job in jobs_generator:
jobs.append(job)
msg = "There should be 1 job."
assert len(jobs) == 1, msg
job = jobs[0]
msg = "It sould be the same project."
assert job['args'][0] == project.id, msg
msg = "It sould be created as an auto import job."
assert job['args'][1] == True, msg
msg = "There sould be the kwargs."
assert job['kwargs'] == 'foobar', msg
def test_rackspace_uploader_init(self, Mock):
"""Test RACKSPACE UPLOADER init works."""
new_extensions = ['pdf', 'doe']
with patch('pybossa.uploader.rackspace.pyrax.cloudfiles',
return_value=cloudfiles_mock):
with patch.dict(self.flask_app.config,
{'ALLOWED_EXTENSIONS': new_extensions}):
with patch('pybossa.uploader.rackspace.pyrax.cloudfiles') as mycf:
mycf.get_container.return_value = True
u = RackspaceUploader()
res = u.init_app(self.flask_app, cont_name='mycontainer')
err_msg = "It should return the container."
assert res is True, err_msg
err_msg = "The container name should be updated."
assert u.cont_name == 'mycontainer', err_msg
for ext in new_extensions:
err_msg = "The .%s extension should be allowed" % ext
assert ext in u.allowed_extensions, err_msg
def test_local_uploader_init(self):
"""Test LOCAL UPLOADER init works."""
u = LocalUploader()
u.init_app(self.flask_app)
new_extensions = ['pdf', 'doe']
new_upload_folder = '/tmp/'
new_config_ext = {'ALLOWED_EXTENSIONS': new_extensions}
new_config_uf = {'UPLOAD_FOLDER': new_upload_folder}
with patch.dict(self.flask_app.config, new_config_ext):
with patch.dict(self.flask_app.config, new_config_uf):
new_uploader = LocalUploader()
new_uploader.init_app(self.flask_app)
expected_extensions = set.union(u.allowed_extensions,
new_extensions)
err_msg = "The new uploader should support two extra extensions"
assert expected_extensions == new_uploader.allowed_extensions, err_msg
err_msg = "Upload folder /tmp should be existing"
assert os.path.isdir(new_uploader.upload_folder) is True, err_msg
err_msg = "Upload folder by default is /tmp/"
assert new_uploader.upload_folder == '/tmp/', err_msg
def test_stdout_chunk(self, get_stdout, get_info_for_rj):
"""Test RunningJob() stdout_chunk"""
get_info_for_rj.return_value = dict([('stdout_size', 10)])
get_stdout.return_value = "line1\nline2\nline3\nline4\nline5\nline6"
running_job = pygenie.jobs.RunningJob('1234-stdout-chunk',
info={'status': 'SUCCEEDED'})
running_job.stdout_chunk(10, offset=0)
assert_equals(
[
call('1234-stdout-chunk', headers={'Range': 'bytes=0-10'})
],
get_stdout.call_args_list
)
def test_stdout_chunk_zero_size(self, get_stdout, get_info_for_rj):
"""Test RunningJob() stdout_chunk_zero_size"""
get_info_for_rj.return_value = dict([('stdout_size', 0)])
get_stdout.return_value = "line1\nline2\nline3\nline4\nline5\nline6"
running_job = pygenie.jobs.RunningJob('1234-stdout-chunk',
info={'status': 'SUCCEEDED'})
chunk = running_job.stdout_chunk(10, offset=0)
assert_equals(
[],
get_stdout.call_args_list
)
assert_equals(chunk, None)
def invoke_3play_callback(self, state='complete'):
"""
Make request to 3PlayMedia callback handler, this invokes
callback with all the necessary parameters.
Arguments:
state(str): state of the callback
"""
response = self.client.post(
# `build_url` strips `/`, putting it back and add necessary query params.
'/{}'.format(build_url(
self.url, edx_video_id=self.video.studio_id,
org=self.org, lang_code=self.video_source_language
)),
content_type='application/x-www-form-urlencoded',
data=urllib.urlencode(dict(file_id=self.file_id, status=state))
)
return response
def test_adapter_mro(context):
from pylogctx import log_adapter
class Parent(object):
pass
class Child(Parent):
pass
@log_adapter(Parent)
def parent_log_maker(instance):
return dict(parent=id(instance))
context.update(Child())
data = context.as_dict()
assert 'parent' in data
def test_adapter_manager(context):
from pylogctx import log_adapter
class Parent(object):
pass
class Child(Parent):
pass
@log_adapter(Parent)
def parent_log_maker(instance):
return dict(parent=id(instance))
with context(Child()):
data = context.as_dict()
assert 'parent' in data
