Merge remote-tracking branch 'refs/remotes/tilly-q/OPW-Moderation-Update'

Conflicts: mediagoblin/templates/mediagoblin/user_pages/user.html mediagoblin/tests/test_auth.py mediagoblin/tests/test_submission.py
2013-10-07 15:48:33 -05:00
parent 345b958871 0a24db84c5
commit 56c4ad89eb
59 changed files with 3410 additions and 207 deletions
--- a/mediagoblin/tests/test_api.py
+++ b/mediagoblin/tests/test_api.py
@@ -35,7 +35,8 @@ class TestAPI(object):
        self.db = mg_globals.database

        self.user_password = u'4cc355_70k3N'
-        self.user = fixture_add_user(u'joapi', self.user_password)
+        self.user = fixture_add_user(u'joapi', self.user_password,
+            privileges=[u'active',u'uploader'])

    def login(self, test_app):
        test_app.post(
--- a/mediagoblin/tests/test_auth.py
+++ b/mediagoblin/tests/test_auth.py
@@ -84,22 +84,26 @@ def test_register_views(test_app):
    template.clear_test_template_context()
    response = test_app.post(
        '/auth/register/', {
-            'username': u'happygirl',
-            'password': 'iamsohappy',
-            'email': 'happygrrl@example.org'})
+            'username': u'angrygirl',
+            'password': 'iamsoangry',
+            'email': 'angrygrrl@example.org'})
    response.follow()

    ## Did we redirect to the proper page?  Use the right template?
-    assert urlparse.urlsplit(response.location)[2] == '/u/happygirl/'
+    assert urlparse.urlsplit(response.location)[2] == '/u/angrygirl/'
    assert 'mediagoblin/user_pages/user_nonactive.html' in template.TEMPLATE_TEST_CONTEXT

    ## Make sure user is in place
    new_user = mg_globals.database.User.query.filter_by(
-        username=u'happygirl').first()
+        username=u'angrygirl').first()
    assert new_user
-    assert new_user.status == u'needs_email_verification'
-    assert new_user.email_verified == False

+    ## Make sure that the proper privileges are granted on registration
+
+    assert new_user.has_privilege(u'commenter')
+    assert new_user.has_privilege(u'uploader')
+    assert new_user.has_privilege(u'reporter')
+    assert not new_user.has_privilege(u'active')
    ## Make sure user is logged in
    request = template.TEMPLATE_TEST_CONTEXT[
        'mediagoblin/user_pages/user_nonactive.html']['request']
@@ -108,7 +112,7 @@ def test_register_views(test_app):
    ## Make sure we get email confirmation, and try verifying
    assert len(mail.EMAIL_TEST_INBOX) == 1
    message = mail.EMAIL_TEST_INBOX.pop()
-    assert message['To'] == 'happygrrl@example.org'
+    assert message['To'] == 'angrygrrl@example.org'
    email_context = template.TEMPLATE_TEST_CONTEXT[
        'mediagoblin/auth/verification_email.txt']
    assert email_context['verification_url'] in message.get_payload(decode=True)
@@ -130,10 +134,8 @@ def test_register_views(test_app):
    # assert context['verification_successful'] == True
    # TODO: Would be good to test messages here when we can do so...
    new_user = mg_globals.database.User.query.filter_by(
-        username=u'happygirl').first()
+        username=u'angrygirl').first()
    assert new_user
-    assert new_user.status == u'needs_email_verification'
-    assert new_user.email_verified == False

    ## Verify the email activation works
    template.clear_test_template_context()
@@ -144,10 +146,8 @@ def test_register_views(test_app):
    # assert context['verification_successful'] == True
    # TODO: Would be good to test messages here when we can do so...
    new_user = mg_globals.database.User.query.filter_by(
-        username=u'happygirl').first()
+        username=u'angrygirl').first()
    assert new_user
-    assert new_user.status == u'active'
-    assert new_user.email_verified == True

    # Uniqueness checks
    # -----------------
@@ -155,9 +155,9 @@ def test_register_views(test_app):
    template.clear_test_template_context()
    response = test_app.post(
        '/auth/register/', {
-            'username': u'happygirl',
-            'password': 'iamsohappy2',
-            'email': 'happygrrl2@example.org'})
+            'username': u'angrygirl',
+            'password': 'iamsoangry2',
+            'email': 'angrygrrl2@example.org'})

    context = template.TEMPLATE_TEST_CONTEXT[
        'mediagoblin/auth/register.html']
@@ -172,7 +172,7 @@ def test_register_views(test_app):
    template.clear_test_template_context()
    response = test_app.post(
        '/auth/forgot_password/',
-        {'username': u'happygirl'})
+        {'username': u'angrygirl'})
    response.follow()

    ## Did we redirect to the proper page?  Use the right template?
@@ -182,7 +182,7 @@ def test_register_views(test_app):
    ## Make sure link to change password is sent by email
    assert len(mail.EMAIL_TEST_INBOX) == 1
    message = mail.EMAIL_TEST_INBOX.pop()
-    assert message['To'] == 'happygrrl@example.org'
+    assert message['To'] == 'angrygrrl@example.org'
    email_context = template.TEMPLATE_TEST_CONTEXT[
        'mediagoblin/plugins/basic_auth/fp_verification_email.txt']
    #TODO - change the name of verification_url to something forgot-password-ish
@@ -212,7 +212,7 @@ def test_register_views(test_app):
    template.clear_test_template_context()
    response = test_app.post(
        '/auth/forgot_password/verify/', {
-            'password': 'iamveryveryhappy',
+            'password': 'iamveryveryangry',
            'token': parsed_get_params['token']})
    response.follow()
    assert 'mediagoblin/auth/login.html' in template.TEMPLATE_TEST_CONTEXT
@@ -221,8 +221,8 @@ def test_register_views(test_app):
    template.clear_test_template_context()
    response = test_app.post(
        '/auth/login/', {
-            'username': u'happygirl',
-            'password': 'iamveryveryhappy'})
+            'username': u'angrygirl',
+            'password': 'iamveryveryangry'})

    # User should be redirected
    response.follow()
@@ -235,7 +235,7 @@ def test_authentication_views(test_app):
    Test logging in and logging out
    """
    # Make a new user
-    test_user = fixture_add_user(active_user=False)
+    test_user = fixture_add_user()


    # Get login
@@ -332,7 +332,6 @@ def test_authentication_views(test_app):
            'next' : '/u/chris/'})
    assert urlparse.urlsplit(response.location)[2] == '/u/chris/'

-
@pytest.fixture()
 def authentication_disabled_app(request):
    return get_app(
@@ -344,6 +343,7 @@ def authentication_disabled_app(request):

 def test_authentication_disabled_app(authentication_disabled_app):
    # app.auth should = false
+    assert mg_globals
    assert mg_globals.app.auth is False

    # Try to visit register page
--- a/mediagoblin/tests/test_edit.py
+++ b/mediagoblin/tests/test_edit.py
@@ -27,7 +27,8 @@ class TestUserEdit(object):
    def setup(self):
        # set up new user
        self.user_password = u'toast'
-        self.user = fixture_add_user(password = self.user_password)
+        self.user = fixture_add_user(password = self.user_password,
+                               privileges=[u'active'])

    def login(self, test_app):
        test_app.post(
@@ -52,7 +53,8 @@ class TestUserEdit(object):
        # deleted too. Perhaps in submission test?

        #Restore user at end of test
-        self.user = fixture_add_user(password = self.user_password)
+        self.user = fixture_add_user(password = self.user_password,
+                               privileges=[u'active'])
        self.login(test_app)


@@ -80,7 +82,8 @@ class TestUserEdit(object):
        assert test_user.url == u'http://dustycloud.org/'

        # change a different user than the logged in (should fail with 403)
-        fixture_add_user(username=u"foo")
+        fixture_add_user(username=u"foo",
+                         privileges=[u'active'])
        res = test_app.post(
            '/u/foo/edit/', {
                'bio': u'I love toast!',
--- a/mediagoblin/tests/test_modelmethods.py
+++ b/mediagoblin/tests/test_modelmethods.py
@@ -18,7 +18,7 @@
 # methods, and so it makes sense to test them here.

 from mediagoblin.db.base import Session
-from mediagoblin.db.models import MediaEntry
+from mediagoblin.db.models import MediaEntry, User, Privilege

 from mediagoblin.tests.tools import fixture_add_user

@@ -47,7 +47,7 @@ class TestMediaEntrySlugs(object):
        entry.id = this_id
        entry.uploader = uploader or self.chris_user.id
        entry.media_type = u'image'
-        
+
        if save:
            entry.save()

@@ -99,7 +99,7 @@ class TestMediaEntrySlugs(object):
                u"Beware, I exist!!", this_id=9000, save=False)
            entry.generate_slug()
            assert entry.slug == u"beware-i-exist-test"
-            
+
        _real_test()

    def test_existing_slug_cant_use_id_extra_junk(self, test_app):
@@ -151,6 +151,44 @@ class TestMediaEntrySlugs(object):
        qbert_entry.generate_slug()
        assert qbert_entry.slug is None

+class TestUserHasPrivilege:
+    def _setup(self):
+        fixture_add_user(u'natalie',
+            privileges=[u'admin',u'moderator',u'active'])
+        fixture_add_user(u'aeva',
+            privileges=[u'moderator',u'active'])
+        self.natalie_user = User.query.filter(
+            User.username==u'natalie').first()
+        self.aeva_user = User.query.filter(
+            User.username==u'aeva').first()
+
+    def test_privilege_added_correctly(self, test_app):
+        self._setup()
+        admin = Privilege.query.filter(
+            Privilege.privilege_name == u'admin').one()
+        # first make sure the privileges were added successfully
+
+        assert admin in self.natalie_user.all_privileges
+        assert admin not in self.aeva_user.all_privileges
+
+    def test_user_has_privilege_one(self, test_app):
+        self._setup()
+
+        # then test out the user.has_privilege method for one privilege
+        assert not self.natalie_user.has_privilege(u'commenter')
+        assert self.aeva_user.has_privilege(u'active')
+
+
+    def test_user_has_privileges_multiple(self, test_app):
+        self._setup()
+
+        # when multiple args are passed to has_privilege,  the method returns
+        # True if the user has ANY of the privileges
+        assert self.natalie_user.has_privilege(u'admin',u'commenter')
+        assert self.aeva_user.has_privilege(u'moderator',u'active')
+        assert not self.natalie_user.has_privilege(u'commenter',u'uploader')
+
+

 def test_media_data_init(test_app):
    Session.rollback()
@@ -165,3 +203,4 @@ def test_media_data_init(test_app):
        obj_in_session += 1
        print repr(obj)
    assert obj_in_session == 0
+
--- a/mediagoblin/tests/test_moderation.py
+++ b/mediagoblin/tests/test_moderation.py
@@ -0,0 +1,242 @@
+# GNU MediaGoblin -- federated, autonomous media hosting
+# Copyright (C) 2011, 2012 MediaGoblin contributors.  See AUTHORS.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import pytest
+
+from mediagoblin.tests.tools import (fixture_add_user,
+            fixture_add_comment_report, fixture_add_comment)
+from mediagoblin.db.models import User, CommentReport, MediaComment, UserBan
+from mediagoblin.tools import template, mail
+from webtest import AppError
+
+class TestModerationViews:
+    @pytest.fixture(autouse=True)
+    def _setup(self, test_app):
+        self.test_app = test_app
+
+        fixture_add_user(u'admin',
+            privileges=[u'admin',u'active'])
+        fixture_add_user(u'moderator',
+            privileges=[u'moderator',u'active'])
+        fixture_add_user(u'regular',
+            privileges=[u'active',u'commenter'])
+        self.query_for_users()
+
+    def login(self, username):
+        self.test_app.post(
+            '/auth/login/', {
+                'username': username,
+                'password': 'toast'})
+        self.query_for_users()
+
+    def logout(self):
+        self.test_app.get('/auth/logout/')
+        self.query_for_users()
+
+    def query_for_users(self):
+        self.admin_user = User.query.filter(User.username==u'admin').first()
+        self.mod_user = User.query.filter(User.username==u'moderator').first()
+        self.user = User.query.filter(User.username==u'regular').first()
+
+    def do_post(self, data, *context_keys, **kwargs):
+        url = kwargs.pop('url', '/submit/')
+        do_follow = kwargs.pop('do_follow', False)
+        template.clear_test_template_context()
+        response = self.test_app.post(url, data, **kwargs)
+        if do_follow:
+            response.follow()
+        context_data = template.TEMPLATE_TEST_CONTEXT
+        for key in context_keys:
+            context_data = context_data[key]
+        return response, context_data
+
+    def testGiveOrTakeAwayPrivileges(self):
+        self.login(u'admin')
+        # First, test an admin taking away a privilege from a user
+        #----------------------------------------------------------------------
+        response, context = self.do_post({'privilege_name':u'commenter'},
+            url='/mod/users/{0}/privilege/'.format(self.user.username))
+        assert response.status == '302 FOUND'
+        self.query_for_users()
+        assert not self.user.has_privilege(u'commenter')
+
+        # Then, test an admin giving a privilege to a user
+        #----------------------------------------------------------------------
+        response, context = self.do_post({'privilege_name':u'commenter'},
+            url='/mod/users/{0}/privilege/'.format(self.user.username))
+        assert response.status == '302 FOUND'
+        self.query_for_users()
+        assert self.user.has_privilege(u'commenter')
+
+        # Then, test a mod trying to take away a privilege from a user
+        # they are not allowed to do this, so this will raise an error
+        #----------------------------------------------------------------------
+        self.logout()
+        self.login(u'moderator')
+
+        with pytest.raises(AppError) as excinfo:
+            response, context = self.do_post({'privilege_name':u'commenter'},
+                url='/mod/users/{0}/privilege/'.format(self.user.username))
+        assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+        self.query_for_users()
+
+        assert self.user.has_privilege(u'commenter')
+
+    def testReportResolution(self):
+        self.login(u'moderator')
+
+        # First, test a moderators taking away a user's privilege in response
+        # to a reported comment
+        #----------------------------------------------------------------------
+        fixture_add_comment_report(reported_user=self.user)
+        comment_report = CommentReport.query.filter(
+            CommentReport.reported_user==self.user).first()
+
+        response = self.test_app.get('/mod/reports/{0}/'.format(
+            comment_report.id))
+        assert response.status == '200 OK'
+        self.query_for_users()
+        comment_report = CommentReport.query.filter(
+            CommentReport.reported_user==self.user).first()
+
+        response, context = self.do_post({'action_to_resolve':[u'takeaway'],
+            'take_away_privileges':[u'commenter'],
+            'targeted_user':self.user.id},
+            url='/mod/reports/{0}/'.format(comment_report.id))
+
+        self.query_for_users()
+        comment_report = CommentReport.query.filter(
+            CommentReport.reported_user==self.user).first()
+        assert response.status == '302 FOUND'
+        assert not self.user.has_privilege(u'commenter')
+        assert comment_report.is_archived_report() is True
+
+        fixture_add_comment_report(reported_user=self.user)
+        comment_report = CommentReport.query.filter(
+            CommentReport.reported_user==self.user).first()
+
+        # Then, test a moderator sending an email to a user in response to a
+        # reported comment
+        #----------------------------------------------------------------------
+        self.query_for_users()
+
+        response, context = self.do_post({'action_to_resolve':[u'sendmessage'],
+            'message_to_user':'This is your last warning, regular....',
+            'targeted_user':self.user.id},
+            url='/mod/reports/{0}/'.format(comment_report.id))
+
+        self.query_for_users()
+        comment_report = CommentReport.query.filter(
+            CommentReport.reported_user==self.user).first()
+        assert response.status == '302 FOUND'
+        assert mail.EMAIL_TEST_MBOX_INBOX ==  [{'to': [u'regular@example.com'],
+            'message': 'Content-Type: text/plain; charset="utf-8"\n\
+MIME-Version: 1.0\nContent-Transfer-Encoding: base64\nSubject: Warning from- \
+moderator \nFrom: notice@mediagoblin.example.org\nTo: regular@example.com\n\n\
+VGhpcyBpcyB5b3VyIGxhc3Qgd2FybmluZywgcmVndWxhci4uLi4=\n',
+            'from': 'notice@mediagoblin.example.org'}]
+        assert comment_report.is_archived_report() is True
+
+        # Then test a moderator banning a user AND a moderator deleting the
+        # offending comment. This also serves as a test for taking multiple
+        # actions to resolve a report
+        #----------------------------------------------------------------------
+        self.query_for_users()
+        fixture_add_comment(author=self.user.id,
+            comment=u'Comment will be removed')
+        test_comment = MediaComment.query.filter(
+            MediaComment.author==self.user.id).first()
+        fixture_add_comment_report(comment=test_comment,
+            reported_user=self.user)
+        comment_report = CommentReport.query.filter(
+            CommentReport.comment==test_comment).filter(
+            CommentReport.resolved==None).first()
+
+        response, context = self.do_post(
+            {'action_to_resolve':[u'userban', u'delete'],
+            'targeted_user':self.user.id,
+            'why_user_was_banned':u'',
+            'user_banned_until':u''},
+            url='/mod/reports/{0}/'.format(comment_report.id))
+        assert response.status == '302 FOUND'
+        self.query_for_users()
+        test_user_ban = UserBan.query.filter(
+            UserBan.user_id == self.user.id).first()
+        assert test_user_ban is not None
+        test_comment = MediaComment.query.filter(
+            MediaComment.author==self.user.id).first()
+        assert test_comment is None
+
+        # Then, test what happens when a moderator attempts to punish an admin
+        # from a reported comment on an admin.
+        #----------------------------------------------------------------------
+        fixture_add_comment_report(reported_user=self.admin_user)
+        comment_report = CommentReport.query.filter(
+            CommentReport.reported_user==self.admin_user).filter(
+            CommentReport.resolved==None).first()
+
+        response, context = self.do_post({'action_to_resolve':[u'takeaway'],
+            'take_away_privileges':[u'active'],
+            'targeted_user':self.admin_user.id},
+            url='/mod/reports/{0}/'.format(comment_report.id))
+        self.query_for_users()
+
+        assert response.status == '200 OK'
+        assert self.admin_user.has_privilege(u'active')
+
+    def testAllModerationViews(self):
+        self.login(u'moderator')
+        username = self.user.username
+        self.query_for_users()
+        fixture_add_comment_report(reported_user=self.admin_user)
+        response = self.test_app.get('/mod/reports/')
+        assert response.status == "200 OK"
+
+        response = self.test_app.get('/mod/reports/1/')
+        assert response.status == "200 OK"
+
+        response = self.test_app.get('/mod/users/')
+        assert response.status == "200 OK"
+
+        user_page_url = '/mod/users/{0}/'.format(username)
+        response = self.test_app.get(user_page_url)
+        assert response.status == "200 OK"
+
+        self.test_app.get('/mod/media/')
+        assert response.status == "200 OK"
+
+    def testBanUnBanUser(self):
+        self.login(u'admin')
+        username = self.user.username
+        user_id = self.user.id
+        ban_url = '/mod/users/{0}/ban/'.format(username)
+        response, context = self.do_post({
+            'user_banned_until':u'',
+            'why_user_was_banned':u'Because I said so'},
+            url=ban_url)
+
+        assert response.status == "302 FOUND"
+        user_banned = UserBan.query.filter(UserBan.user_id==user_id).first()
+        assert user_banned is not None
+        assert user_banned.expiration_date is None
+        assert user_banned.reason == u'Because I said so'
+
+        response, context = self.do_post({},
+            url=ban_url)
+
+        assert response.status == "302 FOUND"
+        user_banned = UserBan.query.filter(UserBan.user_id==user_id).first()
+        assert user_banned is None
--- a/mediagoblin/tests/test_notifications.py
+++ b/mediagoblin/tests/test_notifications.py
@@ -38,7 +38,7 @@ class TestNotifications:

        # TODO: Possibly abstract into a decorator like:
        # @as_authenticated_user('chris')
-        self.test_user = fixture_add_user()
+        self.test_user = fixture_add_user(privileges=[u'active',u'commenter'])

        self.current_user = None

@@ -75,7 +75,10 @@ class TestNotifications:

        '''
        user = fixture_add_user('otherperson', password='nosreprehto',
-                                wants_comment_notification=wants_email)
+                                wants_comment_notification=wants_email,
+                                privileges=[u'active',u'commenter'])
+
+        assert user.wants_comment_notification == wants_email

        user_id = user.id

@@ -124,6 +127,7 @@ otherperson@example.com\n\nSGkgb3RoZXJwZXJzb24sCmNocmlzIGNvbW1lbnRlZCBvbiB5b3VyI
        else:
            assert mail.EMAIL_TEST_MBOX_INBOX == []

+
        # Save the ids temporarily because of DetachedInstanceError
        notification_id = notification.id
        comment_id = notification.subject.id
@@ -153,7 +157,8 @@ otherperson@example.com\n\nSGkgb3RoZXJwZXJzb24sCmNocmlzIGNvbW1lbnRlZCBvbiB5b3VyI
    def test_mark_all_comment_notifications_seen(self):
        """ Test that mark_all_comments_seen works"""

-        user = fixture_add_user('otherperson', password='nosreprehto')
+        user = fixture_add_user('otherperson', password='nosreprehto',
+                        privileges=[u'active'])

        media_entry = fixture_media_entry(uploader=user.id, state=u'processed')

--- a/mediagoblin/tests/test_oauth2.py
+++ b/mediagoblin/tests/test_oauth2.py
@@ -38,7 +38,8 @@ class TestOAuth(object):
        self.pman = pluginapi.PluginManager()

        self.user_password = u'4cc355_70k3N'
-        self.user = fixture_add_user(u'joauth', self.user_password)
+        self.user = fixture_add_user(u'joauth', self.user_password,
+            privileges=[u'active'])

        self.login()

--- a/mediagoblin/tests/test_openid.py
+++ b/mediagoblin/tests/test_openid.py
@@ -238,7 +238,7 @@ class TestOpenIDPlugin(object):
    def test_add_delete(self, openid_plugin_app):
        """Test adding and deleting openids"""
        # Add user
-        test_user = fixture_add_user(password='')
+        test_user = fixture_add_user(password='', privileges=[u'active'])
        openid = OpenIDUserURL()
        openid.openid_url = 'http://real.myopenid.com'
        openid.user_id = test_user.id
--- a/mediagoblin/tests/test_persona.py
+++ b/mediagoblin/tests/test_persona.py
@@ -22,6 +22,7 @@ pytest.importorskip("requests")

 from mediagoblin import mg_globals
 from mediagoblin.db.base import Session
+from mediagoblin.db.models import Privilege
 from mediagoblin.tests.tools import get_app
 from mediagoblin.tools import template

@@ -112,8 +113,9 @@ class TestPersonaPlugin(object):
            # Get user and detach from session
            test_user = mg_globals.database.User.query.filter_by(
                username=u'chris').first()
-            test_user.email_verified = True
-            test_user.status = u'active'
+            active_privilege = Privilege.query.filter(
+                Privilege.privilege_name==u'active').first()
+            test_user.all_privileges.append(active_privilege)
            test_user.save()
            test_user = mg_globals.database.User.query.filter_by(
                username=u'chris').first()
--- a/mediagoblin/tests/test_privileges.py
+++ b/mediagoblin/tests/test_privileges.py
@@ -0,0 +1,205 @@
+# GNU MediaGoblin -- federated, autonomous media hosting
+# Copyright (C) 2011, 2012 MediaGoblin contributors.  See AUTHORS.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import pytest
+from datetime import date, timedelta
+from webtest import AppError
+
+from mediagoblin.tests.tools import fixture_add_user, fixture_media_entry
+
+from mediagoblin.db.models import User, UserBan
+from mediagoblin.tools import template
+
+from .resources import GOOD_JPG
+
+class TestPrivilegeFunctionality:
+
+    @pytest.fixture(autouse=True)
+    def _setup(self, test_app):
+        self.test_app = test_app
+
+        fixture_add_user(u'alex',
+            privileges=[u'admin',u'active'])
+        fixture_add_user(u'meow',
+            privileges=[u'moderator',u'active',u'reporter'])
+        fixture_add_user(u'natalie',
+            privileges=[u'active'])
+        self.query_for_users()
+
+    def login(self, username):
+        self.test_app.post(
+            '/auth/login/', {
+                'username': username,
+                'password': 'toast'})
+        self.query_for_users()
+
+    def logout(self):
+        self.test_app.get('/auth/logout/')
+        self.query_for_users()
+
+    def do_post(self, data, *context_keys, **kwargs):
+        url = kwargs.pop('url', '/submit/')
+        do_follow = kwargs.pop('do_follow', False)
+        template.clear_test_template_context()
+        response = self.test_app.post(url, data, **kwargs)
+        if do_follow:
+            response.follow()
+        context_data = template.TEMPLATE_TEST_CONTEXT
+        for key in context_keys:
+            context_data = context_data[key]
+        return response, context_data
+
+    def query_for_users(self):
+        self.admin_user = User.query.filter(User.username==u'alex').first()
+        self.mod_user = User.query.filter(User.username==u'meow').first()
+        self.user = User.query.filter(User.username==u'natalie').first()
+
+    def testUserBanned(self):
+        self.login(u'natalie')
+        uid = self.user.id
+        # First, test what happens when a user is banned indefinitely
+        #----------------------------------------------------------------------
+        user_ban = UserBan(user_id=uid,
+            reason=u'Testing whether user is banned',
+            expiration_date=None)
+        user_ban.save()
+
+        response = self.test_app.get('/')
+        assert response.status == "200 OK"
+        assert "You are Banned" in response.body
+        # Then test what happens when that ban has an expiration date which
+        # hasn't happened yet
+        #----------------------------------------------------------------------
+        user_ban = UserBan.query.get(uid)
+        user_ban.delete()
+        user_ban = UserBan(user_id=uid,
+            reason=u'Testing whether user is banned',
+            expiration_date= date.today() + timedelta(days=20))
+        user_ban.save()
+
+        response = self.test_app.get('/')
+        assert response.status == "200 OK"
+        assert "You are Banned" in response.body
+
+        # Then test what happens when that ban has an expiration date which
+        # has already happened
+        #----------------------------------------------------------------------
+        user_ban = UserBan.query.get(uid)
+        user_ban.delete()
+        exp_date = date.today() - timedelta(days=20)
+        user_ban = UserBan(user_id=uid,
+            reason=u'Testing whether user is banned',
+            expiration_date= exp_date)
+        user_ban.save()
+
+        response = self.test_app.get('/')
+        assert response.status == "302 FOUND"
+        assert not "You are Banned" in response.body
+
+    def testVariousPrivileges(self):
+        # The various actions that require privileges (ex. reporting,
+        # commenting, moderating...) are tested in other tests. This method
+        # will be used to ensure that those actions are impossible for someone
+        # without the proper privileges.
+        # For other tests that show what happens when a user has the proper
+        # privileges, check out:
+        #       tests/test_moderation.py                moderator
+        #       tests/test_notifications.py             commenter
+        #       tests/test_reporting.py                 reporter
+        #       tests/test_submission.py                uploader
+        #----------------------------------------------------------------------
+        self.login(u'natalie')
+
+        # First test the get and post requests of submission/uploading
+        #----------------------------------------------------------------------
+        with pytest.raises(AppError) as excinfo:
+            response = self.test_app.get('/submit/')
+        assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+
+
+        with pytest.raises(AppError) as excinfo:
+            response = self.do_post({'upload_files':[('file',GOOD_JPG)],
+                'title':u'Normal Upload 1'},
+                url='/submit/')
+        assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+
+        # Test that a user cannot comment without the commenter privilege
+        #----------------------------------------------------------------------
+        self.query_for_users()
+
+        media_entry = fixture_media_entry(uploader=self.admin_user.id,
+            state=u'processed')
+
+        media_entry_id = media_entry.id
+        media_uri_id = '/u/{0}/m/{1}/'.format(self.admin_user.username,
+                                              media_entry.id)
+        media_uri_slug = '/u/{0}/m/{1}/'.format(self.admin_user.username,
+                                                media_entry.slug)
+        response = self.test_app.get(media_uri_slug)
+        assert not "Add a comment" in response.body
+
+        self.query_for_users()
+        with pytest.raises(AppError) as excinfo:
+            response = self.test_app.post(
+                media_uri_id + 'comment/add/',
+                {'comment_content': u'Test comment #42'})
+        assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+
+        # Test that a user cannot report without the reporter privilege
+        #----------------------------------------------------------------------
+        with pytest.raises(AppError) as excinfo:
+            response = self.test_app.get(media_uri_slug+"report/")
+        assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+
+        with pytest.raises(AppError) as excinfo:
+            response = self.do_post(
+                {'report_reason':u'Testing Reports #1',
+                'reporter_id':u'3'},
+                url=(media_uri_slug+"report/"))
+        assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+
+        # Test that a user cannot access the moderation pages w/o moderator
+        # or admin privileges
+        #----------------------------------------------------------------------
+        with pytest.raises(AppError) as excinfo:
+            response = self.test_app.get("/mod/users/")
+        assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+
+        with pytest.raises(AppError) as excinfo:
+            response = self.test_app.get("/mod/reports/")
+        assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+
+        with pytest.raises(AppError) as excinfo:
+            response = self.test_app.get("/mod/media/")
+        assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+
+        with pytest.raises(AppError) as excinfo:
+            response = self.test_app.get("/mod/users/1/")
+        assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+
+        with pytest.raises(AppError) as excinfo:
+            response = self.test_app.get("/mod/reports/1/")
+        assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
+
+        self.query_for_users()
+
+        with pytest.raises(AppError) as excinfo:
+            response, context = self.do_post({'action_to_resolve':[u'takeaway'],
+                'take_away_privileges':[u'active'],
+                'targeted_user':self.admin_user.id},
+                url='/mod/reports/1/')
+            self.query_for_users()
+        assert 'Bad response: 403 FORBIDDEN' in str(excinfo)
--- a/mediagoblin/tests/test_reporting.py
+++ b/mediagoblin/tests/test_reporting.py
@@ -0,0 +1,167 @@
+# GNU MediaGoblin -- federated, autonomous media hosting
+# Copyright (C) 2011, 2012 MediaGoblin contributors.  See AUTHORS.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import pytest
+
+from mediagoblin.tools import template
+from mediagoblin.tests.tools import (fixture_add_user, fixture_media_entry,
+        fixture_add_comment, fixture_add_comment_report)
+from mediagoblin.db.models import (MediaReport, CommentReport, User,
+    MediaComment)
+
+
+class TestReportFiling:
+    @pytest.fixture(autouse=True)
+    def _setup(self, test_app):
+        self.test_app = test_app
+
+        fixture_add_user(u'allie',
+            privileges=[u'reporter',u'active'])
+        fixture_add_user(u'natalie',
+            privileges=[u'active', u'moderator'])
+
+    def login(self, username):
+        self.test_app.post(
+            '/auth/login/', {
+                'username': username,
+                'password': 'toast'})
+
+    def logout(self):
+        self.test_app.get('/auth/logout/')
+
+    def do_post(self, data, *context_keys, **kwargs):
+        url = kwargs.pop('url', '/submit/')
+        do_follow = kwargs.pop('do_follow', False)
+        template.clear_test_template_context()
+        response = self.test_app.post(url, data, **kwargs)
+        if do_follow:
+            response.follow()
+        context_data = template.TEMPLATE_TEST_CONTEXT
+        for key in context_keys:
+            context_data = context_data[key]
+        return response, context_data
+
+    def query_for_users(self):
+        return (User.query.filter(User.username==u'allie').first(),
+        User.query.filter(User.username==u'natalie').first())
+
+    def testMediaReports(self):
+        self.login(u'allie')
+        allie_user, natalie_user = self.query_for_users()
+        allie_id = allie_user.id
+
+        media_entry = fixture_media_entry(uploader=natalie_user.id,
+            state=u'processed')
+
+        mid = media_entry.id
+        media_uri_slug = '/u/{0}/m/{1}/'.format(natalie_user.username,
+                                                media_entry.slug)
+
+        response = self.test_app.get(media_uri_slug + "report/")
+        assert response.status == "200 OK"
+
+        response, context = self.do_post(
+            {'report_reason':u'Testing Media Report',
+            'reporter_id':unicode(allie_id)},url= media_uri_slug + "report/")
+
+        assert response.status == "302 FOUND"
+
+        media_report = MediaReport.query.first()
+
+        allie_user, natalie_user = self.query_for_users()
+        assert media_report is not None
+        assert media_report.report_content == u'Testing Media Report'
+        assert media_report.reporter_id == allie_id
+        assert media_report.reported_user_id == natalie_user.id
+        assert media_report.created is not None
+        assert media_report.discriminator == 'media_report'
+
+    def testCommentReports(self):
+        self.login(u'allie')
+        allie_user, natalie_user = self.query_for_users()
+        allie_id = allie_user.id
+
+        media_entry = fixture_media_entry(uploader=natalie_user.id,
+            state=u'processed')
+        mid = media_entry.id
+        fixture_add_comment(media_entry=mid,
+            author=natalie_user.id)
+        comment = MediaComment.query.first()
+
+        comment_uri_slug = '/u/{0}/m/{1}/c/{2}/'.format(natalie_user.username,
+                                                media_entry.slug,
+                                                comment.id)
+
+        response = self.test_app.get(comment_uri_slug + "report/")
+        assert response.status == "200 OK"
+
+        response, context = self.do_post({
+            'report_reason':u'Testing Comment Report',
+            'reporter_id':unicode(allie_id)},url= comment_uri_slug + "report/")
+
+        assert response.status == "302 FOUND"
+
+        comment_report = CommentReport.query.first()
+
+        allie_user, natalie_user = self.query_for_users()
+        assert comment_report is not None
+        assert comment_report.report_content == u'Testing Comment Report'
+        assert comment_report.reporter_id == allie_id
+        assert comment_report.reported_user_id == natalie_user.id
+        assert comment_report.created is not None
+        assert comment_report.discriminator == 'comment_report'
+
+    def testArchivingReports(self):
+        self.login(u'natalie')
+        allie_user, natalie_user = self.query_for_users()
+        allie_id, natalie_id = allie_user.id, natalie_user.id
+
+        fixture_add_comment(author=allie_user.id,
+            comment=u'Comment will be removed')
+        test_comment = MediaComment.query.filter(
+            MediaComment.author==allie_user.id).first()
+        fixture_add_comment_report(comment=test_comment,
+            reported_user=allie_user,
+            report_content=u'Testing Archived Reports #1',
+            reporter=natalie_user)
+        comment_report = CommentReport.query.filter(
+            CommentReport.reported_user==allie_user).first()
+
+        assert comment_report.report_content == u'Testing Archived Reports #1'
+        response, context = self.do_post(
+            {'action_to_resolve':[u'userban', u'delete'],
+            'targeted_user':allie_user.id,
+            'resolution_content':u'This is a test of archiving reports.'},
+            url='/mod/reports/{0}/'.format(comment_report.id))
+
+        assert response.status == "302 FOUND"
+        allie_user, natalie_user = self.query_for_users()
+
+        archived_report = CommentReport.query.filter(
+            CommentReport.reported_user==allie_user).first()
+
+        assert CommentReport.query.count() != 0
+        assert archived_report is not None
+        assert archived_report.report_content == u'Testing Archived Reports #1'
+        assert archived_report.reporter_id == natalie_id
+        assert archived_report.reported_user_id == allie_id
+        assert archived_report.created is not None
+        assert archived_report.resolved is not None
+        assert archived_report.result == u'''This is a test of archiving reports.
+natalie banned user allie indefinitely.
+natalie deleted the comment.'''
+        assert archived_report.discriminator == 'comment_report'
+
--- a/mediagoblin/tests/test_submission.py
+++ b/mediagoblin/tests/test_submission.py
@@ -47,12 +47,22 @@ class TestSubmission:

        # TODO: Possibly abstract into a decorator like:
        # @as_authenticated_user('chris')
-        test_user = fixture_add_user()
-
-        self.test_user = test_user
+        fixture_add_user(privileges=[u'active',u'uploader', u'commenter'])

        self.login()

+    def our_user(self):
+        """
+        Fetch the user we're submitting with.  Every .get() or .post()
+        invalidates the session; this is a hacky workaround.
+        """
+        #### FIXME: Pytest collects this as a test and runs this.
+        ####   ... it shouldn't.  At least it passes, but that's
+        ####   totally stupid.
+        ####   Also if we found a way to make this run it should be a
+        ####   property.
+        return User.query.filter(User.username==u'chris').first()
+
    def login(self):
        self.test_app.post(
            '/auth/login/', {
@@ -98,10 +108,10 @@ class TestSubmission:
    def check_normal_upload(self, title, filename):
        response, context = self.do_post({'title': title}, do_follow=True,
                                         **self.upload_data(filename))
-        self.check_url(response, '/u/{0}/'.format(self.test_user.username))
+        self.check_url(response, '/u/{0}/'.format(self.our_user().username))
        assert 'mediagoblin/user_pages/user.html' in context
        # Make sure the media view is at least reachable, logged in...
-        url = '/u/{0}/m/{1}/'.format(self.test_user.username,
+        url = '/u/{0}/m/{1}/'.format(self.our_user().username,
                                     title.lower().replace(' ', '-'))
        self.test_app.get(url)
        # ... and logged out too.
@@ -148,7 +158,7 @@ class TestSubmission:
        response, context = self.do_post({'title': u'Normal upload 3 (pdf)'},
                                         do_follow=True,
                                         **self.upload_data(GOOD_PDF))
-        self.check_url(response, '/u/{0}/'.format(self.test_user.username))
+        self.check_url(response, '/u/{0}/'.format(self.our_user().username))
        assert 'mediagoblin/user_pages/user.html' in context

    def test_default_upload_limits(self):
@@ -264,7 +274,7 @@ class TestSubmission:
        # render and post to the edit page.
        edit_url = request.urlgen(
            'mediagoblin.edit.edit_media',
-            user=self.test_user.username, media_id=media_id)
+            user=self.our_user().username, media_id=media_id)
        self.test_app.get(edit_url)
        self.test_app.post(edit_url,
            {'title': u'Balanced Goblin',
@@ -277,7 +287,7 @@ class TestSubmission:
        self.check_comments(request, media_id, 0)
        comment_url = request.urlgen(
            'mediagoblin.user_pages.media_post_comment',
-            user=self.test_user.username, media_id=media_id)
+            user=self.our_user().username, media_id=media_id)
        response = self.do_post({'comment_content': 'i love this test'},
                                url=comment_url, do_follow=True)[0]
        self.check_comments(request, media_id, 1)
@@ -286,7 +296,7 @@ class TestSubmission:
        # ---------------------------------------------------
        delete_url = request.urlgen(
            'mediagoblin.user_pages.media_confirm_delete',
-            user=self.test_user.username, media_id=media_id)
+            user=self.our_user().username, media_id=media_id)
        # Empty data means don't confirm
        response = self.do_post({}, do_follow=True, url=delete_url)[0]
        media = self.check_media(request, {'title': u'Balanced Goblin'}, 1)
@@ -359,7 +369,7 @@ class TestSubmission:
        #   they'll be caught as failures during the processing step.
        response, context = self.do_post({'title': title}, do_follow=True,
                                         **self.upload_data(filename))
-        self.check_url(response, '/u/{0}/'.format(self.test_user.username))
+        self.check_url(response, '/u/{0}/'.format(self.our_user().username))
        entry = mg_globals.database.MediaEntry.query.filter_by(title=title).first()
        assert entry.state == 'failed'
        assert entry.fail_error == u'mediagoblin.processing:BadMediaFail'
--- a/mediagoblin/tests/tools.py
+++ b/mediagoblin/tests/tools.py
@@ -25,7 +25,7 @@ from webtest import TestApp

 from mediagoblin import mg_globals
 from mediagoblin.db.models import User, MediaEntry, Collection, MediaComment, \
-    CommentSubscription, CommentNotification
+    CommentSubscription, CommentNotification, Privilege, CommentReport
 from mediagoblin.tools import testing
 from mediagoblin.init.config import read_mediagoblin_config
 from mediagoblin.db.base import Session
@@ -33,6 +33,8 @@ from mediagoblin.meddleware import BaseMeddleware
 from mediagoblin.auth import gen_password_hash
 from mediagoblin.gmg_commands.dbupdate import run_dbupdate

+from datetime import datetime
+

 MEDIAGOBLIN_TEST_DB_NAME = u'__mediagoblin_tests__'
 TEST_SERVER_CONFIG = pkg_resources.resource_filename(
@@ -133,7 +135,6 @@ def get_app(request, paste_config=None, mgoblin_config=None):
    mg_globals.app.meddleware.insert(0, TestingMeddleware(mg_globals.app))

    app = TestApp(test_app)
-
    return app


@@ -170,7 +171,7 @@ def assert_db_meets_expected(db, expected):


 def fixture_add_user(username=u'chris', password=u'toast',
-                     active_user=True, wants_comment_notification=True):
+                     privileges=[], wants_comment_notification=True):
    # Reuse existing user or create a new one
    test_user = User.query.filter_by(username=username).first()
    if test_user is None:
@@ -179,14 +180,13 @@ def fixture_add_user(username=u'chris', password=u'toast',
    test_user.email = username + u'@example.com'
    if password is not None:
        test_user.pw_hash = gen_password_hash(password)
-    if active_user:
-        test_user.email_verified = True
-        test_user.status = u'active'
-
    test_user.wants_comment_notification = wants_comment_notification
+    for privilege in privileges:
+        query = Privilege.query.filter(Privilege.privilege_name==privilege)
+        if query.count():
+            test_user.all_privileges.append(query.one())

    test_user.save()
-
    # Reload
    test_user = User.query.filter_by(username=username).first()

@@ -314,3 +314,32 @@ def fixture_add_comment(author=None, media_entry=None, comment=None):

    return comment

+def fixture_add_comment_report(comment=None, reported_user=None,
+        reporter=None, created=None, report_content=None):
+    if comment is None:
+        comment = fixture_add_comment()
+
+    if reported_user is None:
+        reported_user = fixture_add_user()
+
+    if reporter is None:
+        reporter = fixture_add_user()
+
+    if created is None:
+        created=datetime.now()
+
+    if report_content is None:
+        report_content = \
+            'Auto-generated test report'
+
+    comment_report = CommentReport(comment=comment,
+        reported_user = reported_user,
+        reporter = reporter,
+        created = created,
+        report_content=report_content)
+
+    comment_report.save()
+
+    Session.expunge(comment_report)
+
+    return comment_report