Plan each property's renames before applying any of them 🟩

The walk now yields a PropertyPlan per address-list row, and rename()
applies one. run() composes the two, so single-pass callers are unchanged.
Knowing a property's size before the first rename is what makes "when is
this property finished?" answerable.

Tests that reached through the old private _process_folder are rewritten
against the public run(), so they survive this kind of change.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Daniel Roth 2026-07-29 09:50:28 +00:00
parent 737d07a951
commit cc07cc0114
2 changed files with 212 additions and 223 deletions

View file

@ -153,18 +153,6 @@ class SharepointRenamerOrchestrator:
walked, so a run that is killed part-way still leaves everything it
reached already renamed and, on the task lane, a completed sub_task
recording it."""
raise NotImplementedError
def rename(self, plan: PropertyPlan) -> RenameSummary:
"""Apply one property's plan, returning what actually happened.
Never raises for a file SharePoint rejects: the rejection is recorded
in the summary so one locked file cannot fail the property, let alone
the run."""
raise NotImplementedError
def run(self) -> RenameSummary:
summary = RenameSummary()
roots = self._discover_roots()
with open(self._csv_path, newline="", encoding="utf-8-sig") as f:
reader = csv.DictReader(f)
@ -175,19 +163,65 @@ class SharepointRenamerOrchestrator:
)
for row in reader:
summary.merge_in(
self._process_row(
roots,
uprn=row["UPRN"].strip(),
address=row["Address"].strip(),
postcode=row["Postcode"].strip(),
yield self._plan_row(
roots,
uprn=row["UPRN"].strip(),
address=row["Address"].strip(),
postcode=row["Postcode"].strip(),
)
def rename(self, plan: PropertyPlan) -> RenameSummary:
"""Apply one property's plan, returning what actually happened.
Never raises for a file SharePoint rejects: the rejection is recorded
in the summary so one locked file cannot fail the property, let alone
the run."""
summary = RenameSummary(
skipped_already_canonical=plan.skipped_already_canonical,
skipped_images=plan.skipped_images,
missing_folders=[plan.uprn] if plan.missing_folder else [],
)
for intent in plan.intents:
if self._dry_run:
summary.would_rename += 1
logger.info(
f'Would rename: "{intent.original_name}"'
f'"{intent.new_name}" (UPRN: {intent.uprn})'
)
continue
try:
self._sp_client.rename_file(intent.item_id, intent.new_name)
summary.renamed += 1
logger.info(
f'Renamed: "{intent.original_name}"'
f'"{intent.new_name}" (UPRN: {intent.uprn})'
)
except Exception as e:
summary.failed.append(
RenameFailure(
uprn=intent.uprn,
original_name=intent.original_name,
new_name=intent.new_name,
error=str(e),
)
)
logger.error(
f'Failed to rename "{intent.original_name}"'
f'"{intent.new_name}" (UPRN: {intent.uprn}): {e}'
)
return summary
def _process_row(
def run(self) -> RenameSummary:
"""Plan and rename every property in one pass — the entry point for
callers that want no per-property bookkeeping."""
summary = RenameSummary()
for plan in self.plan():
summary.merge_in(self.rename(plan))
return summary
def _plan_row(
self, roots: list[str], uprn: str, address: str, postcode: str
) -> RenameSummary:
) -> PropertyPlan:
# Batch folders name properties "{uprn}_{address}"; filenames must not
# repeat the UPRN, so the folder lookup and the canonical name diverge.
display_address = address.removeprefix(f"{uprn}_")
@ -196,67 +230,54 @@ class SharepointRenamerOrchestrator:
f"{root}/{address}, {postcode}"
f"/{SharepointSubfolders.ASSESSMENT.value}/{ASSESSMENT_SUBFOLDER}"
)
found = self._process_folder(folder_path, uprn, display_address, postcode)
found = self._plan_folder(folder_path, uprn, display_address, postcode)
if found is not None:
return found
logger.warning(
f"Missing folder for UPRN {uprn} in any root: {address}, {postcode}"
)
return RenameSummary(missing_folders=[uprn])
return PropertyPlan(
uprn=uprn,
address=display_address,
postcode=postcode,
missing_folder=True,
)
def _process_folder(
def _plan_folder(
self, folder_path: str, uprn: str, address: str, postcode: str
) -> Optional[RenameSummary]:
"""What renaming this folder (and its subfolders) did, or None when the
folder does not exist under this root."""
) -> Optional[PropertyPlan]:
"""What this folder (and its subfolders) needs renaming, or None when
the folder does not exist under this root."""
try:
contents = self._sp_client.get_folders_in_path(folder_path)
except ValueError:
return None
summary = RenameSummary()
plan = PropertyPlan(uprn=uprn, address=address, postcode=postcode)
for item in contents.get("value", []):
if "folder" in item:
child = self._process_folder(
child = self._plan_folder(
f"{folder_path}/{item['name']}", uprn, address, postcode
)
if child is not None:
summary.merge_in(child)
plan.merge_in(child)
elif "file" in item:
original_name: str = item["name"]
if os.path.splitext(original_name)[1].lower() in {".jpg", ".heic"}:
summary.skipped_images += 1
plan.skipped_images += 1
continue
new_name = build_canonical_filename(
uprn, address, postcode, original_name
)
if new_name is None:
summary.skipped_already_canonical += 1
plan.skipped_already_canonical += 1
continue
if self._dry_run:
summary.would_rename += 1
logger.info(
f'Would rename: "{original_name}""{new_name}" (UPRN: {uprn})'
plan.intents.append(
RenameIntent(
uprn=uprn,
item_id=item["id"],
original_name=original_name,
new_name=new_name,
)
else:
try:
self._sp_client.rename_file(item["id"], new_name)
summary.renamed += 1
logger.info(
f'Renamed: "{original_name}""{new_name}" (UPRN: {uprn})'
)
except Exception as e:
summary.failed.append(
RenameFailure(
uprn=uprn,
original_name=original_name,
new_name=new_name,
error=str(e),
)
)
logger.error(
f'Failed to rename "{original_name}""{new_name}" (UPRN: {uprn}): {e}'
)
return summary
)
return plan

View file

@ -1,5 +1,5 @@
from pathlib import Path
from typing import Any, Optional
from typing import Any
from unittest.mock import MagicMock
import pytest
@ -52,26 +52,42 @@ def _make_orchestrator(
return orchestrator
# The folder walk and root discovery are protected, but they are the unit these
# tests pin: one file can be walked without driving a whole CSV through run().
# Routed through these two helpers so the access is declared once here rather
# than at every call site.
def _process_folder(
orchestrator: SharepointRenamerOrchestrator,
folder_path: str,
uprn: str,
address: str,
postcode: str,
) -> Optional[RenameSummary]:
return orchestrator._process_folder( # pyright: ignore[reportPrivateUsage]
folder_path, uprn, address, postcode
)
def _discover_roots(orchestrator: SharepointRenamerOrchestrator) -> list[str]:
# Root discovery is protected but is its own unit; the access is declared
# once here rather than at the call site.
return orchestrator._discover_roots() # pyright: ignore[reportPrivateUsage]
def _one_property_site(
tmp_path: Path,
folders: dict[str, list[dict[str, Any]]],
uprn: str = "100",
address: str = "1 High St",
postcode: str = "AB1 2CD",
) -> tuple[MagicMock, str]:
"""A SharePoint site holding exactly one property, plus its address-list CSV.
``folders`` maps a path relative to the property's assessment folder ("" for
the folder itself, "SubA" for a subfolder) to its Graph listing. Anything
else raises ValueError, as the real client does for a folder that is not
there."""
assessment = _assessment_path(BASE_PATH, address, postcode)
listings: dict[str, dict[str, Any]] = {BASE_PATH: {"value": []}}
for relative, items in folders.items():
path = assessment if relative == "" else f"{assessment}/{relative}"
listings[path] = {"value": items}
sp = MagicMock()
def fake_get(path: str) -> dict[str, Any]:
if path not in listings:
raise ValueError(f"not found: {path}")
return listings[path]
sp.get_folders_in_path.side_effect = fake_get
return sp, _write_csv(tmp_path, [(uprn, address, postcode)])
# ---------------------------------------------------------------------------
# build_canonical_filename
# ---------------------------------------------------------------------------
@ -262,123 +278,101 @@ def test_run_uprn_prefixed_address_yields_single_uprn_filename(
# ---------------------------------------------------------------------------
# _process_folder — files only at root level
# Walking one property's folder tree — driven through the public run()
# ---------------------------------------------------------------------------
def test_renames_top_level_files(caplog: pytest.LogCaptureFixture) -> None:
sp = MagicMock()
sp.get_folders_in_path.return_value = {
"value": [
_make_file("Survey.pdf", "id-1"),
_make_file("Report.docx", "id-2"),
]
}
def test_renames_every_file_in_a_property_folder(tmp_path: Path) -> None:
# Arrange
sp, csv_path = _one_property_site(
tmp_path,
{"": [_make_file("Survey.pdf", "id-1"), _make_file("Report.docx", "id-2")]},
)
_process_folder(_make_orchestrator(sp), "some/path", "100", "1 High St", "AB1 2CD")
# Act
SharepointRenamerOrchestrator(sp, csv_path).run()
# Assert
assert sp.rename_file.call_count == 2
sp.rename_file.assert_any_call("id-1", "100_1 High St AB1 2CD_Survey.pdf")
sp.rename_file.assert_any_call("id-2", "100_1 High St AB1 2CD_Report.docx")
# ---------------------------------------------------------------------------
# _process_folder — recursive two-level hierarchy
# ---------------------------------------------------------------------------
def test_recurses_into_subfolders_and_renames_all_files() -> None:
sp = MagicMock()
root_contents: dict[str, Any] = {
"value": [
_make_file("Root.pdf", "root-file"),
_make_folder("SubA"),
]
}
suba_contents: dict[str, Any] = {
"value": [
_make_file("Sub.pdf", "sub-file"),
]
}
def fake_get(path: str) -> dict[str, Any]:
return root_contents if path == "base/path" else suba_contents
sp.get_folders_in_path.side_effect = fake_get
_process_folder(_make_orchestrator(sp), "base/path", "200", "2 Main Rd", "XY9 8ZW")
assert sp.rename_file.call_count == 2
sp.rename_file.assert_any_call("root-file", "200_2 Main Rd XY9 8ZW_Root.pdf")
sp.rename_file.assert_any_call("sub-file", "200_2 Main Rd XY9 8ZW_Sub.pdf")
sp.get_folders_in_path.assert_any_call("base/path/SubA")
def test_subfolder_results_merge_into_the_parent_summary() -> None:
def test_recurses_into_subfolders_and_renames_all_files(tmp_path: Path) -> None:
# Arrange
sp = MagicMock()
contents: dict[str, dict[str, Any]] = {
"base/path": {
"value": [_make_file("Root.pdf", "root-file"), _make_folder("SubA")]
sp, csv_path = _one_property_site(
tmp_path,
{
"": [_make_file("Root.pdf", "root-file"), _make_folder("SubA")],
"SubA": [_make_file("Sub.pdf", "sub-file")],
},
"base/path/SubA": {
"value": [
_make_file("Sub.pdf", "sub-file"),
_make_file("200_2 Main Rd XY9 8ZW_Done.pdf", "canonical-file"),
]
},
}
def fake_get(path: str) -> dict[str, Any]:
return contents[path]
sp.get_folders_in_path.side_effect = fake_get
)
# Act
summary = _process_folder(
_make_orchestrator(sp), "base/path", "200", "2 Main Rd", "XY9 8ZW"
SharepointRenamerOrchestrator(sp, csv_path).run()
# Assert
assert sp.rename_file.call_count == 2
sp.rename_file.assert_any_call("root-file", "100_1 High St AB1 2CD_Root.pdf")
sp.rename_file.assert_any_call("sub-file", "100_1 High St AB1 2CD_Sub.pdf")
def test_subfolder_results_merge_into_the_parent_summary(tmp_path: Path) -> None:
# Arrange
sp, csv_path = _one_property_site(
tmp_path,
{
"": [_make_file("Root.pdf", "root-file"), _make_folder("SubA")],
"SubA": [
_make_file("Sub.pdf", "sub-file"),
_make_file("100_1 High St AB1 2CD_Done.pdf", "canonical-file"),
],
},
)
# Act
summary = SharepointRenamerOrchestrator(sp, csv_path).run()
# Assert
assert summary == RenameSummary(renamed=2, skipped_already_canonical=1)
# ---------------------------------------------------------------------------
# _process_folder — non-file, non-folder items are skipped
# ---------------------------------------------------------------------------
def test_ignores_package_items() -> None:
sp = MagicMock()
sp.get_folders_in_path.return_value = {"value": [_make_package("Notebook")]}
_process_folder(_make_orchestrator(sp), "some/path", "300", "3 Oak Ave", "ZZ1 1ZZ")
sp.rename_file.assert_not_called()
assert sp.get_folders_in_path.call_count == 1
# ---------------------------------------------------------------------------
# _process_folder — missing folder
# ---------------------------------------------------------------------------
def test_missing_folder_returns_nothing_without_warning(
caplog: pytest.LogCaptureFixture,
) -> None:
def test_ignores_package_items(tmp_path: Path) -> None:
# Arrange
sp = MagicMock()
sp.get_folders_in_path.side_effect = ValueError("not found")
sp, csv_path = _one_property_site(tmp_path, {"": [_make_package("Notebook")]})
# Act
found = _process_folder(
_make_orchestrator(sp), "missing/path", "400", "4 Elm St", "AA2 2BB"
)
SharepointRenamerOrchestrator(sp, csv_path).run()
# Assert
assert found is None
sp.rename_file.assert_not_called()
def test_property_found_under_a_later_root_is_not_reported_missing(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""A miss on an earlier root is silent — only exhausting every root warns."""
# Arrange
csv_path = _write_csv(tmp_path, [("100", "1 High St", "AB1 2CD")])
batch_assessment = _assessment_path(
f"{BASE_PATH}/_Sero Batch 2", "1 High St", "AB1 2CD"
)
sp = MagicMock()
def fake_get(path: str) -> dict[str, Any]:
if path == BASE_PATH:
return {"value": [_make_folder("_Sero Batch 2")]}
if path == batch_assessment:
return {"value": [_make_file("Survey.pdf", "id-1")]}
raise ValueError("not found")
sp.get_folders_in_path.side_effect = fake_get
# Act
summary = SharepointRenamerOrchestrator(sp, csv_path).run()
# Assert
assert summary.missing_folders == []
assert not any("Missing folder" in r.message for r in caplog.records)
@ -434,16 +428,19 @@ def test_run_summary_lists_uprns_with_no_folder_under_any_root(
# ---------------------------------------------------------------------------
# _process_folder — already-canonical files are skipped
# Skips, failures and dry runs — driven through the public run()
# ---------------------------------------------------------------------------
def test_rejected_file_is_recorded_with_its_error_and_siblings_still_rename() -> None:
def test_rejected_file_is_recorded_with_its_error_and_siblings_still_rename(
tmp_path: Path,
) -> None:
# Arrange
sp = MagicMock()
sp.get_folders_in_path.return_value = {
"value": [_make_file("Locked.pdf", "id-1"), _make_file("Survey.pdf", "id-2")]
}
sp, csv_path = _one_property_site(
tmp_path,
{"": [_make_file("Locked.pdf", "id-1"), _make_file("Survey.pdf", "id-2")]},
)
def fake_rename(item_id: str, _new_name: str) -> None:
if item_id == "id-1":
raise PermissionError("locked by another user")
@ -451,9 +448,7 @@ def test_rejected_file_is_recorded_with_its_error_and_siblings_still_rename() ->
sp.rename_file.side_effect = fake_rename
# Act
summary = _process_folder(
_make_orchestrator(sp), "some/path", "100", "1 High St", "AB1 2CD"
)
summary = SharepointRenamerOrchestrator(sp, csv_path).run()
# Assert
assert summary == RenameSummary(
@ -469,67 +464,40 @@ def test_rejected_file_is_recorded_with_its_error_and_siblings_still_rename() ->
)
def test_skips_already_canonical_files() -> None:
sp = MagicMock()
sp.get_folders_in_path.return_value = {
"value": [_make_file("500_Pine Ln BB3 3CC_Doc.pdf", "id-y")]
}
_process_folder(_make_orchestrator(sp), "some/path", "500", "5 Pine Ln", "BB3 3CC")
sp.rename_file.assert_not_called()
def test_summary_counts_already_canonical_and_image_skips_apart() -> None:
def test_summary_counts_already_canonical_and_image_skips_apart(tmp_path: Path) -> None:
# Arrange
sp = MagicMock()
sp.get_folders_in_path.return_value = {
"value": [
_make_file("500_Pine Ln BB3 3CC_Doc.pdf", "id-1"),
_make_file("Front elevation.JPG", "id-2"),
_make_file("Rear elevation.heic", "id-3"),
]
}
sp, csv_path = _one_property_site(
tmp_path,
{
"": [
_make_file("100_1 High St AB1 2CD_Doc.pdf", "id-1"),
_make_file("Front elevation.JPG", "id-2"),
_make_file("Rear elevation.heic", "id-3"),
]
},
)
# Act
summary = _process_folder(
_make_orchestrator(sp), "some/path", "500", "5 Pine Ln", "BB3 3CC"
)
summary = SharepointRenamerOrchestrator(sp, csv_path).run()
# Assert
assert summary == RenameSummary(skipped_already_canonical=1, skipped_images=2)
# ---------------------------------------------------------------------------
# _process_folder — dry_run=True logs intent but never calls rename_file
# ---------------------------------------------------------------------------
def test_dry_run_logs_would_rename_without_calling_api(
caplog: pytest.LogCaptureFixture,
) -> None:
sp = MagicMock()
sp.get_folders_in_path.return_value = {"value": [_make_file("Survey.pdf", "id-1")]}
_process_folder(
_make_orchestrator(sp, dry_run=True), "some/path", "100", "1 High St", "AB1 2CD"
)
sp.rename_file.assert_not_called()
assert any("Would rename" in r.message for r in caplog.records)
def test_dry_run_counts_intents_separately_from_renames() -> None:
def test_dry_run_logs_intent_and_counts_it_without_calling_the_api(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
# Arrange
sp = MagicMock()
sp.get_folders_in_path.return_value = {
"value": [_make_file("Survey.pdf", "id-1"), _make_file("Report.docx", "id-2")]
}
sp, csv_path = _one_property_site(
tmp_path,
{"": [_make_file("Survey.pdf", "id-1"), _make_file("Report.docx", "id-2")]},
)
# Act
summary = _process_folder(
_make_orchestrator(sp, dry_run=True), "some/path", "100", "1 High St", "AB1 2CD"
)
summary = SharepointRenamerOrchestrator(sp, csv_path, dry_run=True).run()
# Assert
assert summary == RenameSummary(renamed=0, would_rename=2)
sp.rename_file.assert_not_called()
assert any("Would rename" in r.message for r in caplog.records)