# 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 six
from base64 import b64encode
import logging
import importlib.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', __task_flush_immediately=True)
assert M.MonQTask.query.get(task_name='allura.tasks.event_tasks.event', args=['my_event1'])
g.post_event('my_event2', __task_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([str(importlib.resources.files('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()