[scm_request_processor] Disable webhook before creating branch
This commit is adding new functionality when creating a new branch. First checks for enabled plugins (webhooks). If plugin preventing creation of new branches is enabled it will disable it, create branch and then re-enable it. Fixes #147
This commit is contained in:
Michal Konecny
zlopez
parent
8e6f930968
commit
9813c33494
4 changed files with 507 additions and 0 deletions
|
|
@ -2122,6 +2122,7 @@ class TestCreateNewBranch:
|
|||
self.toddler.dist_git.get_branches.return_value = []
|
||||
self.toddler.dist_git.get_default_branch.return_value = default_branch
|
||||
self.toddler.dist_git._pagure_url = "https://fp.o"
|
||||
self.toddler.dist_git.get_plugins.return_value = []
|
||||
mock_fedora_account.user_member_of.return_value = True
|
||||
|
||||
mock_dir = MagicMock()
|
||||
|
|
@ -2172,6 +2173,96 @@ class TestCreateNewBranch:
|
|||
reason="Processed",
|
||||
)
|
||||
|
||||
@patch("toddlers.plugins.scm_request_processor.TemporaryDirectory")
|
||||
@patch("toddlers.plugins.scm_request_processor.git")
|
||||
@patch("toddlers.plugins.scm_request_processor.fedora_account")
|
||||
def test_create_new_branch_plugin_enabled(
|
||||
self, mock_fedora_account, mock_git, mock_temp_dir
|
||||
):
|
||||
"""
|
||||
Assert that plugin will be disabled before creating the branch.
|
||||
"""
|
||||
issue = {"id": 100, "user": {"name": "zlopez"}}
|
||||
|
||||
repo = "repo"
|
||||
default_branch = "rawhide"
|
||||
branch = "f36"
|
||||
namespace = "rpms"
|
||||
action = "new_branch"
|
||||
sls = {"rawhide": "2050-06-01"}
|
||||
json = {
|
||||
"repo": repo,
|
||||
"branch": branch,
|
||||
"namespace": namespace,
|
||||
"action": action,
|
||||
"sls": sls,
|
||||
}
|
||||
self.toddler.dist_git.get_project_contributors.return_value = {
|
||||
"users": {"admin": [], "commit": [], "collaborators": []},
|
||||
"groups": {"admin": ["group"], "commit": [], "collaborators": []},
|
||||
}
|
||||
self.toddler.dist_git.get_branches.return_value = []
|
||||
self.toddler.dist_git.get_default_branch.return_value = default_branch
|
||||
self.toddler.dist_git._pagure_url = "https://fp.o"
|
||||
self.toddler.dist_git.get_plugins.return_value = [
|
||||
scm_request_processor.PLUGIN_NAME
|
||||
]
|
||||
mock_fedora_account.user_member_of.return_value = True
|
||||
|
||||
mock_dir = MagicMock()
|
||||
mock_dir.__enter__.return_value = "dir"
|
||||
mock_temp_dir.return_value = mock_dir
|
||||
|
||||
mock_git_repo = Mock()
|
||||
mock_git_repo.first_commit.return_value = "SHA256"
|
||||
|
||||
mock_git.clone_repo.return_value = mock_git_repo
|
||||
|
||||
self.toddler.create_new_branch(issue, json)
|
||||
|
||||
# Asserts
|
||||
self.toddler.dist_git.get_project_contributors.assert_called_with(
|
||||
namespace, repo
|
||||
)
|
||||
|
||||
self.toddler.dist_git.get_branches.assert_called_with(namespace, repo)
|
||||
|
||||
mock_fedora_account.user_member_of.assert_called_with(
|
||||
mock_fedora_account.get_user_by_username(), "group"
|
||||
)
|
||||
|
||||
mock_git.clone_repo.assert_called_with(
|
||||
"{0}/{1}/{2}".format(self.toddler.dist_git._pagure_url, namespace, repo),
|
||||
"dir",
|
||||
)
|
||||
|
||||
self.toddler.dist_git.get_default_branch.assert_called_with(namespace, repo)
|
||||
|
||||
mock_git_repo.first_commit.assert_called_with(default_branch)
|
||||
self.toddler.dist_git.disable_plugin.assert_called_with(
|
||||
namespace, repo, scm_request_processor.PLUGIN_NAME
|
||||
)
|
||||
|
||||
self.toddler.dist_git.new_branch.assert_called_with(
|
||||
namespace, repo, branch, from_commit="SHA256"
|
||||
)
|
||||
self.toddler.dist_git.enable_plugin.assert_called_with(
|
||||
namespace, repo, scm_request_processor.PLUGIN_NAME
|
||||
)
|
||||
|
||||
message = (
|
||||
"The branch was created in git. It "
|
||||
"may take up to 10 minutes before you have "
|
||||
"write access on the branch."
|
||||
)
|
||||
|
||||
self.toddler.pagure_io.close_issue.assert_called_with(
|
||||
100,
|
||||
namespace=scm_request_processor.PROJECT_NAMESPACE,
|
||||
message=message,
|
||||
reason="Processed",
|
||||
)
|
||||
|
||||
@patch("toddlers.plugins.scm_request_processor.fedora_account")
|
||||
def test_create_new_branch_create_git_branch_disabled(self, mock_fedora_account):
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ Unit tests for `toddlers.utils.pagure`.
|
|||
import json
|
||||
import logging
|
||||
from unittest.mock import call, MagicMock, Mock, patch
|
||||
import urllib
|
||||
|
||||
import pytest
|
||||
|
||||
|
|
@ -2305,3 +2306,210 @@ class TestUpdateWatcherStatus:
|
|||
data=json.dumps({"status": status, "watcher": watcher}),
|
||||
headers=self.pagure.get_auth_header(),
|
||||
)
|
||||
|
||||
|
||||
class TestPagureGetPlugins:
|
||||
"""
|
||||
Test class for `toddlers.utils.pagure.Pagure.get_plugins` method.
|
||||
"""
|
||||
|
||||
def setup_method(self):
|
||||
"""
|
||||
Setup method for the test class.
|
||||
"""
|
||||
config = {
|
||||
"pagure_url": "https://pagure.io",
|
||||
"pagure_api_key": "Very secret key",
|
||||
}
|
||||
self.pagure = pagure.set_pagure(config)
|
||||
self.pagure._requests_session = Mock()
|
||||
|
||||
def test_get_plugins(self):
|
||||
"""
|
||||
Assert correct retrieval of plugins enabled on repository
|
||||
"""
|
||||
namespace = "rpms"
|
||||
package = "foo"
|
||||
|
||||
response_mock = MagicMock()
|
||||
response_mock.ok.return_value = True
|
||||
response_mock.json.return_value = {
|
||||
"plugins": [
|
||||
{"plugin_01": {}},
|
||||
{"plugin_02": {}},
|
||||
]
|
||||
}
|
||||
|
||||
# mock_request.side_effect = request
|
||||
self.pagure._requests_session.get.return_value = response_mock
|
||||
|
||||
result = self.pagure.get_plugins(namespace, package)
|
||||
assert result == [
|
||||
"plugin_01",
|
||||
"plugin_02",
|
||||
]
|
||||
|
||||
def test_get_plugins_failure(self):
|
||||
"""
|
||||
Assert that failing to get plugins is handled correctly.
|
||||
"""
|
||||
response_mock = MagicMock()
|
||||
response_mock.ok = False
|
||||
response_mock.headers = {"content-type": "application/json"}
|
||||
|
||||
self.pagure._requests_session.get.return_value = response_mock
|
||||
|
||||
namespace = "rpms"
|
||||
package = "foo"
|
||||
|
||||
expected_error = "Failed to get enabled plugins on '{0}/{1}'".format(
|
||||
namespace, package
|
||||
)
|
||||
|
||||
with pytest.raises(PagureError, match=expected_error):
|
||||
self.pagure.get_plugins(namespace, package)
|
||||
|
||||
self.pagure._requests_session.get.assert_called_with(
|
||||
"https://pagure.io/api/0/{0}/{1}/settings/plugins".format(
|
||||
namespace, package
|
||||
),
|
||||
headers=self.pagure.get_auth_header(),
|
||||
)
|
||||
|
||||
|
||||
class TestPagureDisablePlugin:
|
||||
"""
|
||||
Test class for `toddlers.utils.pagure.Pagure.disable_plugin` method.
|
||||
"""
|
||||
|
||||
def setup_method(self):
|
||||
"""
|
||||
Setup method for the test class.
|
||||
"""
|
||||
config = {
|
||||
"pagure_url": "https://pagure.io",
|
||||
"pagure_api_key": "Very secret key",
|
||||
}
|
||||
self.pagure = pagure.set_pagure(config)
|
||||
self.pagure._requests_session = Mock()
|
||||
|
||||
def test_disable_plugin(self):
|
||||
"""
|
||||
Assert plugin is correctly disabled on repository
|
||||
"""
|
||||
namespace = "rpms"
|
||||
package = "foo"
|
||||
plugin = "Some plugin"
|
||||
|
||||
response_mock = MagicMock()
|
||||
response_mock.ok.return_value = True
|
||||
|
||||
# mock_request.side_effect = request
|
||||
self.pagure._requests_session.post.return_value = response_mock
|
||||
|
||||
self.pagure.disable_plugin(namespace, package, plugin)
|
||||
self.pagure._requests_session.post.assert_called_with(
|
||||
"https://pagure.io/api/0/{0}/{1}/settings/{2}/remove".format(
|
||||
namespace, package, urllib.parse.quote(plugin)
|
||||
),
|
||||
data=json.dumps({}),
|
||||
headers=self.pagure.get_auth_header(),
|
||||
)
|
||||
|
||||
def test_disable_plugin_failure(self):
|
||||
"""
|
||||
Assert that failing to disable plugin is handled correctly.
|
||||
"""
|
||||
response_mock = MagicMock()
|
||||
response_mock.ok = False
|
||||
response_mock.headers = {"content-type": "application/json"}
|
||||
|
||||
self.pagure._requests_session.post.return_value = response_mock
|
||||
|
||||
namespace = "rpms"
|
||||
package = "foo"
|
||||
plugin = "Some plugin"
|
||||
|
||||
expected_error = "Failed to disable plugin on '{0}/{1}'".format(
|
||||
namespace, package
|
||||
)
|
||||
|
||||
with pytest.raises(PagureError, match=expected_error):
|
||||
self.pagure.disable_plugin(namespace, package, plugin)
|
||||
|
||||
self.pagure._requests_session.post.assert_called_with(
|
||||
"https://pagure.io/api/0/{0}/{1}/settings/{2}/remove".format(
|
||||
namespace, package, urllib.parse.quote(plugin)
|
||||
),
|
||||
data=json.dumps({}),
|
||||
headers=self.pagure.get_auth_header(),
|
||||
)
|
||||
|
||||
|
||||
class TestPagureEnablePlugin:
|
||||
"""
|
||||
Test class for `toddlers.utils.pagure.Pagure.enable_plugin` method.
|
||||
"""
|
||||
|
||||
def setup_method(self):
|
||||
"""
|
||||
Setup method for the test class.
|
||||
"""
|
||||
config = {
|
||||
"pagure_url": "https://pagure.io",
|
||||
"pagure_api_key": "Very secret key",
|
||||
}
|
||||
self.pagure = pagure.set_pagure(config)
|
||||
self.pagure._requests_session = Mock()
|
||||
|
||||
def test_enable_plugin(self):
|
||||
"""
|
||||
Assert plugin is correctly enabled on repository
|
||||
"""
|
||||
namespace = "rpms"
|
||||
package = "foo"
|
||||
plugin = "Some plugin"
|
||||
|
||||
response_mock = MagicMock()
|
||||
response_mock.ok.return_value = True
|
||||
|
||||
# mock_request.side_effect = request
|
||||
self.pagure._requests_session.post.return_value = response_mock
|
||||
|
||||
self.pagure.enable_plugin(namespace, package, plugin)
|
||||
self.pagure._requests_session.post.assert_called_with(
|
||||
"https://pagure.io/api/0/{0}/{1}/settings/{2}/install".format(
|
||||
namespace, package, urllib.parse.quote(plugin)
|
||||
),
|
||||
data=json.dumps({}),
|
||||
headers=self.pagure.get_auth_header(),
|
||||
)
|
||||
|
||||
def test_enable_plugin_failure(self):
|
||||
"""
|
||||
Assert that failing to enable plugin is handled correctly.
|
||||
"""
|
||||
response_mock = MagicMock()
|
||||
response_mock.ok = False
|
||||
response_mock.headers = {"content-type": "application/json"}
|
||||
|
||||
self.pagure._requests_session.post.return_value = response_mock
|
||||
|
||||
namespace = "rpms"
|
||||
package = "foo"
|
||||
plugin = "Some plugin"
|
||||
|
||||
expected_error = "Failed to enable plugin on '{0}/{1}'".format(
|
||||
namespace, package
|
||||
)
|
||||
|
||||
with pytest.raises(PagureError, match=expected_error):
|
||||
self.pagure.enable_plugin(namespace, package, plugin)
|
||||
|
||||
self.pagure._requests_session.post.assert_called_with(
|
||||
"https://pagure.io/api/0/{0}/{1}/settings/{2}/install".format(
|
||||
namespace, package, urllib.parse.quote(plugin)
|
||||
),
|
||||
data=json.dumps({}),
|
||||
headers=self.pagure.get_auth_header(),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -33,6 +33,9 @@ PROJECT_NAME_REGEX = r"^[a-zA-Z0-9_][a-zA-Z0-9-_.+]*$"
|
|||
# Regex for epel branch validation
|
||||
EPEL_REGEX = r"^epel\d+(?:\.\d+)?(?:-next)?$"
|
||||
|
||||
# Plugin preventing branch creation
|
||||
PLUGIN_NAME = "Prevent creating new branches by git push"
|
||||
|
||||
# Where to look for the scm-requests tickets
|
||||
PROJECT_NAMESPACE = "releng/fedora-scm-requests"
|
||||
|
||||
|
|
@ -921,9 +924,19 @@ class SCMRequestProcessor(ToddlerBase):
|
|||
return
|
||||
commit = git_repo.first_commit(default_branch)
|
||||
if commit:
|
||||
# Check if the plugin preventing branch creation is enabled
|
||||
plugins = self.dist_git.get_plugins(namespace, repo)
|
||||
plugin_enabled = False
|
||||
# Plugin is enabled, disabling it
|
||||
if PLUGIN_NAME in plugins:
|
||||
plugin_enabled = True
|
||||
self.dist_git.disable_plugin(namespace, repo, PLUGIN_NAME)
|
||||
self.dist_git.new_branch(
|
||||
namespace, repo, branch_name, from_commit=commit
|
||||
)
|
||||
# Re-enable the plugin
|
||||
if plugin_enabled:
|
||||
self.dist_git.enable_plugin(namespace, repo, PLUGIN_NAME)
|
||||
new_branch_comment = (
|
||||
"The branch was created in git. It "
|
||||
"may take up to 10 minutes before you have "
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ Examples:
|
|||
import json
|
||||
import logging
|
||||
from typing import Any, List, Optional
|
||||
import urllib
|
||||
|
||||
from toddlers.exceptions import PagureError
|
||||
from toddlers.utils import requests
|
||||
|
|
@ -1526,3 +1527,197 @@ class Pagure:
|
|||
watcher,
|
||||
status,
|
||||
)
|
||||
|
||||
def get_plugins(
|
||||
self,
|
||||
namespace: str,
|
||||
project: str,
|
||||
) -> List[str]:
|
||||
"""
|
||||
Get enabled plugins on project.
|
||||
|
||||
Params:
|
||||
namespace: Namespace of the project
|
||||
project: Project name
|
||||
|
||||
Raises:
|
||||
`toddlers.utils.exceptions.PagureError``: When getting plugins fails.
|
||||
"""
|
||||
api_url = "{0}/api/0/{1}/{2}/settings/plugins".format(
|
||||
self._pagure_url, namespace, project
|
||||
)
|
||||
headers = self.get_auth_header()
|
||||
|
||||
log.debug(
|
||||
"Get enabled plugins on '%s/%s'",
|
||||
namespace,
|
||||
project,
|
||||
)
|
||||
response = self._requests_session.get(api_url, headers=headers)
|
||||
|
||||
if not response.ok:
|
||||
log.error(
|
||||
"Getting enabled plugins on '%s/%s' failed. " "Got status_code '%s'.",
|
||||
namespace,
|
||||
project,
|
||||
response.status_code,
|
||||
)
|
||||
|
||||
response_json = None
|
||||
print(response.headers)
|
||||
if response.headers.get("content-type") == "application/json":
|
||||
response_json = response.json()
|
||||
log.error("Received response: %s", response.json())
|
||||
|
||||
raise PagureError(
|
||||
(
|
||||
"Failed to get enabled plugins on '{0}/{1}'\n\n"
|
||||
"Request to '{2}:\n\n"
|
||||
"Response:\n"
|
||||
"{3}\n\n"
|
||||
"Status code: {4}"
|
||||
).format(
|
||||
namespace,
|
||||
project,
|
||||
api_url,
|
||||
response_json,
|
||||
response.status_code,
|
||||
)
|
||||
)
|
||||
|
||||
response_json = response.json()
|
||||
plugins = []
|
||||
for plugin in response_json["plugins"]:
|
||||
plugins.extend(plugin.keys())
|
||||
|
||||
log.debug(
|
||||
"Retrieved enabled plugins on '%s/%s': %s", namespace, project, plugins
|
||||
)
|
||||
|
||||
return plugins
|
||||
|
||||
def disable_plugin(
|
||||
self,
|
||||
namespace: str,
|
||||
project: str,
|
||||
plugin: str,
|
||||
) -> None:
|
||||
"""
|
||||
Disable plugin on project.
|
||||
|
||||
Params:
|
||||
namespace: Namespace of the project
|
||||
project: Project name
|
||||
plugin: Plugin to disable
|
||||
|
||||
Raises:
|
||||
`toddlers.utils.exceptions.PagureError``: When disabling plugins fails.
|
||||
"""
|
||||
api_url = "{0}/api/0/{1}/{2}/settings/{3}/remove".format(
|
||||
self._pagure_url, namespace, project, urllib.parse.quote(plugin)
|
||||
)
|
||||
headers = self.get_auth_header()
|
||||
|
||||
log.debug(
|
||||
"Disable plugin '%s' on '%s/%s'",
|
||||
plugin,
|
||||
namespace,
|
||||
project,
|
||||
)
|
||||
response = self._requests_session.post(
|
||||
api_url, data=json.dumps({}), headers=headers
|
||||
)
|
||||
|
||||
if not response.ok:
|
||||
log.error(
|
||||
"Disabling plugin on '%s/%s' failed. " "Got status_code '%s'.",
|
||||
namespace,
|
||||
project,
|
||||
response.status_code,
|
||||
)
|
||||
|
||||
response_json = None
|
||||
if response.headers.get("content-type") == "application/json":
|
||||
response_json = response.json()
|
||||
log.error("Received response: %s", response.json())
|
||||
|
||||
raise PagureError(
|
||||
(
|
||||
"Failed to disable plugin on '{0}/{1}'\n\n"
|
||||
"Request to '{2}:\n\n"
|
||||
"Response:\n"
|
||||
"{3}\n\n"
|
||||
"Status code: {4}"
|
||||
).format(
|
||||
namespace,
|
||||
project,
|
||||
api_url,
|
||||
response_json,
|
||||
response.status_code,
|
||||
)
|
||||
)
|
||||
|
||||
log.debug("Disabled plugin '%s' on '%s/%s': %s", plugin, namespace, project)
|
||||
|
||||
def enable_plugin(
|
||||
self,
|
||||
namespace: str,
|
||||
project: str,
|
||||
plugin: str,
|
||||
) -> None:
|
||||
"""
|
||||
Enable plugin on project.
|
||||
|
||||
Params:
|
||||
namespace: Namespace of the project
|
||||
project: Project name
|
||||
plugin: Plugin to enable
|
||||
|
||||
Raises:
|
||||
`toddlers.utils.exceptions.PagureError``: When disabling plugins fails.
|
||||
"""
|
||||
api_url = "{0}/api/0/{1}/{2}/settings/{3}/install".format(
|
||||
self._pagure_url, namespace, project, urllib.parse.quote(plugin)
|
||||
)
|
||||
headers = self.get_auth_header()
|
||||
|
||||
log.debug(
|
||||
"Enable plugin '%s' on '%s/%s'",
|
||||
plugin,
|
||||
namespace,
|
||||
project,
|
||||
)
|
||||
response = self._requests_session.post(
|
||||
api_url, data=json.dumps({}), headers=headers
|
||||
)
|
||||
|
||||
if not response.ok:
|
||||
log.error(
|
||||
"Enabling plugin on '%s/%s' failed. " "Got status_code '%s'.",
|
||||
namespace,
|
||||
project,
|
||||
response.status_code,
|
||||
)
|
||||
|
||||
response_json = None
|
||||
if response.headers.get("content-type") == "application/json":
|
||||
response_json = response.json()
|
||||
log.error("Received response: %s", response.json())
|
||||
|
||||
raise PagureError(
|
||||
(
|
||||
"Failed to enable plugin on '{0}/{1}'\n\n"
|
||||
"Request to '{2}:\n\n"
|
||||
"Response:\n"
|
||||
"{3}\n\n"
|
||||
"Status code: {4}"
|
||||
).format(
|
||||
namespace,
|
||||
project,
|
||||
api_url,
|
||||
response_json,
|
||||
response.status_code,
|
||||
)
|
||||
)
|
||||
|
||||
log.debug("Enabled plugin '%s' on '%s/%s': %s", plugin, namespace, project)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue