# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
import email.parser
import email.iterators
import operator
import shutil
from textwrap import dedent
import unittest
import six
from base64 import b64encode
import logging
import pkg_resources
import tg
import mock
from tg import tmpl_context as c, app_globals as g
import pytest
from ming.odm import FieldProperty, Mapper
from ming.odm import ThreadLocalODMSession
from testfixtures import LogCapture
from alluratest.controller import setup_basic_test, setup_global_objects, TestController
from allura import model as M
from allura.command.taskd import TaskdCommand
from allura.lib import helpers as h
from allura.lib.mail_util import MAX_MAIL_LINE_OCTETS, email_policy
from allura.tasks import event_tasks
from allura.tasks import index_tasks
from allura.tasks import mail_tasks
from allura.tasks import notification_tasks
from allura.tasks import repo_tasks
from allura.tasks import export_tasks
from allura.tasks import admin_tasks
from allura.tests import decorators as td
from allura.tests.exclude_from_rewrite_hook import raise_compound_exception
from allura.lib.decorators import event_handler, task
class TestRepoTasks:
def setup_method(self, method):
setup_basic_test()
setup_global_objects()
@mock.patch('allura.tasks.repo_tasks.c.app')
@mock.patch('allura.tasks.repo_tasks.g.post_event')
def test_clone_posts_event_on_failure(self, post_event, app):
fake_source_url = 'fake_source_url'
fake_traceback = 'fake_traceback'
app.repo.init_as_clone.side_effect = Exception(fake_traceback)
repo_tasks.clone(None, None, fake_source_url)
assert post_event.call_args[0][0] == 'repo_clone_task_failed'
assert post_event.call_args[0][1] == fake_source_url
assert post_event.call_args[0][2] is None
# ignore args[3] which is a traceback string
@mock.patch('allura.tasks.repo_tasks.session', autospec=True)
@mock.patch.object(M, 'MergeRequest')
def test_merge(self, MR, session):
mr = mock.Mock(_id='_id', activity_name='merge req', activity_url='/fake/url', activity_extras={}, node_id=None,
app=mock.Mock(activity_name='code merge', activity_url='/fake/url', activity_extras={},
node_id=None))
MR.query.get.return_value = mr
repo_tasks.merge(mr._id)
mr.app.repo.merge.assert_called_once_with(mr)
assert mr.status == 'merged'
session.assert_called_once_with(mr)
session.return_value.flush.assert_called_once_with(mr)
@mock.patch.object(M, 'MergeRequest')
def test_can_merge(self, MR):
mr = M.MergeRequest(_id='_id')
MR.query.get.return_value = mr
repo_tasks.can_merge(mr._id)
mr.app.repo.can_merge.assert_called_once_with(mr)
val = mr.app.repo.can_merge.return_value
mr.set_can_merge_cache.assert_called_once_with(val)
# used in test_post_event_from_within_task below
@task
def _task_that_creates_event(event_name,):
g.post_event(event_name)
# event does not get flushed to db right away (at end of task, ming middleware will flush it)
assert not M.MonQTask.query.get(task_name='allura.tasks.event_tasks.event', args=[event_name])
class TestEventTasks:
def setup_method(self, method):
setup_basic_test()
setup_global_objects()
self.called_with = []
def test_fire_event(self):
event_tasks.event('my_event', self, 1, 2, a=5)
assert self.called_with == [((1, 2), {'a': 5})], self.called_with
def test_post_event_explicit_flush(self):
g.post_event('my_event1', flush_immediately=True)
assert M.MonQTask.query.get(task_name='allura.tasks.event_tasks.event', args=['my_event1'])
g.post_event('my_event2', flush_immediately=False)
assert not M.MonQTask.query.get(task_name='allura.tasks.event_tasks.event', args=['my_event2'])
ThreadLocalODMSession.flush_all()
assert M.MonQTask.query.get(task_name='allura.tasks.event_tasks.event', args=['my_event2'])
def test_post_event_from_script(self):
# simulate post_event being called from a paster script command:
with mock.patch.dict(tg.request.environ, PATH_INFO='--script--'):
g.post_event('my_event3')
# event task is flushed to db right away:
assert M.MonQTask.query.get(task_name='allura.tasks.event_tasks.event', args=['my_event3'])
def test_post_event_from_within_task(self):
# instead of M.MonQTask.run_ready() run real 'taskd' so we get all the setup we need
taskd = TaskdCommand('taskd')
taskd.parse_args([pkg_resources.resource_filename('allura', '../test.ini')])
taskd.keep_running = True
taskd.restart_when_done = False
_task_that_creates_event.post('my_event4')
with mock.patch('allura.command.taskd.setproctitle') as setproctitle:
def stop_taskd_after_this_task(*args):
taskd.keep_running = False
setproctitle.side_effect = stop_taskd_after_this_task # avoid proc title change; useful hook to stop taskd
taskd.worker()
# after the initial task is done, the event task has been persisted:
assert M.MonQTask.query.get(task_name='allura.tasks.event_tasks.event', args=['my_event4'])
def test_compound_error(self):
t = raise_compound_exception.post()
with LogCapture(level=logging.ERROR) as lc, \
mock.patch.dict(tg.config, {'monq.raise_errors': False}): # match normal non-test behavior
t()
# lc.check() would be nice, but string is too detailed to check
assert lc.records[0].name == 'allura.model.monq_model'
msg = lc.records[0].getMessage()
assert "AssertionError('assert 0'" in msg
assert "AssertionError('assert 5'" in msg
assert ' on job <MonQTask ' in msg
assert ' (error) P:10 allura.tests.exclude_from_rewrite_hook.raise_compound_exception ' in msg
for x in range(10):
assert ('assert %d' % x) in t.result
class TestIndexTasks:
def setup_method(self, method):
setup_basic_test()
setup_global_objects()
def test_add_projects(self):
g.solr.db.clear()
old_solr_size = len(g.solr.db)
projects = M.Project.query.find().all()
index_tasks.add_projects.post([p._id for p in projects])
M.MonQTask.run_ready()
new_solr_size = len(g.solr.db)
assert old_solr_size + len(projects) == new_solr_size
@td.with_wiki
def test_del_projects(self):
projects = M.Project.query.find().all()
index_tasks.add_projects([p._id for p in projects])
with mock.patch('allura.tasks.index_tasks.g.solr') as solr:
index_tasks.del_projects([p.index_id() for p in projects])
assert solr.delete.call_count, 1
for project in projects:
assert project.index_id() in solr.delete.call_args[1]['q']
@td.with_wiki
def test_add_artifacts(self):
from allura.lib.search import find_shortlinks
with mock.patch('allura.lib.search.find_shortlinks') as find_slinks:
find_slinks.side_effect = lambda s: find_shortlinks(s)
old_shortlinks = M.Shortlink.query.find().count()
old_solr_size = len(g.solr.db)
artifacts = [_TestArtifact() for x in range(5)]
for i, a in enumerate(artifacts):
a._shorthand_id = 't%d' % i
a.text = 'This is a reference to [t3]'
arefs = [M.ArtifactReference.from_artifact(a) for a in artifacts]
ref_ids = [r._id for r in arefs]
M.artifact_orm_session.flush()
index_tasks.add_artifacts(ref_ids)
new_shortlinks = M.Shortlink.query.find().count()
new_solr_size = len(g.solr.db)
assert old_shortlinks + \
5 == new_shortlinks, 'Shortlinks not created'
assert old_solr_size + \
5 == new_solr_size, "Solr additions didn't happen"
M.main_orm_session.flush()
M.main_orm_session.clear()
t3 = _TestArtifact.query.get(_shorthand_id='t3')
assert len(t3.backrefs) == 5, t3.backrefs
assert (find_slinks.call_args_list ==
[mock.call(a.index().get('text')) for a in artifacts])
@td.with_wiki
@mock.patch('allura.tasks.index_tasks.g.solr')
def test_del_artifacts(self, solr):
old_shortlinks = M.Shortlink.query.find().count()
artifacts = [_TestArtifact(_shorthand_id='ta_%s' % x)
for x in range(5)]
M.artifact_orm_session.flush()
arefs = [M.ArtifactReference.from_artifact(a) for a in artifacts]
ref_ids = [r._id for r in arefs]
M.artifact_orm_session.flush()
index_tasks.add_artifacts(ref_ids)
M.main_orm_session.flush()
M.main_orm_session.clear()
new_shortlinks = M.Shortlink.query.find().count()
assert old_shortlinks + 5 == new_shortlinks, 'Shortlinks not created'
assert solr.add.call_count == 1
sort_key = operator.itemgetter('id')
assert (
sorted(solr.add.call_args[0][0], key=sort_key) ==
sorted((ref.artifact.solarize() for ref in arefs),
key=sort_key))
index_tasks.del_artifacts(ref_ids)
M.main_orm_session.flush()
M.main_orm_session.clear()
new_shortlinks = M.Shortlink.query.find().count()
assert old_shortlinks == new_shortlinks, 'Shortlinks not deleted'
solr_query = 'id:({})'.format(' || '.join(ref_ids))
solr.delete.assert_called_once_with(q=solr_query)
class TestMailTasks:
def setup_method(self, method):
setup_basic_test()
setup_global_objects()
# these tests go down through the mail_util.SMTPClient.sendmail method
# since usage is generally through the task, and not using mail_util
# directly
def test_send_email_ascii_with_user_lookup(self):
c.user = M.User.by_username('test-admin')
with mock.patch.object(mail_tasks.smtp_client, '_client') as _client:
mail_tasks.sendmail(
fromaddr=str(c.user._id),
destinations=[str(c.user._id)],
text='This is a test',
reply_to=g.noreply,
subject='Test subject',
message_id=h.gen_message_id())
assert _client.sendmail.call_count == 1
return_path, rcpts, body = _client.sendmail.call_args[0]
body = body.split(email_policy.linesep)
assert rcpts == [c.user.get_pref('email_address')]
assert 'Reply-To: %s' % g.noreply in body
assert 'From: "Test Admin" <test-admin@users.localhost>' in body
assert 'Subject: Test subject' in body
# plain
assert 'This is a test' in body
# html
assert (
'<div class="markdown_content"><p>This is a test</p></div>' in body)
def test_send_email_nonascii(self):
with mock.patch.object(mail_tasks.smtp_client, '_client') as _client:
mail_tasks.sendmail(
fromaddr='"По" <foo@bar.com>',
destinations=['blah@blah.com'],
text='Громады стройные теснятся',
reply_to=g.noreply,
subject='По оживлённым берегам',
message_id=h.gen_message_id())
assert _client.sendmail.call_count == 1
return_path, rcpts, body = _client.sendmail.call_args[0]
body = body.split(email_policy.linesep)
assert rcpts == ['blah@blah.com']
assert 'Reply-To: %s' % g.noreply in body
# The address portion must not be encoded, only the name portion can be.
unquoted_cyrillic_No = '=?utf-8?q?=D0=9F=D0=BE?=' # По
quoted_cyrillic_No = '=?utf-8?b?ItCf0L4i?=' # "По"
assert (f'From: {quoted_cyrillic_No} <foo@bar.com>' in body or
f'From: {unquoted_cyrillic_No} <foo@bar.com>' in body), body
assert (
'Subject: =?utf-8?b?0J/QviDQvtC20LjQstC70ZHQvdC90YvQvCDQsdC10YDQtdCz0LDQvA==?=' in body)
assert 'Content-Type: text/plain; charset="utf-8"' in body
assert 'Content-Transfer-Encoding: base64' in body
assert six.ensure_text(b64encode('Громады стройные теснятся'.encode())) in body
def test_send_email_with_disabled_user(self):
c.user = M.User.by_username('test-admin')
c.user.disabled = True
destination_user = M.User.by_username('test-user-1')
destination_user.preferences['email_address'] = 'user1@mail.com'
ThreadLocalODMSession.flush_all()
with mock.patch.object(mail_tasks.smtp_client, '_client') as _client:
mail_tasks.sendmail(
fromaddr=str(c.user._id),
destinations=[str(destination_user._id)],
text='This is a test',
reply_to=g.noreply,
subject='Test subject',
message_id=h.gen_message_id())
assert _client.sendmail.call_count == 1
return_path, rcpts, body = _client.sendmail.call_args[0]
body = body.split(email_policy.linesep)
assert 'From: %s' % g.noreply in body
def test_send_email_with_disabled_destination_user(self):
c.user = M.User.by_username('test-admin')
destination_user = M.User.by_username('test-user-1')
destination_user.preferences['email_address'] = 'user1@mail.com'
destination_user.disabled = True
ThreadLocalODMSession.flush_all()
with mock.patch.object(mail_tasks.smtp_client, '_client') as _client:
mail_tasks.sendmail(
fromaddr=str(c.user._id),
destinations=[str(destination_user._id)],
text='This is a test',
reply_to=g.noreply,
subject='Test subject',
message_id=h.gen_message_id())
assert _client.sendmail.call_count == 0
def test_sendsimplemail_with_disabled_user(self):
c.user = M.User.by_username('test-admin')
with mock.patch.object(mail_tasks.smtp_client, '_client') as _client:
mail_tasks.sendsimplemail(
fromaddr=str(c.user._id),
toaddr='test@mail.com',
text='This is a test',
reply_to=g.noreply,
subject='Test subject',
message_id=h.gen_message_id())
assert _client.sendmail.call_count == 1
return_path, rcpts, body = _client.sendmail.call_args[0]
body = body.split(email_policy.linesep)
assert 'From: "Test Admin" <test-admin@users.localhost>' in body
c.user.disabled = True
ThreadLocalODMSession.flush_all()
mail_tasks.sendsimplemail(
fromaddr=str(c.user._id),
toaddr='test@mail.com',
text='This is a test',
reply_to=g.noreply,
subject='Test subject',
message_id=h.gen_message_id())
assert _client.sendmail.call_count == 2
return_path, rcpts, body = _client.sendmail.call_args[0]
body = body.split(email_policy.linesep)
assert 'From: %s' % g.noreply in body
def test_email_sender_to_headers(self):
c.user = M.User.by_username('test-admin')
with mock.patch.object(mail_tasks.smtp_client, '_client') as _client:
mail_tasks.sendsimplemail(
fromaddr=str(c.user._id),
toaddr='test@mail.com',
text='This is a test',
reply_to=g.noreply,
subject='Test subject',
sender='tickets@test.p.domain.net',
message_id=h.gen_message_id())
assert _client.sendmail.call_count == 1
return_path, rcpts, body = _client.sendmail.call_args[0]
body = body.split(email_policy.linesep)
assert 'From: "Test Admin" <test-admin@users.localhost>' in body
assert 'Sender: tickets@test.p.domain.net' in body
assert 'To: test@mail.com' in body
_client.reset_mock()
mail_tasks.sendmail(
fromaddr=str(c.user._id),
destinations=[str(c.user._id)],
text='This is a test',
reply_to='123@tickets.test.p.domain.net',
subject='Test subject',
sender='tickets@test.p.domain.net',
message_id=h.gen_message_id())
assert _client.sendmail.call_count == 1
return_path, rcpts, body = _client.sendmail.call_args[0]
body = body.split(email_policy.linesep)
assert 'From: "Test Admin" <test-admin@users.localhost>' in body
assert 'Sender: tickets@test.p.domain.net' in body
assert 'To: 123@tickets.test.p.domain.net' in body
def test_email_references_header(self):
c.user = M.User.by_username('test-admin')
with mock.patch.object(mail_tasks.smtp_client, '_client') as _client:
mail_tasks.sendsimplemail(
fromaddr=str(c.user._id),
toaddr='test@mail.com',
text='This is a test',
reply_to=g.noreply,
subject='Test subject',
references=['a', 'b', 'c'],
message_id=h.gen_message_id())
assert _client.sendmail.call_count == 1
return_path, rcpts, body = _client.sendmail.call_args[0]
body = body.split(email_policy.linesep)
assert 'From: "Test Admin" <test-admin@users.localhost>' in body
assert 'References: <a> <b> <c>' in body
_client.reset_mock()
mail_tasks.sendmail(
fromaddr=str(c.user._id),
destinations=[str(c.user._id)],
text='This is a test',
reply_to=g.noreply,
subject='Test subject',
references='ref',
message_id=h.gen_message_id())
assert _client.sendmail.call_count == 1
return_path, rcpts, body = _client.sendmail.call_args[0]
body = body.split(email_policy.linesep)
assert 'From: "Test Admin" <test-admin@users.localhost>' in body
assert 'References: <ref>' in body
def test_cc(self):
c.user = M.User.by_username('test-admin')
with mock.patch.object(mail_tasks.smtp_client, '_client') as _client:
mail_tasks.sendsimplemail(
fromaddr=str(c.user._id),
toaddr='test@mail.com',
text='This is a test',
reply_to=g.noreply,
subject='Test subject',
cc='someone@example.com',
message_id=h.gen_message_id())
assert _client.sendmail.call_count == 1
return_path, rcpts, body = _client.sendmail.call_args[0]
assert 'CC: someone@example.com' in body
assert 'someone@example.com' in rcpts
def test_fromaddr_objectid_not_str(self):
c.user = M.User.by_username('test-admin')
with mock.patch.object(mail_tasks.smtp_client, '_client') as _client:
mail_tasks.sendsimplemail(
fromaddr=c.user._id,
toaddr='test@mail.com',
text='This is a test',
reply_to=g.noreply,
subject='Test subject',
message_id=h.gen_message_id())
assert _client.sendmail.call_count == 1
return_path, rcpts, body = _client.sendmail.call_args[0]
assert 'From: "Test Admin" <test-admin@users.localhost>' in body
@pytest.mark.parametrize('bodychars', [
'0123456789', # plain ascii is handled different since it doesn't necessarily need to be encoded
'Громады стро ',
])
def test_send_email_long_lines(self, bodychars):
with mock.patch.object(mail_tasks.smtp_client, '_client') as _client:
mail_tasks.sendsimplemail(
fromaddr='"По" <foo@bar.com>',
toaddr='blah@blah.com',
text=bodychars * 100,
reply_to=g.noreply,
subject='123451234512345' * 100,
references=['foo@example.com'] * 100, # needs to handle really long headers as well
message_id=h.gen_message_id())
return_path, rcpts, body = _client.sendmail.call_args[0]
body_lines = body.split(email_policy.linesep)
for line in body_lines:
assert len(line) <= MAX_MAIL_LINE_OCTETS
msg = email.parser.Parser().parsestr(body)
plain_subpart = next(email.iterators.typed_subpart_iterator(msg, 'text', 'plain'))
plain = plain_subpart.get_payload(decode=True).decode('utf-8')
html_subpart = next(email.iterators.typed_subpart_iterator(msg, 'text', 'html')).get_payload(decode=True)
html = html_subpart.decode('utf-8')
assert (bodychars + bodychars) in plain
assert f'<div class="markdown_content"><p>{bodychars}{bodychars}' in html
@td.with_wiki
def test_receive_email_ok(self):
c.user = M.User.by_username('test-admin')
import forgewiki
with mock.patch.object(forgewiki.wiki_main.ForgeWikiApp, 'handle_message') as f:
mail_tasks.route_email(
'0.0.0.0', c.user.email_addresses[0],
['Page@wiki.test.p.in.localhost'],
'This is a mail message')
args, kwargs = f.call_args
assert args[0] == 'Page'
assert len(args) == 2
@td.with_tool('test', 'Tickets', 'bugs')
def test_receive_autoresponse(self):
message = dedent('''\
Date: Wed, 30 Oct 2013 01:38:40 -0700
From: <test-admin@domain.net>
To: <1@bugs.test.p.in.localhost>
Message-ID: <super-unique-id>
Subject: Not here Re: Message notification
Precedence: bulk
X-Autoreply: yes
Auto-Submitted: auto-replied
I'm not here''')
import forgetracker
c.user = M.User.by_username('test-admin')
with mock.patch.object(forgetracker.tracker_main.ForgeTrackerApp, 'handle_message') as hm:
mail_tasks.route_email(
'0.0.0.0',
c.user.email_addresses[0],
['1@bugs.test.p.in.localhost'],
message)
assert hm.call_count == 0
@td.with_tool('test', 'Tickets', 'bugs')
def test_email_posting_disabled(self):
message = 'Hello, world!'
import forgetracker
c.user = M.User.by_username('test-admin')
with mock.patch.object(forgetracker.tracker_main.ForgeTrackerApp, 'handle_message') as hm:
c.app.config.options = {'AllowEmailPosting': False}
mail_tasks.route_email(
'0.0.0.0',
c.user.email_addresses[0],
['1@bugs.test.p.in.localhost'],
message)
assert hm.call_count == 0
class TestUserNotificationTasks(TestController):
def setup_method(self, method):
super().setup_method(method)
self.setup_with_tools()
@td.with_wiki
def setup_with_tools(self):
pass
def test_send_usermentions_notification(self):
c.user = M.User.by_username('test-admin')
test_user = M.User.by_username('test-user-1')
test_user.set_pref('mention_notifications', True)
M.MonQTask.query.remove()
d = dict(title='foo', text='Hey @test-user-1!')
self.app.post('/wiki/foo/update', params=d)
M.MonQTask.run_ready()
# check email notification
tasks = M.MonQTask.query.find(
dict(task_name='allura.tasks.mail_tasks.sendsimplemail')).all()
assert len(tasks) == 1
assert (tasks[0].kwargs['subject'] ==
'[test:wiki] Your name was mentioned')
assert tasks[0].kwargs['toaddr'] == 'test-user-1@allura.local'
assert tasks[0].kwargs['reply_to'] == g.noreply
text = tasks[0].kwargs['text']
assert 'Your name was mentioned at [foo]' in text
assert 'by Test Admin' in text
assert 'auth/subscriptions#notifications' in text
class TestNotificationTasks:
def setup_method(self, method):
setup_basic_test()
setup_global_objects()
def test_delivers_messages(self):
with mock.patch.object(M.Mailbox, 'deliver') as deliver, \
mock.patch.object(M.Mailbox, 'fire_ready') as fire_ready:
notification_tasks.notify('42', ['52'], 'none')
deliver.assert_called_with('42', ['52'], 'none')
fire_ready.assert_called_with()
@event_handler('my_event')
def _my_event(event_type, testcase, *args, **kwargs):
testcase.called_with.append((args, kwargs))
class _TestArtifact(M.Artifact):
_shorthand_id = FieldProperty(str)
text = FieldProperty(str)
def url(self):
return ''
def shorthand_id(self):
return getattr(self, '_shorthand_id', self._id)
def index(self):
return dict(
super().index(),
text=self.text)
class TestExportTasks:
def setup_method(self, method):
setup_basic_test()
setup_global_objects()
project = M.Project.query.get(shortname='test')
shutil.rmtree(project.bulk_export_path(tg.config['bulk_export_path']), ignore_errors=True)
def teardown_method(self, method):
project = M.Project.query.get(shortname='test')
shutil.rmtree(project.bulk_export_path(tg.config['bulk_export_path']), ignore_errors=True)
def test_bulk_export_filter_exportable(self):
exportable = mock.Mock(exportable=True)
not_exportable = mock.Mock(exportable=False)
BE = export_tasks.BulkExport()
assert BE.filter_exportable([None, exportable, not_exportable]) == [exportable]
def test_bulk_export_filter_successful(self):
BE = export_tasks.BulkExport()
assert BE.filter_successful(['foo', None, '0']) == ['foo', '0']
@mock.patch('allura.tasks.export_tasks.shutil')
@mock.patch('allura.tasks.export_tasks.zipdir')
@mock.patch.dict(tg.config, {'bulk_export_filename': '{project}.zip'})
@td.with_wiki
def test_bulk_export(self, zipdir, shutil):
M.MonQTask.query.remove()
export_tasks.bulk_export(['wiki'])
temp = '/tmp/bulk_export/p/test/test'
zipfn = '/tmp/bulk_export/p/test/test.zip'
zipdir.assert_called_with(temp, zipfn)
shutil.rmtree.assert_called_once_with(six.ensure_binary(temp))
# check notification
tasks = M.MonQTask.query.find(
dict(task_name='allura.tasks.mail_tasks.sendsimplemail')).all()
assert len(tasks) == 1
assert (tasks[0].kwargs['subject'] ==
'Bulk export for project test completed')
assert tasks[0].kwargs['fromaddr'] == '"Allura" <noreply@localhost>'
assert tasks[0].kwargs['reply_to'] == g.noreply
text = tasks[0].kwargs['text']
assert 'The bulk export for project test is completed.' in text
assert 'The following tools were exported:\n- wiki' in text
assert 'Sample instructions for test' in text
def test_bulk_export_status(self):
assert c.project.bulk_export_status() is None
export_tasks.bulk_export.post(['wiki'])
assert c.project.bulk_export_status() == 'busy'
class TestAdminTasks:
def test_install_app_docstring(self):
assert 'ep_name, mount_point=None' in admin_tasks.install_app.__doc__
Mapper.compile_all()