Diff (tree)

@@ -0,0 +1,29 @@

+ """Add enforced_flags field to projects

+ 

+ Revision ID: b534625b531a

+ Revises: 4255158a6913

+ Create Date: 2017-04-05 12:12:48.077097

+ 

+ """

+ 

+ # revision identifiers, used by Alembic.

+ revision = 'b534625b531a'

+ down_revision = '4255158a6913'

+ 

+ from alembic import op

+ import sqlalchemy as sa

+ 

+ 

+ def upgrade():

+     ''' Add the column _enforced_flags to the table projects.

+     '''

+     op.add_column(

+         'projects',

+         sa.Column('_enforced_flags', sa.Text, nullable=True)

+     )

+ 

+ 

+ def downgrade():

+     ''' Add the column _enforced_flags to the table projects.

+     '''

+     op.drop_column('projects', '_enforced_flags')
file changed

@@ -1,7 +1,7 @@

  # -*- coding: utf-8 -*-

  

  """

-  (c) 2015 - Copyright Red Hat Inc

+  (c) 2015-2017 - Copyright Red Hat Inc

  

   Authors:

     Pierre-Yves Chibon <pingou@pingoured.fr>
file changed

@@ -1,7 +1,7 @@

  # -*- coding: utf-8 -*-

  

  """

-  (c) 2015-2016 - Copyright Red Hat Inc

+  (c) 2015-2017 - Copyright Red Hat Inc

  

   Authors:

     Pierre-Yves Chibon <pingou@pingoured.fr>
file changed

@@ -369,6 +369,7 @@

      _reports = sa.Column(sa.Text, nullable=True)

      _notifications = sa.Column(sa.Text, nullable=True)

      _close_status = sa.Column(sa.Text, nullable=True)

+     _enforced_flags = sa.Column(sa.Text, nullable=True)

  

      date_created = sa.Column(sa.DateTime, nullable=False,

                               default=datetime.datetime.utcnow)

@@ -616,6 +617,23 @@

          self._close_status = json.dumps(close_status)

  

      @property

+     def enforced_flags(self):

+         """ Return the dict stored as string in the database as an actual

+         dict object.

+         """

+         enforced_flags = {}

+ 

+         if self._enforced_flags:

+             enforced_flags = json.loads(self._enforced_flags)

+ 

+         return enforced_flags

+ 

+     @enforced_flags.setter

+     def enforced_flags(self, enforced_flags):

+         ''' Ensures the enforced_flags are properly saved. '''

+         self._enforced_flags = json.dumps(enforced_flags)

+ 

+     @property

      def open_requests(self):

          ''' Returns the number of open pull-requests for this project. '''

          return BASE.metadata.bind.query(

@@ -1690,10 +1708,15 @@

          ''' Returns if a PR can be merged based on whether it has the

          mandatory flags.

          '''

-         if mandatory_flags:

+         if mandatory_flags or self.project.enforced_flags:

+             repo_flags = self.project.enforced_flags

              flag_ok = False

              for flag in self.flags:

-                 if flag.token_id in mandatory_flags:

+                 if flag.token_id in mandatory_flags \

+                         and flag.username == mandatory_flags[flag.token_id]:

+                     flag_ok = flag.percent == 100

+                 elif flag.token_id in repo_flags \

+                         and flag.username == repo_flags[flag.token_id]:

                      flag_ok = flag.percent == 100

          else:

              flag_ok = True

@@ -632,7 +632,11 @@

                  requestid=requestid)

            }}" method="POST">

            {{ mergeform.csrf_token }}

-           <button id="merge_btn" type="submit" class="btn btn-block">Merge</button>

+           <button id="merge_btn" type="submit" class="btn btn-block"{%

+             if not pull_request.mandatory_flags_ok(config.get('MANDATORY_FLAGS'))

+           %} disabled title="Required flag missing"{% endif %}>

+             Merge

+           </button>

          </form>

         </div>

        {% endif %}
file changed

@@ -221,6 +221,101 @@

        </div>

      </div>

  

+     {% if repo.tokens %}

+     <div class="col-md-8 col-md-offset-2">

+       <div class="card">

+         <div class="card-header">

+           <strong>Enforced Flags</strong>

+         </div>

+         <div class="card-block">

+           <p>

+             This mechanism allows you to require a certain flag associated

+             with a specific API token and with a specific username to allow

+             merging a pull-request.

+           </p>

+         </div>

+         <form action="{{ url_for(

+               '.update_enforced_flags',

+               repo=repo.name,

+               username=username,

+               namespace=repo.namespace) }}"

+             method="post" class="icon">

+           {{ tag_form.csrf_token }}

+           <div class="card-block">

+             <div class="row">

+               <div class="col-sm-2">

+                 <strong>Title</strong>

+               </div>

+               <div class="col-sm-10">

+                 <strong>API token</strong>

+               </div>

+             </div>

+ 

+             {% for flag in repo.enforced_flags | sort %}

+               <div class="row p-t-1">

+                 <div class="col-sm-2 p-r-0">

+                   <input type="text" name="title"

+                     value="{{ repo.enforced_flags[flag] }}" size="3" class="form-control"/>

+                 </div>

+                 <div class="col-sm-9 p-r-0">

+                   <select name="token" class="form-control">

+                     <option value=""></option>

+                     {% for token in repo.tokens %}

+                       {% if token.user.username == g.fas_user.username

+                             and 'pull_request_flag' in token.acls_list %}

+                         <option value="{{ token.id }}" {%

+                           if token.id == flag %}selected{%

+                           endif %}>

+                           {{ token.id }}

+                         </option>

+                       {% endif %}

+                     {% endfor %}

+                   </select>

+                 </div>

+               </div>

+             {% endfor %}

+ 

+             <div id="enforced_flags">

+               <div class="row p-t-1">

+                 <div class="col-sm-2 p-r-0">

+                   <input type="text" name="title" size="3" class="form-control"/>

+                 </div>

+                 <div class="col-sm-9 p-r-0">

+                   <select name="token" class="form-control c-select">

+                     <option value=""></option>

+                     {% for token in repo.tokens %}

+                       {% if token.user.username == g.fas_user.username

+                             and 'pull_request_flag' in token.acls_list %}

+                         <option value="{{ token.id }}">

+                           {{ token.id }}

+                         </option>

+                       {% endif %}

+                     {% endfor %}

+                   </select>

+                 </div>

+               </div>

+             </div>

+             <div class="row p-t-1">

+               <div class="col-sm-6">

+                 <a class="btn btn-secondary btn-sm btn-block extend-form" data-target="#enforced_flags">

+                     Add new requirement

+                 </a>

+               </div>

+             </div>

+             <div class="row p-t-1">

+               <div class="col-sm-12">

+                 <button class="btn btn-primary" type="submit"

+                     title="Update the required flags">

+                   Update

+                 </button>

+               </div>

+             </div>

+           </div>

+         </form>

+       </div>

+     </div>

+     {% endif %}

+ 

      <div class="col-md-8 col-md-offset-2">

        <div class="card">

          <div class="card-header">
file changed

@@ -2559,3 +2559,88 @@

      return flask.redirect(flask.url_for(

          'view_settings', username=username, repo=repo.name,

          namespace=namespace))

+ 

+ 

+ @APP.route('/<repo>/update/enforced_flags', methods=['POST'])

+ @APP.route('/<namespace>/<repo>/update/enforced_flags', methods=['POST'])

+ @APP.route('/fork/<username>/<repo>/update/enforced_flags', methods=['POST'])

+ @APP.route(

+     '/fork/<username>/<namespace>/<repo>/update/enforced_flags',

+     methods=['POST'])

+ @login_required

+ def update_enforced_flags(repo, username=None, namespace=None):

+     """ Update the enforced_flags of a project.

+     """

+     if admin_session_timedout():

+         flask.flash('Action canceled, try it again', 'error')

+         url = flask.url_for(

+             'view_settings', username=username, repo=repo,

+             namespace=namespace)

+         return flask.redirect(

+             flask.url_for('auth_login', next=url))

+ 

+     repo = flask.g.repo

+ 

+     if not repo.settings.get('pull_requests', True):

+         flask.abort(404, 'Pull-Requests are disabled for this project')

+ 

+     if not flask.g.repo_admin:

+         flask.abort(

+             403,

+             'You are not allowed to change the settings for this project')

+ 

+     form = pagure.forms.ConfirmationForm()

+ 

+     if form.validate_on_submit():

+         tokens = [

+             w.strip() for w in flask.request.form.getlist('token')

+             if w.strip()

+         ]

+         titles = [

+             w.strip() for w in flask.request.form.getlist('title')

+             if w.strip()

+         ]

+         error = False

+ 

+         if len(titles) != len(tokens):

+             flask.flash(

+                 'The number of tokens and titles provided do not match',

+                 'error')

+             error = True

+ 

+         for input_token in tokens:

+             seen = False

+             try:

+                 for saved_token in repo.tokens:

+                     if saved_token.id == input_token:

+                         seen = True

+                         if 'pull_request_flag' not in saved_token.acls_list:

+                             raise pagure.exceptions.PagureException(

+                                 'Token %s does not have the permission to '

+                                 'flag a pull-request.' % input_token)

+                         break

+                 if not seen:

+                     raise pagure.exceptions.PagureException(

+                         'No token %s found associated with this '

+                         'project' % input_token)

+             except pagure.exceptions.PagureException as err:

+                 error = True

+                 flask.flash(str(err), 'error')

+ 

+         if not error:

+             enforced_flags = {}

+             for cnt in range(len(tokens)):

+                 if tokens[cnt].strip():

+                     enforced_flags[tokens[cnt]] = titles[cnt]

+             try:

+                 repo.enforced_flags = enforced_flags

+                 SESSION.add(repo)

+                 SESSION.commit()

+                 flask.flash('Enforced flags updated')

+             except SQLAlchemyError as err:  # pragma: no cover

+                 SESSION.rollback()

+                 flask.flash(str(err), 'error')

+ 

+     return flask.redirect(flask.url_for(

+         'view_settings', username=username, repo=repo.name,

+         namespace=namespace))

@@ -1227,23 +1227,23 @@

          self.assertEqual(request.flags[0].comment, 'Tests passed')

          self.assertEqual(request.flags[0].percent, 100)

  

-     @patch('pagure.lib.notify.send_email')

-     @patch('pagure.lib.git.merge_pull_request')

-     @patch.dict('pagure.APP.config', {'MANDATORY_FLAGS': 'foobar'})

+     @patch('pagure.lib.notify.send_email', return_value=True)

+     @patch(

+         'pagure.lib.git.merge_pull_request', return_value='Changes merged!')

+     @patch.dict(

+         'pagure.APP.config', {'MANDATORY_FLAGS': {'foobar': 'koji'}})

      def test_api_pull_request_merge_no_flag(self, mpr, send_email):

          """ Test the api_pull_request_merge method of the flask api when

          the PR is missing one of the mandatory flag.

          """

-         mpr.return_value = 'Changes merged!'

-         send_email.return_value = True

  

          tests.create_projects(self.session)

          tests.create_tokens(self.session)

          tests.create_tokens_acl(self.session)

  

          # Create the pull-request to close

-         repo = pagure.lib.get_project(self.session, 'test')

-         forked_repo = pagure.lib.get_project(self.session, 'test')

+         repo = pagure.get_authorized_project(self.session, 'test')

+         forked_repo = pagure.get_authorized_project(self.session, 'test')

          req = pagure.lib.new_pull_request(

              session=self.session,

              repo_from=forked_repo,

@@ -1305,7 +1305,8 @@

  

      @patch('pagure.lib.notify.send_email')

      @patch('pagure.lib.git.merge_pull_request')

-     @patch.dict('pagure.APP.config', {'MANDATORY_FLAGS': 'foobar'})

+     @patch.dict(

+         'pagure.APP.config', {'MANDATORY_FLAGS': {'foobar': 'koji'}})

      def test_api_pull_request_merge_w_invalid_flag(self, mpr, send_email):

          """ Test the api_pull_request_merge method of the flask api when

          the PR is missing one of the mandatory flag.

@@ -1318,8 +1319,8 @@

          tests.create_tokens_acl(self.session)

  

          # Create the pull-request to close

-         repo = pagure.lib.get_project(self.session, 'test')

-         forked_repo = pagure.lib.get_project(self.session, 'test')

+         repo = pagure.get_authorized_project(self.session, 'test')

+         forked_repo = pagure.get_authorized_project(self.session, 'test')

          req = pagure.lib.new_pull_request(

              session=self.session,

              repo_from=forked_repo,

@@ -1368,7 +1369,8 @@

  

      @patch('pagure.lib.notify.send_email')

      @patch('pagure.lib.git.merge_pull_request')

-     @patch.dict('pagure.APP.config', {'MANDATORY_FLAGS': 'foobar'})

+     @patch.dict(

+         'pagure.APP.config', {'MANDATORY_FLAGS': {'foobar': 'koji'}})

      def test_api_pull_request_merge_w_flag(self, mpr, send_email):

          """ Test the api_pull_request_merge method of the flask api when

          the PR is missing one of the mandatory flag.

@@ -1381,8 +1383,8 @@

          tests.create_tokens_acl(self.session)

  

          # Create the pull-request to close

-         repo = pagure.lib.get_project(self.session, 'test')

-         forked_repo = pagure.lib.get_project(self.session, 'test')

+         repo = pagure.get_authorized_project(self.session, 'test')

+         forked_repo = pagure.get_authorized_project(self.session, 'test')

          req = pagure.lib.new_pull_request(

              session=self.session,

              repo_from=forked_repo,

@@ -1428,7 +1430,8 @@

  

      @patch('pagure.lib.notify.send_email')

      @patch('pagure.lib.git.merge_pull_request')

-     @patch.dict('pagure.APP.config', {'MANDATORY_FLAGS': 'foobar'})

+     @patch.dict(

+         'pagure.APP.config', {'MANDATORY_FLAGS': {'foobar': 'koji'}})

      def test_api_pull_request_merge_w_two_flags(self, mpr, send_email):

          """ Test the api_pull_request_merge method of the flask api when

          the PR is missing one of the mandatory flag.

@@ -1441,8 +1444,9 @@

          tests.create_tokens_acl(self.session)

  

          # Create the pull-request to close

-         repo = pagure.lib.get_project(self.session, 'test')

-         forked_repo = pagure.lib.get_project(self.session, 'test')

+         repo = pagure.get_authorized_project(self.session, 'test')

+         forked_repo = pagure.get_authorized_project(

+             self.session, 'test')

          req = pagure.lib.new_pull_request(

              session=self.session,

              repo_from=forked_repo,

@@ -497,10 +497,9 @@

                  '</button>\n                      Merge conflicts!',

                  output.data)

  

-     @patch('pagure.lib.notify.send_email')

+     @patch('pagure.lib.notify.send_email', return_value=True)

      def test_merge_request_pull_nochange(self, send_email):

          """ Test the merge_request_pull endpoint. """

-         send_email.return_value = True

  

          tests.create_projects(self.session)

          tests.create_projects_git(

@@ -0,0 +1,409 @@

+ # -*- coding: utf-8 -*-

+ 

+ """

+  (c) 2017 - Copyright Red Hat Inc

+ 

+  Authors:

+    Pierre-Yves Chibon <pingou@pingoured.fr>

+ 

+ """

+ 

+ __requires__ = ['SQLAlchemy >= 0.8']

+ import pkg_resources

+ 

+ import datetime

+ import json

+ import unittest

+ import shutil

+ import sys

+ import tempfile

+ import os

+ 

+ import pygit2

+ from mock import patch, MagicMock

+ 

+ sys.path.insert(0, os.path.join(os.path.dirname(

+     os.path.abspath(__file__)), '..'))

+ 

+ import pagure.lib

+ import tests

+ from pagure.lib.repo import PagureRepo

+ 

+ 

+ class PagureFlaskRoadmaptests(tests.Modeltests):

+     """ Tests for the pagure's roadmap """

+ 

+     @patch('pagure.lib.notify.send_email', MagicMock(return_value=True))

+     def setUp(self):

+         """ Set up the environnment, ran before every tests. """

+         super(PagureFlaskRoadmaptests, self).setUp()

+ 

+         pagure.APP.config['TESTING'] = True

+         pagure.SESSION = self.session

+         pagure.ui.SESSION = self.session

+         pagure.ui.app.SESSION = self.session

+         pagure.ui.filters.SESSION = self.session

+         pagure.ui.repo.SESSION = self.session

+         pagure.ui.issues.SESSION = self.session

+ 

+         pagure.APP.config['GIT_FOLDER'] = self.path

+         pagure.APP.config['REQUESTS_FOLDER'] = os.path.join(

+             self.path, 'requests')

+         pagure.APP.config['TICKETS_FOLDER'] = os.path.join(

+             self.path, 'tickets')

+         pagure.APP.config['DOCS_FOLDER'] = os.path.join(

+             self.path, 'docs')

+         self.app = pagure.APP.test_client()

+ 

+         tests.create_projects(self.session)

+         tests.create_projects_git(os.path.join(self.path), bare=True)

+ 

+         # Create the pull-request to close

+         repo = pagure.get_authorized_project(self.session, 'test')

+         req = pagure.lib.new_pull_request(

+             session=self.session,

+             repo_from=repo,

+             branch_from='feature',

+             repo_to=repo,

+             branch_to='master',

+             title='test pull-request',

+             user='pingou',

+             requestfolder=None,

+         )

+         self.session.commit()

+         self.assertEqual(req.id, 1)

+         self.assertEqual(req.title, 'test pull-request')

+ 

+         # Add an API token to flag PR

+         msg = pagure.lib.add_token_to_user(

+             self.session,

+             repo,

+             description='Flag PR',

+             acls=['pull_request_flag'],

+             username='pingou',

+         )

+         self.assertEqual(msg, 'Token created')

+         self.good_token = repo.tokens[0].id

+         self.assertEqual(len(self.good_token), 64)

+ 

+         # Add an API token to comment on PR

+         msg = pagure.lib.add_token_to_user(

+             self.session,

+             repo,

+             description='Flag PR',

+             acls=['pull_request_comment'],

+             username='pingou',

+         )

+         self.assertEqual(msg, 'Token created')

+ 

+         self.bad_token = repo.tokens[1].id

+         self.assertEqual(len(self.bad_token), 64)

+ 

+         self.assertTrue(self.bad_token != self.good_token)

+ 

+     def test_update_enforced_flags_invalid_token(self):

+         """ Test update_enforced_flags of a repo. """

+ 

+         # Set some required flags

+         repo = pagure.get_authorized_project(self.session, 'test')

+         self.assertEqual(repo.enforced_flags, {})

+ 

+         user = tests.FakeUser()

+         user.username = 'pingou'

+         with tests.user_set(pagure.APP, user):

+ 

+             # Get the CSRF token

+             output = self.app.get('/test/settings')

+             self.assertEqual(output.status_code, 200)

+             self.assertIn(

+                 u'<title>Settings - test - Pagure</title>', output.data)

+             self.assertIn(u'<h3>Settings for test</h3>', output.data)

+ 

+             csrf_token = output.data.split(

+                 'name="csrf_token" type="hidden" value="')[1].split('">')[0]

+ 

+             data = {

+                 'token': 1,

+                 'title': 'Tomorrow',

+             }

+             output = self.app.post(

+                 '/test/update/enforced_flags', data=data, follow_redirects=True)

+             self.assertEqual(output.status_code, 200)

+             # Check the redirect

+             self.assertIn(

+                 u'<title>Settings - test - Pagure</title>', output.data)

+             self.assertIn('<h3>Settings for test</h3>', output.data)

+ 

+             # Check the result of the action -- None, no CSRF

+             repo = pagure.get_authorized_project(self.session, 'test')

+             self.assertEqual(repo.enforced_flags, {})

+ 

+             data = {

+                 'token': 1,

+                 'title': 'Tomorrow',

+                 'csrf_token': csrf_token,

+             }

+             output = self.app.post(

+                 '/test/update/enforced_flags', data=data, follow_redirects=True)

+             self.assertEqual(output.status_code, 200)

+             self.assertIn(

+                 u'<title>Settings - test - Pagure</title>', output.data)

+             self.assertIn(u'<h3>Settings for test</h3>', output.data)

+             self.assertIn(

+                 u'No token 1 found associated with this project',

+                 output.data)

+ 

+             # Check the result of the action -- None, invalid token

+             repo = pagure.get_authorized_project(self.session, 'test')

+             self.assertEqual(repo.enforced_flags, {})

+ 

+     def test_update_enforced_flags_no_token(self):

+         """ Test update_enforced_flags of a repo. """

+ 

+         # Set some required flags

+         repo = pagure.get_authorized_project(self.session, 'test')

+         self.assertEqual(repo.enforced_flags, {})

+ 

+         user = tests.FakeUser()

+         user.username = 'pingou'

+         with tests.user_set(pagure.APP, user):

+ 

+             # Get the CSRF token

+             output = self.app.get('/test/settings')

+             self.assertEqual(output.status_code, 200)

+             self.assertIn(

+                 u'<title>Settings - test - Pagure</title>', output.data)

+             self.assertIn(u'<h3>Settings for test</h3>', output.data)

+ 

+             csrf_token = output.data.split(

+                 'name="csrf_token" type="hidden" value="')[1].split('">')[0]

+ 

+             data = {

+                 'title': 'Tomorrow',

+             }

+             output = self.app.post(

+                 '/test/update/enforced_flags', data=data, follow_redirects=True)

+             self.assertEqual(output.status_code, 200)

+             # Check the redirect

+             self.assertIn(

+                 u'<title>Settings - test - Pagure</title>', output.data)

+             self.assertIn('<h3>Settings for test</h3>', output.data)

+ 

+             # Check the result of the action -- None, no CSRF

+             repo = pagure.get_authorized_project(self.session, 'test')

+             self.assertEqual(repo.enforced_flags, {})

+ 

+             data = {

+                 'title': 'Tomorrow',

+                 'csrf_token': csrf_token,

+             }

+             output = self.app.post(

+                 '/test/update/enforced_flags', data=data, follow_redirects=True)

+             self.assertEqual(output.status_code, 200)

+             self.assertIn(

+                 u'<title>Settings - test - Pagure</title>', output.data)

+             self.assertIn(u'<h3>Settings for test</h3>', output.data)

+             self.assertIn(

+                 u'The number of tokens and titles provided do not match',

+                 output.data)

+ 

+             # Check the result of the action -- None, invalid token

+             repo = pagure.get_authorized_project(self.session, 'test')

+             self.assertEqual(repo.enforced_flags, {})

+ 

+     def test_update_enforced_flags_good_token_bad_acl(self):

+         """ Test update_enforced_flags of a repo. """

+ 

+         # Set some required flags

+         repo = pagure.get_authorized_project(self.session, 'test')

+         self.assertEqual(repo.enforced_flags, {})

+ 

+         user = tests.FakeUser()

+         user.username = 'pingou'

+         with tests.user_set(pagure.APP, user):

+ 

+             # Get the CSRF token

+             output = self.app.get('/test/settings')

+             self.assertEqual(output.status_code, 200)

+             self.assertIn(

+                 u'<title>Settings - test - Pagure</title>', output.data)

+             self.assertIn(u'<h3>Settings for test</h3>', output.data)

+ 

+             csrf_token = output.data.split(

+                 'name="csrf_token" type="hidden" value="')[1].split('">')[0]

+ 

+             data = {

+                 'token': self.bad_token,

+                 'title': 'jenkins',

+             }

+             output = self.app.post(

+                 '/test/update/enforced_flags', data=data, follow_redirects=True)

+             self.assertEqual(output.status_code, 200)

+             # Check the redirect

+             self.assertIn(

+                 u'<title>Settings - test - Pagure</title>', output.data)

+             self.assertIn('<h3>Settings for test</h3>', output.data)

+ 

+             # Check the result of the action -- None, no CSRF

+             repo = pagure.get_authorized_project(self.session, 'test')

+             self.assertEqual(repo.enforced_flags, {})

+ 

+             data = {

+                 'token': self.bad_token,

+                 'title': 'jenkins',

+                 'csrf_token': csrf_token,

+             }

+             output = self.app.post(

+                 '/test/update/enforced_flags', data=data, follow_redirects=True)

+             self.assertEqual(output.status_code, 200)

+             self.assertIn(

+                 u'<title>Settings - test - Pagure</title>', output.data)

+             self.assertIn(u'<h3>Settings for test</h3>', output.data)

+             self.assertIn(

+                 u'Token %s does not have the permission to flag a '

+                 'pull-request.' % self.bad_token, output.data)

+ 

+             # Check the result of the action -- None, token exists but has

+             # the wrong ACL

+             repo = pagure.get_authorized_project(self.session, 'test')

+             self.assertEqual(repo.enforced_flags, {})

+ 

+     def test_update_enforced_flags(self):

+         """ Test update_enforced_flags of a repo. """

+ 

+         # Set some required flags

+         repo = pagure.get_authorized_project(self.session, 'test')

+         self.assertEqual(repo.enforced_flags, {})

+ 

+         user = tests.FakeUser()

+         user.username = 'pingou'

+         with tests.user_set(pagure.APP, user):

+ 

+             # Get the CSRF token

+             output = self.app.get('/test/settings')

+             self.assertEqual(output.status_code, 200)

+             self.assertIn(

+                 u'<title>Settings - test - Pagure</title>', output.data)

+             self.assertIn(u'<h3>Settings for test</h3>', output.data)

+ 

+             csrf_token = output.data.split(

+                 'name="csrf_token" type="hidden" value="')[1].split('">')[0]

+ 

+             data = {

+                 'token': self.good_token,

+                 'title': 'jenkins',

+             }

+             output = self.app.post(

+                 '/test/update/enforced_flags', data=data, follow_redirects=True)

+             self.assertEqual(output.status_code, 200)

+             # Check the redirect

+             self.assertIn(

+                 u'<title>Settings - test - Pagure</title>', output.data)

+             self.assertIn('<h3>Settings for test</h3>', output.data)

+ 

+             # Check the result of the action -- None, no CSRF

+             repo = pagure.get_authorized_project(self.session, 'test')

+             self.assertEqual(repo.enforced_flags, {})

+ 

+             data = {

+                 'token': self.good_token,

+                 'title': 'jenkins',

+                 'csrf_token': csrf_token,

+             }

+             output = self.app.post(

+                 '/test/update/enforced_flags', data=data, follow_redirects=True)

+             self.assertEqual(output.status_code, 200)

+             self.assertIn(

+                 u'<title>Settings - test - Pagure</title>', output.data)

+             self.assertIn(u'<h3>Settings for test</h3>', output.data)

+             self.assertIn(u'Enforced flags updated', output.data)

+ 

+             # Check the result of the action -- Saved

+             repo = pagure.get_authorized_project(self.session, 'test')

+             self.assertEqual(

+                 repo.enforced_flags, {self.good_token: u'jenkins'})

+ 

+             # Test re-sending the same data

+             output = self.app.post(

+                 '/test/update/enforced_flags', data=data, follow_redirects=True)

+             self.assertEqual(output.status_code, 200)

+             self.assertIn(

+                 u'<title>Settings - test - Pagure</title>', output.data)

+             self.assertIn(u'<h3>Settings for test</h3>', output.data)

+             self.assertIn(u'Enforced flags updated', output.data)

+ 

+             # Check the result of the action -- Saved

+             repo = pagure.get_authorized_project(self.session, 'test')

+             self.assertEqual(

+                 repo.enforced_flags, {self.good_token: u'jenkins'})

+ 

+             # Test clearing the data

+             data = {'csrf_token': csrf_token}

+             output = self.app.post(

+                 '/test/update/enforced_flags', data=data, follow_redirects=True)

+             self.assertEqual(output.status_code, 200)

+             self.assertIn(

+                 u'<title>Settings - test - Pagure</title>', output.data)

+             self.assertIn(u'<h3>Settings for test</h3>', output.data)

+             self.assertIn(u'Enforced flags updated', output.data)

+ 

+             # Check the result of the action -- nothing saved now

+             repo = pagure.get_authorized_project(self.session, 'test')

+             self.assertEqual(repo.enforced_flags, {})

+ 

+     def test_update_enforced_flags_invalid_project(self):

+         """ Test update_enforced_flags of a repo. """

+ 

+         # Set some required flags

+         repo = pagure.get_authorized_project(self.session, 'test')

+         self.assertEqual(repo.enforced_flags, {})

+ 

+         user = tests.FakeUser()

+         user.username = 'pingou'

+         with tests.user_set(pagure.APP, user):

+ 

+             # Check for an invalid project

+             output = self.app.post(

+                 '/foo/update/enforced_flags', data={})

+             self.assertEqual(output.status_code, 404)

+ 

+     def test_update_enforced_flags_pr_disabled(self):

+         """ Test update_enforced_flags of a repo when the project disabled

+         PR. """

+ 

+         repo = pagure.get_authorized_project(self.session, 'test')

+         self.assertEqual(repo.enforced_flags, {})

+ 

+         user = tests.FakeUser()

+         user.username = 'pingou'

+         with tests.user_set(pagure.APP, user):

+ 

+             # Check the behavior if the project disabled the issue tracker

+             settings = repo.settings

+             settings['pull_requests'] = False

+             repo.settings = settings

+             self.session.add(repo)

+             self.session.commit()

+ 

+             output = self.app.post(

+                 '/test/update/enforced_flags', data={})

+             self.assertEqual(output.status_code, 404)

+ 

+     def test_update_enforced_flags_non_admin(self):

+         """ Test update_enforced_flags of a repo with a non-autorized user.

+         """

+         repo = pagure.get_authorized_project(self.session, 'test')

+         self.assertEqual(repo.enforced_flags, {})

+ 

+         user = tests.FakeUser()

+ 

+         # Check for a non-admin user

+         user.username = 'ralph'

+         with tests.user_set(pagure.APP, user):

+             output = self.app.post(

+                 '/test/update/enforced_flags', data={})

+             self.assertEqual(output.status_code, 403)

+ 

+ 

+ if __name__ == '__main__':

+     unittest.main(verbosity=2)