# -*- coding: utf-8 -*-
"""
(c) 2018 - Copyright Red Hat Inc
Authors:
Pierre-Yves Chibon <pingou@pingoured.fr>
"""
from __future__ import unicode_literals, absolute_import
import datetime
import os
import shutil
import sys
import tempfile
import time
import unittest
import pygit2
import six
from mock import ANY, patch, MagicMock, call
sys.path.insert(
0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")
)
import pagure.lib.tasks_services
import pagure.lib.query
import tests
import pagure.lib.tasks_services
class PagureLibTaskServicestests(tests.Modeltests):
""" Tests for pagure.lib.task_services """
maxDiff = None
def setUp(self):
""" Set up the environnment, ran before every tests. """
super(PagureLibTaskServicestests, self).setUp()
tests.create_projects(self.session)
# Create a fork of test for foo
item = pagure.lib.model.Project(
user_id=2, # foo
name="test",
is_fork=True,
parent_id=1,
description="test project #1",
hook_token="aaabbbccc_foo",
)
item.close_status = [
"Invalid",
"Insufficient data",
"Fixed",
"Duplicate",
]
self.session.add(item)
self.session.commit()
def test_webhook_notification_invalid_project(self):
""" Test the webhook_notification method. """
self.assertRaises(
RuntimeError,
pagure.lib.tasks_services.webhook_notification,
topic="topic",
msg={"payload": ["a", "b", "c"]},
namespace=None,
name="invalid",
user=None,
)
@patch("pagure.lib.tasks_services.call_web_hooks")
def test_webhook_notification_no_webhook(self, call_wh):
""" Test the webhook_notification method. """
output = pagure.lib.tasks_services.webhook_notification(
topic="topic",
msg={"payload": ["a", "b", "c"]},
namespace=None,
name="test",
user=None,
)
self.assertIsNone(output)
call_wh.assert_not_called()
@patch("pagure.lib.git.log_commits_to_db")
def test_log_commit_send_notifications_invalid_project(self, log):
""" Test the log_commit_send_notifications method. """
output = pagure.lib.tasks_services.log_commit_send_notifications(
name="invalid",
commits=[],
abspath=None,
branch=None,
default_branch=None,
namespace=None,
username=None,
)
self.assertIsNone(output)
log.assert_not_called()
@patch("pagure.lib.notify.notify_new_commits")
@patch("pagure.lib.git.log_commits_to_db")
def test_log_commit_send_notifications_valid_project(self, log, notif):
""" Test the log_commit_send_notifications method. """
output = pagure.lib.tasks_services.log_commit_send_notifications(
name="test",
commits=["hash1", "hash2"],
abspath="/path/to/git",
branch="master",
default_branch="master",
namespace=None,
username=None,
)
self.assertIsNone(output)
log.assert_called_once_with(
ANY, ANY, ["hash1", "hash2"], "/path/to/git"
)
notif.assert_called_once_with(
"/path/to/git", ANY, "master", ["hash1", "hash2"]
)
@patch("pagure.lib.tasks_services.trigger_jenkins_build")
def test_trigger_ci_build_invalid_project(self, trigger_jenk):
""" Test the trigger_ci_build method. """
output = pagure.lib.tasks_services.trigger_ci_build(
project_name="invalid",
cause="PR#ID",
branch="feature",
branch_to="master",
ci_type="jenkins",
)
self.assertIsNone(output)
trigger_jenk.assert_not_called()
@patch("pagure.lib.tasks_services.trigger_jenkins_build")
def test_trigger_ci_build_not_configured_project(self, trigger_jenk):
""" Test the trigger_ci_build method. """
self.assertRaises(
pagure.exceptions.PagureException,
pagure.lib.tasks_services.trigger_ci_build,
project_name="test",
cause="PR#ID",
branch="feature",
branch_to="master",
ci_type="jenkins",
)
trigger_jenk.assert_not_called()
@patch("pagure.lib.tasks_services.trigger_jenkins_build")
def test_trigger_ci_build_not_configured_project_fork(self, trigger_jenk):
""" Test the trigger_ci_build method. """
self.assertRaises(
pagure.exceptions.PagureException,
pagure.lib.tasks_services.trigger_ci_build,
project_name="forks/foo/test",
cause="PR#ID",
branch="feature",
branch_to="master",
ci_type="jenkins",
)
trigger_jenk.assert_not_called()
@patch("pagure.lib.query._get_project")
def test_load_json_commits_to_db_invalid_data_type(self, get_project):
""" Test the load_json_commits_to_db method. """
output = pagure.lib.tasks_services.load_json_commits_to_db(
name="test",
commits=["hash1", "hash2"],
abspath="/path/to/git",
data_type="invalid",
agent="pingou",
namespace=None,
username=None,
)
self.assertIsNone(output)
get_project.assert_not_called()
@patch("pagure.lib.tasks_services.get_files_to_load")
def test_load_json_commits_to_db_invalid_project(self, get_files):
""" Test the load_json_commits_to_db method. """
output = pagure.lib.tasks_services.load_json_commits_to_db(
name="invalid",
commits=["hash1", "hash2"],
abspath="/path/to/git",
data_type="ticket",
agent="pingou",
namespace=None,
username=None,
)
self.assertIsNone(output)
get_files.assert_not_called()
@patch("pagure.lib.git.update_request_from_git")
@patch("pagure.lib.git.update_ticket_from_git")
def test_load_json_commits_to_db_invalid_path(self, up_issue, up_pr):
""" Test the load_json_commits_to_db method. """
output = pagure.lib.tasks_services.load_json_commits_to_db(
name="test",
commits=["hash1", "hash2"],
abspath=self.path,
data_type="ticket",
agent="pingou",
namespace=None,
username=None,
)
self.assertIsNone(output)
up_issue.assert_not_called()
up_pr.assert_not_called()
@patch("pagure.lib.git.update_request_from_git")
@patch("pagure.lib.git.update_ticket_from_git")
def test_load_json_commits_to_db_invalid_path_one_commit(
self, up_issue, up_pr
):
""" Test the load_json_commits_to_db method. """
output = pagure.lib.tasks_services.load_json_commits_to_db(
name="test",
commits=["hash1"],
abspath=self.path,
data_type="ticket",
agent="pingou",
namespace=None,
username=None,
)
self.assertIsNone(output)
up_issue.assert_not_called()
up_pr.assert_not_called()
@patch("pagure.lib.notify.send_email")
@patch("pagure.lib.git.update_request_from_git")
@patch("pagure.lib.git.update_ticket_from_git")
def test_load_json_commits_to_db_no_agent(self, up_issue, up_pr, send):
""" Test the load_json_commits_to_db method. """
output = pagure.lib.tasks_services.load_json_commits_to_db(
name="test",
commits=[],
abspath=None,
data_type="ticket",
agent=None,
namespace=None,
username=None,
)
self.assertIsNone(output)
up_issue.assert_not_called()
up_pr.assert_not_called()
send.assert_not_called()
@patch("pagure.lib.notify.send_email")
@patch("pagure.lib.git.update_request_from_git")
@patch("pagure.lib.git.update_ticket_from_git")
@patch("pagure.lib.git.read_git_lines")
def test_load_json_commits_to_db_no_agent(
self, git, up_issue, up_pr, send
):
""" Test the load_json_commits_to_db method. """
git.side_effect = [["file1"], ["file2"], ["files/image"], ["file1"]]
output = pagure.lib.tasks_services.load_json_commits_to_db(
name="test",
commits=["hash1", "hash2"],
abspath=self.path,
data_type="ticket",
agent=None,
namespace=None,
username=None,
)
self.assertIsNone(output)
up_issue.assert_not_called()
up_pr.assert_not_called()
send.assert_not_called()
@patch("json.loads")
@patch("pagure.lib.notify.send_email")
@patch("pagure.lib.git.update_request_from_git")
@patch("pagure.lib.git.update_ticket_from_git")
@patch("pagure.lib.git.read_git_lines")
def test_load_json_commits_to_db_tickets(
self, git, up_issue, up_pr, send, json_loads
):
""" Test the load_json_commits_to_db method. """
git.side_effect = [["file1"], ["file2"], ["files/image"], ["file1"]]
json_loads.return_value = "foobar"
output = pagure.lib.tasks_services.load_json_commits_to_db(
name="test",
commits=["hash1", "hash2"],
abspath=self.path,
data_type="ticket",
agent=None,
namespace=None,
username=None,
)
self.assertIsNone(output)
calls = [
call(
ANY,
agent=None,
issue_uid="file1",
json_data="foobar",
namespace=None,
reponame="test",
username=None,
),
call(
ANY,
agent=None,
issue_uid="file2",
json_data="foobar",
namespace=None,
reponame="test",
username=None,
),
]
self.assertEqual(calls, up_issue.mock_calls)
up_pr.assert_not_called()
send.assert_not_called()
@patch("json.loads")
@patch("pagure.lib.notify.send_email")
@patch("pagure.lib.git.update_request_from_git")
@patch("pagure.lib.git.update_ticket_from_git")
@patch("pagure.lib.git.read_git_lines")
def test_load_json_commits_to_db_prs(
self, git, up_issue, up_pr, send, json_loads
):
""" Test the load_json_commits_to_db method. """
git.side_effect = [["file1"], ["file2"], ["files/image"], ["file1"]]
json_loads.return_value = "foobar"
output = pagure.lib.tasks_services.load_json_commits_to_db(
name="test",
commits=["hash1", "hash2"],
abspath=self.path,
data_type="pull-request",
agent="pingou",
namespace=None,
username=None,
)
self.assertIsNone(output)
calls = [
call(
ANY,
json_data="foobar",
namespace=None,
reponame="test",
request_uid="file1",
username=None,
),
call(
ANY,
json_data="foobar",
namespace=None,
reponame="test",
request_uid="file2",
username=None,
),
]
up_issue.assert_not_called()
self.assertEqual(calls, up_pr.mock_calls)
calls = [
call(
"Good Morning\n\n"
"This is the log of loading all the files pushed in the git "
"repo into\n"
"the database. It should ignore files that are not JSON files,"
" this\nis fine.\n\n"
"Loading: file1 -- 1/2 ... ... Done\n"
"Loading: file2 -- 2/2 ... ... Done",
"Issue import report",
"bar@pingou.com",
)
]
self.assertEqual(calls, send.mock_calls)
@patch("json.loads")
@patch("pagure.lib.notify.send_email")
@patch("pagure.lib.git.update_request_from_git")
@patch("pagure.lib.git.update_ticket_from_git")
@patch("pagure.lib.git.read_git_lines")
def test_load_json_commits_to_db_prs_raises_error(
self, git, up_issue, up_pr, send, json_loads
):
""" Test the load_json_commits_to_db method. """
git.side_effect = [["file1"], ["file2"], ["files/image"], ["file1"]]
json_loads.return_value = "foobar"
up_pr.side_effect = Exception("foo error")
output = pagure.lib.tasks_services.load_json_commits_to_db(
name="test",
commits=["hash1", "hash2"],
abspath=self.path,
data_type="pull-request",
agent="pingou",
namespace=None,
username=None,
)
self.assertIsNone(output)
calls = [
call(
ANY,
json_data="foobar",
namespace=None,
reponame="test",
request_uid="file1",
username=None,
)
]
up_issue.assert_not_called()
self.assertEqual(calls, up_pr.mock_calls)
calls = [
call(
"Good Morning\n\n"
"This is the log of loading all the files pushed in the git "
"repo into\n"
"the database. It should ignore files that are not JSON files,"
" this\nis fine.\n\n"
"Loading: file1 -- 1/2 ... ... FAILED\n",
"Issue import report",
"bar@pingou.com",
)
]
self.assertEqual(calls, send.mock_calls)
class PagureLibTaskServicesWithWebHooktests(tests.Modeltests):
""" Tests for pagure.lib.task_services """
maxDiff = None
def setUp(self):
""" Set up the environnment, ran before every tests. """
super(PagureLibTaskServicesWithWebHooktests, self).setUp()
pagure.config.config["REQUESTS_FOLDER"] = None
self.sshkeydir = os.path.join(self.path, "sshkeys")
pagure.config.config["MIRROR_SSHKEYS_FOLDER"] = self.sshkeydir
tests.create_projects(self.session)
project = pagure.lib.query._get_project(self.session, "test")
settings = project.settings
settings["Web-hooks"] = "http://foo.com/api/flag\nhttp://bar.org/bar"
project.settings = settings
self.session.add(project)
self.session.commit()
@patch("pagure.lib.tasks_services.call_web_hooks")
def test_webhook_notification_no_webhook(self, call_wh):
""" Test the webhook_notification method. """
output = pagure.lib.tasks_services.webhook_notification(
topic="topic",
msg={"payload": ["a", "b", "c"]},
namespace=None,
name="test",
user=None,
)
self.assertIsNone(output)
project = pagure.lib.query._get_project(self.session, "test")
call_wh.assert_called_once_with(
ANY,
"topic",
{"payload": ["a", "b", "c"]},
["http://foo.com/api/flag", "http://bar.org/bar"],
)
@patch("time.time", MagicMock(return_value=2))
@patch("uuid.uuid4", MagicMock(return_value="not_so_random"))
@patch("datetime.datetime")
@patch("requests.post")
def test_webhook_notification_no_webhook(self, post, dt):
""" Test the webhook_notification method. """
post.return_value = False
utcnow = MagicMock()
utcnow.year = 2018
dt.utcnow.return_value = utcnow
output = pagure.lib.tasks_services.webhook_notification(
topic="topic",
msg={"payload": ["a", "b", "c"]},
namespace=None,
name="test",
user=None,
)
self.assertIsNone(output)
project = pagure.lib.query._get_project(self.session, "test")
self.assertEqual(post.call_count, 2)
calls = [
call(
"http://bar.org/bar",
data="{"
'"i": 1, '
'"msg": {'
'"pagure_instance": "http://localhost.localdomain/", '
'"payload": ["a", "b", "c"], '
'"project_fullname": "test"}, '
'"msg_id": "2018-not_so_random", '
'"timestamp": 2, '
'"topic": "topic"}',
headers={
"X-Pagure": "http://localhost.localdomain/",
"X-Pagure-project": "test",
"X-Pagure-Signature": "74b12f0b25bf7767014a0c0de9f3c10"
"191e943d8",
"X-Pagure-Signature-256": "f3d757796554466eac49a5282b2"
"4ee32a1ecfb65dedd6c6231fb207240a9fe58",
"X-Pagure-Topic": b"topic",
"Content-Type": "application/json",
},
timeout=60,
),
call(
"http://foo.com/api/flag",
data="{"
'"i": 1, '
'"msg": {'
'"pagure_instance": "http://localhost.localdomain/", '
'"payload": ["a", "b", "c"], '
'"project_fullname": "test"}, '
'"msg_id": "2018-not_so_random", '
'"timestamp": 2, '
'"topic": "topic"}',
headers={
"X-Pagure": "http://localhost.localdomain/",
"X-Pagure-project": "test",
"X-Pagure-Signature": "74b12f0b25bf7767014a0c0de9f3c10"
"191e943d8",
"X-Pagure-Signature-256": "f3d757796554466eac49a5282b2"
"4ee32a1ecfb65dedd6c6231fb207240a9fe58",
"X-Pagure-Topic": b"topic",
"Content-Type": "application/json",
},
timeout=60,
),
]
print(post.mock_calls)
self.assertEqual(calls, post.mock_calls)
class PagureLibTaskServicesJenkinsCItests(tests.Modeltests):
""" Tests for pagure.lib.task_services """
maxDiff = None
def setUp(self):
""" Set up the environnment, ran before every tests. """
super(PagureLibTaskServicesJenkinsCItests, self).setUp()
pagure.config.config["REQUESTS_FOLDER"] = None
self.sshkeydir = os.path.join(self.path, "sshkeys")
pagure.config.config["MIRROR_SSHKEYS_FOLDER"] = self.sshkeydir
tests.create_projects(self.session)
project = pagure.lib.query.get_authorized_project(self.session, "test")
# Install the plugin at the DB level
plugin = pagure.lib.plugins.get_plugin("Pagure CI")
dbobj = plugin.db_object()
dbobj.ci_url = "https://ci.server.org/"
dbobj.ci_job = "pagure"
dbobj.pagure_ci_token = "random_token"
dbobj.project_id = project.id
self.session.add(dbobj)
self.session.commit()
# Create a fork of test for foo
item = pagure.lib.model.Project(
user_id=2, # foo
name="test",
is_fork=True,
parent_id=1,
description="test project #1",
hook_token="aaabbbccc_foo",
)
item.close_status = [
"Invalid",
"Insufficient data",
"Fixed",
"Duplicate",
]
self.session.add(item)
self.session.commit()
@patch("pagure.lib.tasks_services.trigger_jenkins_build")
def test_trigger_ci_build_invalid_ci(self, trigger_jenk):
""" Test the trigger_ci_build method. """
output = pagure.lib.tasks_services.trigger_ci_build(
project_name="test",
cause="PR#ID",
branch="feature",
branch_to="master",
ci_type="travis",
)
self.assertIsNone(output)
trigger_jenk.assert_not_called()
@patch("pagure.lib.tasks_services.trigger_jenkins_build")
def test_trigger_ci_build_invalid_ci_fork(self, trigger_jenk):
""" Test the trigger_ci_build method. """
output = pagure.lib.tasks_services.trigger_ci_build(
project_name="forks/foo/test",
cause="PR#ID",
branch="feature",
branch_to="master",
ci_type="travis",
)
self.assertIsNone(output)
trigger_jenk.assert_not_called()
@patch("pagure.lib.tasks_services.trigger_jenkins_build")
def test_trigger_ci_build_valid_project(self, trigger_jenk):
""" Test the trigger_ci_build method. """
output = pagure.lib.tasks_services.trigger_ci_build(
project_name="test",
cause="PR#ID",
branch="feature",
branch_to="master",
ci_type="jenkins",
)
self.assertIsNone(output)
trigger_jenk.assert_called_once_with(
branch="feature",
cause="PR#ID",
ci_password=None,
ci_username=None,
job="pagure",
project_path="test.git",
token="random_token",
url="https://ci.server.org/",
branch_to="master",
)
@patch("pagure.lib.tasks_services.trigger_jenkins_build")
def test_trigger_ci_build_valid_project_fork(self, trigger_jenk):
""" Test the trigger_ci_build method. """
output = pagure.lib.tasks_services.trigger_ci_build(
project_name="forks/foo/test",
cause="PR#ID",
branch="feature",
branch_to="master",
ci_type="jenkins",
)
self.assertIsNone(output)
trigger_jenk.assert_called_once_with(
branch="feature",
cause="PR#ID",
ci_password=None,
ci_username=None,
job="pagure",
project_path="forks/foo/test.git",
token="random_token",
url="https://ci.server.org/",
branch_to="master",
)
class PagureLibTaskServicesJenkinsCIAuthtests(tests.Modeltests):
""" Tests for pagure.lib.task_services """
maxDiff = None
def setUp(self):
""" Set up the environnment, ran before every tests. """
super(PagureLibTaskServicesJenkinsCIAuthtests, self).setUp()
pagure.config.config["REQUESTS_FOLDER"] = None
self.sshkeydir = os.path.join(self.path, "sshkeys")
pagure.config.config["MIRROR_SSHKEYS_FOLDER"] = self.sshkeydir
tests.create_projects(self.session)
project = pagure.lib.query.get_authorized_project(self.session, "test")
# Install the plugin at the DB level
plugin = pagure.lib.plugins.get_plugin("Pagure CI")
dbobj = plugin.db_object()
dbobj.ci_url = "https://ci.server.org/"
dbobj.ci_job = "pagure"
dbobj.ci_username = "jenkins_username"
dbobj.ci_password = "jenkins_password"
dbobj.pagure_ci_token = "random_token"
dbobj.project_id = project.id
self.session.add(dbobj)
self.session.commit()
# Create a fork of test for foo
item = pagure.lib.model.Project(
user_id=2, # foo
name="test",
is_fork=True,
parent_id=1,
description="test project #1",
hook_token="aaabbbccc_foo",
)
item.close_status = [
"Invalid",
"Insufficient data",
"Fixed",
"Duplicate",
]
self.session.add(item)
self.session.commit()
@patch("pagure.lib.tasks_services.trigger_jenkins_build")
def test_trigger_ci_build_invalid_ci(self, trigger_jenk):
""" Test the trigger_ci_build method. """
output = pagure.lib.tasks_services.trigger_ci_build(
project_name="test",
cause="PR#ID",
branch="feature",
branch_to="master",
ci_type="travis",
)
self.assertIsNone(output)
trigger_jenk.assert_not_called()
@patch("pagure.lib.tasks_services.trigger_jenkins_build")
def test_trigger_ci_build_invalid_ci_fork(self, trigger_jenk):
""" Test the trigger_ci_build method. """
output = pagure.lib.tasks_services.trigger_ci_build(
project_name="forks/foo/test",
cause="PR#ID",
branch="feature",
branch_to="master",
ci_type="travis",
)
self.assertIsNone(output)
trigger_jenk.assert_not_called()
@patch("pagure.lib.tasks_services.trigger_jenkins_build")
def test_trigger_ci_build_valid_project(self, trigger_jenk):
""" Test the trigger_ci_build method. """
output = pagure.lib.tasks_services.trigger_ci_build(
project_name="test",
cause="PR#ID",
branch="feature",
branch_to="master",
ci_type="jenkins",
)
self.assertIsNone(output)
trigger_jenk.assert_called_once_with(
branch="feature",
cause="PR#ID",
ci_password="jenkins_password",
ci_username="jenkins_username",
job="pagure",
project_path="test.git",
token="random_token",
url="https://ci.server.org/",
branch_to="master",
)
@patch("pagure.lib.tasks_services.trigger_jenkins_build")
def test_trigger_ci_build_valid_project_fork(self, trigger_jenk):
""" Test the trigger_ci_build method. """
output = pagure.lib.tasks_services.trigger_ci_build(
project_name="forks/foo/test",
cause="PR#ID",
branch="feature",
branch_to="master",
ci_type="jenkins",
)
self.assertIsNone(output)
trigger_jenk.assert_called_once_with(
branch="feature",
cause="PR#ID",
ci_password="jenkins_password",
ci_username="jenkins_username",
job="pagure",
project_path="forks/foo/test.git",
token="random_token",
url="https://ci.server.org/",
branch_to="master",
)
class PagureLibTaskServicesLoadJsonTickettests(tests.Modeltests):
""" Tests for pagure.lib.task_services """
maxDiff = None
def setUp(self):
""" Set up the environnment, ran before every tests. """
super(PagureLibTaskServicesLoadJsonTickettests, self).setUp()
tests.create_projects(self.session)
self.gitrepo = os.path.join(self.path, "repos", "tickets", "test.git")
repopath = os.path.join(self.path, "repos", "tickets")
os.makedirs(self.gitrepo)
self.repo_obj = pygit2.init_repository(self.gitrepo, bare=True)
project = pagure.lib.query.get_authorized_project(self.session, "test")
# Create an issue to play with
msg = pagure.lib.query.new_issue(
session=self.session,
repo=project,
title="Test issue",
content="We should work on this",
user="pingou",
)
self.assertEqual(msg.title, "Test issue")
issue = pagure.lib.query.search_issues(
self.session, project, issueid=1
)
# Add a couple of comment on the ticket
msg = pagure.lib.query.add_issue_comment(
session=self.session,
issue=issue,
comment="Hey look a comment!",
user="foo",
)
self.session.commit()
self.assertEqual(msg, "Comment added")
commits = [
commit
for commit in self.repo_obj.walk(
self.repo_obj.head.target, pygit2.GIT_SORT_NONE
)
]
# 2 commits: creation - new comment
self.assertEqual(len(commits), 2)
issue = pagure.lib.query.search_issues(
self.session, project, issueid=1
)
self.assertEqual(len(issue.comments), 1)
@patch("pagure.lib.notify.send_email")
@patch("pagure.lib.git.update_request_from_git")
def test_loading_issue_json(self, up_pr, send):
""" Test loading the JSON file of a ticket. """
project = pagure.lib.query.get_authorized_project(self.session, "test")
issue = pagure.lib.query.search_issues(
self.session, project, issueid=1
)
commits = [
commit.oid.hex
for commit in self.repo_obj.walk(
self.repo_obj.head.target, pygit2.GIT_SORT_NONE
)
]
output = pagure.lib.tasks_services.load_json_commits_to_db(
name="test",
commits=commits,
abspath=self.gitrepo,
data_type="ticket",
agent="pingou",
namespace=None,
username=None,
)
self.assertIsNone(output)
up_pr.assert_not_called()
calls = [
call(
"Good Morning\n\n"
"This is the log of loading all the files pushed in the git "
"repo into\n"
"the database. It should ignore files that are not JSON files,"
" this\nis fine.\n\n"
"Loading: %s -- 1/1 ... ... Done" % issue.uid,
"Issue import report",
"bar@pingou.com",
)
]
self.assertEqual(calls, send.mock_calls)
project = pagure.lib.query.get_authorized_project(self.session, "test")
issue = pagure.lib.query.search_issues(
self.session, project, issueid=1
)
self.assertEqual(len(issue.comments), 1)
if __name__ == "__main__":
unittest.main(verbosity=2)