Skip to content
Open
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
49 changes: 49 additions & 0 deletions mdfluence/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -274,6 +274,17 @@ def get_parser():
help="only upload pages and attachments that have changed. "
"This adds a hash of the page or attachment contents to the update message",
)
parser.add_argument(
"--clear",
action="store_true",
help="delete all subpages of the parent page before uploading new ones",
)
parser.add_argument(
"--yes",
"-y",
action="store_true",
help="skip confirmation prompts (e.g. for --clear)",
)
parser.add_argument(
"file_list",
type=Path,
Expand Down Expand Up @@ -349,6 +360,30 @@ def main():
)
sys.exit(1)

if args.clear:
if not args.parent_id:
error_console.log(
":x: --clear requires --parent-id to identify which "
"subpages to delete\n",
markup=True,
)
sys.exit(1)

child_pages = confluence.get_child_pages(args.parent_id)
if child_pages and not args.yes:
console.log(
f"[bold red]WARNING:[/] --clear will delete "
f"{len(child_pages)} top-level child page(s) "
f"and all their descendants under parent {args.parent_id}.",
markup=True,
)
confirm = input("Continue? [y/N] ").strip().lower()
if confirm not in ("y", "yes"):
console.log("Aborted.")
sys.exit(0)

_delete_subpages(confluence, args.parent_id)

pages_to_upload = collect_pages_to_upload(args)

page_title_counts = Counter([page.title for page in pages_to_upload])
Expand Down Expand Up @@ -789,5 +824,19 @@ def collect_pages_to_upload(args):
return pages_to_upload


def _delete_subpages(confluence, page_id):
"""Recursively delete all child pages under page_id."""
child_pages = confluence.get_child_pages(page_id)
for child in child_pages:
try:
_delete_subpages(confluence, child.id)
confluence.delete_page(child.id)
console.log(f"Deleted page {child.title}")
except HTTPError as e:
error_console.log(
f"Failed to delete page {child.title}: {e.response.content}"
)


if __name__ == "__main__":
main()
28 changes: 28 additions & 0 deletions mdfluence/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,8 @@ def __init__(
def _request(self, method, path, **kwargs):
r = self.api.request(method, urljoin(self.host, path), **kwargs)
r.raise_for_status()
if r.status_code == 204:
return None
return bunchify(r.json())

def _get(self, path, **kwargs):
Expand All @@ -74,6 +76,9 @@ def _post(self, path, **kwargs):
def _put(self, path, **kwargs):
return self._request("PUT", path, **kwargs)

def _delete(self, path, **kwargs):
return self._request("DELETE", path, **kwargs)

def get_page(
self,
title=None,
Expand Down Expand Up @@ -209,6 +214,29 @@ def update_page(

return self._put(f"content/{page.id}", json=update_structure)

def delete_page(self, page_id):
return self._delete(f"content/{page_id}")

def get_child_pages(self, page_id):
"""Return all child pages, following pagination."""
all_results = []
params = {"limit": 100}
while True:
result = self._get(f"content/{page_id}/child/page", params=params)
if not result or not result.results:
break
all_results.extend(result.results)
next_link = getattr(getattr(result, "_links", None), "next", None)
if not next_link:
break
# next_link is a relative URL with query params — extract start param
from urllib.parse import parse_qs, urlparse

parsed = urlparse(next_link)
qs = parse_qs(parsed.query)
params["start"] = int(qs.get("start", [0])[0])
return all_results

def get_attachment(self, confluence_page, name):
existing_attachments = self._get(
f"content/{confluence_page.id}/child/attachment",
Expand Down
135 changes: 135 additions & 0 deletions test_package/unit/test_clear.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,135 @@
import mdfluence.__main__ as main_module
from mdfluence.api import MinimalConfluence as Confluence


def test_delete_subpages_empty_parent(mocker):
"""No children — nothing to delete."""
confluence = mocker.Mock(spec=Confluence)
confluence.get_child_pages.return_value = []

main_module._delete_subpages(confluence, "parent-123")

confluence.get_child_pages.assert_called_once_with("parent-123")
confluence.delete_page.assert_not_called()


def test_delete_subpages_flat_children(mocker):
"""Deletes all direct children (no nesting)."""
confluence = mocker.Mock(spec=Confluence)

child1 = mocker.Mock()
child1.id = "c1"
child1.title = "Child 1"
child2 = mocker.Mock()
child2.id = "c2"
child2.title = "Child 2"

# First call returns children, recursive calls return empty
confluence.get_child_pages.side_effect = [
[child1, child2], # children of parent
[], # children of c1
[], # children of c2
]

main_module._delete_subpages(confluence, "parent-123")

assert confluence.delete_page.call_count == 2
confluence.delete_page.assert_any_call("c1")
confluence.delete_page.assert_any_call("c2")


def test_delete_subpages_nested_children(mocker):
"""Recursively deletes nested children (depth-first)."""
confluence = mocker.Mock(spec=Confluence)

grandchild = mocker.Mock()
grandchild.id = "gc1"
grandchild.title = "Grandchild"
child = mocker.Mock()
child.id = "c1"
child.title = "Child"

confluence.get_child_pages.side_effect = [
[child], # children of parent
[grandchild], # children of c1
[], # children of gc1
]

main_module._delete_subpages(confluence, "parent-123")

assert confluence.delete_page.call_count == 2
# Grandchild deleted before child (depth-first)
calls = confluence.delete_page.call_args_list
assert calls[0].args[0] == "gc1"
assert calls[1].args[0] == "c1"


def test_delete_subpages_partial_failure(mocker):
"""Continues deleting remaining pages when one fails."""
from requests import HTTPError

confluence = mocker.Mock(spec=Confluence)

child1 = mocker.Mock()
child1.id = "c1"
child1.title = "Child 1"
child2 = mocker.Mock()
child2.id = "c2"
child2.title = "Child 2"

confluence.get_child_pages.side_effect = [
[child1, child2],
[],
[],
]

response_mock = mocker.Mock()
response_mock.content = b"permission denied"
confluence.delete_page.side_effect = [
HTTPError(response=response_mock), # c1 fails
None, # c2 succeeds
]

main_module._delete_subpages(confluence, "parent-123")

assert confluence.delete_page.call_count == 2


def test_get_child_pages_pagination(requests_mock):
"""get_child_pages follows pagination links."""
confluence = Confluence(host="http://example.com/api/", token="test")

requests_mock.get(
"http://example.com/api/content/123/child/page",
json={
"results": [{"id": "1", "title": "Page 1"}],
"_links": {
"next": "/api/content/123/child/page?start=1&limit=100",
},
},
)
requests_mock.get(
"http://example.com/api/content/123/child/page?start=1&limit=100",
json={
"results": [{"id": "2", "title": "Page 2"}],
"_links": {},
},
)

results = confluence.get_child_pages("123")
assert len(results) == 2
assert results[0].title == "Page 1"
assert results[1].title == "Page 2"


def test_get_child_pages_empty(requests_mock):
"""get_child_pages returns empty list for no children."""
confluence = Confluence(host="http://example.com/api/", token="test")

requests_mock.get(
"http://example.com/api/content/123/child/page",
json={"results": [], "_links": {}},
)

results = confluence.get_child_pages("123")
assert results == []
Loading