# -*- coding: utf-8 -*- """ (c) 2015-2017 - Copyright Red Hat Inc Authors: Pierre-Yves Chibon """ from __future__ import unicode_literals, absolute_import import datetime import json import unittest import re import shutil import sys import tempfile import time import os cchardet = None try: import cchardet except ImportError: pass import pygit2 import six from mock import ANY, patch, MagicMock sys.path.insert( 0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..") ) import pagure.lib.query import tests from pagure.lib.repo import PagureRepo from pagure.utils import __get_file_in_tree as get_file_in_tree class PagureFlaskRepotests(tests.Modeltests): """ Tests for flask app controller of pagure """ def setUp(self): """ Set up the environnment, ran before every tests. """ super(PagureFlaskRepotests, self).setUp() pagure.config.config["VIRUS_SCAN_ATTACHMENTS"] = False pagure.config.config["UPLOAD_FOLDER_URL"] = "/releases/" pagure.config.config["UPLOAD_FOLDER_PATH"] = os.path.join( self.path, "releases" ) @patch("pagure.decorators.admin_session_timedout") def test_add_user_when_user_mngt_off(self, ast): """ Test the add_user endpoint when user management is turned off in the pagure instance """ pagure.config.config["ENABLE_USER_MNGT"] = False ast.return_value = False # No Git repo output = self.app.get("/foo/adduser") self.assertEqual(output.status_code, 404) tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "repos")) # User not logged in output = self.app.get("/test/adduser") self.assertEqual(output.status_code, 302) user = tests.FakeUser(username="pingou") with tests.user_set(self.app.application, user): output = self.app.get("/test/adduser") self.assertEqual(output.status_code, 404) # just get the csrf token pagure.config.config["ENABLE_USER_MNGT"] = True output = self.app.get("/test/adduser") output_text = output.get_data(as_text=True) csrf_token = output_text.split( 'name="csrf_token" type="hidden" value="' )[1].split('">')[0] pagure.config.config["ENABLE_USER_MNGT"] = False data = {"user": "ralph"} output = self.app.post("/test/adduser", data=data) self.assertEqual(output.status_code, 404) data["csrf_token"] = csrf_token output = self.app.post("/test/adduser", data=data) self.assertEqual(output.status_code, 404) data["user"] = "foo" tests.create_projects_git(os.path.join(self.path, "repos")) output = self.app.post( "/test/adduser", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 404) pagure.config.config["ENABLE_USER_MNGT"] = True @patch("pagure.decorators.admin_session_timedout") def test_add_deploykey(self, ast): """ Test the add_deploykey endpoint. """ ast.return_value = False # No git repo output = self.app.get("/foo/adddeploykey") self.assertEqual(output.status_code, 404) tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "repos")) # User not logged in output = self.app.get("/test/adddeploykey") self.assertEqual(output.status_code, 302) user = tests.FakeUser() with tests.user_set(self.app.application, user): output = self.app.get("/test/adddeploykey") self.assertEqual(output.status_code, 403) ast.return_value = True output = self.app.get("/test/adddeploykey") self.assertEqual(output.status_code, 302) # Redirect also happens for POST request output = self.app.post("/test/adddeploykey") self.assertEqual(output.status_code, 302) # Need to do this un-authentified since our fake user isn't in the DB # Check the message flashed during the redirect output = self.app.get("/", follow_redirects=True) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn("Action canceled, try it " "again", output_text) ast.return_value = False user.username = "pingou" with tests.user_set(self.app.application, user): output = self.app.get("/test/adddeploykey") self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn("Add deploy key to the", output_text) csrf_token = output_text.split( 'name="csrf_token" type="hidden" value="' )[1].split('">')[0] data = {"ssh_key": "asdf", "pushaccess": "false"} # No CSRF token output = self.app.post("/test/adddeploykey", data=data) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn("Add deploy key to the", output_text) data["csrf_token"] = csrf_token # First, invalid SSH key output = self.app.post("/test/adddeploykey", data=data) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn("Add deploy key to the", output_text) self.assertIn("SSH key invalid", output_text) # Next up, multiple SSH keys data[ "ssh_key" ] = "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQDAzBMSIlvPRaEiLOTVInErkRIw9CzQQcnslDekAn1jFnGf+SNa1acvbTiATbCX71AA03giKrPxPH79dxcC7aDXerc6zRcKjJs6MAL9PrCjnbyxCKXRNNZU5U9X/DLaaL1b3caB+WD6OoorhS3LTEtKPX8xyjOzhf3OQSzNjhJp5Q==\nssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQDAzBMSIlvPRaEiLOTVInErkRIw9CzQQcnslDekAn1jFnGf+SNa1acvbTiATbCX71AA03giKrPxPH79dxcC7aDXerc6zRcKjJs6MAL9PrCjnbyxCKXRNNZU5U9X/DLaaL1b3caB+WD6OoorhS3LTEtKPX8xyjOzhf3OQSzNjhJp5Q==" output = self.app.post( "/test/adddeploykey", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn("Please add single SSH keys.", output_text) # Now, a valid SSH key data[ "ssh_key" ] = "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQDAzBMSIlvPRaEiLOTVInErkRIw9CzQQcnslDekAn1jFnGf+SNa1acvbTiATbCX71AA03giKrPxPH79dxcC7aDXerc6zRcKjJs6MAL9PrCjnbyxCKXRNNZU5U9X/DLaaL1b3caB+WD6OoorhS3LTEtKPX8xyjOzhf3OQSzNjhJp5Q==" output = self.app.post( "/test/adddeploykey", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn("SSH key added", output_text) self.assertNotIn("Push Access", output_text) # And now, adding the same key output = self.app.post( "/test/adddeploykey", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn("SSH key already exists", output_text) # And next, a key with push access data[ "ssh_key" ] = "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC9Xwc2RDzPBhlEDARfHldGjudIVoa04tqT1JVKGQmyllTFz7Rb8CngQL3e7zyNzotnhwYKHdoiLlPkVEiDee4dWMUe48ilqId+FJZQGhyv8fu4BoFdE1AJUVylzmltbLg14VqG5gjTpXgtlrEva9arKwBMHJjRYc8ScaSn3OgyQw==" data["pushaccess"] = "true" output = self.app.post( "/test/adddeploykey", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn("SSH key added", output_text) self.assertIn("Push Access", output_text) @patch("pagure.decorators.admin_session_timedout") @patch.dict("pagure.config.config", {"DEPLOY_KEY": False}) def test_add_deploykey_disabled(self, ast): """ Test the add_deploykey endpoint when it's disabled in the config. """ ast.return_value = False tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "repos")) user = tests.FakeUser(username="pingou") with tests.user_set(self.app.application, user): output = self.app.get("/test/adddeploykey") self.assertEqual(output.status_code, 404) output = self.app.post("/test/adddeploykey") self.assertEqual(output.status_code, 404) @patch("pagure.decorators.admin_session_timedout") @patch("pagure.lib.notify.log") def test_add_user(self, mock_log, ast): """ Test the add_user endpoint. """ ast.return_value = False # No git repo output = self.app.get("/foo/adduser") self.assertEqual(output.status_code, 404) tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "repos")) # User not logged in output = self.app.get("/test/adduser") self.assertEqual(output.status_code, 302) user = tests.FakeUser() with tests.user_set(self.app.application, user): output = self.app.get("/test/adduser") self.assertEqual(output.status_code, 403) ast.return_value = True output = self.app.get("/test/adduser") self.assertEqual(output.status_code, 302) # Redirect also happens for POST request output = self.app.post("/test/adduser") self.assertEqual(output.status_code, 302) # Need to do this un-authentified since our fake user isn't in the DB # Check the message flashed during the redirect output = self.app.get("/", follow_redirects=True) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn("Action canceled, try it " "again", output_text) ast.return_value = False user.username = "pingou" with tests.user_set(self.app.application, user): output = self.app.get("/test/adduser") self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn("Add user to the", output_text) csrf_token = output_text.split( 'name="csrf_token" type="hidden" value="' )[1].split('">')[0] data = {"user": "ralph"} # Missing access and no CSRF output = self.app.post("/test/adduser", data=data) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Add user - test - Pagure", output_text ) self.assertIn("Add user to the", output_text) # No CSRF output = self.app.post("/test/adduser", data=data) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Add user - test - Pagure", output_text ) # Missing access data["csrf_token"] = csrf_token output = self.app.post("/test/adduser", data=data) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Add user - test - Pagure", output_text ) self.assertIn("Add user to the", output_text) # Unknown user data["access"] = "commit" output = self.app.post("/test/adduser", data=data) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Add user - test - Pagure", output_text ) self.assertIn("Add user to the", output_text) self.assertIn("No user "ralph" found", output_text) # All correct data["user"] = "foo" output = self.app.post( "/test/adduser", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( '
Project Settings
', output_text, ) self.assertIn("User added", output_text) mock_log.assert_called_with(ANY, topic="project.user.added", msg=ANY) @patch("pagure.decorators.admin_session_timedout") def test_add_group_project_when_user_mngt_off(self, ast): """ Test the add_group_project endpoint when user management is turned off in the pagure instance""" pagure.config.config["ENABLE_USER_MNGT"] = False ast.return_value = False # No Git repo output = self.app.get("/foo/addgroup") self.assertEqual(output.status_code, 404) tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "repos")) # User not logged in output = self.app.get("/test/addgroup") self.assertEqual(output.status_code, 302) msg = pagure.lib.query.add_group( self.session, group_name="foo", group_type="bar", display_name="foo group", description=None, user="pingou", is_admin=False, blacklist=pagure.config.config["BLACKLISTED_GROUPS"], ) self.session.commit() self.assertEqual(msg, "User `pingou` added to the group `foo`.") user = tests.FakeUser(username="pingou") with tests.user_set(self.app.application, user): # just get the csrf token pagure.config.config["ENABLE_USER_MNGT"] = True output = self.app.get("/test/addgroup") output_text = output.get_data(as_text=True) csrf_token = output_text.split( 'name="csrf_token" type="hidden" value="' )[1].split('">')[0] pagure.config.config["ENABLE_USER_MNGT"] = False data = {"group": "ralph"} output = self.app.post("/test/addgroup", data=data) self.assertEqual(output.status_code, 404) data["csrf_token"] = csrf_token output = self.app.post("/test/addgroup", data=data) self.assertEqual(output.status_code, 404) data["group"] = "foo" output = self.app.post( "/test/addgroup", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 404) pagure.config.config["ENABLE_USER_MNGT"] = True @patch.dict("pagure.config.config", {"ENABLE_GROUP_MNGT": False}) @patch("pagure.decorators.admin_session_timedout") def test_add_group_project_grp_mngt_off(self, ast): """ Test the add_group_project endpoint when group management is turned off in the pagure instance""" ast.return_value = False tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "repos")) user = tests.FakeUser(username="pingou") with tests.user_set(self.app.application, user): data = { "group": "ralph", "access": "ticket", "csrf_token": self.get_csrf(), } output = self.app.post( "/test/addgroup", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Add group - test - Pagure", output_text ) self.assertIn("No group ralph found.", output_text) @patch("pagure.decorators.admin_session_timedout") def test_add_group_project(self, ast): """ Test the add_group_project endpoint. """ ast.return_value = False # No Git repo output = self.app.get("/foo/addgroup") self.assertEqual(output.status_code, 404) tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "repos")) # User not logged in output = self.app.get("/test/addgroup") self.assertEqual(output.status_code, 302) user = tests.FakeUser() with tests.user_set(self.app.application, user): output = self.app.get("/test/addgroup") self.assertEqual(output.status_code, 403) ast.return_value = True output = self.app.get("/test/addgroup") self.assertEqual(output.status_code, 302) # Redirect also happens for POST request output = self.app.post("/test/addgroup") self.assertEqual(output.status_code, 302) # Need to do this un-authentified since our fake user isn't in the DB # Check the message flashed during the redirect output = self.app.get("/", follow_redirects=True) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn("Action canceled, try it " "again", output_text) ast.return_value = False msg = pagure.lib.query.add_group( self.session, group_name="foo", display_name="foo group", description=None, group_type="bar", user="pingou", is_admin=False, blacklist=pagure.config.config["BLACKLISTED_GROUPS"], ) self.session.commit() self.assertEqual(msg, "User `pingou` added to the group `foo`.") user.username = "pingou" with tests.user_set(self.app.application, user): output = self.app.get("/test/addgroup") self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn("Add group to the", output_text) csrf_token = output_text.split( 'name="csrf_token" type="hidden" value="' )[1].split('">')[0] data = {"group": "ralph"} # Missing CSRF output = self.app.post("/test/addgroup", data=data) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Add group - test - Pagure", output_text ) self.assertIn("Add group to the", output_text) # Missing access data["csrf_token"] = csrf_token output = self.app.post("/test/addgroup", data=data) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Add group - test - Pagure", output_text ) self.assertIn("Add group to the", output_text) # All good data["access"] = "ticket" output = self.app.post( "/test/addgroup", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn("Group added", output_text) @patch("pagure.decorators.admin_session_timedout") def test_remove_user_when_user_mngt_off(self, ast): """ Test the remove_user endpoint when user management is turned off in the pagure instance""" pagure.config.config["ENABLE_USER_MNGT"] = False ast.return_value = False # Git repo not found output = self.app.post("/foo/dropuser/1") self.assertEqual(output.status_code, 404) user = tests.FakeUser(username="pingou") with tests.user_set(self.app.application, user): tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "repos")) output = self.app.post("/test/settings") output_text = output.get_data(as_text=True) csrf_token = output_text.split( 'name="csrf_token" type="hidden" value="' )[1].split('">')[0] data = {"csrf_token": csrf_token} output = self.app.post( "/test/dropuser/2", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 404) # User not logged in output = self.app.post("/test/dropuser/1") self.assertEqual(output.status_code, 302) # Add an user to a project repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.add_user_to_project( session=self.session, project=repo, new_user="foo", user="pingou" ) self.session.commit() self.assertEqual(msg, "User added") with tests.user_set(self.app.application, user): output = self.app.post("/test/dropuser/2", follow_redirects=True) self.assertEqual(output.status_code, 404) data = {"csrf_token": csrf_token} output = self.app.post( "/test/dropuser/2", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 404) pagure.config.config["ENABLE_USER_MNGT"] = True @patch("pagure.decorators.admin_session_timedout") def test_remove_deploykey(self, ast): """ Test the remove_deploykey endpoint. """ ast.return_value = False # Git repo not found output = self.app.post("/foo/dropdeploykey/1") self.assertEqual(output.status_code, 404) user = tests.FakeUser() with tests.user_set(self.app.application, user): output = self.app.post("/foo/dropdeploykey/1") self.assertEqual(output.status_code, 404) tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "repos")) output = self.app.post("/test/dropdeploykey/1") self.assertEqual(output.status_code, 403) ast.return_value = True output = self.app.post("/test/dropdeploykey/1") self.assertEqual(output.status_code, 302) ast.return_value = False # User not logged in output = self.app.post("/test/dropdeploykey/1") self.assertEqual(output.status_code, 302) user.username = "pingou" with tests.user_set(self.app.application, user): output = self.app.post("/test/settings") output_text = output.get_data(as_text=True) csrf_token = output_text.split( 'name="csrf_token" type="hidden" value="' )[1].split('">')[0] data = {"csrf_token": csrf_token} output = self.app.post( "/test/dropdeploykey/1", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn("Deploy key does not exist in project", output_text) # Add a deploy key to a project repo = pagure.lib.query.get_authorized_project(self.session, "test") pingou = pagure.lib.query.get_user(self.session, "pingou") msg = pagure.lib.query.add_sshkey_to_project_or_user( session=self.session, project=repo, ssh_key="ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQDAzBMSIlvPRaEiLOTVInErkRIw9CzQQcnslDekAn1jFnGf+SNa1acvbTiATbCX71AA03giKrPxPH79dxcC7aDXerc6zRcKjJs6MAL9PrCjnbyxCKXRNNZU5U9X/DLaaL1b3caB+WD6OoorhS3LTEtKPX8xyjOzhf3OQSzNjhJp5Q==", pushaccess=True, creator=pingou, ) self.session.commit() self.assertEqual(msg, "SSH key added") with tests.user_set(self.app.application, user): output = self.app.post( "/test/dropdeploykey/1", follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertNotIn("Deploy key removed", output_text) data = {"csrf_token": csrf_token} output = self.app.post( "/test/dropdeploykey/1", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn("Deploy key removed", output_text) @patch("pagure.decorators.admin_session_timedout") @patch.dict("pagure.config.config", {"DEPLOY_KEY": False}) def test_remove_deploykey_disabled(self, ast): """ Test the remove_deploykey endpoint when it's disabled in the config. """ ast.return_value = False tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "repos")) user = tests.FakeUser(username="pingou") with tests.user_set(self.app.application, user): output = self.app.post("/test/dropdeploykey/1") self.assertEqual(output.status_code, 404) @patch("pagure.decorators.admin_session_timedout") @patch("pagure.lib.notify.log") def test_remove_user(self, mock_log, ast): """ Test the remove_user endpoint. """ ast.return_value = False # Git repo not found output = self.app.post("/foo/dropuser/1") self.assertEqual(output.status_code, 404) user = tests.FakeUser() with tests.user_set(self.app.application, user): output = self.app.post("/foo/dropuser/1") self.assertEqual(output.status_code, 404) tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "repos")) output = self.app.post("/test/dropuser/1") self.assertEqual(output.status_code, 403) ast.return_value = True output = self.app.post("/test/dropuser/1") self.assertEqual(output.status_code, 302) ast.return_value = False # User not logged in output = self.app.post("/test/dropuser/1") self.assertEqual(output.status_code, 302) user.username = "pingou" with tests.user_set(self.app.application, user): output = self.app.post("/test/settings") output_text = output.get_data(as_text=True) csrf_token = output_text.split( 'name="csrf_token" type="hidden" value="' )[1].split('">')[0] data = {"csrf_token": csrf_token} output = self.app.post( "/test/dropuser/2", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn( "User does not have any " "access on the repo", output_text ) # Add an user to a project repo = pagure.lib.query.get_authorized_project(self.session, "test") self.assertEqual(len(repo.users), 0) msg = pagure.lib.query.add_user_to_project( session=self.session, project=repo, new_user="foo", user="pingou" ) self.session.commit() self.assertEqual(msg, "User added") self.assertEqual(len(repo.users), 1) with tests.user_set(self.app.application, user): output = self.app.post("/test/dropuser/2", follow_redirects=True) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertNotIn("User removed", output_text) self.assertIn('action="/test/dropuser/2">', output_text) repo = pagure.lib.query.get_authorized_project( self.session, "test" ) self.assertEqual(len(repo.users), 1) data = {"csrf_token": csrf_token} output = self.app.post( "/test/dropuser/2", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn("User removed", output_text) self.assertNotIn('action="/test/dropuser/2">', output_text) self.session.commit() repo = pagure.lib.query.get_authorized_project( self.session, "test" ) self.assertEqual(len(repo.users), 0) mock_log.assert_called_with(ANY, topic="project.user.removed", msg=ANY) @patch("pagure.decorators.admin_session_timedout") @patch("pagure.lib.notify.log") def test_remove_user_self(self, mock_log, ast): """ Test the remove_user endpoint when removing themselves. """ ast.return_value = False tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "repos")) # Add an user to a project repo = pagure.lib.query.get_authorized_project(self.session, "test") self.assertEqual(len(repo.users), 0) msg = pagure.lib.query.add_user_to_project( session=self.session, project=repo, new_user="foo", user="pingou" ) self.session.commit() self.assertEqual(msg, "User added") self.assertEqual(len(repo.users), 1) # Let user foo remove themselves user = tests.FakeUser(username="foo") with tests.user_set(self.app.application, user): csrf_token = self.get_csrf() data = {"csrf_token": csrf_token} output = self.app.post( "/test/dropuser/2", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Overview - test - Pagure", output_text ) self.assertIn( '

\ntest' "\n

", output_text, ) self.assertIn("User removed", output_text) self.session.commit() repo = pagure.lib.query.get_authorized_project(self.session, "test") self.assertEqual(len(repo.users), 0) mock_log.assert_called_with(ANY, topic="project.user.removed", msg=ANY) @patch("pagure.decorators.admin_session_timedout") def test_remove_group_project_when_user_mngt_off(self, ast): """ Test the remove_group_project endpoint when user management is turned off in the pagure instance""" pagure.config.config["ENABLE_USER_MNGT"] = False ast.return_value = False # No Git repo output = self.app.post("/foo/dropgroup/1") self.assertEqual(output.status_code, 404) tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "repos")) # User not logged in output = self.app.post("/test/dropgroup/1") self.assertEqual(output.status_code, 302) user = tests.FakeUser() user.username = "pingou" with tests.user_set(self.app.application, user): output = self.app.post("/test/settings") output_text = output.get_data(as_text=True) csrf_token = output_text.split( 'name="csrf_token" type="hidden" value="' )[1].split('">')[0] data = {"csrf_token": csrf_token} output = self.app.post( "/test/dropgroup/2", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 404) # Create the new group msg = pagure.lib.query.add_group( session=self.session, group_name="testgrp", group_type="user", display_name="testgrp group", description=None, user="pingou", is_admin=False, blacklist=[], ) self.assertEqual(msg, "User `pingou` added to the group `testgrp`.") self.session.commit() repo = pagure.lib.query.get_authorized_project(self.session, "test") # Add the group to a project msg = pagure.lib.query.add_group_to_project( session=self.session, project=repo, new_group="testgrp", user="pingou", ) self.session.commit() self.assertEqual(msg, "Group added") with tests.user_set(self.app.application, user): output = self.app.post("/test/dropgroup/1", follow_redirects=True) self.assertEqual(output.status_code, 404) data = {"csrf_token": csrf_token} output = self.app.post( "/test/dropgroup/1", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 404) pagure.config.config["ENABLE_USER_MNGT"] = True @patch("pagure.decorators.admin_session_timedout") def test_remove_group_project(self, ast): """ Test the remove_group_project endpoint. """ ast.return_value = False # No Git repo output = self.app.post("/foo/dropgroup/1") self.assertEqual(output.status_code, 404) user = tests.FakeUser() with tests.user_set(self.app.application, user): output = self.app.post("/foo/dropgroup/1") self.assertEqual(output.status_code, 404) tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "repos")) output = self.app.post("/test/dropgroup/1") self.assertEqual(output.status_code, 403) ast.return_value = True output = self.app.post("/test/dropgroup/1") self.assertEqual(output.status_code, 302) ast.return_value = False # User not logged in output = self.app.post("/test/dropgroup/1") self.assertEqual(output.status_code, 302) user.username = "pingou" with tests.user_set(self.app.application, user): output = self.app.post("/test/settings") output_text = output.get_data(as_text=True) csrf_token = output_text.split( 'name="csrf_token" type="hidden" value="' )[1].split('">')[0] data = {"csrf_token": csrf_token} output = self.app.post( "/test/dropgroup/2", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn( "" "Group does not seem to be part of this project", output_text, ) # Create the new group msg = pagure.lib.query.add_group( session=self.session, group_name="testgrp", group_type="user", display_name="testgrp group", description=None, user="pingou", is_admin=False, blacklist=[], ) self.assertEqual(msg, "User `pingou` added to the group `testgrp`.") self.session.commit() repo = pagure.lib.query.get_authorized_project(self.session, "test") # Add the group to a project msg = pagure.lib.query.add_group_to_project( session=self.session, project=repo, new_group="testgrp", user="pingou", ) self.session.commit() self.assertEqual(msg, "Group added") repo = pagure.lib.query.get_authorized_project(self.session, "test") self.assertEqual(len(repo.groups), 1) with tests.user_set(self.app.application, user): output = self.app.post("/test/dropgroup/1", follow_redirects=True) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn('action="/test/dropgroup/1">', output_text) self.assertNotIn("Group removed", output_text) repo = pagure.lib.query.get_authorized_project( self.session, "test" ) self.assertEqual(len(repo.groups), 1) data = {"csrf_token": csrf_token} output = self.app.post( "/test/dropgroup/1", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn("Group removed", output_text) self.assertNotIn('action="/test/dropgroup/1">', output_text) self.session.commit() repo = pagure.lib.query.get_authorized_project( self.session, "test" ) self.assertEqual(len(repo.groups), 0) @patch("pagure.decorators.admin_session_timedout") def test_update_project(self, ast): """ Test the update_project endpoint. """ ast.return_value = True # Git repo not found output = self.app.post("/foo/update") self.assertEqual(output.status_code, 404) user = tests.FakeUser() with tests.user_set(self.app.application, user): # Project does not exist output = self.app.post("/foo/update") self.assertEqual(output.status_code, 404) tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "repos")) # Session timed-out output = self.app.post("/test/update") self.assertEqual(output.status_code, 302) ast.return_value = False # Not allowed output = self.app.post("/test/update") self.assertEqual(output.status_code, 403) # User not logged in output = self.app.post("/test/update") self.assertEqual(output.status_code, 302) user.username = "pingou" with tests.user_set(self.app.application, user): output = self.app.post("/test/update", follow_redirects=True) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) csrf_token = output_text.split( 'name="csrf_token" type="hidden" value="' )[1].split('">')[0] data = { "description": "new description for test project #1", "csrf_token": csrf_token, } output = self.app.post( "/test/update", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn( '', output_text, ) self.assertIn("Project updated", output_text) # Edit the avatar_email data = { "description": "new description for test project #1", "avatar_email": "pingou@fp.o", "csrf_token": csrf_token, } output = self.app.post( "/test/update", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn( '', output_text, ) self.assertIn("Project updated", output_text) # Reset the avatar_email data = { "description": "new description for test project #1", "avatar_email": "", "csrf_token": csrf_token, } output = self.app.post( "/test/update", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn( '', output_text, ) self.assertIn("Project updated", output_text) @patch("pagure.decorators.admin_session_timedout") def test_update_project_update_tag(self, ast): """ Test the view_settings endpoint when updating the project's tags. We had an issue where when you add an existing tag to a project we were querying the wrong table in the database. It would thus not find the tag, would try to add it, and (rightfully) complain about duplicated content. This test ensure we are behaving properly. """ ast.return_value = False tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "repos")) user = tests.FakeUser(username="pingou") with tests.user_set(self.app.application, user): output = self.app.get("/test/settings") self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) csrf_token = output_text.split( 'name="csrf_token" type="hidden" value="' )[1].split('">')[0] # Add tag to a project so that they are added to the database data = { "csrf_token": csrf_token, "description": "Test project", "tags": "test,pagure,tag", } output = self.app.post( "/test/update", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn("Project updated", output_text) # Remove two of the tags of the project, they will still be in # the DB but not associated to this project data = { "csrf_token": csrf_token, "description": "Test project", "tags": "tag", } output = self.app.post( "/test/update", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn("Project updated", output_text) # Try re-adding the two tags, this used to fail before we fixed # it data = { "csrf_token": csrf_token, "description": "Test project", "tags": "test,pagure,tag", } output = self.app.post( "/test/update", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn("Project updated", output_text) @patch("pagure.decorators.admin_session_timedout") def test_view_settings(self, ast): """ Test the view_settings endpoint. """ ast.return_value = False # No Git repo output = self.app.get("/foo/settings") self.assertEqual(output.status_code, 404) user = tests.FakeUser() with tests.user_set(self.app.application, user): output = self.app.get("/foo/settings") self.assertEqual(output.status_code, 404) tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "repos")) output = self.app.get("/test/settings") self.assertEqual(output.status_code, 403) # User not logged in output = self.app.get("/test/settings") self.assertEqual(output.status_code, 302) user.username = "pingou" with tests.user_set(self.app.application, user): ast.return_value = True output = self.app.get("/test/settings") self.assertEqual(output.status_code, 302) ast.return_value = False output = self.app.get("/test/settings") self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) # Both checkbox checked before self.assertIn( '', output_text, ) self.assertIn( '', output_text, ) csrf_token = output_text.split( 'name="csrf_token" type="hidden" value="' )[1].split('">')[0] data = {} output = self.app.post( "/test/settings", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) # Both checkbox are still checked output = self.app.get("/test/settings", follow_redirects=True) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn( '', output_text, ) self.assertIn( '', output_text, ) data = {"csrf_token": csrf_token} output = self.app.post( "/test/settings", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Overview - test - Pagure", output_text ) self.assertIn( "Edited successfully " "settings of repo: test", output_text ) # Both checkbox are now un-checked output = self.app.get("/test/settings", follow_redirects=True) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn( '', output_text, ) self.assertIn( '', output_text, ) data = { "csrf_token": csrf_token, "pull_requests": "y", "issue_tracker": "y", } output = self.app.post( "/test/settings", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Overview - test - Pagure", output_text ) self.assertIn( "Edited successfully " "settings of repo: test", output_text ) # Both checkbox are again checked output = self.app.get("/test/settings", follow_redirects=True) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn( '', output_text, ) self.assertIn( '', output_text, ) @patch( "pagure.decorators.admin_session_timedout", MagicMock(return_value=False), ) def test_view_settings_custom_fields(self): """ Test the view_settings endpoint when the project has some custom field for issues. """ tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "repos")) repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.set_custom_key_fields( self.session, repo, ["bugzilla", "upstream", "reviewstatus"], ["link", "boolean", "list"], [ "unused data for non-list type", "", "ack", "nack", "needs review", ], [None, None, None], ) self.session.commit() self.assertEqual(msg, "List of custom fields updated") self.assertIsNotNone(repo.issue_keys) user = tests.FakeUser(username="pingou") with tests.user_set(self.app.application, user): output = self.app.get("/test/settings") self.assertEqual(output.status_code, 200) @patch("pagure.lib.git.generate_gitolite_acls") @patch("pagure.decorators.admin_session_timedout") def test_view_settings_pr_only(self, ast, gen_acl): """ Test the view_settings endpoint when turning on PR only. """ ast.return_value = False tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "repos")) user = tests.FakeUser(username="pingou") with tests.user_set(self.app.application, user): output = self.app.get("/test/settings") self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) csrf_token = self.get_csrf(output=output) data = { "csrf_token": csrf_token, "pull_requests": "y", "issue_tracker": "y", "pull_request_access_only": "y", } output = self.app.post( "/test/settings", data=data, follow_redirects=True ) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Overview - test - Pagure", output_text ) self.assertIn( "Edited successfully " "settings of repo: test", output_text ) # Both checkbox are again checked output = self.app.get("/test/settings", follow_redirects=True) self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) self.assertIn( '', output_text, ) self.assertIn( '', output_text, ) self.assertIn( '', output_text, ) repo = pagure.lib.query.get_authorized_project( self.session, "test" ) self.assertEqual(gen_acl.call_count, 1) args = gen_acl.call_args self.assertEqual(args[0], tuple()) self.assertListEqual(list(args[1]), ["project"]) self.assertEqual(args[1]["project"].fullname, "test") @patch("pagure.decorators.admin_session_timedout") def test_fields_in_view_settings(self, ast): """ Test the default fields in view_settings endpoint. """ ast.return_value = False # No Git repo output = self.app.get("/foo/settings") self.assertEqual(output.status_code, 404) user = tests.FakeUser() with tests.user_set(self.app.application, user): output = self.app.get("/foo/settings") self.assertEqual(output.status_code, 404) item = pagure.lib.model.Project( user_id=1, # pingou name="test", description="test project #1", hook_token="aaabbbccc", ) self.session.add(item) self.session.commit() tests.create_projects_git(os.path.join(self.path, "repos")) output = self.app.get("/test/settings") self.assertEqual(output.status_code, 403) # User not logged in output = self.app.get("/test/settings") self.assertEqual(output.status_code, 302) user.username = "pingou" with tests.user_set(self.app.application, user): ast.return_value = True output = self.app.get("/test/settings") self.assertEqual(output.status_code, 302) ast.return_value = False output = self.app.get("/test/settings") self.assertEqual(output.status_code, 200) output_text = output.get_data(as_text=True) self.assertIn( "Settings - test - Pagure", output_text ) self.assertIn( '
Project Settings
', output_text, ) # Check that the priorities have their empty fields self.assertIn( """