#       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()