Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 44 additions & 0 deletions atlassian/bitbucket/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -366,6 +366,50 @@ def delete_repo_hook_script(self, project_key: str, repository_slug: str, script
url = f"{self._url_repo(project_key, repository_slug, api_version='latest')}/hook-scripts/{script_id}"
return self.delete(url)

def get_hook_script(self, script_id: int):
"""Return the registered global hook script with ``script_id``.

The result only contains the script metadata; use
``get_hook_script_content()`` to fetch the script body.
"""
return self.get(f"{self._url_hook_scripts()}/{script_id}")

def get_hook_script_content(self, script_id: int):
"""Return the raw body of the registered global hook script.

``not_json_response`` is used because the endpoint returns the
executable script as plain text rather than a JSON payload.
"""
return self.get(
f"{self._url_hook_scripts()}/{script_id}/content",
not_json_response=True,
)

def update_hook_script(
self, script_id: int, content: bytes, name: str, hook_type: str, description: Optional[str] = None
):
"""Update a registered global hook script on Bitbucket Data Center.

``hook_type`` must be ``PRE`` or ``POST``. Like ``create_hook_script``,
the API expects a multipart form; ``content`` is the executable script
bytes.
"""
if hook_type not in ("PRE", "POST"):
raise ValueError("hook_type must be 'PRE' or 'POST'")

files: Dict[str, Any] = {
"content": ("hook-script", content, "application/octet-stream"),
"name": (None, name),
"type": (None, hook_type),
}
if description is not None:
files["description"] = (None, description)
return self.put(f"{self._url_hook_scripts()}/{script_id}", files=files, headers=self.no_check_headers)

def delete_hook_script(self, script_id: int):
"""Delete the registered global hook script with ``script_id``."""
return self.delete(f"{self._url_hook_scripts()}/{script_id}")

def get_categories(self, project_key, repository_slug=None):
"""
Get a list of categories assigned to a project or repository.
Expand Down
19 changes: 19 additions & 0 deletions docs/bitbucket.rst
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,25 @@ configured scripts. ``delete_project_hook_script()`` and
``delete_repo_hook_script()`` remove only a scope configuration; they do not
delete the global registered script.

The global registered script itself is managed with the server-level methods.
``get_hook_script()`` returns a script's metadata, ``get_hook_script_content()``
returns its raw body, ``update_hook_script()`` replaces its body or metadata,
and ``delete_hook_script()`` removes it entirely:

.. code-block:: python

metadata = bitbucket.get_hook_script(hook_script["id"])
script = bitbucket.get_hook_script_content(hook_script["id"])

bitbucket.update_hook_script(
hook_script["id"],
content=script.replace(b"old", b"new"),
name="Audit pushes (v2)",
hook_type="POST",
description="Records every push",
)
bitbucket.delete_hook_script(hook_script["id"])

Release report from two refs (Server/Data Center)
-------------------------------------------------

Expand Down
44 changes: 44 additions & 0 deletions tests/test_bitbucket_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,50 @@ def test_configure_repo_hook_script(self, mock_put):
"rest/api/latest/projects/PROJ/repos/repository/hook-scripts/12", data={"triggerIds": []}
)

@patch.object(Bitbucket, "get")
def test_get_hook_script(self, mock_get):
mock_get.return_value = {"id": 12, "name": "Audit pushes"}

result = self.bitbucket.get_hook_script(12)

self.assertEqual(result["id"], 12)
mock_get.assert_called_once_with("rest/api/latest/hook-scripts/12")

@patch.object(Bitbucket, "get")
def test_get_hook_script_content_is_not_json(self, mock_get):
mock_get.return_value = b"#!/bin/sh\necho hook\n"

result = self.bitbucket.get_hook_script_content(12)

self.assertEqual(result, b"#!/bin/sh\necho hook\n")
mock_get.assert_called_once_with("rest/api/latest/hook-scripts/12/content", not_json_response=True)

@patch.object(Bitbucket, "put")
def test_update_hook_script_uses_multipart_latest_endpoint(self, mock_put):
script = b"#!/bin/sh\necho hook v2\n"
mock_put.return_value = {"id": 12}

result = self.bitbucket.update_hook_script(12, script, "Audit pushes", "POST", "Audit every push")

self.assertEqual(result, {"id": 12})
files = mock_put.call_args.kwargs["files"]
self.assertEqual(files["content"], ("hook-script", script, "application/octet-stream"))
self.assertEqual(files["name"], (None, "Audit pushes"))
self.assertEqual(files["type"], (None, "POST"))
self.assertEqual(files["description"], (None, "Audit every push"))
self.assertEqual(mock_put.call_args.args[0], "rest/api/latest/hook-scripts/12")
self.assertEqual(mock_put.call_args.kwargs["headers"], self.bitbucket.no_check_headers)

def test_update_hook_script_rejects_unknown_hook_type(self):
with self.assertRaisesRegex(ValueError, "PRE.*POST"):
self.bitbucket.update_hook_script(12, b"#!/bin/sh", "Invalid", "PRE_RECEIVE")

@patch.object(Bitbucket, "delete")
def test_delete_hook_script(self, mock_delete):
self.bitbucket.delete_hook_script(12)

mock_delete.assert_called_once_with("rest/api/latest/hook-scripts/12")


class TestPersonalRepositories(TestCase):
def setUp(self):
Expand Down
Loading