# -*- coding: utf-8 -*- """ (c) 2015-2018 - Copyright Red Hat Inc Authors: Pierre-Yves Chibon """ from __future__ import unicode_literals, absolute_import import datetime import os import pytz import shutil import sys import unittest import json from mock import patch sys.path.insert( 0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..") ) import pagure.api import pagure.config import pagure.lib.model as model import pagure.lib.query import tests class PagureFlaskApiUSertests(tests.Modeltests): """ Tests for the flask API of pagure for issue """ maxDiff = None def setUp(self): """ Set up the environnment, ran before every tests. """ super(PagureFlaskApiUSertests, self).setUp() pagure.config.config["REQUESTS_FOLDER"] = None def test_api_users(self): """ Test the api_users function. """ output = self.app.get("/api/0/users") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(sorted(data["users"]), ["foo", "pingou"]) self.assertEqual( sorted(data.keys()), ["mention", "total_users", "users"] ) self.assertEqual(data["total_users"], 2) output = self.app.get("/api/0/users?pattern=p") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(data["users"], ["pingou"]) self.assertEqual( sorted(data.keys()), ["mention", "total_users", "users"] ) self.assertEqual(data["total_users"], 1) def test_api_view_user(self): """ Test the api_view_user method of the flask api The tested user has no project or forks. """ output = self.app.get("/api/0/user/pingou") self.assertEqual(output.status_code, 200) exp = { "forks": [], "forks_pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "forkpage": 1, "pages": 0, "per_page": 20, "prev": None, }, "repos": [], "repos_pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "repopage": 1, "pages": 0, "per_page": 20, "prev": None, }, "user": { "fullname": "PY C", "name": "pingou", "url_path": "user/pingou", "avatar_url": "https://seccdn.libravatar.org/avatar/...", }, } data = json.loads(output.get_data(as_text=True)) data["user"]["avatar_url"] = "https://seccdn.libravatar.org/avatar/..." for k in ["forks_pagination", "repos_pagination"]: for k2 in ["first", "last"]: self.assertIsNotNone(data[k][k2]) data[k][k2] = "http://localhost..." self.assertEqual(data, exp) def test_api_view_user_with_project(self): """ Test the api_view_user method of the flask api, this time the user has some project defined. """ tests.create_projects(self.session) output = self.app.get("/api/0/user/pingou") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) data["repos"][0]["date_created"] = "1490272832" data["repos"][0]["date_modified"] = "1490272832" data["repos"][1]["date_created"] = "1490272832" data["repos"][1]["date_modified"] = "1490272832" data["repos"][2]["date_created"] = "1490272832" data["repos"][2]["date_modified"] = "1490272832" for k in ["forks_pagination", "repos_pagination"]: for k2 in ["first", "last"]: self.assertIsNotNone(data[k][k2]) data[k][k2] = "http://localhost..." expected_data = { "forks": [], "forks_pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "forkpage": 1, "pages": 0, "per_page": 20, "prev": None, }, "repos": [ { "access_groups": {"admin": [], "commit": [], "ticket": []}, "access_users": { "admin": [], "commit": [], "owner": ["pingou"], "ticket": [], }, "close_status": [ "Invalid", "Insufficient data", "Fixed", "Duplicate", ], "custom_keys": [], "date_created": "1490272832", "date_modified": "1490272832", "description": "test project #1", "fullname": "test", "url_path": "test", "id": 1, "milestones": {}, "name": "test", "namespace": None, "parent": None, "priorities": {}, "tags": [], "user": { "fullname": "PY C", "name": "pingou", "url_path": "user/pingou", }, }, { "access_groups": {"admin": [], "commit": [], "ticket": []}, "access_users": { "admin": [], "commit": [], "owner": ["pingou"], "ticket": [], }, "close_status": [ "Invalid", "Insufficient data", "Fixed", "Duplicate", ], "custom_keys": [], "date_created": "1490272832", "date_modified": "1490272832", "description": "test project #2", "fullname": "test2", "url_path": "test2", "id": 2, "milestones": {}, "name": "test2", "namespace": None, "parent": None, "priorities": {}, "tags": [], "user": { "fullname": "PY C", "name": "pingou", "url_path": "user/pingou", }, }, { "access_groups": {"admin": [], "commit": [], "ticket": []}, "access_users": { "admin": [], "commit": [], "owner": ["pingou"], "ticket": [], }, "close_status": [ "Invalid", "Insufficient data", "Fixed", "Duplicate", ], "custom_keys": [], "date_created": "1490272832", "date_modified": "1490272832", "description": "namespaced test project", "fullname": "somenamespace/test3", "url_path": "somenamespace/test3", "id": 3, "milestones": {}, "name": "test3", "namespace": "somenamespace", "parent": None, "priorities": {}, "tags": [], "user": { "fullname": "PY C", "name": "pingou", "url_path": "user/pingou", }, }, ], "repos_pagination": { "first": "http://localhost...", "last": "http://localhost...", "next": None, "repopage": 1, "pages": 1, "per_page": 20, "prev": None, }, "user": { "fullname": "PY C", "name": "pingou", "url_path": "user/pingou", "avatar_url": "https://seccdn.libravatar.org/avatar/...", }, } data["user"]["avatar_url"] = "https://seccdn.libravatar.org/avatar/..." self.assertEqual(data, expected_data) @patch("pagure.lib.notify.send_email") def test_api_view_user_activity_stats(self, mockemail): """ Test the api_view_user_activity_stats method of the flask user api. """ mockemail.return_value = True tests.create_projects(self.session) tests.create_tokens(self.session) tests.create_tokens_acl(self.session) headers = {"Authorization": "token aaabbbcccddd"} # Create a pull-request repo = pagure.lib.query._get_project(self.session, "test") forked_repo = pagure.lib.query._get_project(self.session, "test") req = pagure.lib.query.new_pull_request( session=self.session, repo_from=forked_repo, branch_from="master", repo_to=repo, branch_to="master", title="test pull-request", user="pingou", ) self.session.commit() self.assertEqual(req.id, 1) self.assertEqual(req.title, "test pull-request") # Check comments before self.session.commit() request = pagure.lib.query.search_pull_requests( self.session, project_id=1, requestid=1 ) self.assertEqual(len(request.comments), 0) data = {"comment": "This is a very interesting question"} # Valid request output = self.app.post( "/api/0/test/pull-request/1/comment", data=data, headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual(data, {"message": "Comment added"}) # One comment added self.session.commit() request = pagure.lib.query.search_pull_requests( self.session, project_id=1, requestid=1 ) self.assertEqual(len(request.comments), 1) # Close PR output = self.app.post( "/api/0/test/pull-request/1/close", headers=headers ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertDictEqual(data, {"message": "Pull-request closed!"}) # PR closed self.session.commit() request = pagure.lib.query.search_pull_requests( self.session, project_id=1, requestid=1 ) self.assertEqual(request.status, "Closed") # Finally retrieve the user's logs output = self.app.get("/api/0/user/pingou/activity/stats") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) date = datetime.datetime.utcnow().date().strftime("%Y-%m-%d") self.assertDictEqual(data, {date: 4}) @patch("pagure.lib.notify.send_email") def test_api_view_user_activity_date(self, mockemail): """ Test the api_view_user_activity_date method of the flask user api. """ self.test_api_view_user_activity_stats() # Invalid date output = self.app.get("/api/0/user/pingou/activity/AABB") self.assertEqual(output.status_code, 400) # Invalid date, arrow >= 0.15 throws an exception, # previous versions parsed it output = self.app.get("/api/0/user/pingou/activity/2016asd") arrow_version = self.get_arrow_version() if arrow_version >= (0, 15, 6): self.assertEqual(output.status_code, 400) exp = { "error": "Could not match input '2016asd' to any of the following formats: " "YYYY-MM-DD, YYYY-M-DD, YYYY-M-D, YYYY/MM/DD, YYYY/M/DD, YYYY/M/D, " "YYYY.MM.DD, YYYY.M.DD, YYYY.M.D, YYYYMMDD, YYYY-DDDD, YYYYDDDD, " "YYYY-MM, YYYY/MM, YYYY.MM, YYYY, W", "error_code": "ENOCODE", } self.assertEqual(json.loads(output.get_data(as_text=True)), exp) elif arrow_version >= (0, 15): self.assertEqual(output.status_code, 400) exp = { "error": "Could not match input '2016asd' to any of the following formats: " "YYYY-MM-DD, YYYY-M-DD, YYYY-M-D, YYYY/MM/DD, YYYY/M/DD, YYYY/M/D, " "YYYY.MM.DD, YYYY.M.DD, YYYY.M.D, YYYYMMDD, YYYY-DDDD, YYYYDDDD, " "YYYY-MM, YYYY/MM, YYYY.MM, YYYY", "error_code": "ENOCODE", } self.assertEqual(json.loads(output.get_data(as_text=True)), exp) else: self.assertEqual(output.status_code, 200) exp = {"activities": [], "date": "2016-01-01"} self.assertEqual(json.loads(output.get_data(as_text=True)), exp) # Invalid date, arrow => throws an exception, # previous versions parsed it just not really as expected output = self.app.get("/api/0/user/pingou/activity/20161245") if self.get_arrow_version() >= (0, 15): self.assertEqual(output.status_code, 400) exp = { "error": "day is out of range for month", "error_code": "ENOCODE", } self.assertEqual(json.loads(output.get_data(as_text=True)), exp) else: self.assertEqual(output.status_code, 200) exp = {"activities": [], "date": "1970-08-22"} self.assertEqual(json.loads(output.get_data(as_text=True)), exp) date = datetime.datetime.utcnow().date().strftime("%Y-%m-%d") # Retrieve the user's logs for today output = self.app.get("/api/0/user/pingou/activity/%s" % date) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) exp = { "activities": [ { "date": date, "date_created": "1477558752", "type": "pull-request", "description_mk": '

pingou created PR test#1

', "id": 1, "ref_id": "1", "type": "created", "user": { "fullname": "PY C", "name": "pingou", "url_path": "user/pingou", }, }, { "date": date, "date_created": "1477558752", "type": "pull-request", "description_mk": '

pingou commented on PR test#1

', "id": 2, "ref_id": "1", "type": "commented", "user": { "fullname": "PY C", "name": "pingou", "url_path": "user/pingou", }, }, { "date": date, "date_created": "1477558752", "type": "pull-request", "description_mk": '

pingou closed PR test#1

', "id": 3, "ref_id": "1", "type": "closed", "user": { "fullname": "PY C", "name": "pingou", "url_path": "user/pingou", }, }, { "date": date, "date_created": "1477558752", "type": "pull-request", "description_mk": '

pingou commented on PR test#1

', "id": 4, "ref_id": "1", "type": "commented", "user": { "fullname": "PY C", "name": "pingou", "url_path": "user/pingou", }, }, ], "date": date, } for idx, act in enumerate(data["activities"]): act["date_created"] = "1477558752" data["activities"][idx] = act self.assertEqual(data, exp) @patch("pagure.lib.notify.send_email") def test_api_view_user_activity_date_1_activity(self, mockemail): """ Test the api_view_user_activity_date method of the flask user api when the user only did one action. """ tests.create_projects(self.session) repo = pagure.lib.query._get_project(self.session, "test") now = datetime.datetime.utcnow() date = now.date().strftime("%Y-%m-%d") # Create a single commit log log = model.PagureLog( user_id=1, user_email="foo@bar.com", project_id=1, log_type="committed", ref_id="githash", date=now.date(), date_created=now, ) self.session.add(log) self.session.commit() # Retrieve the user's logs for today output = self.app.get( "/api/0/user/pingou/activity/%s?grouped=1" % date ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) exp = { "activities": [ { "description_mk": '

pingou committed on test#githash

' } ], "date": date, } self.assertEqual(data, exp) @patch("pagure.lib.notify.send_email") def test_api_view_user_activity_timezone_negative(self, mockemail): """Test api_view_user_activity{_stats,_date} with the America/ New York timezone, which is 5 hours behind UTC in winter and 4 hours behind UTC in summer (daylight savings). The events will occur on XXXX-02-15 in UTC, but on XXXX-02-14 local. """ tests.create_projects(self.session) today = datetime.datetime.utcnow() # stats/activty returns data for now()-1year time range, so alter # year as needed to select 15th feb of current or previous year if today < datetime.datetime(today.year, 2, 15, 3, 30): year = today.year - 1 else: year = today.year dateobj = datetime.datetime(year, 2, 15, 3, 30) utcdate = "%s-02-15" % year # the Unix timestamp for YYYY-02-15 12:00 UTC utcts = str( int( ( datetime.datetime(year, 2, 15, 12, 0, tzinfo=pytz.UTC) - datetime.datetime(1970, 1, 1, tzinfo=pytz.UTC) ).total_seconds() ) ) localdate = "%s-02-14" % year # the Unix timestamp for YYYY-02-15 18:00 America/New_York localts = str( int( ( datetime.datetime( year, 2, 14, 17, 0, tzinfo=pytz.timezone("America/New_York"), ) - datetime.datetime( 1970, 1, 1, tzinfo=pytz.timezone("America/New_York") ) ).total_seconds() ) ) # Create a single commit log log = model.PagureLog( user_id=1, user_email="foo@bar.com", project_id=1, log_type="committed", ref_id="githash", date=dateobj.date(), date_created=dateobj, ) self.session.add(log) self.session.commit() # Retrieve the user's stats with no timezone specified (==UTC) output = self.app.get("/api/0/user/pingou/activity/stats") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) # date in output should be UTC date self.assertDictEqual(data, {utcdate: 1}) # Now in timestamp format... output = self.app.get( "/api/0/user/pingou/activity/stats?format=timestamp" ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) # timestamp in output should be UTC ts self.assertDictEqual(data, {utcts: 1}) # Retrieve the user's stats with local timezone specified output = self.app.get( "/api/0/user/pingou/activity/stats?tz=America/New_York" ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) # date in output should be local date self.assertDictEqual(data, {localdate: 1}) # Now in timestamp format... output = self.app.get( "/api/0/user/pingou/activity/stats?format=timestamp&tz=America/New_York" ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) # timestamp in output should be local ts self.assertDictEqual(data, {localts: 1}) # Retrieve the user's logs for 2018-02-15 with no timezone output = self.app.get( "/api/0/user/pingou/activity/%s?grouped=1" % utcdate ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) exp = { "activities": [ { "description_mk": '

pingou committed on test#githash

' } ], "date": utcdate, } self.assertEqual(data, exp) # Now retrieve the user's logs for 2018-02-14 with local time output = self.app.get( "/api/0/user/pingou/activity/%s?grouped=1&tz=America/New_York" % localdate ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) exp["date"] = localdate self.assertEqual(data, exp) @patch("pagure.lib.notify.send_email") def test_api_view_user_activity_timezone_positive(self, mockemail): """Test api_view_user_activity{_stats,_date} with the Asia/ Dubai timezone, which is 4 hours ahead of UTC. The events will occur on XXXX-02-15 in UTC, but on XXXX-02-16 in local time. """ tests.create_projects(self.session) today = datetime.datetime.utcnow() # stats/activty returns data for now()-1year time range, so alter # year as needed to select 15th feb of current or previous year if today < datetime.datetime(today.year, 2, 15, 3, 30): year = today.year - 1 else: year = today.year dateobj = datetime.datetime(year, 2, 15, 22, 30) utcdate = "%s-02-15" % year # the Unix timestamp for YYYY-02-15 12:00 UTC utcts = str( int( ( datetime.datetime(year, 2, 15, 12, 0, tzinfo=pytz.UTC) - datetime.datetime(1970, 1, 1, tzinfo=pytz.UTC) ).total_seconds() ) ) localdate = "%s-02-16" % year # the Unix timestamp for YYYY-02-16 9:00 Asia/Dubai localts = str( int( ( datetime.datetime( year, 2, 16, 8, 0, tzinfo=pytz.timezone("Asia/Dubai") ) - datetime.datetime( 1970, 1, 1, tzinfo=pytz.timezone("Asia/Dubai") ) ).total_seconds() ) ) # Create a single commit log log = model.PagureLog( user_id=1, user_email="foo@bar.com", project_id=1, log_type="committed", ref_id="githash", date=dateobj.date(), date_created=dateobj, ) self.session.add(log) self.session.commit() # Retrieve the user's stats with no timezone specified (==UTC) output = self.app.get("/api/0/user/pingou/activity/stats") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) # date in output should be UTC date self.assertDictEqual(data, {utcdate: 1}) # Now in timestamp format... output = self.app.get( "/api/0/user/pingou/activity/stats?format=timestamp" ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) # timestamp in output should be UTC ts self.assertDictEqual(data, {utcts: 1}) # Retrieve the user's stats with local timezone specified output = self.app.get( "/api/0/user/pingou/activity/stats?tz=Asia/Dubai" ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) # date in output should be local date self.assertDictEqual(data, {localdate: 1}) # Now in timestamp format... output = self.app.get( "/api/0/user/pingou/activity/stats?format=timestamp&tz=Asia/Dubai" ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) # timestamp in output should be local ts self.assertDictEqual(data, {localts: 1}) # Retrieve the user's logs for 2018-02-15 with no timezone output = self.app.get( "/api/0/user/pingou/activity/%s?grouped=1" % utcdate ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) exp = { "activities": [ { "description_mk": '

pingou committed on test#githash

' } ], "date": utcdate, } self.assertEqual(data, exp) # Now retrieve the user's logs for 2018-02-16 with local time output = self.app.get( "/api/0/user/pingou/activity/%s?grouped=1&tz=Asia/Dubai" % localdate ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) exp["date"] = localdate self.assertEqual(data, exp) class PagureFlaskApiUsertestrequests(tests.Modeltests): """ Tests for the user requests endpoints """ maxDiff = None def setUp(self): """ Set up the environnment, ran before every tests. """ super(PagureFlaskApiUsertestrequests, self).setUp() pagure.config.config["REQUESTS_FOLDER"] = None tests.create_projects(self.session) # Create few pull-requests repo = pagure.lib.query.get_authorized_project(self.session, "test") forked_repo = pagure.lib.query.get_authorized_project( self.session, "test" ) pagure.lib.query.new_pull_request( session=self.session, repo_from=forked_repo, branch_from="master", repo_to=repo, branch_to="master", title="open pullrequest by user foo on repo test", user="foo", ) repo = pagure.lib.query.get_authorized_project(self.session, "test2") forked_repo = pagure.lib.query.get_authorized_project( self.session, "test2" ) pagure.lib.query.new_pull_request( session=self.session, repo_from=forked_repo, branch_from="master", repo_to=repo, branch_to="master", title="open pullrequest by user foo on repo test2", user="foo", ) self.session.commit() repo = pagure.lib.query.get_authorized_project(self.session, "test") forked_repo = pagure.lib.query.get_authorized_project( self.session, "test" ) pagure.lib.query.new_pull_request( session=self.session, repo_from=forked_repo, branch_from="master", repo_to=repo, branch_to="master", title="closed pullrequest by user foo on repo test", user="foo", status="Closed", ) repo = pagure.lib.query.get_authorized_project(self.session, "test2") forked_repo = pagure.lib.query.get_authorized_project( self.session, "test2" ) pagure.lib.query.new_pull_request( session=self.session, repo_from=forked_repo, branch_from="master", repo_to=repo, branch_to="master", title="closed pullrequest by user foo on repo test2", user="foo", status="Closed", ) self.session.commit() repo = pagure.lib.query.get_authorized_project(self.session, "test") forked_repo = pagure.lib.query.get_authorized_project( self.session, "test" ) pagure.lib.query.new_pull_request( session=self.session, repo_from=forked_repo, branch_from="master", repo_to=repo, branch_to="master", title="merged pullrequest by user foo on repo test", user="foo", status="Merged", ) repo = pagure.lib.query.get_authorized_project(self.session, "test2") forked_repo = pagure.lib.query.get_authorized_project( self.session, "test2" ) pagure.lib.query.new_pull_request( session=self.session, repo_from=forked_repo, branch_from="master", repo_to=repo, branch_to="master", title="merged pullrequest by user foo on repo test2", user="foo", status="Merged", ) self.session.commit() repo = pagure.lib.query.get_authorized_project(self.session, "test") forked_repo = pagure.lib.query.get_authorized_project( self.session, "test" ) pagure.lib.query.new_pull_request( session=self.session, repo_from=forked_repo, branch_from="master", repo_to=repo, branch_to="master", title="open pullrequest by user pingou on repo test", user="pingou", ) self.session.commit() repo = pagure.lib.query.get_authorized_project(self.session, "test2") forked_repo = pagure.lib.query.get_authorized_project( self.session, "test2" ) pagure.lib.query.new_pull_request( session=self.session, repo_from=forked_repo, branch_from="master", repo_to=repo, branch_to="master", title="open pullrequest by user pingou on repo test2", user="pingou", ) self.session.commit() repo = pagure.lib.query.get_authorized_project(self.session, "test") forked_repo = pagure.lib.query.get_authorized_project( self.session, "test" ) pagure.lib.query.new_pull_request( session=self.session, repo_from=forked_repo, branch_from="master", repo_to=repo, branch_to="master", title="closed pullrequest by user pingou on repo test", user="pingou", status="Closed", ) self.session.commit() repo = pagure.lib.query.get_authorized_project(self.session, "test2") forked_repo = pagure.lib.query.get_authorized_project( self.session, "test2" ) pagure.lib.query.new_pull_request( session=self.session, repo_from=forked_repo, branch_from="master", repo_to=repo, branch_to="master", title="closed pullrequest by user pingou on repo test2", user="pingou", status="Closed", ) self.session.commit() repo = pagure.lib.query.get_authorized_project(self.session, "test") forked_repo = pagure.lib.query.get_authorized_project( self.session, "test" ) pagure.lib.query.new_pull_request( session=self.session, repo_from=forked_repo, branch_from="master", repo_to=repo, branch_to="master", title="merged pullrequest by user pingou on repo test", user="pingou", status="Merged", ) self.session.commit() repo = pagure.lib.query.get_authorized_project(self.session, "test2") forked_repo = pagure.lib.query.get_authorized_project( self.session, "test2" ) pagure.lib.query.new_pull_request( session=self.session, repo_from=forked_repo, branch_from="master", repo_to=repo, branch_to="master", title="merged pullrequest by user pingou on repo test2", user="pingou", status="Merged", ) self.session.commit() @patch("pagure.lib.notify.send_email") def test_api_view_user_requests_filed(self, mockemail): """ Test the api_view_user_requests_filed method of the flask user api """ # First we test without the status parameter. It should default to `open` output = self.app.get("/api/0/user/pingou/requests/filed") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 2) self.assertEqual( sorted(data.keys()), ["args", "pagination", "requests", "total_requests"], ) self.assertEqual(data["requests"][0]["user"]["name"], "pingou") self.assertEqual(data["requests"][1]["user"]["name"], "pingou") self.assertEqual(data["requests"][0]["status"], "Open") self.assertEqual(data["requests"][1]["status"], "Open") self.assertEqual( data["requests"][0]["title"], "open pullrequest by user pingou on repo test2", ) self.assertEqual( data["requests"][1]["title"], "open pullrequest by user pingou on repo test", ) self.assertEqual(data["args"]["status"], "open") self.assertEqual(data["args"]["page"], 1) # Next test with the status parameter set to `open`. output = self.app.get("/api/0/user/pingou/requests/filed?status=open") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 2) self.assertEqual( sorted(data.keys()), ["args", "pagination", "requests", "total_requests"], ) self.assertEqual(data["requests"][0]["user"]["name"], "pingou") self.assertEqual(data["requests"][1]["user"]["name"], "pingou") self.assertEqual(data["requests"][0]["status"], "Open") self.assertEqual(data["requests"][1]["status"], "Open") self.assertEqual( data["requests"][0]["title"], "open pullrequest by user pingou on repo test2", ) self.assertEqual( data["requests"][1]["title"], "open pullrequest by user pingou on repo test", ) self.assertEqual(data["args"]["status"], "open") self.assertEqual(data["args"]["page"], 1) # Next test with the status parameter set to `closed`. output = self.app.get( "/api/0/user/pingou/requests/filed?status=closed" ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 2) self.assertEqual( sorted(data.keys()), ["args", "pagination", "requests", "total_requests"], ) self.assertEqual(data["requests"][0]["user"]["name"], "pingou") self.assertEqual(data["requests"][1]["user"]["name"], "pingou") self.assertEqual(data["requests"][0]["status"], "Closed") self.assertEqual(data["requests"][1]["status"], "Closed") self.assertEqual( data["requests"][0]["title"], "closed pullrequest by user pingou on repo test2", ) self.assertEqual( data["requests"][1]["title"], "closed pullrequest by user pingou on repo test", ) self.assertEqual(data["args"]["status"], "closed") self.assertEqual(data["args"]["page"], 1) # Next test with the status parameter set to `merged`. output = self.app.get( "/api/0/user/pingou/requests/filed?status=merged" ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 2) self.assertEqual( sorted(data.keys()), ["args", "pagination", "requests", "total_requests"], ) self.assertEqual(data["requests"][0]["user"]["name"], "pingou") self.assertEqual(data["requests"][1]["user"]["name"], "pingou") self.assertEqual(data["requests"][0]["status"], "Merged") self.assertEqual(data["requests"][1]["status"], "Merged") self.assertEqual( data["requests"][0]["title"], "merged pullrequest by user pingou on repo test2", ) self.assertEqual( data["requests"][1]["title"], "merged pullrequest by user pingou on repo test", ) self.assertEqual(data["args"]["status"], "merged") self.assertEqual(data["args"]["page"], 1) # Finally, test with the status parameter set to `all`. output = self.app.get("/api/0/user/pingou/requests/filed?status=all") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 6) self.assertEqual( sorted(data.keys()), ["args", "pagination", "requests", "total_requests"], ) self.assertEqual(data["requests"][0]["user"]["name"], "pingou") self.assertEqual(data["requests"][1]["user"]["name"], "pingou") self.assertEqual(data["requests"][2]["user"]["name"], "pingou") self.assertEqual(data["requests"][3]["user"]["name"], "pingou") self.assertEqual(data["requests"][4]["user"]["name"], "pingou") self.assertEqual(data["requests"][5]["user"]["name"], "pingou") self.assertEqual(data["requests"][0]["status"], "Merged") self.assertEqual(data["requests"][1]["status"], "Merged") self.assertEqual(data["requests"][2]["status"], "Closed") self.assertEqual(data["requests"][3]["status"], "Closed") self.assertEqual(data["requests"][4]["status"], "Open") self.assertEqual(data["requests"][5]["status"], "Open") self.assertEqual( data["requests"][0]["title"], "merged pullrequest by user pingou on repo test2", ) self.assertEqual( data["requests"][1]["title"], "merged pullrequest by user pingou on repo test", ) self.assertEqual( data["requests"][2]["title"], "closed pullrequest by user pingou on repo test2", ) self.assertEqual( data["requests"][3]["title"], "closed pullrequest by user pingou on repo test", ) self.assertEqual( data["requests"][4]["title"], "open pullrequest by user pingou on repo test2", ) self.assertEqual( data["requests"][5]["title"], "open pullrequest by user pingou on repo test", ) self.assertEqual(data["args"]["status"], "all") self.assertEqual(data["args"]["page"], 1) # Test page 2 with the status parameter set to `all`. output = self.app.get( "/api/0/user/pingou/requests/filed?status=all&page=2" ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) self.assertEqual( sorted(data.keys()), ["args", "pagination", "requests", "total_requests"], ) self.assertEqual(data["args"]["page"], 2) @patch("pagure.lib.notify.send_email") def test_api_view_user_requests_filed_created(self, mockemail): """ Test the api_view_user_requests_filed method of the flask user api with the created parameter """ today = datetime.datetime.utcnow().date() output = self.app.get( "/api/0/user/pingou/requests/filed?status=all&created=%s" % (today.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 6) yesterday = today - datetime.timedelta(days=1) output = self.app.get( "/api/0/user/pingou/requests/filed?status=all&created=%s" % (yesterday.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 6) tomorrow = today + datetime.timedelta(days=1) output = self.app.get( "/api/0/user/pingou/requests/filed?status=all&created=%s" % (tomorrow.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) output = self.app.get( "/api/0/user/pingou/requests/filed?status=all&created=..%s" % (today.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) output = self.app.get( "/api/0/user/pingou/requests/filed?status=all&created=..%s" % (yesterday.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) thedaybefore = today - datetime.timedelta(days=1) output = self.app.get( "/api/0/user/pingou/requests/filed?status=all&created=..%s" % (thedaybefore.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) output = self.app.get( "/api/0/user/pingou/requests/filed?status=all&created=..%s" % (tomorrow.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 6) output = self.app.get( "/api/0/user/pingou/requests/filed?status=all&created=%s..%s" % (thedaybefore.isoformat(), tomorrow.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 6) @patch("pagure.lib.notify.send_email") def test_api_view_user_requests_filed_updated(self, mockemail): """ Test the api_view_user_requests_filed method of the flask user api with the created parameter """ today = datetime.datetime.utcnow().date() output = self.app.get( "/api/0/user/pingou/requests/filed?status=all&updated=%s" % (today.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) yesterday = today - datetime.timedelta(days=1) output = self.app.get( "/api/0/user/pingou/requests/filed?status=all&updated=%s" % (yesterday.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) tomorrow = today + datetime.timedelta(days=1) output = self.app.get( "/api/0/user/pingou/requests/filed?status=all&updated=%s" % (tomorrow.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 6) @patch("pagure.lib.notify.send_email") def test_api_view_user_requests_filed_closed(self, mockemail): """ Test the api_view_user_requests_filed method of the flask user api with the created parameter """ today = datetime.datetime.utcnow().date() output = self.app.get( "/api/0/user/pingou/requests/filed?status=all&closed=%s" % (today.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) yesterday = today - datetime.timedelta(days=1) output = self.app.get( "/api/0/user/pingou/requests/filed?status=all&closed=%s" % (yesterday.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) tomorrow = today + datetime.timedelta(days=1) output = self.app.get( "/api/0/user/pingou/requests/filed?status=all&closed=%s" % (tomorrow.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) @patch("pagure.lib.notify.send_email") def test_api_view_user_requests_filed_foo(self, mockemail): """ Test the api_view_user_requests_filed method of the flask user api """ # Default data returned output = self.app.get( "/api/0/user/foo/requests/filed?status=all&per_page=6" ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 6) self.assertEqual( sorted(data.keys()), ["args", "pagination", "requests", "total_requests"], ) # There are 6 PRs, that's 1 page at 6 results per page self.assertEqual(data["pagination"]["pages"], 1) @patch("pagure.lib.notify.send_email") def test_api_view_user_requests_filed_foo_grp_access(self, mockemail): """ Test when the user has accessed to some PRs via a group. """ # Add the user to a group msg = pagure.lib.query.add_group( self.session, group_name="some_group", display_name="Some Group", description=None, group_type="bar", user="pingou", is_admin=False, blacklist=[], ) self.session.commit() # Add the group to the project `test2` project = pagure.lib.query._get_project(self.session, "test2") msg = pagure.lib.query.add_group_to_project( session=self.session, project=project, new_group="some_group", user="pingou", ) self.session.commit() self.assertEqual(msg, "Group added") # Add foo to the group group = pagure.lib.query.search_groups( self.session, group_name="some_group" ) result = pagure.lib.query.add_user_to_group( self.session, "foo", group, "pingou", True ) self.session.commit() self.assertEqual(result, "User `foo` added to the group `some_group`.") # Query the API for foo's filed PRs output = self.app.get( "/api/0/user/foo/requests/filed?status=all&per_page=6" ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 6) self.assertEqual( sorted(data.keys()), ["args", "pagination", "requests", "total_requests"], ) # There are 6 PRs, that's 1 page at 6 results per page self.assertEqual(data["pagination"]["pages"], 1) @patch("pagure.lib.notify.send_email") def test_api_view_user_requests_actionable(self, mockemail): """ Test the api_view_user_requests_actionable method of the flask user api """ # First we test without the status parameter. It should default to `open` output = self.app.get("/api/0/user/pingou/requests/actionable") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 2) self.assertEqual( sorted(data.keys()), ["args", "pagination", "requests", "total_requests"], ) self.assertEqual(data["requests"][0]["user"]["name"], "foo") self.assertEqual(data["requests"][1]["user"]["name"], "foo") self.assertEqual(data["requests"][0]["status"], "Open") self.assertEqual(data["requests"][1]["status"], "Open") self.assertEqual( data["requests"][0]["title"], "open pullrequest by user foo on repo test2", ) self.assertEqual( data["requests"][1]["title"], "open pullrequest by user foo on repo test", ) self.assertEqual(data["args"]["status"], "open") self.assertEqual(data["args"]["page"], 1) # Next test with the status parameter set to `open`. output = self.app.get( "/api/0/user/pingou/requests/actionable?status=open" ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 2) self.assertEqual( sorted(data.keys()), ["args", "pagination", "requests", "total_requests"], ) self.assertEqual(data["requests"][0]["user"]["name"], "foo") self.assertEqual(data["requests"][1]["user"]["name"], "foo") self.assertEqual(data["requests"][0]["status"], "Open") self.assertEqual(data["requests"][1]["status"], "Open") self.assertEqual( data["requests"][0]["title"], "open pullrequest by user foo on repo test2", ) self.assertEqual( data["requests"][1]["title"], "open pullrequest by user foo on repo test", ) self.assertEqual(data["args"]["status"], "open") self.assertEqual(data["args"]["page"], 1) # Next test with the status parameter set to `closed`. output = self.app.get( "/api/0/user/pingou/requests/actionable?status=closed" ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 2) self.assertEqual( sorted(data.keys()), ["args", "pagination", "requests", "total_requests"], ) self.assertEqual(data["requests"][0]["user"]["name"], "foo") self.assertEqual(data["requests"][1]["user"]["name"], "foo") self.assertEqual(data["requests"][0]["status"], "Closed") self.assertEqual(data["requests"][1]["status"], "Closed") self.assertEqual( data["requests"][0]["title"], "closed pullrequest by user foo on repo test2", ) self.assertEqual( data["requests"][1]["title"], "closed pullrequest by user foo on repo test", ) self.assertEqual(data["args"]["status"], "closed") self.assertEqual(data["args"]["page"], 1) # Next test with the status parameter set to `merged`. output = self.app.get( "/api/0/user/pingou/requests/actionable?status=merged" ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 2) self.assertEqual( sorted(data.keys()), ["args", "pagination", "requests", "total_requests"], ) self.assertEqual(data["requests"][0]["user"]["name"], "foo") self.assertEqual(data["requests"][1]["user"]["name"], "foo") self.assertEqual(data["requests"][0]["status"], "Merged") self.assertEqual(data["requests"][1]["status"], "Merged") self.assertEqual( data["requests"][0]["title"], "merged pullrequest by user foo on repo test2", ) self.assertEqual( data["requests"][1]["title"], "merged pullrequest by user foo on repo test", ) self.assertEqual(data["args"]["status"], "merged") self.assertEqual(data["args"]["page"], 1) # Finally, test with the status parameter set to `all`. output = self.app.get( "/api/0/user/pingou/requests/actionable?status=all" ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 6) self.assertEqual( sorted(data.keys()), ["args", "pagination", "requests", "total_requests"], ) self.assertEqual(data["requests"][0]["user"]["name"], "foo") self.assertEqual(data["requests"][1]["user"]["name"], "foo") self.assertEqual(data["requests"][2]["user"]["name"], "foo") self.assertEqual(data["requests"][3]["user"]["name"], "foo") self.assertEqual(data["requests"][4]["user"]["name"], "foo") self.assertEqual(data["requests"][5]["user"]["name"], "foo") self.assertEqual(data["requests"][0]["status"], "Merged") self.assertEqual(data["requests"][1]["status"], "Merged") self.assertEqual(data["requests"][2]["status"], "Closed") self.assertEqual(data["requests"][3]["status"], "Closed") self.assertEqual(data["requests"][4]["status"], "Open") self.assertEqual(data["requests"][5]["status"], "Open") self.assertEqual( data["requests"][0]["title"], "merged pullrequest by user foo on repo test2", ) self.assertEqual( data["requests"][1]["title"], "merged pullrequest by user foo on repo test", ) self.assertEqual( data["requests"][2]["title"], "closed pullrequest by user foo on repo test2", ) self.assertEqual( data["requests"][3]["title"], "closed pullrequest by user foo on repo test", ) self.assertEqual( data["requests"][4]["title"], "open pullrequest by user foo on repo test2", ) self.assertEqual( data["requests"][5]["title"], "open pullrequest by user foo on repo test", ) self.assertEqual(data["args"]["status"], "all") self.assertEqual(data["args"]["page"], 1) # Test page 2 with the status parameter set to `all`. output = self.app.get( "/api/0/user/pingou/requests/actionable?status=all&page=2" ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) self.assertEqual( sorted(data.keys()), ["args", "pagination", "requests", "total_requests"], ) self.assertEqual(data["args"]["page"], 2) @patch("pagure.lib.notify.send_email") def test_api_view_user_requests_actionable_created(self, mockemail): """ Test the api_view_user_requests_filed method of the flask user api with the created parameter """ today = datetime.datetime.utcnow().date() output = self.app.get( "/api/0/user/pingou/requests/actionable?status=all&created=%s" % (today.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 6) yesterday = today - datetime.timedelta(days=1) output = self.app.get( "/api/0/user/pingou/requests/actionable?status=all&created=%s" % (yesterday.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 6) tomorrow = today + datetime.timedelta(days=1) output = self.app.get( "/api/0/user/pingou/requests/actionable?status=all&created=%s" % (tomorrow.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) output = self.app.get( "/api/0/user/pingou/requests/actionable?status=all&created=..%s" % (today.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) output = self.app.get( "/api/0/user/pingou/requests/actionable?status=all&created=..%s" % (yesterday.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) thedaybefore = today - datetime.timedelta(days=1) output = self.app.get( "/api/0/user/pingou/requests/actionable?status=all&created=..%s" % (thedaybefore.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) output = self.app.get( "/api/0/user/pingou/requests/actionable?status=all&created=..%s" % (tomorrow.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 6) output = self.app.get( "/api/0/user/pingou/requests/actionable?status=all&created=%s..%s" % (thedaybefore.isoformat(), tomorrow.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 6) @patch("pagure.lib.notify.send_email") def test_api_view_user_requests_actionable_updated(self, mockemail): """ Test the api_view_user_requests_filed method of the flask user api with the created parameter """ today = datetime.datetime.utcnow().date() output = self.app.get( "/api/0/user/pingou/requests/actionable?status=all&updated=%s" % (today.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) yesterday = today - datetime.timedelta(days=1) output = self.app.get( "/api/0/user/pingou/requests/actionable?status=all&updated=%s" % (yesterday.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) tomorrow = today + datetime.timedelta(days=1) output = self.app.get( "/api/0/user/pingou/requests/actionable?status=all&updated=%s" % (tomorrow.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 6) @patch("pagure.lib.notify.send_email") def test_api_view_user_requests_actionable_closed(self, mockemail): """ Test the api_view_user_requests_filed method of the flask user api with the created parameter """ today = datetime.datetime.utcnow().date() output = self.app.get( "/api/0/user/pingou/requests/actionable?status=all&closed=%s" % (today.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) yesterday = today - datetime.timedelta(days=1) output = self.app.get( "/api/0/user/pingou/requests/actionable?status=all&closed=%s" % (yesterday.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) tomorrow = today + datetime.timedelta(days=1) output = self.app.get( "/api/0/user/pingou/requests/actionable?status=all&closed=%s" % (tomorrow.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(len(data["requests"]), 0) class PagureFlaskApiUsertestissues(tests.Modeltests): """ Tests for the user issues endpoints """ maxDiff = None def setUp(self): """ Set up the environnment, ran before every tests. """ super(PagureFlaskApiUsertestissues, self).setUp() pagure.config.config["REQUESTS_FOLDER"] = None tests.create_projects(self.session) repo = pagure.lib.query._get_project(self.session, "test") # Create issues to play with msg = pagure.lib.query.new_issue( session=self.session, repo=repo, title="Test issue", content="We should work on this", user="pingou", ) self.session.commit() self.assertEqual(msg.title, "Test issue") def test_user_issues_empty(self): """ Return the list of issues associated with the specified user. """ output = self.app.get("/api/0/user/foo/issues") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) for k in ["pagination_issues_assigned", "pagination_issues_created"]: for k2 in ["first", "last"]: self.assertIsNotNone(data[k][k2]) data[k][k2] = None self.assertEqual( data, { "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, }, "issues_assigned": [], "issues_created": [], "pagination_issues_assigned": { "first": None, "last": None, "next": None, "page": 1, "pages": 0, "per_page": 20, "prev": None, }, "pagination_issues_created": { "first": None, "last": None, "next": None, "page": 1, "pages": 0, "per_page": 20, "prev": None, }, "total_issues_assigned": 0, "total_issues_assigned_pages": 1, "total_issues_created": 0, "total_issues_created_pages": 1, }, ) def test_user_issues(self): """ Return the list of issues associated with the specified user. """ output = self.app.get("/api/0/user/pingou/issues") self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) issues = [] for issue in data["issues_created"]: issue["date_created"] = "1513111778" issue["last_updated"] = "1513111778" issue["project"]["date_created"] = "1513111778" issue["project"]["date_modified"] = "1513111778" issues.append(issue) data["issues_created"] = issues for k in ["pagination_issues_assigned", "pagination_issues_created"]: for k2 in ["first", "last"]: self.assertIsNotNone(data[k][k2]) data[k][k2] = None self.assertEqual( data, { "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, }, "issues_assigned": [], "issues_created": [ { "assignee": None, "blocks": [], "close_status": None, "closed_at": None, "closed_by": None, "comments": [], "content": "We should work on this", "custom_fields": [], "date_created": "1513111778", "depends": [], "id": 1, "last_updated": "1513111778", "milestone": None, "priority": None, "private": False, "project": { "access_groups": { "admin": [], "commit": [], "ticket": [], }, "access_users": { "admin": [], "commit": [], "owner": ["pingou"], "ticket": [], }, "close_status": [ "Invalid", "Insufficient data", "Fixed", "Duplicate", ], "custom_keys": [], "date_created": "1513111778", "date_modified": "1513111778", "description": "test project #1", "fullname": "test", "id": 1, "milestones": {}, "name": "test", "namespace": None, "parent": None, "priorities": {}, "tags": [], "url_path": "test", "user": { "fullname": "PY C", "name": "pingou", "url_path": "user/pingou", }, }, "related_prs": [], "status": "Open", "tags": [], "title": "Test issue", "user": { "fullname": "PY C", "name": "pingou", "url_path": "user/pingou", }, } ], "pagination_issues_assigned": { "first": None, "last": None, "next": None, "page": 1, "pages": 0, "per_page": 20, "prev": None, }, "pagination_issues_created": { "first": None, "last": None, "next": None, "page": 1, "pages": 1, "per_page": 20, "prev": None, }, "total_issues_assigned": 0, "total_issues_assigned_pages": 1, "total_issues_created": 1, "total_issues_created_pages": 1, }, ) def test_user_issues_created(self): """ Return the list of issues associated with the specified user and play with the created filter. """ today = datetime.datetime.utcnow().date() output = self.app.get( "/api/0/user/pingou/issues?created=%s" % (today.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(data["total_issues_assigned"], 0) self.assertEqual(data["total_issues_created"], 1) yesterday = today - datetime.timedelta(days=1) output = self.app.get( "/api/0/user/pingou/issues?created=%s" % (yesterday.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(data["total_issues_assigned"], 0) self.assertEqual(data["total_issues_created"], 1) tomorrow = today + datetime.timedelta(days=1) output = self.app.get( "/api/0/user/pingou/issues?created=%s" % (tomorrow.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(data["total_issues_assigned"], 0) self.assertEqual(data["total_issues_created"], 0) output = self.app.get( "/api/0/user/pingou/issues?created=..%s" % (yesterday.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(data["total_issues_assigned"], 0) self.assertEqual(data["total_issues_created"], 0) output = self.app.get( "/api/0/user/pingou/issues?created=%s..%s" % (yesterday.isoformat(), today.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(data["total_issues_assigned"], 0) self.assertEqual(data["total_issues_created"], 0) output = self.app.get( "/api/0/user/pingou/issues?created=%s..%s" % (yesterday.isoformat(), tomorrow.isoformat()) ) self.assertEqual(output.status_code, 200) data = json.loads(output.get_data(as_text=True)) self.assertEqual(data["total_issues_assigned"], 0) self.assertEqual(data["total_issues_created"], 1) if __name__ == "__main__": unittest.main(verbosity=2)