# -*- coding: utf-8 -*- """ (c) 2015 - Copyright Red Hat Inc Authors: Pierre-Yves Chibon """ from __future__ import unicode_literals, absolute_import import arrow import copy import datetime import unittest import shutil import sys import time import os import flask import json import munch from mock import patch, MagicMock from sqlalchemy.exc import SQLAlchemyError sys.path.insert( 0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..") ) import pagure.lib.query import tests FULL_ISSUE_LIST = [ { "assignee": None, "blocks": [], "close_status": None, "closed_at": None, "closed_by": None, "comments": [], "content": "We should work on this", "custom_fields": [], "date_created": "1431414800", "depends": [], "id": 2, "last_updated": "1431414800", "milestone": None, "priority": None, "private": True, "related_prs": [], "status": "Closed", "tags": [], "title": "Test issue", "user": {"fullname": "PY C", "name": "pingou"}, }, { "assignee": {"fullname": "foo bar", "name": "foo"}, "blocks": [], "close_status": None, "closed_at": None, "closed_by": None, "comments": [], "content": "This issue needs attention", "custom_fields": [], "date_created": "1431414800", "depends": [], "id": 8, "last_updated": "1431414800", "milestone": None, "priority": None, "private": True, "related_prs": [], "status": "Open", "tags": [], "title": "test issue1", "user": {"fullname": "PY C", "name": "pingou"}, }, { "assignee": None, "blocks": [], "close_status": None, "closed_at": None, "closed_by": None, "comments": [], "content": "This issue needs attention", "custom_fields": [], "date_created": "1431414800", "depends": [], "id": 7, "last_updated": "1431414800", "milestone": None, "priority": None, "private": True, "related_prs": [], "status": "Open", "tags": [], "title": "test issue", "user": {"fullname": "PY C", "name": "pingou"}, }, { "assignee": None, "blocks": [], "close_status": None, "closed_at": None, "closed_by": None, "comments": [], "content": "This issue needs attention", "custom_fields": [], "date_created": "1431414800", "depends": [], "id": 6, "last_updated": "1431414800", "milestone": None, "priority": None, "private": False, "related_prs": [], "status": "Open", "tags": [], "title": "test issue", "user": {"fullname": "PY C", "name": "pingou"}, }, { "assignee": None, "blocks": [], "close_status": None, "closed_at": None, "closed_by": None, "comments": [], "content": "This issue needs attention", "custom_fields": [], "date_created": "1431414800", "depends": [], "id": 5, "last_updated": "1431414800", "milestone": None, "priority": None, "private": False, "related_prs": [], "status": "Open", "tags": [], "title": "test issue", "user": {"fullname": "PY C", "name": "pingou"}, }, { "assignee": None, "blocks": [], "close_status": None, "closed_at": None, "closed_by": None, "comments": [], "content": "This issue needs attention", "custom_fields": [], "date_created": "1431414800", "depends": [], "id": 4, "last_updated": "1431414800", "milestone": None, "priority": None, "private": False, "related_prs": [], "status": "Open", "tags": [], "title": "test issue", "user": {"fullname": "PY C", "name": "pingou"}, }, { "assignee": None, "blocks": [], "close_status": None, "closed_at": None, "closed_by": None, "comments": [], "content": "This issue needs attention", "custom_fields": [], "date_created": "1431414800", "depends": [], "id": 3, "last_updated": "1431414800", "milestone": None, "priority": None, "private": False, "related_prs": [], "status": "Open", "tags": [], "title": "test issue", "user": {"fullname": "PY C", "name": "pingou"}, }, { "assignee": None, "blocks": [], "close_status": None, "closed_at": None, "closed_by": None, "comments": [], "content": "This issue needs attention", "custom_fields": [], "date_created": "1431414800", "depends": [], "id": 2, "last_updated": "1431414800", "milestone": "milestone-1.0", "priority": None, "private": False, "related_prs": [], "status": "Open", "tags": [], "title": "test issue", "user": {"fullname": "PY C", "name": "pingou"}, }, { "assignee": None, "blocks": [], "close_status": None, "closed_at": None, "closed_by": None, "comments": [], "content": "This issue needs attention", "custom_fields": [], "date_created": "1431414800", "depends": [], "id": 1, "last_updated": "1431414800", "milestone": None, "priority": None, "private": False, "related_prs": [], "status": "Open", "tags": [], "title": "test issue", "user": {"fullname": "PY C", "name": "pingou"}, }, ] LCL_ISSUES = [ { "assignee": None, "blocks": [], "close_status": None, "closed_at": None, "closed_by": None, "comments": [], "content": "Description", "custom_fields": [], "date_created": "1431414800", "depends": [], "id": 2, "last_updated": "1431414800", "milestone": None, "priority": None, "private": False, "related_prs": [], "status": "Open", "tags": [], "title": "Issue #2", "user": {"fullname": "PY C", "name": "pingou"}, }, { "assignee": None, "blocks": [], "close_status": None, "closed_at": None, "closed_by": None, "comments": [], "content": "Description", "custom_fields": [], "date_created": "1431414800", "depends": [], "id": 1, "last_updated": "1431414800", "milestone": None, "priority": None, "private": False, "related_prs": [], "status": "Open", "tags": [], "title": "Issue #1", "user": {"fullname": "PY C", "name": "pingou"}, }, ] class PagureFlaskApiIssuetests(tests.SimplePagureTest): """ Tests for the flask API of pagure for issue """ maxDiff = None def setUp(self): """ Set up the environnment, ran before every tests. """ super(PagureFlaskApiIssuetests, self).setUp() pagure.config.config["TICKETS_FOLDER"] = None def test_api_new_issue_wrong_token(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # Valid token, wrong project output = self.app.post("/api/0/test2/new_issue", headers=headers) self.assertEqual(output.status_code, 401) data = json.loads(output.get_data(as_text=True)) self.assertEqual(sorted(data.keys()), ["error", "error_code"]) self.assertEqual(pagure.api.APIERROR.EINVALIDTOK.value, data["error"]) self.assertEqual( pagure.api.APIERROR.EINVALIDTOK.name, data["error_code"] ) @patch.dict( "pagure.config.config", {"ENABLE_TICKETS_NAMESPACE": ["foobar"]} ) def test_api_new_issue_wrong_namespace(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # Valid token, wrong project output = self.app.post( "/api/0/somenamespace/test3/new_issue", headers=headers ) self.assertEqual(output.status_code, 404) data = json.loads(output.get_data(as_text=True)) self.assertEqual(sorted(data.keys()), ["error", "error_code"]) self.assertEqual( pagure.api.APIERROR.ETRACKERDISABLED.value, data["error"] ) self.assertEqual( pagure.api.APIERROR.ETRACKERDISABLED.name, data["error_code"] ) def test_api_new_issue_no_input(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # No input output = self.app.post("/api/0/test/new_issue", headers=headers) self.assertEqual(output.status_code, 400) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, { "error": "Invalid or incomplete input submitted", "error_code": "EINVALIDREQ", "errors": { "issue_content": ["This field is required."], "title": ["This field is required."], }, }, ) def test_api_new_issue_invalid_repo(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} data = {"title": "test issue"} # Invalid repo output = self.app.post( "/api/0/foo/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 404) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"error": "Project not found", "error_code": "ENOPROJECT"} ) def test_api_new_issue_invalid_request(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # Incomplete request output = self.app.post( "/api/0/test/new_issue", data={}, headers=headers ) self.assertEqual(output.status_code, 400) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, { "error": "Invalid or incomplete input submitted", "error_code": "EINVALIDREQ", "errors": { "issue_content": ["This field is required."], "title": ["This field is required."], }, }, ) def test_api_new_issue(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} data = { "title": "test issue", "issue_content": "This issue needs attention", } # Valid request output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" self.assertDictEqual( data, {"issue": FULL_ISSUE_LIST[8], "message": "Issue created"} ) def test_api_new_issue_img(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} with open(os.path.join(tests.HERE, "placebo.png"), "rb") as stream: data = { "title": "test issue", "issue_content": "This issue needs attention ", "filestream": stream, } # Valid request output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" issue = copy.deepcopy(FULL_ISSUE_LIST[8]) issue["id"] = 1 self.assertIn( "_tests_placebo.png)](/test/issue/raw/files/" "8a06845923010b27bfd8e7e75acff7badc40d1021b4994e01f5e11ca" "40bc3abe", data["issue"]["content"], ) data["issue"]["content"] = "This issue needs attention" self.assertDictEqual( data, {"issue": issue, "message": "Issue created"} ) def test_api_new_issue_invalid_milestone(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # Valid request but invalid milestone data = { "title": "test issue", "issue_content": "This issue needs attention", "milestone": ["milestone-1.0"], } output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 400) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, { "error": "Invalid or incomplete input submitted", "error_code": "EINVALIDREQ", "errors": {"milestone": ["Not a valid choice"]}, }, ) def test_api_new_issue_milestone(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # Set some milestones repo = pagure.lib.query.get_authorized_project(self.session, "test") repo.milestones = {"milestone-1.0": "", "milestone-2.0": "Tomorrow!"} self.session.add(repo) self.session.commit() # Valid request with milestone data = { "title": "test issue", "issue_content": "This issue needs attention", "milestone": ["milestone-1.0"], } output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" issue = copy.deepcopy(FULL_ISSUE_LIST[7]) issue["id"] = 1 self.assertDictEqual( data, {"issue": issue, "message": "Issue created"} ) def test_api_new_issue_public(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # Valid request, with private='false' data = { "title": "test issue", "issue_content": "This issue needs attention", "private": "false", } output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" issue = copy.deepcopy(FULL_ISSUE_LIST[6]) issue["id"] = 1 self.assertDictEqual( data, {"issue": issue, "message": "Issue created"} ) # Valid request, with private=False data = { "title": "test issue", "issue_content": "This issue needs attention", "private": False, } output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" issue = copy.deepcopy(FULL_ISSUE_LIST[5]) issue["id"] = 2 self.assertDictEqual( data, {"issue": issue, "message": "Issue created"} ) # Valid request, with private='False' data = { "title": "test issue", "issue_content": "This issue needs attention", "private": "False", } output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" issue = copy.deepcopy(FULL_ISSUE_LIST[4]) issue["id"] = 3 self.assertDictEqual( data, {"issue": issue, "message": "Issue created"} ) # Valid request, with private=0 data = { "title": "test issue", "issue_content": "This issue needs attention", "private": 0, } output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" issue = copy.deepcopy(FULL_ISSUE_LIST[3]) issue["id"] = 4 self.assertDictEqual( data, {"issue": issue, "message": "Issue created"} ) def test_api_new_issue_private(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # Private issue: True data = { "title": "test issue", "issue_content": "This issue needs attention", "private": True, } output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" issue = copy.deepcopy(FULL_ISSUE_LIST[2]) issue["id"] = 1 self.assertDictEqual( data, {"issue": issue, "message": "Issue created"} ) # Private issue: 1 data = { "title": "test issue1", "issue_content": "This issue needs attention", "private": 1, "assignee": "foo", } output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" exp = copy.deepcopy(FULL_ISSUE_LIST[1]) exp["id"] = 2 self.assertDictEqual(data, {"issue": exp, "message": "Issue created"}) @patch("pagure.utils.check_api_acls", MagicMock(return_value=None)) def test_api_new_issue_raise_db_error(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} data = { "title": "test issue", "issue_content": "This issue needs attention", } with self._app.test_request_context("/") as ctx: flask.g.session = self.session flask.g.fas_user = tests.FakeUser(username="foo") with patch( "flask.g.session.commit", MagicMock(side_effect=SQLAlchemyError("DB error")), ): output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 400) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, { "error": "An error occurred at the database " "level and prevent the action from reaching " "completion", "error_code": "EDBERROR", }, ) def test_api_new_issue_user_token_no_input(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session, project_id=None) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # Valid token, invalid request - No input output = self.app.post("/api/0/test2/new_issue", headers=headers) self.assertEqual(output.status_code, 400) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, { "error": "Invalid or incomplete input submitted", "error_code": "EINVALIDREQ", "errors": { "issue_content": ["This field is required."], "title": ["This field is required."], }, }, ) def test_api_new_issue_user_token_invalid_user(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session, project_id=None) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # Another project, still an invalid request - No input output = self.app.post("/api/0/test/new_issue", headers=headers) self.assertEqual(output.status_code, 400) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, { "error": "Invalid or incomplete input submitted", "error_code": "EINVALIDREQ", "errors": { "issue_content": ["This field is required."], "title": ["This field is required."], }, }, ) def test_api_new_issue_user_token_invalid_repo(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session, project_id=None) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} data = {"title": "test issue"} # Invalid repo output = self.app.post( "/api/0/foo/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 404) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"error": "Project not found", "error_code": "ENOPROJECT"} ) def test_api_new_issue_user_token_invalid_request(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session, project_id=None) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # Incomplete request output = self.app.post( "/api/0/test/new_issue", data={}, headers=headers ) self.assertEqual(output.status_code, 400) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, { "error": "Invalid or incomplete input submitted", "error_code": "EINVALIDREQ", "errors": { "issue_content": ["This field is required."], "title": ["This field is required."], }, }, ) def test_api_new_issue_user_token(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session, project_id=None) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} data = { "title": "test issue", "issue_content": "This issue needs attention", } # Valid request output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" self.assertDictEqual( data, {"issue": FULL_ISSUE_LIST[8], "message": "Issue created"} ) def test_api_new_issue_user_token_milestone(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session, project_id=None) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # Set some milestones repo = pagure.lib.query.get_authorized_project(self.session, "test") repo.milestones = {"milestone-1.0": "", "milestone-2.0": "Tomorrow!"} self.session.add(repo) self.session.commit() # Valid request with milestone data = { "title": "test issue", "issue_content": "This issue needs attention", "milestone": ["milestone-1.0"], } output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" issue = copy.deepcopy(FULL_ISSUE_LIST[7]) issue["id"] = 1 self.assertDictEqual( data, {"issue": issue, "message": "Issue created"} ) def test_api_new_issue_user_token_public(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session, project_id=None) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # Valid request, with private='false' data = { "title": "test issue", "issue_content": "This issue needs attention", "private": "false", } output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" issue = copy.deepcopy(FULL_ISSUE_LIST[6]) issue["id"] = 1 self.assertDictEqual( data, {"issue": issue, "message": "Issue created"} ) # Valid request, with private=False data = { "title": "test issue", "issue_content": "This issue needs attention", "private": False, } output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" issue = copy.deepcopy(FULL_ISSUE_LIST[5]) issue["id"] = 2 self.assertDictEqual( data, {"issue": issue, "message": "Issue created"} ) # Valid request, with private='False' data = { "title": "test issue", "issue_content": "This issue needs attention", "private": "False", } output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" issue = copy.deepcopy(FULL_ISSUE_LIST[4]) issue["id"] = 3 self.assertDictEqual( data, {"issue": issue, "message": "Issue created"} ) # Valid request, with private=0 data = { "title": "test issue", "issue_content": "This issue needs attention", "private": 0, } output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" issue = copy.deepcopy(FULL_ISSUE_LIST[4]) issue["id"] = 4 self.assertDictEqual( data, {"issue": issue, "message": "Issue created"} ) def test_api_new_issue_user_token_private(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session, project_id=None) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # Private issue: True data = { "title": "test issue", "issue_content": "This issue needs attention", "private": True, } output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" issue = copy.deepcopy(FULL_ISSUE_LIST[2]) issue["id"] = 1 self.assertDictEqual( data, {"issue": issue, "message": "Issue created"} ) # Private issue: 1 data = { "title": "test issue1", "issue_content": "This issue needs attention", "private": 1, "assignee": "foo", } output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" issue = copy.deepcopy(FULL_ISSUE_LIST[1]) issue["id"] = 2 self.assertDictEqual( data, {"issue": issue, "message": "Issue created"} ) # Private issue: 'true' data = { "title": "test issue1", "issue_content": "This issue needs attention", "private": "true", } output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" exp = copy.deepcopy(FULL_ISSUE_LIST[1]) exp["id"] = 3 exp["assignee"] = None self.assertDictEqual(data, {"issue": exp, "message": "Issue created"}) def test_api_view_issues(self): """ Test the api_view_issues method of the flask api. """ self.test_api_new_issue() # Invalid repo output = self.app.get("/api/0/foo/issues") self.assertEqual(output.status_code, 404) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"error": "Project not found", "error_code": "ENOPROJECT"} ) # List all opened issues output = self.app.get("/api/0/test/issues") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": None, "status": None, "tags": [], }, "issues": [FULL_ISSUE_LIST[8]], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 1, }, ) # Create private issue repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue", content="We should work on this", user="pingou", private=True, status="Closed", ) self.session.commit() self.assertEqual(msg.title, "Test issue") # Access issues un-authenticated output = self.app.get("/api/0/test/issues") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": None, "status": None, "tags": [], }, "issues": [FULL_ISSUE_LIST[8]], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 1, }, ) headers = {"Authorization": "token aaabbbccc"} # Access issues authenticated but non-existing token output = self.app.get("/api/0/test/issues", headers=headers) self.assertEqual(output.status_code, 401) # Create a new token for another user item = pagure.lib.model.Token( id="bar_token", user_id=2, project_id=1, expiration=datetime.datetime.utcnow() + datetime.timedelta(days=30), ) self.session.add(item) self.session.commit() headers = {"Authorization": "token bar_token"} # Access issues authenticated but wrong token output = self.app.get("/api/0/test/issues", headers=headers) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": None, "status": None, "tags": [], }, "issues": [FULL_ISSUE_LIST[8]], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 1, }, ) headers = {"Authorization": "token aaabbbcccddd"} # Access issues authenticated correctly output = self.app.get("/api/0/test/issues", headers=headers) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": None, "status": None, "tags": [], }, "issues": [FULL_ISSUE_LIST[8]], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 1, }, ) headers = {"Authorization": "token aaabbbccc"} # Access issues authenticated but non-existing token output = self.app.get("/api/0/test/issues", headers=headers) self.assertEqual(output.status_code, 401) # Create a new token for another user item = pagure.lib.model.Token( id="bar_token_foo", user_id=2, project_id=1, expiration=datetime.datetime.utcnow() + datetime.timedelta(days=30), ) self.session.add(item) self.session.commit() headers = {"Authorization": "token bar_token_foo"} # Access issues authenticated but wrong token output = self.app.get("/api/0/test/issues", headers=headers) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": None, "status": None, "tags": [], }, "issues": [FULL_ISSUE_LIST[8]], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 1, }, ) headers = {"Authorization": "token aaabbbcccddd"} # Access issues authenticated correctly output = self.app.get("/api/0/test/issues", headers=headers) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": None, "status": None, "tags": [], }, "issues": [FULL_ISSUE_LIST[8]], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 1, }, ) # List closed issue output = self.app.get( "/api/0/test/issues?status=Closed", headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issues"][0]["date_created"] = "1431414800" data["issues"][0]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": None, "status": "Closed", "tags": [], }, "issues": [FULL_ISSUE_LIST[0]], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 1, }, ) # List closed issue output = self.app.get( "/api/0/test/issues?status=Invalid", headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": None, "status": "Invalid", "tags": [], }, "issues": [], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 0, "per_page": 20, "prev": None, }, "total_issues": 0, }, ) # List all issues output = self.app.get("/api/0/test/issues?status=All", headers=headers) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["last_updated"] = "1431414800" data["issues"][idx]["date_created"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": None, "status": "All", "tags": [], }, "issues": [FULL_ISSUE_LIST[0], FULL_ISSUE_LIST[8]], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 2, }, ) def test_api_view_issues_user_token(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session, project_id=None) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} data = { "title": "test issue", "issue_content": "This issue needs attention", } # Create an issue output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" self.assertDictEqual( data, {"issue": FULL_ISSUE_LIST[8], "message": "Issue created"} ) # List all opened issues output = self.app.get("/api/0/test/issues") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": None, "status": None, "tags": [], }, "issues": [FULL_ISSUE_LIST[8]], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 1, }, ) def test_api_view_issues_private_user_token(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session, project_id=None) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} data = { "title": "test issue", "issue_content": "This issue needs attention", "private": True, } # Create an issue output = self.app.post( "/api/0/test/new_issue", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) lcl_issue = copy.deepcopy(FULL_ISSUE_LIST[8]) lcl_issue["private"] = True data["issue"]["date_created"] = "1431414800" data["issue"]["last_updated"] = "1431414800" self.assertDictEqual( data, {"issue": lcl_issue, "message": "Issue created"} ) # List all opened issues - unauth output = self.app.get("/api/0/test/issues") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": None, "status": None, "tags": [], }, "issues": [], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 0, "per_page": 20, "prev": None, }, "total_issues": 0, }, ) # List all opened issues - auth output = self.app.get("/api/0/test/issues", headers=headers) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": None, "status": None, "tags": [], }, "issues": [lcl_issue], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 1, }, ) def test_api_view_issues_since_invalid_format(self): """ Test the api_view_issues method of the flask api. """ self.test_api_new_issue() # Invalid repo output = self.app.get("/api/0/test/issues?since=12-13") self.assertEqual(output.status_code, 400) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"error": "Invalid datetime format", "error_code": "EDATETIME"}, ) def test_api_view_issues_since_invalid_timestamp(self): """ Test the api_view_issues method of the flask api. """ self.test_api_new_issue() # Invalid repo output = self.app.get("/api/0/test/issues?since=100000000000000") self.assertEqual(output.status_code, 400) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"error": "Invalid timestamp format", "error_code": "ETIMESTAMP"}, ) def test_api_view_issues_reversed(self): """ Test the api_view_issues method of the flask api. in reversed order. """ self.test_api_new_issue() headers = {"Authorization": "token aaabbbcccddd"} # List issues in reverse order output = self.app.get("/api/0/test/issues?order=asc", headers=headers) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["last_updated"] = "1431414800" data["issues"][idx]["date_created"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." expected = { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": "asc", "priority": None, "since": None, "status": None, "tags": [], }, "issues": [FULL_ISSUE_LIST[8]], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 1, } self.assertDictEqual(data, expected) def test_api_view_issues_milestone(self): """ Test the api_view_issues method of the flask api when filtering for a milestone. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) repo = pagure.lib.query.get_authorized_project(self.session, "test") # Create 2 tickets but only 1 has a milestone start = arrow.utcnow().timestamp issue = pagure.lib.model.Issue( id=pagure.lib.query.get_next_id(self.session, repo.id), project_id=repo.id, title="Issue #1", content="Description", user_id=1, # pingou uid="issue#1", private=False, ) self.session.add(issue) self.session.commit() issue = pagure.lib.model.Issue( id=pagure.lib.query.get_next_id(self.session, repo.id), project_id=repo.id, title="Issue #2", content="Description", user_id=1, # pingou uid="issue#2", private=False, milestone="v1.0", ) self.session.add(issue) self.session.commit() # List all opened issues output = self.app.get("/api/0/test/issues") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." lcl_issues = copy.deepcopy(LCL_ISSUES) lcl_issues[0]["milestone"] = "v1.0" self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": None, "status": None, "tags": [], }, "issues": lcl_issues, "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 2, }, ) # List all issues of the milestone v1.0 output = self.app.get("/api/0/test/issues?milestones=v1.0") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": ["v1.0"], "no_stones": None, "order": None, "priority": None, "since": None, "status": None, "tags": [], }, "issues": [lcl_issues[0]], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 1, }, ) def test_api_view_issues_priority(self): """ Test the api_view_issues method of the flask api when filtering for a priority. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) repo = pagure.lib.query.get_authorized_project(self.session, "test") # Create 2 tickets but only 1 has a priority start = arrow.utcnow().timestamp issue = pagure.lib.model.Issue( id=pagure.lib.query.get_next_id(self.session, repo.id), project_id=repo.id, title="Issue #1", content="Description", user_id=1, # pingou uid="issue#1", private=False, ) self.session.add(issue) self.session.commit() issue = pagure.lib.model.Issue( id=pagure.lib.query.get_next_id(self.session, repo.id), project_id=repo.id, title="Issue #2", content="Description", user_id=1, # pingou uid="issue#2", private=False, priority=1, ) self.session.add(issue) self.session.commit() # Set some priorities to the project repo.priorities = {"1": "High", "2": "Normal"} self.session.add(repo) self.session.commit() # List all opened issues output = self.app.get("/api/0/test/issues") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." lcl_issues = copy.deepcopy(LCL_ISSUES) lcl_issues[0]["priority"] = 1 self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": None, "status": None, "tags": [], }, "issues": lcl_issues, "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 2, }, ) # List all issues of the priority high (ie: 1) output = self.app.get("/api/0/test/issues?priority=high") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": "high", "since": None, "status": None, "tags": [], }, "issues": [lcl_issues[0]], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 1, }, ) output = self.app.get("/api/0/test/issues?priority=1") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": "1", "since": None, "status": None, "tags": [], }, "issues": [lcl_issues[0]], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 1, }, ) def test_api_view_issues_priority_invalid(self): """ Test the api_view_issues method of the flask api when filtering for an invalid priority. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) # Try getting issues with an invalid priority output = self.app.get("/api/0/test/issues?priority=foobar") self.assertEqual(output.status_code, 400) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, { "error": "Invalid priority submitted", "error_code": "EINVALIDPRIORITY", }, ) def test_api_view_issues_no_stones(self): """ Test the api_view_issues method of the flask api when filtering with no_stones. """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) repo = pagure.lib.query.get_authorized_project(self.session, "test") # Create 2 tickets but only 1 has a milestone start = arrow.utcnow().timestamp issue = pagure.lib.model.Issue( id=pagure.lib.query.get_next_id(self.session, repo.id), project_id=repo.id, title="Issue #1", content="Description", user_id=1, # pingou uid="issue#1", private=False, ) self.session.add(issue) self.session.commit() issue = pagure.lib.model.Issue( id=pagure.lib.query.get_next_id(self.session, repo.id), project_id=repo.id, title="Issue #2", content="Description", user_id=1, # pingou uid="issue#2", private=False, milestone="v1.0", ) self.session.add(issue) self.session.commit() # List all opened issues output = self.app.get("/api/0/test/issues") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." lcl_issues = copy.deepcopy(LCL_ISSUES) lcl_issues[0]["milestone"] = "v1.0" self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": None, "status": None, "tags": [], }, "issues": lcl_issues, "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 2, }, ) # List all issues with no milestone output = self.app.get("/api/0/test/issues?no_stones=1") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": True, "order": None, "priority": None, "since": None, "status": None, "tags": [], }, "issues": [lcl_issues[1]], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 1, }, ) # List all issues with a milestone output = self.app.get("/api/0/test/issues?no_stones=0") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": False, "order": None, "priority": None, "since": None, "status": None, "tags": [], }, "issues": [lcl_issues[0]], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 1, }, ) def test_api_view_issues_since(self): """ Test the api_view_issues method of the flask api for since option """ tests.create_projects(self.session) tests.create_projects_git( os.path.join(self.path, "tickets"), bare=True ) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) repo = pagure.lib.query.get_authorized_project(self.session, "test") # Create 1st tickets start = arrow.utcnow().timestamp issue = pagure.lib.model.Issue( id=pagure.lib.query.get_next_id(self.session, repo.id), project_id=repo.id, title="Issue #1", content="Description", user_id=1, # pingou uid="issue#1", private=False, ) self.session.add(issue) self.session.commit() time.sleep(1) middle = arrow.utcnow().timestamp # Create 2nd tickets issue = pagure.lib.model.Issue( id=pagure.lib.query.get_next_id(self.session, repo.id), project_id=repo.id, title="Issue #2", content="Description", user_id=1, # pingou uid="issue#2", private=False, ) self.session.add(issue) self.session.commit() time.sleep(1) final = arrow.utcnow().timestamp # Create private issue issue = pagure.lib.model.Issue( id=pagure.lib.query.get_next_id(self.session, repo.id), project_id=repo.id, title="Issue #3", content="Description", user_id=1, # pingou uid="issue#3", private=True, ) self.session.add(issue) self.session.commit() # Invalid repo output = self.app.get("/api/0/foo/issues") self.assertEqual(output.status_code, 404) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"error": "Project not found", "error_code": "ENOPROJECT"} ) # List all opened issues output = self.app.get("/api/0/test/issues") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": None, "status": None, "tags": [], }, "issues": LCL_ISSUES, "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 2, }, ) time.sleep(1) late = arrow.utcnow().timestamp # List all opened issues from the start output = self.app.get("/api/0/test/issues?since=%s" % start) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": str(start), "status": None, "tags": [], }, "issues": LCL_ISSUES, "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 2, }, ) # List all opened issues from the middle output = self.app.get("/api/0/test/issues?since=%s" % middle) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": str(middle), "status": None, "tags": [], }, "issues": LCL_ISSUES[:1], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 1, }, ) # List all opened issues at the end output = self.app.get("/api/0/test/issues?since=%s" % final) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["date_created"] = "1431414800" data["issues"][idx]["last_updated"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": str(final), "status": None, "tags": [], }, "issues": [], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 0, "per_page": 20, "prev": None, }, "total_issues": 0, }, ) headers = {"Authorization": "token aaabbbcccddd"} # Test since for a value before creation of issues output = self.app.get( "/api/0/test/issues?since=%s" % final, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for idx in range(len(data["issues"])): data["issues"][idx]["last_updated"] = "1431414800" data["issues"][idx]["date_created"] = "1431414800" for k in ["first", "last"]: self.assertIsNotNone(data["pagination"][k]) data["pagination"][k] = "http://localhost..." self.assertDictEqual( data, { "args": { "assignee": None, "author": None, "milestones": [], "no_stones": None, "order": None, "priority": None, "since": str(final), "status": None, "tags": [], }, "issues": [ { "assignee": None, "blocks": [], "close_status": None, "closed_at": None, "closed_by": None, "comments": [], "content": "Description", "custom_fields": [], "date_created": "1431414800", "depends": [], "id": 3, "last_updated": "1431414800", "milestone": None, "priority": None, "private": True, "related_prs": [], "status": "Open", "tags": [], "title": "Issue #3", "user": {"fullname": "PY C", "name": "pingou"}, } ], "pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues": 1, }, ) def test_api_view_issue(self): """ Test the api_view_issue method of the flask api. """ self.test_api_new_issue() # Invalid repo output = self.app.get("/api/0/foo/issue/1") self.assertEqual(output.status_code, 404) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"error": "Project not found", "error_code": "ENOPROJECT"} ) # Invalid issue for this repo output = self.app.get("/api/0/test2/issue/1") self.assertEqual(output.status_code, 404) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"error": "Issue not found", "error_code": "ENOISSUE"} ) # Valid issue output = self.app.get("/api/0/test/issue/1") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["date_created"] = "1431414800" data["last_updated"] = "1431414800" self.assertDictEqual( data, { "assignee": None, "blocks": [], "comments": [], "content": "This issue needs attention", "custom_fields": [], "date_created": "1431414800", "close_status": None, "closed_at": None, "closed_by": None, "depends": [], "id": 1, "last_updated": "1431414800", "milestone": None, "priority": None, "private": False, "related_prs": [], "status": "Open", "tags": [], "title": "test issue", "user": {"fullname": "PY C", "name": "pingou"}, }, ) # Create private issue repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue", content="We should work on this", user="pingou", private=True, issue_uid="aaabbbccc", ) self.session.commit() self.assertEqual(msg.title, "Test issue") # Access private issue un-authenticated output = self.app.get("/api/0/test/issue/2") self.assertEqual(output.status_code, 403) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, { "error": "You are not allowed to view this issue", "error_code": "EISSUENOTALLOWED", }, ) headers = {"Authorization": "token aaabbbccc"} # Access private issue authenticated but non-existing token output = self.app.get("/api/0/test/issue/2", headers=headers) self.assertEqual(output.status_code, 401) data = json.loads(output.get_data(as_text=True)) self.assertEqual( sorted(data.keys()), ["error", "error_code", "errors"] ) self.assertEqual(pagure.api.APIERROR.EINVALIDTOK.value, data["error"]) self.assertEqual( pagure.api.APIERROR.EINVALIDTOK.name, data["error_code"] ) self.assertEqual(data["errors"], "Invalid token") # Create a new token for another user item = pagure.lib.model.Token( id="bar_token", user_id=2, project_id=1, expiration=datetime.datetime.utcnow() + datetime.timedelta(days=30), ) self.session.add(item) self.session.commit() headers = {"Authorization": "token bar_token"} # Access private issue authenticated but wrong token output = self.app.get("/api/0/test/issue/2", headers=headers) self.assertEqual(output.status_code, 403) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, { "error": "You are not allowed to view this issue", "error_code": "EISSUENOTALLOWED", }, ) headers = {"Authorization": "token aaabbbcccddd"} # Access private issue authenticated correctly output = self.app.get("/api/0/test/issue/2", headers=headers) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["date_created"] = "1431414800" data["last_updated"] = "1431414800" self.assertDictEqual( data, { "assignee": None, "blocks": [], "comments": [], "content": "We should work on this", "custom_fields": [], "date_created": "1431414800", "close_status": None, "closed_at": None, "closed_by": None, "depends": [], "id": 2, "last_updated": "1431414800", "milestone": None, "priority": None, "private": True, "related_prs": [], "status": "Open", "tags": [], "title": "Test issue", "user": {"fullname": "PY C", "name": "pingou"}, }, ) # Access private issue authenticated correctly using the issue's uid output = self.app.get("/api/0/test/issue/aaabbbccc", headers=headers) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["date_created"] = "1431414800" data["last_updated"] = "1431414800" self.assertDictEqual( data, { "assignee": None, "blocks": [], "comments": [], "content": "We should work on this", "custom_fields": [], "date_created": "1431414800", "close_status": None, "closed_at": None, "closed_by": None, "depends": [], "id": 2, "last_updated": "1431414800", "milestone": None, "priority": None, "private": True, "related_prs": [], "status": "Open", "tags": [], "title": "Test issue", "user": {"fullname": "PY C", "name": "pingou"}, }, ) def test_api_change_milestone_issue_invalid_project(self): """ Test the api_change_milestone_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "tickets")) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) # Set some milestones to the project repo = pagure.lib.query.get_authorized_project(self.session, "test") repo.milestones = {"v1.0": None, "v2.0": "Soon"} self.session.add(repo) self.session.commit() headers = {"Authorization": "token aaabbbcccddd"} # Invalid project output = self.app.post("/api/0/foo/issue/1/milestone", headers=headers) self.assertEqual(output.status_code, 404) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"error": "Project not found", "error_code": "ENOPROJECT"} ) @patch.dict( "pagure.config.config", {"ENABLE_TICKETS_NAMESPACE": ["foobar"]} ) def test_api_change_milestone_issue_wrong_namespace(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "tickets")) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) # Set some milestones to the project repo = pagure.lib.query.get_authorized_project( self.session, "test3", namespace="somenamespace" ) repo.milestones = {"v1.0": None, "v2.0": "Soon"} self.session.add(repo) self.session.commit() # Create normal issue repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue #1", content="We should work on this", user="pingou", private=False, ) self.session.commit() self.assertEqual(msg.title, "Test issue #1") headers = {"Authorization": "token aaabbbcccddd"} # Valid token, wrong project output = self.app.post( "/api/0/somenamespace/test3/issue/1/milestone", headers=headers ) self.assertEqual(output.status_code, 404) data = json.loads(output.get_data(as_text=True)) self.assertEqual(sorted(data.keys()), ["error", "error_code"]) self.assertEqual( pagure.api.APIERROR.ETRACKERDISABLED.value, data["error"] ) self.assertEqual( pagure.api.APIERROR.ETRACKERDISABLED.name, data["error_code"] ) def test_api_change_milestone_issue_wrong_token(self): """ Test the api_change_milestone_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "tickets")) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) # Set some milestones to the project repo = pagure.lib.query.get_authorized_project(self.session, "test") repo.milestones = {"v1.0": None, "v2.0": "Soon"} self.session.add(repo) self.session.commit() headers = {"Authorization": "token aaabbbcccddd"} # Valid token, wrong project output = self.app.post( "/api/0/test2/issue/1/milestone", headers=headers ) self.assertEqual(output.status_code, 401) data = json.loads(output.get_data(as_text=True)) self.assertEqual(sorted(data.keys()), ["error", "error_code"]) self.assertEqual(pagure.api.APIERROR.EINVALIDTOK.value, data["error"]) self.assertEqual( pagure.api.APIERROR.EINVALIDTOK.name, data["error_code"] ) def test_api_change_milestone_issue_no_issue(self): """ Test the api_change_milestone_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "tickets")) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) # Set some milestones to the project repo = pagure.lib.query.get_authorized_project(self.session, "test") repo.milestones = {"v1.0": None, "v2.0": "Soon"} self.session.add(repo) self.session.commit() headers = {"Authorization": "token aaabbbcccddd"} # No issue output = self.app.post( "/api/0/test/issue/1/milestone", headers=headers ) self.assertEqual(output.status_code, 404) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"error": "Issue not found", "error_code": "ENOISSUE"} ) def test_api_change_milestone_issue_no_milestone(self): """ Test the api_change_milestone_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "tickets")) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) # Set some milestones to the project repo = pagure.lib.query.get_authorized_project(self.session, "test") repo.milestones = {"v1.0": None, "v2.0": "Soon"} self.session.add(repo) self.session.commit() headers = {"Authorization": "token aaabbbcccddd"} # Create normal issue repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue #1", content="We should work on this", user="pingou", private=False, ) self.session.commit() self.assertEqual(msg.title, "Test issue #1") # Check milestone before repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(issue.milestone, None) data = {"milestone": ""} # Valid request but no milestone specified output = self.app.post( "/api/0/test/issue/1/milestone", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual(data, {"message": "No changes"}) # No change repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(issue.milestone, None) def test_api_change_milestone_issue_invalid_milestone(self): """ Test the api_change_milestone_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "tickets")) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) # Set some milestones to the project repo = pagure.lib.query.get_authorized_project(self.session, "test") repo.milestones = {"v1.0": None, "v2.0": "Soon"} self.session.add(repo) self.session.commit() headers = {"Authorization": "token aaabbbcccddd"} # Create normal issue repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue #1", content="We should work on this", user="pingou", private=False, ) self.session.commit() self.assertEqual(msg.title, "Test issue #1") # Check milestone before repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(issue.milestone, None) data = {"milestone": "milestone-1-0"} # Invalid milestone specified output = self.app.post( "/api/0/test/issue/1/milestone", data=data, headers=headers ) self.assertEqual(output.status_code, 400) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, { "error": "Invalid or incomplete input submitted", "error_code": "EINVALIDREQ", "errors": {"milestone": ["Not a valid choice"]}, }, ) def test_api_change_milestone_issue(self): """ Test the api_change_milestone_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "tickets")) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) # Set some milestones to the project repo = pagure.lib.query.get_authorized_project(self.session, "test") repo.milestones = {"v1.0": None, "v2.0": "Soon"} self.session.add(repo) self.session.commit() headers = {"Authorization": "token aaabbbcccddd"} # Create normal issue repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue #1", content="We should work on this", user="pingou", private=False, ) self.session.commit() self.assertEqual(msg.title, "Test issue #1") # Check milestone before repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(issue.milestone, None) data = {"milestone": "v1.0"} # Valid requests output = self.app.post( "/api/0/test/issue/1/milestone", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"message": ["Issue set to the milestone: v1.0"]} ) def test_api_change_milestone_issue_remove_milestone(self): """ Test the api_change_milestone_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "tickets")) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) # Set some milestones to the project repo = pagure.lib.query.get_authorized_project(self.session, "test") repo.milestones = {"v1.0": None, "v2.0": "Soon"} self.session.add(repo) self.session.commit() headers = {"Authorization": "token aaabbbcccddd"} # Create normal issue repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue #1", content="We should work on this", user="pingou", private=False, ) self.session.commit() self.assertEqual(msg.title, "Test issue #1") # Check milestone before repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(issue.milestone, None) data = {"milestone": "v1.0"} # Valid requests output = self.app.post( "/api/0/test/issue/1/milestone", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"message": ["Issue set to the milestone: v1.0"]} ) # remove milestone data = {"milestone": ""} # Valid requests output = self.app.post( "/api/0/test/issue/1/milestone", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"message": ["Issue set to the milestone: None (was: v1.0)"]} ) # Change recorded repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(issue.milestone, None) def test_api_change_milestone_issue_remove_milestone2(self): """ Test the api_change_milestone_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "tickets")) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) # Set some milestones to the project repo = pagure.lib.query.get_authorized_project(self.session, "test") repo.milestones = {"v1.0": None, "v2.0": "Soon"} self.session.add(repo) self.session.commit() headers = {"Authorization": "token aaabbbcccddd"} # Create normal issue repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue #1", content="We should work on this", user="pingou", private=False, ) self.session.commit() self.assertEqual(msg.title, "Test issue #1") # Check milestone before repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(issue.milestone, None) data = {"milestone": "v1.0"} # Valid requests output = self.app.post( "/api/0/test/issue/1/milestone", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"message": ["Issue set to the milestone: v1.0"]} ) # remove milestone by using no milestone in JSON data = {} # Valid requests output = self.app.post( "/api/0/test/issue/1/milestone", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"message": ["Issue set to the milestone: None (was: v1.0)"]} ) # Change recorded repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(issue.milestone, None) def test_api_change_milestone_issue_unauthorized(self): """ Test the api_change_milestone_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "tickets")) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) # Set some milestones to the project repo = pagure.lib.query.get_authorized_project(self.session, "test") repo.milestones = {"v1.0": None, "v2.0": "Soon"} self.session.add(repo) self.session.commit() headers = {"Authorization": "token aaabbbcccddd"} # Create normal issue repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue #1", content="We should work on this", user="pingou", private=False, ) self.session.commit() self.assertEqual(msg.title, "Test issue #1") headers = {"Authorization": "token pingou_foo"} data = {"milestone": "v1.0"} # Un-authorized issue output = self.app.post( "/api/0/foo/issue/1/milestone", data={}, headers=headers ) self.assertEqual(output.status_code, 401) data = json.loads(output.get_data(as_text=True)) self.assertEqual( sorted(data.keys()), ["error", "error_code", "errors"] ) self.assertEqual(pagure.api.APIERROR.EINVALIDTOK.value, data["error"]) self.assertEqual( pagure.api.APIERROR.EINVALIDTOK.name, data["error_code"] ) self.assertEqual(data["errors"], "Invalid token") @patch("pagure.lib.notify.send_email", MagicMock(return_value=True)) @patch( "pagure.lib.query.add_metadata_update_notif", MagicMock(side_effect=pagure.exceptions.PagureException("error")), ) def test_api_change_milestone_issue_raises_exception(self): """ Test the api_change_milestone_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "tickets")) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) # Set some milestones to the project repo = pagure.lib.query.get_authorized_project(self.session, "test") repo.milestones = {"v1.0": None, "v2.0": "Soon"} self.session.add(repo) self.session.commit() headers = {"Authorization": "token aaabbbcccddd"} # Create normal issue repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue #1", content="We should work on this", user="pingou", private=False, ) self.session.commit() self.assertEqual(msg.title, "Test issue #1") data = {"milestone": "v1.0"} # Valid requests output = self.app.post( "/api/0/test/issue/1/milestone", data=data, headers=headers ) self.assertEqual(output.status_code, 400) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual(data, {"error": "error", "error_code": "ENOCODE"}) def test_api_view_issue_related_prs(self): tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "tickets")) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) # Create issue repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue #1", content="We should work on this", user="pingou", private=False, issue_uid="aaabbbccc1", ) self.session.commit() self.assertEqual(msg.title, "Test issue #1") self.assertEqual(msg.related_prs, []) self.assertEqual(msg.id, 1) # Create pull request repo_to = pagure.lib.query.get_authorized_project(self.session, "test") repo_from = pagure.lib.query.get_authorized_project( self.session, "test" ) msg = pagure.lib.query.new_pull_request( session=self.session, repo_from=repo_from, repo_to=repo_to, branch_from="master", branch_to="master", title="New shiny feature", user="pingou", initial_comment="Fixes #1", ) self.session.commit() self.assertEqual(msg.id, 2) self.assertEqual(msg.title, "New shiny feature") output = self.app.get("/api/0/test/pull-request/2") self.assertEqual( json.loads(output.get_data(as_text=True)).get("initial_comment"), "Fixes #1", ) output = self.app.get("/api/0/test/issue/1") data = json.loads(output.get_data(as_text=True)) self.assertEqual( data.get("related_prs"), [{"id": 2, "title": "New shiny feature"}] ) @patch("pagure.lib.git.update_git") @patch("pagure.lib.notify.send_email") def test_api_view_issue_comment(self, p_send_email, p_ugt): """ Test the api_view_issue_comment endpoint. """ p_send_email.return_value = True p_ugt.return_value = True tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "tickets")) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) # Create normal issue in test repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue #1", content="We should work on this", user="pingou", private=False, issue_uid="aaabbbccc1", ) self.session.commit() self.assertEqual(msg.title, "Test issue #1") headers = {"Authorization": "token aaabbbcccddd"} data = {"comment": "This is a very interesting question"} # Valid request output = self.app.post( "/api/0/test/issue/1/comment", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["avatar_url"] = "https://seccdn.libravatar.org/avatar/..." self.assertDictEqual( data, { "message": "Comment added", "avatar_url": "https://seccdn.libravatar.org/avatar/...", "user": "pingou", }, ) # One comment added repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(len(issue.comments), 1) # View a comment that does not exist output = self.app.get("/api/0/foo/issue/100/comment/2") self.assertEqual(output.status_code, 404) # Issue exists but not the comment output = self.app.get("/api/0/test/issue/1/comment/2") self.assertEqual(output.status_code, 404) # Issue and comment exists output = self.app.get("/api/0/test/issue/1/comment/1") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["date_created"] = "1435821770" data["comment_date"] = "2015-07-02 09:22" data["avatar_url"] = "https://seccdn.libravatar.org/avatar/..." self.assertDictEqual( data, { "avatar_url": "https://seccdn.libravatar.org/avatar/...", "comment": "This is a very interesting question", "comment_date": "2015-07-02 09:22", "date_created": "1435821770", "edited_on": None, "editor": None, "notification": False, "id": 1, "parent": None, "reactions": {}, "user": {"fullname": "PY C", "name": "pingou"}, }, ) # Issue and comment exists, using UID output = self.app.get("/api/0/test/issue/aaabbbccc1/comment/1") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["date_created"] = "1435821770" data["comment_date"] = "2015-07-02 09:22" data["avatar_url"] = "https://seccdn.libravatar.org/avatar/..." self.assertDictEqual( data, { "avatar_url": "https://seccdn.libravatar.org/avatar/...", "comment": "This is a very interesting question", "comment_date": "2015-07-02 09:22", "date_created": "1435821770", "edited_on": None, "editor": None, "notification": False, "id": 1, "parent": None, "reactions": {}, "user": {"fullname": "PY C", "name": "pingou"}, }, ) @patch("pagure.lib.git.update_git") @patch("pagure.lib.notify.send_email") def test_api_view_issue_comment_private(self, p_send_email, p_ugt): """ Test the api_view_issue_comment endpoint. """ p_send_email.return_value = True p_ugt.return_value = True tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "tickets")) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) # Create normal issue in test repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue #1", content="We should work on this", user="foo", private=True, issue_uid="aaabbbccc1", ) self.session.commit() self.assertEqual(msg.title, "Test issue #1") # Create a token for another user item = pagure.lib.model.Token( id="foo_token_2", user_id=2, project_id=1, expiration=datetime.datetime.utcnow() + datetime.timedelta(days=30), ) self.session.add(item) self.session.commit() tests.create_tokens_acl(self.session, token_id="foo_token_2") # Add a comment to that issue data = {"comment": "This is a very interesting question"} headers = {"Authorization": "token foo_token_2"} output = self.app.post( "/api/0/test/issue/1/comment", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["avatar_url"] = "https://seccdn.libravatar.org/avatar/..." self.assertDictEqual( data, { "message": "Comment added", "avatar_url": "https://seccdn.libravatar.org/avatar/...", "user": "foo", }, ) # Private issue - no auth output = self.app.get("/api/0/test/issue/1/comment/2") self.assertEqual(output.status_code, 403) # Private issue - Auth - Invalid token headers = {"Authorization": "token aaabbbcccdddee"} output = self.app.get("/api/0/test/issue/1/comment/2", headers=headers) self.assertEqual(output.status_code, 401) # Private issue - Auth - valid token - unknown comment headers = {"Authorization": "token foo_token_2"} output = self.app.get("/api/0/test/issue/1/comment/3", headers=headers) self.assertEqual(output.status_code, 404) # Private issue - Auth - valid token - known comment headers = {"Authorization": "token foo_token_2"} output = self.app.get("/api/0/test/issue/1/comment/1", headers=headers) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["date_created"] = "1435821770" data["comment_date"] = "2015-07-02 09:22" data["avatar_url"] = "https://seccdn.libravatar.org/avatar/..." self.assertDictEqual( data, { "avatar_url": "https://seccdn.libravatar.org/avatar/...", "comment": "This is a very interesting question", "comment_date": "2015-07-02 09:22", "date_created": "1435821770", "edited_on": None, "editor": None, "notification": False, "id": 1, "parent": None, "reactions": {}, "user": {"fullname": "foo bar", "name": "foo"}, }, ) @patch.dict( "pagure.config.config", {"ENABLE_TICKETS_NAMESPACE": ["foobar"]} ) def test_api_assign_issue_wrong_namespace(self): """ Test the api_new_issue method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "tickets")) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) # Set some milestones to the project repo = pagure.lib.query.get_authorized_project( self.session, "test3", namespace="somenamespace" ) repo.milestones = {"v1.0": None, "v2.0": "Soon"} self.session.add(repo) self.session.commit() # Create normal issue repo = pagure.lib.query.get_authorized_project( self.session, "test3", namespace="somenamespace" ) msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue #1", content="We should work on this", user="pingou", private=False, ) self.session.commit() self.assertEqual(msg.title, "Test issue #1") headers = {"Authorization": "token aaabbbcccddd"} # Valid token, wrong project output = self.app.post( "/api/0/somenamespace/test3/issue/1/assign", headers=headers ) self.assertEqual(output.status_code, 404) data = json.loads(output.get_data(as_text=True)) self.assertEqual(sorted(data.keys()), ["error", "error_code"]) self.assertEqual( pagure.api.APIERROR.ETRACKERDISABLED.value, data["error"] ) self.assertEqual( pagure.api.APIERROR.ETRACKERDISABLED.name, data["error_code"] ) @patch("pagure.lib.git.update_git") @patch("pagure.lib.notify.send_email") def test_api_assign_issue(self, p_send_email, p_ugt): """ Test the api_assign_issue method of the flask api. """ p_send_email.return_value = True p_ugt.return_value = True tests.create_projects(self.session) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # Invalid project output = self.app.post("/api/0/foo/issue/1/assign", headers=headers) self.assertEqual(output.status_code, 404) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"error": "Project not found", "error_code": "ENOPROJECT"} ) # Valid token, wrong project output = self.app.post("/api/0/test2/issue/1/assign", headers=headers) self.assertEqual(output.status_code, 401) data = json.loads(output.get_data(as_text=True)) self.assertEqual(sorted(data.keys()), ["error", "error_code"]) self.assertEqual(pagure.api.APIERROR.EINVALIDTOK.value, data["error"]) self.assertEqual( pagure.api.APIERROR.EINVALIDTOK.name, data["error_code"] ) # No input output = self.app.post("/api/0/test/issue/1/assign", headers=headers) self.assertEqual(output.status_code, 404) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"error": "Issue not found", "error_code": "ENOISSUE"} ) # Create normal issue repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue #1", content="We should work on this", user="pingou", private=False, issue_uid="aaabbbccc1", ) self.session.commit() self.assertEqual(msg.title, "Test issue #1") # Check comments before repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(len(issue.comments), 0) # No change repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(issue.status, "Open") data = {"assignee": "pingou"} # Valid request output = self.app.post( "/api/0/test/issue/1/assign", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual(data, {"message": "Issue assigned to pingou"}) # Un-assign output = self.app.post( "/api/0/test/issue/1/assign", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual(data, {"message": "Assignee reset"}) # No change repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(issue.assignee, None) # Un-assign data = {"assignee": None} output = self.app.post( "/api/0/test/issue/1/assign", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual(data, {"message": "Nothing to change"}) repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(issue.assignee, None) # Re-assign for the rest of the tests data = {"assignee": "pingou"} output = self.app.post( "/api/0/test/issue/1/assign", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual(data, {"message": "Issue assigned to pingou"}) # Un-assign data = {"assignee": ""} output = self.app.post( "/api/0/test/issue/1/assign", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual(data, {"message": "Assignee reset"}) # Re-assign for the rest of the tests data = {"assignee": "pingou"} output = self.app.post( "/api/0/test/issue/1/assign", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual(data, {"message": "Issue assigned to pingou"}) # One comment added self.session.commit() repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(issue.assignee.user, "pingou") # Create another project item = pagure.lib.model.Project( user_id=2, # foo name="foo", description="test project #3", hook_token="aaabbbdddeee", ) self.session.add(item) self.session.commit() # Create a token for pingou for this project item = pagure.lib.model.Token( id="pingou_foo", user_id=1, project_id=4, expiration=datetime.datetime.utcnow() + datetime.timedelta(days=30), ) self.session.add(item) self.session.commit() # Give `issue_change_status` to this token when `issue_comment` # is required item = pagure.lib.model.TokenAcl(token_id="pingou_foo", acl_id=8) self.session.add(item) self.session.commit() repo = pagure.lib.query.get_authorized_project(self.session, "foo") # Create private issue msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue", content="We should work on this", user="foo", private=True, issue_uid="aaabbbccc#2", ) self.session.commit() self.assertEqual(msg.title, "Test issue") # Check before repo = pagure.lib.query.get_authorized_project(self.session, "foo") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(len(issue.comments), 0) data = {"assignee": "pingou"} headers = {"Authorization": "token pingou_foo"} # Valid request but un-authorized output = self.app.post( "/api/0/foo/issue/1/assign", data=data, headers=headers ) self.assertEqual(output.status_code, 401) data = json.loads(output.get_data(as_text=True)) self.assertEqual( sorted(data.keys()), ["error", "error_code", "errors"] ) self.assertEqual(pagure.api.APIERROR.EINVALIDTOK.value, data["error"]) self.assertEqual( pagure.api.APIERROR.EINVALIDTOK.name, data["error_code"] ) self.assertEqual( data["errors"], "Missing ACLs: issue_assign, issue_update" ) # No comment added repo = pagure.lib.query.get_authorized_project(self.session, "foo") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(len(issue.comments), 0) # Create token for user foo item = pagure.lib.model.Token( id="foo_token2", user_id=2, project_id=4, expiration=datetime.datetime.utcnow() + datetime.timedelta(days=30), ) self.session.add(item) self.session.commit() tests.create_tokens_acl(self.session, token_id="foo_token2") data = {"assignee": "pingou"} headers = {"Authorization": "token foo_token2"} # Valid request and authorized output = self.app.post( "/api/0/foo/issue/1/assign", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual(data, {"message": "Issue assigned to pingou"}) @patch("pagure.lib.git.update_git") @patch("pagure.lib.notify.send_email") def test_api_assign_issue_issuer(self, p_send_email, p_ugt): """ Test the api_assign_issue method of the flask api. """ p_send_email.return_value = True p_ugt.return_value = True tests.create_projects(self.session) tests.create_tokens(self.session, user_id=2) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # Create normal issue repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue #1", content="We should work on this", user="pingou", private=False, issue_uid="aaabbbccc1", assignee="foo", ) self.session.commit() self.assertEqual(msg.title, "Test issue #1") # Check comments before repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(len(issue.comments), 0) # Un-assign data = {"assignee": None} output = self.app.post( "/api/0/test/issue/1/assign", data={}, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual(data, {"message": "Assignee reset"}) # No longer allowed to self-assign since no access data = {"assignee": "foo"} output = self.app.post( "/api/0/test/issue/1/assign", data=data, headers=headers ) self.assertEqual(output.status_code, 403) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, { "error": "You are not allowed to view this issue", "error_code": "EISSUENOTALLOWED", }, ) @patch("pagure.lib.git.update_git") @patch("pagure.lib.notify.send_email") def test_api_subscribe_issue(self, p_send_email, p_ugt): """ Test the api_subscribe_issue method of the flask api. """ p_send_email.return_value = True p_ugt.return_value = True item = pagure.lib.model.User( user="bar", fullname="bar foo", password="foo", default_email="bar@bar.com", ) self.session.add(item) item = pagure.lib.model.UserEmail(user_id=3, email="bar@bar.com") self.session.add(item) self.session.commit() tests.create_projects(self.session) tests.create_tokens(self.session, user_id=3) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # Invalid project output = self.app.post("/api/0/foo/issue/1/subscribe", headers=headers) self.assertEqual(output.status_code, 404) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"error": "Project not found", "error_code": "ENOPROJECT"} ) # Valid token, wrong project output = self.app.post( "/api/0/test2/issue/1/subscribe", headers=headers ) self.assertEqual(output.status_code, 401) data = json.loads(output.get_data(as_text=True)) self.assertEqual(sorted(data.keys()), ["error", "error_code"]) self.assertEqual(pagure.api.APIERROR.EINVALIDTOK.value, data["error"]) self.assertEqual( pagure.api.APIERROR.EINVALIDTOK.name, data["error_code"] ) # No input output = self.app.post( "/api/0/test/issue/1/subscribe", headers=headers ) self.assertEqual(output.status_code, 404) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"error": "Issue not found", "error_code": "ENOISSUE"} ) # Create normal issue repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue #1", content="We should work on this", user="foo", private=False, issue_uid="aaabbbccc1", ) self.session.commit() self.assertEqual(msg.title, "Test issue #1") # Check subscribtion before repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual( pagure.lib.query.get_watch_list(self.session, issue), set(["pingou", "foo"]), ) # Unsubscribe - no changes data = {} output = self.app.post( "/api/0/test/issue/1/subscribe", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["avatar_url"] = "https://seccdn.libravatar.org/avatar/..." self.assertDictEqual( data, { "message": "You are no longer watching this issue", "avatar_url": "https://seccdn.libravatar.org/avatar/...", "user": "bar", }, ) data = {} output = self.app.post( "/api/0/test/issue/1/subscribe", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["avatar_url"] = "https://seccdn.libravatar.org/avatar/..." self.assertDictEqual( data, { "message": "You are no longer watching this issue", "avatar_url": "https://seccdn.libravatar.org/avatar/...", "user": "bar", }, ) # No change repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual( pagure.lib.query.get_watch_list(self.session, issue), set(["pingou", "foo"]), ) # Subscribe data = {"status": True} output = self.app.post( "/api/0/test/issue/1/subscribe", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["avatar_url"] = "https://seccdn.libravatar.org/avatar/..." self.assertDictEqual( data, { "message": "You are now watching this issue", "avatar_url": "https://seccdn.libravatar.org/avatar/...", "user": "bar", }, ) # Subscribe - no changes data = {"status": True} output = self.app.post( "/api/0/test/issue/1/subscribe", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["avatar_url"] = "https://seccdn.libravatar.org/avatar/..." self.assertDictEqual( data, { "message": "You are now watching this issue", "avatar_url": "https://seccdn.libravatar.org/avatar/...", "user": "bar", }, ) repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual( pagure.lib.query.get_watch_list(self.session, issue), set(["pingou", "foo", "bar"]), ) # Unsubscribe data = {} output = self.app.post( "/api/0/test/issue/1/subscribe", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["avatar_url"] = "https://seccdn.libravatar.org/avatar/..." self.assertDictEqual( data, { "message": "You are no longer watching this issue", "avatar_url": "https://seccdn.libravatar.org/avatar/...", "user": "bar", }, ) repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual( pagure.lib.query.get_watch_list(self.session, issue), set(["pingou", "foo"]), ) def test_api_update_custom_field(self): """ Test the api_update_custom_field method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "tickets")) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # Invalid project output = self.app.post( "/api/0/foo/issue/1/custom/bugzilla", headers=headers ) self.assertEqual(output.status_code, 404) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"error": "Project not found", "error_code": "ENOPROJECT"} ) # Valid token, wrong project output = self.app.post( "/api/0/test2/issue/1/custom/bugzilla", headers=headers ) self.assertEqual(output.status_code, 401) data = json.loads(output.get_data(as_text=True)) self.assertEqual(sorted(data.keys()), ["error", "error_code"]) self.assertEqual(pagure.api.APIERROR.EINVALIDTOK.value, data["error"]) self.assertEqual( pagure.api.APIERROR.EINVALIDTOK.name, data["error_code"] ) # No issue output = self.app.post( "/api/0/test/issue/1/custom/bugzilla", headers=headers ) self.assertEqual(output.status_code, 404) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, {"error": "Issue not found", "error_code": "ENOISSUE"} ) # Create normal issue repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue #1", content="We should work on this", user="pingou", private=False, ) self.session.commit() self.assertEqual(msg.title, "Test issue #1") # Project does not have this custom field output = self.app.post( "/api/0/test/issue/1/custom/bugzilla", headers=headers ) self.assertEqual(output.status_code, 400) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, { "error": "Invalid custom field submitted", "error_code": "EINVALIDISSUEFIELD", }, ) # Check the behavior if the project disabled the issue tracker repo = pagure.lib.query.get_authorized_project(self.session, "test") settings = repo.settings settings["issue_tracker"] = False repo.settings = settings self.session.add(repo) self.session.commit() output = self.app.post( "/api/0/test/issue/1/custom/bugzilla", headers=headers ) self.assertEqual(output.status_code, 404) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, { "error": "Issue tracker disabled", "error_code": "ETRACKERDISABLED", }, ) repo = pagure.lib.query.get_authorized_project(self.session, "test") settings = repo.settings settings["issue_tracker"] = True repo.settings = settings self.session.add(repo) self.session.commit() # Invalid API token headers = {"Authorization": "token foobar"} output = self.app.post( "/api/0/test/issue/1/custom/bugzilla", headers=headers ) self.assertEqual(output.status_code, 401) data = json.loads(output.get_data(as_text=True)) self.assertEqual( sorted(data.keys()), ["error", "error_code", "errors"] ) self.assertEqual(pagure.api.APIERROR.EINVALIDTOK.value, data["error"]) self.assertEqual( pagure.api.APIERROR.EINVALIDTOK.name, data["error_code"] ) self.assertEqual(data["errors"], "Invalid token") headers = {"Authorization": "token aaabbbcccddd"} # Set some custom fields repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.set_custom_key_fields( self.session, repo, ["bugzilla", "upstream", "reviewstatus", "duedate"], ["link", "boolean", "list", "date"], ["", "", "ack, nack , needs review", "2018-10-10"], [None, None, None, None], ) self.session.commit() self.assertEqual(msg, "List of custom fields updated") # Check the project custom fields were correctly set for key in repo.issue_keys: # Check that the bugzilla field correctly had its data removed if key.name == "bugzilla": self.assertIsNone(key.data) # Check that the reviewstatus list field still has its list if key.name == "reviewstatus": self.assertEqual( sorted(key.data), ["ack", "nack", "needs review"] ) # Check that the duedate date field still has its date if key.name == "duedate": self.assertEqual(key.data, "2018-10-10") # Check that not setting the value on a non-existing custom field # changes nothing output = self.app.post( "/api/0/test/issue/1/custom/bugzilla", headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual(data, {"message": "No changes"}) repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(issue.other_fields, []) self.assertEqual(len(issue.other_fields), 0) # Invalid value output = self.app.post( "/api/0/test/issue/1/custom/bugzilla", headers=headers, data={"value": "foobar"}, ) self.assertEqual(output.status_code, 400) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, { "error": "Invalid custom field submitted, the value is not " "a link", "error_code": "EINVALIDISSUEFIELD_LINK", }, ) repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(issue.other_fields, []) self.assertEqual(len(issue.other_fields), 0) # All good output = self.app.post( "/api/0/test/issue/1/custom/bugzilla", headers=headers, data={"value": "https://bugzilla.redhat.com/1234"}, ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, { "message": "Custom field bugzilla adjusted to " "https://bugzilla.redhat.com/1234" }, ) self.session.commit() repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(len(issue.other_fields), 1) self.assertEqual(issue.other_fields[0].key.name, "bugzilla") self.assertEqual( issue.other_fields[0].value, "https://bugzilla.redhat.com/1234" ) # Reset the value output = self.app.post( "/api/0/test/issue/1/custom/bugzilla", headers=headers, data={"value": ""}, ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual( data, { "message": "Custom field bugzilla reset " "(from https://bugzilla.redhat.com/1234)" }, ) self.session.commit() repo = pagure.lib.query.get_authorized_project(self.session, "test") issue = pagure.lib.query.search_issues(self.session, repo, issueid=1) self.assertEqual(len(issue.other_fields), 0) @patch( "pagure.lib.query.set_custom_key_value", MagicMock(side_effect=pagure.exceptions.PagureException("error")), ) def test_api_update_custom_field_raises_error(self): """ Test the api_update_custom_field method of the flask api. """ tests.create_projects(self.session) tests.create_projects_git(os.path.join(self.path, "tickets")) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # Create normal issue repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue #1", content="We should work on this", user="pingou", private=False, ) self.session.commit() self.assertEqual(msg.title, "Test issue #1") # Set some custom fields 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") # Check the project custom fields were correctly set for key in repo.issue_keys: # Check that the bugzilla field correctly had its data removed if key.name == "bugzilla": self.assertIsNone(key.data) # Check that the reviewstatus list field still has its list elif key.name == "reviewstatus": self.assertEqual( sorted(key.data), ["ack", "nack", "needs review"] ) # Should work but raises an exception output = self.app.post( "/api/0/test/issue/1/custom/bugzilla", headers=headers, data={"value": "https://bugzilla.redhat.com/1234"}, ) self.assertEqual(output.status_code, 400) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual(data, {"error": "error", "error_code": "ENOCODE"}) def test_api_view_issues_history_stats(self): """ Test the api_view_issues_history_stats method of the flask api. """ self.test_api_new_issue() # Create private issue repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue", content="We should work on this", user="pingou", private=True, status="Closed", ) self.session.commit() self.assertEqual(msg.title, "Test issue") output = self.app.get("/api/0/test/issues/history/stats") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data), 1) self.assertEqual(len(data["stats"]), 53) last_key = sorted(data["stats"].keys())[-1] self.assertEqual(data["stats"][last_key], 0) for k in sorted(data["stats"].keys())[:-1]: self.assertEqual(data["stats"][k], 0) def test_api_view_user_issues_pingou(self): """ Test the api_view_user_issues method of the flask api for pingou. """ self.test_api_new_issue() # Create private issue repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue", content="We should work on this", user="pingou", private=True, status="Closed", ) self.session.commit() self.assertEqual(msg.title, "Test issue") output = self.app.get("/api/0/user/pingou/issues") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) args = { "assignee": True, "author": True, "closed": None, "created": None, "milestones": [], "no_stones": None, "order": None, "order_key": None, "page": 1, "since": None, "status": None, "tags": [], "updated": None, } self.assertEqual(data["args"], args) self.assertEqual(data["issues_assigned"], []) self.assertEqual(len(data["issues_created"]), 1) self.assertEqual(data["total_issues_assigned"], 0) self.assertEqual(data["total_issues_created"], 1) self.assertEqual(data["total_issues_assigned_pages"], 1) self.assertEqual(data["total_issues_created_pages"], 1) # Restrict to a certain, fake milestone output = self.app.get("/api/0/user/pingou/issues?milestones=v1.0") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) args = { "assignee": True, "author": True, "closed": None, "created": None, "milestones": ["v1.0"], "no_stones": None, "order": None, "order_key": None, "page": 1, "since": None, "status": None, "tags": [], "updated": None, } self.assertEqual(data["args"], args) self.assertEqual(data["issues_assigned"], []) self.assertEqual(data["issues_created"], []) self.assertEqual(data["total_issues_assigned"], 0) self.assertEqual(data["total_issues_created"], 0) self.assertEqual(data["total_issues_assigned_pages"], 1) self.assertEqual(data["total_issues_created_pages"], 1) # Restrict to a certain status output = self.app.get("/api/0/user/pingou/issues?status=closed") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) args = { "assignee": True, "author": True, "closed": None, "created": None, "milestones": [], "no_stones": None, "order": None, "order_key": None, "page": 1, "since": None, "status": "closed", "tags": [], "updated": None, } self.assertEqual(data["args"], args) self.assertEqual(data["issues_assigned"], []) self.assertEqual(len(data["issues_created"]), 1) self.assertEqual(data["total_issues_assigned"], 0) self.assertEqual(data["total_issues_created"], 1) self.assertEqual(data["total_issues_assigned_pages"], 1) self.assertEqual(data["total_issues_created_pages"], 1) # Restrict to a certain status output = self.app.get("/api/0/user/pingou/issues?status=all") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) args = { "assignee": True, "author": True, "closed": None, "created": None, "milestones": [], "no_stones": None, "order": None, "order_key": None, "page": 1, "since": None, "status": "all", "tags": [], "updated": None, } self.assertEqual(data["args"], args) self.assertEqual(data["issues_assigned"], []) self.assertEqual(len(data["issues_created"]), 2) self.assertEqual(data["total_issues_assigned"], 0) self.assertEqual(data["total_issues_created"], 2) self.assertEqual(data["total_issues_assigned_pages"], 1) self.assertEqual(data["total_issues_created_pages"], 1) def test_api_view_user_issues_foo(self): """ Test the api_view_user_issues method of the flask api for foo. """ self.test_api_new_issue() # Create private issue repo = pagure.lib.query.get_authorized_project(self.session, "test") msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue", content="We should work on this", user="pingou", private=True, status="Closed", ) self.session.commit() self.assertEqual(msg.title, "Test issue") output = self.app.get("/api/0/user/foo/issues") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) args = { "assignee": True, "author": True, "closed": None, "created": None, "milestones": [], "no_stones": None, "order": None, "order_key": None, "page": 1, "since": None, "status": None, "tags": [], "updated": None, } self.assertEqual(data["args"], args) self.assertEqual(len(data["issues_assigned"]), 0) self.assertEqual(data["issues_created"], []) self.assertEqual(data["total_issues_assigned"], 0) self.assertEqual(data["total_issues_created"], 0) self.assertEqual(data["total_issues_assigned_pages"], 1) self.assertEqual(data["total_issues_created_pages"], 1) def test_api_view_user_issues_foo_invalid_page(self): """ Test the api_view_user_issues method of the flask api for foo. """ self.test_api_new_issue() output = self.app.get("/api/0/user/foo/issues?page=0") self.assertEqual(output.status_code, 400) data = json.loads(output.get_data(as_text=True)) self.assertEqual( data, { "error": "Invalid or incomplete input submitted", "error_code": "EINVALIDREQ", }, ) output = self.app.get("/api/0/user/foo/issues?page=abc") self.assertEqual(output.status_code, 400) data = json.loads(output.get_data(as_text=True)) self.assertEqual( data, { "error": "Invalid or incomplete input submitted", "error_code": "EINVALIDREQ", }, ) def test_api_view_user_issues_foo_no_assignee(self): """ Test the api_view_user_issues method of the flask api for foo. """ self.test_api_new_issue() output = self.app.get("/api/0/user/foo/issues?assignee=0") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) args = { "assignee": False, "author": True, "closed": None, "created": None, "milestones": [], "no_stones": None, "order": None, "order_key": None, "page": 1, "since": None, "status": None, "tags": [], "updated": None, } self.assertEqual(data["args"], args) self.assertEqual(data["issues_assigned"], []) self.assertEqual(data["issues_created"], []) self.assertEqual(data["total_issues_assigned"], 0) self.assertEqual(data["total_issues_created"], 0) self.assertEqual(data["total_issues_assigned_pages"], 1) self.assertEqual(data["total_issues_created_pages"], 1) def test_api_view_user_issues_pingou_no_author(self): """ Test the api_view_user_issues method of the flask api for pingou. """ self.test_api_new_issue() output = self.app.get("/api/0/user/pingou/issues?author=0") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) args = { "assignee": True, "author": False, "closed": None, "created": None, "milestones": [], "no_stones": None, "order": None, "order_key": None, "page": 1, "since": None, "status": None, "tags": [], "updated": None, } self.assertEqual(data["args"], args) self.assertEqual(data["issues_assigned"], []) self.assertEqual(data["issues_created"], []) self.assertEqual(data["total_issues_assigned"], 0) self.assertEqual(data["total_issues_created"], 0) self.assertEqual(data["total_issues_assigned_pages"], 1) self.assertEqual(data["total_issues_created_pages"], 1) if __name__ == "__main__": SUITE = unittest.TestLoader().loadTestsFromTestCase( PagureFlaskApiIssuetests ) unittest.TextTestRunner(verbosity=2).run(SUITE)