def test_attachment_methods():
    d = M.Discussion(shortname='test', name='test')
    t = M.Thread.new(discussion_id=d._id, subject='Test Thread')
    p = t.post('This is a post')
    p_att = p.attach('foo.text', StringIO('Hello, world!'),
                     discussion_id=d._id,
                     thread_id=t._id,
                     post_id=p._id)
    t_att = p.attach('foo2.text', StringIO('Hello, thread!'),
                     discussion_id=d._id,
                     thread_id=t._id)
    d_att = p.attach('foo3.text', StringIO('Hello, discussion!'),
                     discussion_id=d._id)

    ThreadLocalORMSession.flush_all()
    assert p_att.post == p
    assert p_att.thread == t
    assert p_att.discussion == d
    for att in (p_att, t_att, d_att):
        assert 'wiki/_discuss' in att.url()
        assert 'attachment/' in att.url()

    # Test notification in mail
    t = M.Thread.new(discussion_id=d._id, subject='Test comment notification')
    fs = FieldStorage()
    fs.name = 'file_info'
    fs.filename = 'fake.txt'
    fs.type = 'text/plain'
    fs.file = StringIO('this is the content of the fake file\n')
    p = t.post(text=u'test message', forum=None, subject='', file_info=fs)
    ThreadLocalORMSession.flush_all()
    n = M.Notification.query.get(
        subject=u'[test:wiki] Test comment notification')
    assert '\nAttachment: fake.txt (37 Bytes; text/plain)' in n.text

 def test_direct_sub(self):
     self._subscribe()
     self._post_notification(text='A')
     self._post_notification(text='B')
     ThreadLocalORMSession.flush_all()
     ThreadLocalORMSession.close_all()
     M.Mailbox.fire_ready()

def test_make_app_admin_only():
    h.set_context('test', 'wiki', neighborhood='Projects')
    anon = M.User.anonymous()
    dev = M.User.query.get(username='test-user')
    admin = M.User.query.get(username='test-admin')
    c.project.add_user(dev, ['Developer'])
    ThreadLocalORMSession.flush_all()
    Credentials.get().clear()
    assert has_access(c.app, 'read', user=anon)()
    assert has_access(c.app, 'read', user=dev)()
    assert has_access(c.app, 'read', user=admin)()
    assert not has_access(c.app, 'create', user=anon)()
    assert has_access(c.app, 'create', user=dev)()
    assert has_access(c.app, 'create', user=admin)()
    assert c.app.is_visible_to(anon)
    assert c.app.is_visible_to(dev)
    assert c.app.is_visible_to(admin)
    h.make_app_admin_only(c.app)
    ThreadLocalORMSession.flush_all()
    Credentials.get().clear()
    assert not has_access(c.app, 'read', user=anon)()
    assert not has_access(c.app, 'read', user=dev)()
    assert has_access(c.app, 'read', user=admin)()
    assert not has_access(c.app, 'create', user=anon)()
    assert not has_access(c.app, 'create', user=dev)()
    assert has_access(c.app, 'create', user=admin)()
    assert not c.app.is_visible_to(anon)
    assert not c.app.is_visible_to(dev)
    assert c.app.is_visible_to(admin)

 def test_refresh(self):
     committer_name = 'Test Committer'
     committer_email = '[email protected]'
     ci = mock.Mock()
     ci.authored.name = committer_name
     ci.committed.name = committer_name
     ci.committed.email = committer_email
     ci.author_url = '/u/test-committer/'
     self.repo._impl.commit = mock.Mock(return_value=ci)
     self.repo._impl.new_commits = mock.Mock(return_value=['foo%d' % i for i in range(100) ])
     self.repo._impl.all_commit_ids = mock.Mock(return_value=['foo%d' % i for i in range(100) ])
     self.repo.symbolics_for_commit = mock.Mock(return_value=[['master', 'branch'], []])
     def refresh_commit_info(oid, seen, lazy=False):
         M.repo.CommitDoc(dict(
                 authored=dict(
                     name=committer_name,
                     email=committer_email),
                 _id=oid)).m.insert()
     self.repo._impl.refresh_commit_info = refresh_commit_info
     _id = lambda oid: getattr(oid, '_id', str(oid))
     self.repo.shorthand_for_commit = lambda oid: '[' + _id(oid) + ']'
     self.repo.url_for_commit = lambda oid: '/ci/' + _id(oid) + '/'
     self.repo.refresh()
     ThreadLocalORMSession.flush_all()
     notifications = M.Notification.query.find().all()
     for n in notifications:
         if '100 new commits' in n.subject:
             assert "master,branch:  by %s http://localhost/ci/foo99" % committer_name in n.text
             break
     else:
         assert False, 'Did not find notification'
     assert M.Feed.query.find(dict(
         author_name=committer_name)).count() == 100

    def test_email(self):
        self._subscribe()  # as current user: test-admin
        user2 = M.User.query.get(username='test-user-2')
        self._subscribe(user=user2)
        self._post_notification()
        ThreadLocalORMSession.flush_all()

        assert_equal(M.Notification.query.get()['from_address'], '"Test Admin" <[email protected]>')
        assert_equal(M.Mailbox.query.find().count(), 2)

        M.MonQTask.run_ready()  # sends the notification out into "mailboxes", and from mailboxes into email tasks
        mboxes = M.Mailbox.query.find().all()
        assert_equal(len(mboxes), 2)
        assert_equal(len(mboxes[0].queue), 1)
        assert not mboxes[0].queue_empty
        assert_equal(len(mboxes[1].queue), 1)
        assert not mboxes[1].queue_empty

        email_tasks = M.MonQTask.query.find({'state': 'ready'}).all()
        assert_equal(len(email_tasks), 2)  # make sure both subscribers will get an email

        first_destinations = [e.kwargs['destinations'][0] for e in email_tasks]
        assert_in(str(c.user._id), first_destinations)
        assert_in(str(user2._id), first_destinations)
        assert_equal(email_tasks[0].kwargs['fromaddr'], '"Test Admin" <[email protected]>')
        assert_equal(email_tasks[1].kwargs['fromaddr'], '"Test Admin" <[email protected]>')
        assert_equal(email_tasks[0].kwargs['sender'], '[email protected]')
        assert_equal(email_tasks[1].kwargs['sender'], '[email protected]')
        assert email_tasks[0].kwargs['text'].startswith('Home modified by Test Admin')
        assert 'you indicated interest in ' in email_tasks[0].kwargs['text']

def test_hideawards_macro():
    p_nbhd = M.Neighborhood.query.get(name='Projects')

    app_config_id = ObjectId()
    award = M.Award(app_config_id=app_config_id)
    award.short = u'Award short'
    award.full = u'Award full'
    award.created_by_neighborhood_id = p_nbhd._id

    project = M.Project.query.get(
        neighborhood_id=p_nbhd._id, shortname=u'test')

    M.AwardGrant(
        award=award,
        award_url='http://award.org',
        comment='Winner!',
        granted_by_neighborhood=p_nbhd,
        granted_to_project=project)

    ThreadLocalORMSession.flush_all()

    with h.push_context(p_nbhd.neighborhood_project._id):
        r = g.markdown_wiki.convert('[[projects]]')
        assert_in('<div class="feature"> <a href="http://award.org" rel="nofollow" title="Winner!">'
                  'Award short</a> </div>',
                  squish_spaces(r))

        r = g.markdown_wiki.convert('[[projects show_awards_banner=False]]')
        assert_not_in('Award short', r)

 def test_user_search_for_disabled_user(self):
     user = M.User.by_username('test-admin')
     user.disabled = True
     ThreadLocalORMSession.flush_all()
     r = self.app.get('/p/test/user_search?term=test', status=200)
     j = json.loads(r.body)
     assert j == {'users': []}

def main():
    for chunk in utils.chunked_find(M.Project):
        for p in chunk:
            p.install_app('activity')

        ThreadLocalORMSession.flush_all()
        ThreadLocalORMSession.close_all()

    def fix_for_project(self, project):
        c.project = project
        base.log.info(
            'Checking discussion instances for each tracker in project %s' %
            project.shortname)
        trackers = [ac for ac in project.app_configs
                    if ac.tool_name.lower() == 'tickets']
        for tracker in trackers:
            base.log.info('Found tracker %s' % tracker)
            for ticket in Ticket.query.find({'app_config_id': tracker._id}):
                base.log.info('Processing ticket %s [#%s] %s'
                              % (ticket._id, ticket.ticket_num, ticket.summary))
                if ticket.discussion_thread.discussion.app_config_id != tracker._id:
                    # Some tickets were moved from this tracker,
                    # and Discussion instance for entire tracker was moved too.
                    # Should move it back.
                    base.log.info("Some tickets were moved from this tracker. "
                                  "Moving tracker's discussion instance back.")
                    ticket.discussion_thread.discussion.app_config_id = tracker._id

                if ticket.discussion_thread.discussion_id != tracker.discussion_id:
                    # Ticket was moved from another tracker.
                    # Should bind his comment thread to tracker's Discussion
                    base.log.info("Ticket was moved from another tracker. "
                                  "Bind ticket's comment thread to tracker's Discussion instance.")
                    ticket.discussion_thread.discussion_id = tracker.discussion_id
                    for post in ticket.discussion_thread.posts:
                        post.discussion_id = tracker.discussion_id

            ThreadLocalORMSession.flush_all()

def test_post_delete():
    d = M.Discussion(shortname="test", name="test")
    t = M.Thread.new(discussion_id=d._id, subject="Test Thread")
    p = t.post("This is a post")
    p.attach("foo.text", StringIO(""), discussion_id=d._id, thread_id=t._id, post_id=p._id)
    ThreadLocalORMSession.flush_all()
    p.delete()

    def test_export_with_attachments(self):
        project = M.Project.query.get(shortname='test')
        discussion = project.app_instance('discussion')
        post = Forum.query.get(shortname='general').sorted_threads[0].first_post
        test_file1 = FieldStorage()
        test_file1.name = 'file_info'
        test_file1.filename = 'test_file'
        test_file1.file = StringIO('test file1\n')
        post.add_attachment(test_file1)
        ThreadLocalORMSession.flush_all()

        f = tempfile.TemporaryFile()
        temp_dir = tempfile.mkdtemp()
        discussion.bulk_export(f, temp_dir, True)
        f.seek(0)
        discussion = json.loads(f.read())
        forums = sorted(discussion['forums'], key=lambda x: x['name'])
        threads = sorted(forums[0]['threads'], key=lambda x: x['subject'])
        file_path = os.path.join(
            'discussion',
            str(post.discussion_id),
            str(post.thread_id),
            post.slug,
            'test_file'
        )
        assert_equal(threads[0]['posts'][0]['attachments'][0]['path'], file_path)
        os.path.exists(file_path)

def test_attachment_methods():
    d = M.Discussion(shortname="test", name="test")
    t = M.Thread.new(discussion_id=d._id, subject="Test Thread")
    p = t.post("This is a post")
    p_att = p.attach("foo.text", StringIO("Hello, world!"), discussion_id=d._id, thread_id=t._id, post_id=p._id)
    t_att = p.attach("foo2.text", StringIO("Hello, thread!"), discussion_id=d._id, thread_id=t._id)
    d_att = p.attach("foo3.text", StringIO("Hello, discussion!"), discussion_id=d._id)

    ThreadLocalORMSession.flush_all()
    assert p_att.post == p
    assert p_att.thread == t
    assert p_att.discussion == d
    for att in (p_att, t_att, d_att):
        assert "wiki/_discuss" in att.url()
        assert "attachment/" in att.url()

    # Test notification in mail
    t = M.Thread.new(discussion_id=d._id, subject="Test comment notification")
    fs = FieldStorage()
    fs.name = "file_info"
    fs.filename = "fake.txt"
    fs.type = "text/plain"
    fs.file = StringIO("this is the content of the fake file\n")
    p = t.post(text=u"test message", forum=None, subject="", file_info=fs)
    ThreadLocalORMSession.flush_all()
    n = M.Notification.query.get(subject=u"[test:wiki] Test comment notification")
    assert "\nAttachment: fake.txt (37 Bytes; text/plain)" in n.text

 def test_refresh(self):
     ci = mock.Mock()
     ci.count_revisions=mock.Mock(return_value=100)
     ci.authored.name = 'Test Committer'
     ci.author_url = '/u/test-committer/'
     self.repo._impl.commit = mock.Mock(return_value=ci)
     self.repo._impl.new_commits = mock.Mock(return_value=['foo%d' % i for i in range(100) ])
     self.repo._impl.all_commit_ids = mock.Mock(return_value=['foo%d' % i for i in range(100) ])
     self.repo.symbolics_for_commit = mock.Mock(return_value=[['master', 'branch'], []])
     def refresh_commit_info(oid, seen, lazy=False):
         M.repo.CommitDoc(dict(
                 authored=dict(
                     name='Test Committer',
                     email='[email protected]'),
                 _id=oid)).m.insert()
     def set_heads():
         self.repo.heads = [ ming.base.Object(name='head', object_id='foo0', count=100) ]
     self.repo._impl.refresh_commit_info = refresh_commit_info
     self.repo._impl.refresh_heads = mock.Mock(side_effect=set_heads)
     self.repo.shorthand_for_commit = lambda oid: '[' + str(oid) + ']'
     self.repo.url_for_commit = lambda oid: '/ci/' + str(oid) + '/'
     self.repo.refresh()
     ThreadLocalORMSession.flush_all()
     notifications = M.Notification.query.find().all()
     for n in notifications:
         if '100 new commits' in n.subject:
             assert "master,branch:  by Test Committer http://localhost/#" in n.text
             break
     else:
         assert False, 'Did not find notification'
     assert M.Feed.query.find(dict(
         title='New commit',
         author_name='Test Committer')).count()

 def fork(self, project_id=None, mount_point=None, mount_label=None):
     # this shows the form and handles the submission
     security.require_authenticated()
     if not c.app.forkable: raise exc.HTTPNotFound
     from_repo = c.app.repo
     ThreadLocalORMSession.flush_all()
     ThreadLocalORMSession.close_all()
     from_project = c.project
     to_project = M.Project.query.get(_id=ObjectId(project_id))
     mount_label = mount_label or '%s - %s' % (c.project.name, c.app.config.options.mount_label)
     mount_point = (mount_point or from_project.shortname)
     if request.method != 'POST' or not mount_point:
         return dict(from_repo=from_repo,
                     user_project=c.user.private_project(),
                     mount_point=mount_point,
                     mount_label=mount_label)
     else:
         with h.push_config(c, project=to_project):
             if not to_project.database_configured:
                 to_project.configure_project(is_user_project=True)
             security.require(security.has_access(to_project, 'admin'))
             try:
                 to_project.install_app(
                     ep_name=from_repo.tool_name,
                     mount_point=mount_point,
                     mount_label=mount_label,
                     cloned_from_project_id=from_project._id,
                     cloned_from_repo_id=from_repo._id)
                 redirect(to_project.url()+mount_point+'/')
             except exc.HTTPRedirection:
                 raise
             except Exception, ex:
                 flash(str(ex), 'error')
                 redirect(request.referer)

def test_macro_include_no_extra_br():
    p_nbhd = M.Neighborhood.query.get(name='Projects')
    p_test = M.Project.query.get(shortname='test', neighborhood_id=p_nbhd._id)
    wiki = p_test.app_instance('wiki')
    with h.push_context(p_test._id, app_config_id=wiki.config._id):
        p = WM.Page.upsert(title='Include_1')
        p.text = 'included page 1'
        p.commit()
        p = WM.Page.upsert(title='Include_2')
        p.text = 'included page 2'
        p.commit()
        p = WM.Page.upsert(title='Include_3')
        p.text = 'included page 3'
        p.commit()
        ThreadLocalORMSession.flush_all()
        md = '[[include ref=Include_1]]\n[[include ref=Include_2]]\n[[include ref=Include_3]]'
        html = g.markdown_wiki.convert(md)

    expected_html = '''<div class="markdown_content"><p></p><div>
<div class="markdown_content"><p>included page 1</p></div>
</div>
<div>
<div class="markdown_content"><p>included page 2</p></div>
</div>
<div>
<div class="markdown_content"><p>included page 3</p></div>
</div>
<p></p></div>'''
    assert_equal(squish_spaces(html), squish_spaces(expected_html))

    def test_paged_diffs_with_detect_copies(self):
        # setup
        h.set_context('test', 'src-weird', neighborhood='Projects')
        repo_dir = pkg_resources.resource_filename(
            'forgegit', 'tests/data')
        repo = GM.Repository(
            name='weird-chars.git',
            fs_path=repo_dir,
            url_path='/src-weird/',
            tool='git',
            status='creating')
        repo.refresh()
        ThreadLocalORMSession.flush_all()
        ThreadLocalORMSession.close_all()

        diffs = repo.paged_diffs('346c52c1dddc729e2c2711f809336401f0ff925e')  # Test copy
        expected = {
            'added': [],
            'removed': [],
            'copied': [{'new': u'README.copy', 'old': u'README', 'ratio': 1.0}],
            'renamed': [],
            'changed': [u'README'],
            'total': 2,
        }
        assert_equals(diffs, expected)
        diffs = repo.paged_diffs('3cb2bbcd7997f89060a14fe8b1a363f01883087f')  # Test rename
        expected = {
            'added': [],
            'removed': [],
            'copied': [],
            'renamed': [{'new': u'README', 'old': u'README-copy.md', 'ratio': 1.0}],
            'changed': [],
            'total': 1,
        }
        assert_equals(diffs, expected)

def test_macro_include_permissions():
    p_nbhd = M.Neighborhood.query.get(name='Projects')
    p_test = M.Project.query.get(shortname='test', neighborhood_id=p_nbhd._id)
    wiki = p_test.app_instance('wiki')
    wiki2 = p_test.app_instance('wiki2')
    with h.push_context(p_test._id, app_config_id=wiki.config._id):
        p = WM.Page.upsert(title='CanRead')
        p.text = 'Can see this!'
        p.commit()
        ThreadLocalORMSession.flush_all()

    with h.push_context(p_test._id, app_config_id=wiki2.config._id):
        role = M.ProjectRole.by_name('*anonymous')._id
        read_perm = M.ACE.allow(role, 'read')
        acl = c.app.config.acl
        if read_perm in acl:
            acl.remove(read_perm)
        p = WM.Page.upsert(title='CanNotRead')
        p.text = 'Can not see this!'
        p.commit()
        ThreadLocalORMSession.flush_all()

    with h.push_context(p_test._id, app_config_id=wiki.config._id):
        c.user = M.User.anonymous()
        md = '[[include ref=CanRead]]\n[[include ref=wiki2:CanNotRead]]'
        html = g.markdown_wiki.convert(md)
        assert_in('Can see this!', html)
        assert_not_in('Can not see this!', html)
        assert_in("[[include: you don't have a read permission for wiki2:CanNotRead]]", html)

def test_macro_include_extra_br():
    p_nbhd = M.Neighborhood.query.get(name="Projects")
    p_test = M.Project.query.get(shortname="test", neighborhood_id=p_nbhd._id)
    wiki = p_test.app_instance("wiki")
    with h.push_context(p_test._id, app_config_id=wiki.config._id):
        p = WM.Page.upsert(title="Include_1")
        p.text = "included page 1"
        p.commit()
        p = WM.Page.upsert(title="Include_2")
        p.text = "included page 2"
        p.commit()
        p = WM.Page.upsert(title="Include_3")
        p.text = "included page 3"
        p.commit()
        ThreadLocalORMSession.flush_all()
        md = "[[include ref=Include_1]]\n[[include ref=Include_2]]\n[[include ref=Include_3]]"
        html = g.markdown_wiki.convert(md)

    expected_html = """
<div class="markdown_content">
<p>
<div><div class="markdown_content"><p>included page 1</p></div></div>
<div><div class="markdown_content"><p>included page 2</p></div></div>
<div><div class="markdown_content"><p>included page 3</p></div></div>
</p>
</div>
""".strip().replace(
        "\n", ""
    )
    assert html.strip().replace("\n", "") == expected_html, html

    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='[email protected]',
                text=u'This is a test',
                reply_to=g.noreply,
                subject=u'Test subject',
                message_id=h.gen_message_id())
            assert_equal(_client.sendmail.call_count, 1)
            return_path, rcpts, body = _client.sendmail.call_args[0]
            body = body.split('\n')
            assert_in('From: "Test Admin" <[email protected]>', body)

            c.user.disabled = True
            ThreadLocalORMSession.flush_all()
            mail_tasks.sendsimplemail(
                fromaddr=str(c.user._id),
                toaddr='[email protected]',
                text=u'This is a test',
                reply_to=g.noreply,
                subject=u'Test subject',
                message_id=h.gen_message_id())
            assert_equal(_client.sendmail.call_count, 2)
            return_path, rcpts, body = _client.sendmail.call_args[0]
            body = body.split('\n')
            assert_in('From: %s' % g.noreply, body)

    def test_export_with_attachments(self):
        project = M.Project.query.get(shortname='test')
        blog = project.app_instance('blog')
        with h.push_context('test', 'blog', neighborhood='Projects'):
            post = BM.BlogPost.new(
                title='Test title',
                text='test post',
                labels=['the firstlabel', 'the second label'],
                delete=None
            )
            ThreadLocalORMSession.flush_all()
            test_file1 = FieldStorage()
            test_file1.name = 'file_info'
            test_file1.filename = 'test_file'
            test_file1.file = StringIO('test file1\n')
            p = post.discussion_thread.add_post(text='test comment')
            p.add_multiple_attachments(test_file1)
            ThreadLocalORMSession.flush_all()
        f = tempfile.TemporaryFile()
        temp_dir = tempfile.mkdtemp()
        blog.bulk_export(f, temp_dir, True)
        f.seek(0)
        blog = json.loads(f.read())
        blog['posts'] = sorted(
            blog['posts'], key=lambda x: x['title'], reverse=True)

        file_path = 'blog/{}/{}/{}/test_file'.format(
            post._id,
            post.discussion_thread._id,
            list(post.discussion_thread.post_class().query.find())[0].slug
        )
        assert_equal(blog['posts'][0]['discussion_thread']['posts'][0]
                     ['attachments'][0]['path'], file_path)
        assert os.path.exists(os.path.join(temp_dir, file_path))