import sys
import shutil
import unittest

import mock
from pylons import c, g

from ming.orm import FieldProperty, Mapper

from alluratest.controller import setup_basic_test, setup_global_objects

from allura import model as M
from allura.lib import helpers as h
from allura.lib.exceptions import CompoundError
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.tests import decorators as td
from allura.lib.decorators import event_handler, task

class TestEventTasks(unittest.TestCase):

    def setUp(self):
        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_compound_error(self):
        '''test_compound_exception -- make sure our multi-exception return works
        OK
        '''
        setup_basic_test()
        setup_global_objects()
        t = raise_exc.post()
        self.assertRaises(CompoundError, t)
        for x in range(10):
            assert ('assert %d' % x) in t.result

class TestIndexTasks(unittest.TestCase):

    def setUp(self):
        setup_basic_test()
        setup_global_objects()

    @td.with_wiki
    def test_add_artifacts(self):
        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()
        a = _TestArtifact.query.get(_shorthand_id='t3')
        assert len(a.backrefs) == 5, a.backrefs

    @td.with_wiki
    def test_del_artifacts(self):
        old_shortlinks = M.Shortlink.query.find().count()
        old_solr_size = len(g.solr.db)
        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()
        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"
        index_tasks.del_artifacts(ref_ids)
        M.main_orm_session.flush()
        M.main_orm_session.clear()
        new_shortlinks = M.Shortlink.query.find().count()
        new_solr_size = len(g.solr.db)
        assert old_shortlinks == new_shortlinks, 'Shortlinks not deleted'
        assert old_solr_size == new_solr_size, "Solr deletions didn't happen"

class TestMailTasks(unittest.TestCase):

    def setUp(self):
        setup_basic_test()
        setup_global_objects()

    def test_send_email(self):
        c.user = M.User.by_username('test-admin')
        with mock.patch_object(mail_tasks.smtp_client, 'sendmail') as f:
            mail_tasks.sendmail(
                str(c.user._id),
                [ str(c.user._id) ],
                'This is a test',
                'noreply@sf.net',
                'Test subject',
                h.gen_message_id())
            assert len(f.call_args_list)==3, f.call_args_list
            args,kwargs  = f.call_args_list[0]
            assert map(str, args[0]) == [ '"Test Admin" <None>' ]
            assert str(args[1]) == '"Test Admin" <None>'
            assert str(args[2]) == 'noreply@sf.net'
            assert args[3] == 'Test subject'
            assert '@' in args[4], args[4]
            assert args[5] == None
            assert 'This is a test' in str(args[6]), str(args[6])

    @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.sf.net'],
                'This is a mail message')
            args, kwargs = f.call_args
            assert args[0] == 'Page'
            assert len(args) == 2

class TestNotificationTasks(unittest.TestCase):

    def setUp(self):
        setup_basic_test()
        setup_global_objects()

    def test_delivers_messages(self):
        with mock.patch_object(M.Mailbox, 'deliver') as deliver:
            with mock.patch_object(M.Mailbox, 'fire_ready') as fire_ready:
                notification_tasks.notify('42', '52', 'none')
                assert deliver.called_with('42', '52', 'none')
                assert fire_ready.called_with()

class TestRepoTasks(unittest.TestCase):

    def setUp(self):
        setup_basic_test()
        self.setup_with_tools()

    @td.with_svn
    def setup_with_tools(self):
        h.set_context('test', 'src', neighborhood='Projects')

    def test_init(self):
        ns = M.Notification.query.find().count()
        with mock.patch_object(c.app.repo, 'init') as f:
            repo_tasks.init()
            M.main_orm_session.flush()
            assert f.called_with()
            assert ns + 1 == M.Notification.query.find().count()

    def test_clone(self):
        ns = M.Notification.query.find().count()
        with mock.patch_object(c.app.repo, 'init_as_clone') as f:
            repo_tasks.clone('foo', 'bar', 'baz')
            M.main_orm_session.flush()
            f.assert_called_with('foo', 'bar', 'baz')
            assert ns + 1 == M.Notification.query.find().count()

    def test_refresh(self):
        with mock.patch_object(c.app.repo, 'refresh') as f:
            repo_tasks.refresh()
            f.assert_called_with()

    def test_uninstall(self):
        with mock.patch_object(shutil, 'rmtree') as f:
            repo_tasks.uninstall()
            f.assert_called_with('/tmp/svn/p/test/src', ignore_errors=True)

@event_handler('my_event')
def _my_event(event_type, testcase, *args, **kwargs):
    testcase.called_with.append((args, kwargs))

@task
def raise_exc():
    errs = []
    for x in range(10):
        try:
            assert False, 'assert %d' % x
        except:
            errs.append(sys.exc_info())
    raise CompoundError(*errs)
    
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(_TestArtifact, self).index(),
            text=self.text)

Mapper.compile_all()
