mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-08-03 05:18:22 +00:00
cc07cc0114
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>
503 lines
15 KiB
Python
503 lines
15 KiB
Python
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from domain.pashub_fetcher.sharepoint_subfolders import SharepointSubfolders
|
|
from orchestration.sharepoint_renamer_orchestrator import (
|
|
ASSESSMENT_SUBFOLDER,
|
|
BASE_PATH,
|
|
PropertyPlan,
|
|
RenameFailure,
|
|
RenameIntent,
|
|
RenameSummary,
|
|
SharepointRenamerOrchestrator,
|
|
build_canonical_filename,
|
|
)
|
|
|
|
|
|
def _assessment_path(root: str, address: str, postcode: str) -> str:
|
|
return (
|
|
f"{root}/{address}, {postcode}"
|
|
f"/{SharepointSubfolders.ASSESSMENT.value}/{ASSESSMENT_SUBFOLDER}"
|
|
)
|
|
|
|
|
|
def _write_csv(tmp_path: Path, rows: list[tuple[str, str, str]]) -> str:
|
|
lines = ["UPRN,Address,Postcode"] + [",".join(row) for row in rows]
|
|
csv_file = tmp_path / "addresses.csv"
|
|
csv_file.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
return str(csv_file)
|
|
|
|
|
|
def _make_file(name: str, item_id: str = "id-1") -> dict[str, Any]:
|
|
return {"name": name, "id": item_id, "file": {}}
|
|
|
|
|
|
def _make_folder(name: str) -> dict[str, Any]:
|
|
return {"name": name, "folder": {}}
|
|
|
|
|
|
def _make_package(name: str) -> dict[str, Any]:
|
|
return {"name": name, "package": {}}
|
|
|
|
|
|
def _make_orchestrator(
|
|
sp: MagicMock, dry_run: bool = False
|
|
) -> SharepointRenamerOrchestrator:
|
|
orchestrator = SharepointRenamerOrchestrator.__new__(SharepointRenamerOrchestrator)
|
|
orchestrator._sp_client = sp # pyright: ignore[reportPrivateUsage]
|
|
orchestrator._dry_run = dry_run # pyright: ignore[reportPrivateUsage]
|
|
return orchestrator
|
|
|
|
|
|
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
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_already_canonical_returns_none() -> None:
|
|
assert (
|
|
build_canonical_filename(
|
|
"100", "1 High St", "AB1 2CD", "100_High St AB1 2CD_Report.pdf"
|
|
)
|
|
is None
|
|
)
|
|
|
|
|
|
def test_strips_address_prefix_and_adds_uprn() -> None:
|
|
result = build_canonical_filename(
|
|
"100", "1 High St", "AB1 2CD", "1 High St AB1 2CD - Survey.pdf"
|
|
)
|
|
assert result == "100_1 High St AB1 2CD_Survey.pdf"
|
|
|
|
|
|
def test_no_prefix_still_canonical() -> None:
|
|
result = build_canonical_filename("100", "1 High St", "AB1 2CD", "Survey.pdf")
|
|
assert result == "100_1 High St AB1 2CD_Survey.pdf"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _discover_roots — batch subfolders become extra roots
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_discover_roots_includes_sero_batch_folders() -> None:
|
|
# Arrange
|
|
sp = MagicMock()
|
|
sp.get_folders_in_path.return_value = {
|
|
"value": [
|
|
_make_folder("1 High St, AB1 2CD"),
|
|
_make_folder("_Sero Batch 3"),
|
|
_make_folder("_Sero Batch 2"),
|
|
_make_file("stray.pdf"),
|
|
]
|
|
}
|
|
|
|
# Act
|
|
roots = _discover_roots(_make_orchestrator(sp))
|
|
|
|
# Assert
|
|
assert roots == [
|
|
BASE_PATH,
|
|
f"{BASE_PATH}/_Sero Batch 2",
|
|
f"{BASE_PATH}/_Sero Batch 3",
|
|
]
|
|
sp.get_folders_in_path.assert_called_once_with(BASE_PATH)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run — properties under a batch root are still found and renamed
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_run_renames_files_for_property_under_batch_root(tmp_path: Path) -> None:
|
|
# 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
|
|
SharepointRenamerOrchestrator(sp, csv_path).run()
|
|
|
|
# Assert
|
|
sp.rename_file.assert_called_once_with("id-1", "100_1 High St AB1 2CD_Survey.pdf")
|
|
|
|
|
|
def test_plan_lists_each_file_a_property_needs_renamed(tmp_path: Path) -> None:
|
|
# Arrange
|
|
csv_path = _write_csv(tmp_path, [("100", "1 High St", "AB1 2CD")])
|
|
assessment = _assessment_path(BASE_PATH, "1 High St", "AB1 2CD")
|
|
|
|
sp = MagicMock()
|
|
|
|
def fake_get(path: str) -> dict[str, Any]:
|
|
if path == BASE_PATH:
|
|
return {"value": []}
|
|
if path == assessment:
|
|
return {
|
|
"value": [
|
|
_make_file("Survey.pdf", "id-1"),
|
|
_make_file("Report.docx", "id-2"),
|
|
]
|
|
}
|
|
raise ValueError("not found")
|
|
|
|
sp.get_folders_in_path.side_effect = fake_get
|
|
|
|
# Act
|
|
plans = list(SharepointRenamerOrchestrator(sp, csv_path).plan())
|
|
|
|
# Assert
|
|
assert plans == [
|
|
PropertyPlan(
|
|
uprn="100",
|
|
address="1 High St",
|
|
postcode="AB1 2CD",
|
|
intents=[
|
|
RenameIntent(
|
|
uprn="100",
|
|
item_id="id-1",
|
|
original_name="Survey.pdf",
|
|
new_name="100_1 High St AB1 2CD_Survey.pdf",
|
|
),
|
|
RenameIntent(
|
|
uprn="100",
|
|
item_id="id-2",
|
|
original_name="Report.docx",
|
|
new_name="100_1 High St AB1 2CD_Report.docx",
|
|
),
|
|
],
|
|
)
|
|
]
|
|
sp.rename_file.assert_not_called()
|
|
|
|
|
|
def test_run_summary_counts_each_renamed_file(tmp_path: Path) -> None:
|
|
# Arrange
|
|
csv_path = _write_csv(tmp_path, [("100", "1 High St", "AB1 2CD")])
|
|
assessment = _assessment_path(BASE_PATH, "1 High St", "AB1 2CD")
|
|
|
|
sp = MagicMock()
|
|
|
|
def fake_get(path: str) -> dict[str, Any]:
|
|
if path == BASE_PATH:
|
|
return {"value": []}
|
|
if path == assessment:
|
|
return {
|
|
"value": [
|
|
_make_file("Survey.pdf", "id-1"),
|
|
_make_file("Report.docx", "id-2"),
|
|
]
|
|
}
|
|
raise ValueError("not found")
|
|
|
|
sp.get_folders_in_path.side_effect = fake_get
|
|
|
|
# Act
|
|
summary = SharepointRenamerOrchestrator(sp, csv_path).run()
|
|
|
|
# Assert
|
|
assert summary.renamed == 2
|
|
|
|
|
|
def test_run_uprn_prefixed_address_yields_single_uprn_filename(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
# Arrange
|
|
csv_path = _write_csv(tmp_path, [("100", "100_1 High St", "AB1 2CD")])
|
|
batch_assessment = _assessment_path(
|
|
f"{BASE_PATH}/_Sero Batch 2", "100_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
|
|
SharepointRenamerOrchestrator(sp, csv_path).run()
|
|
|
|
# Assert
|
|
sp.rename_file.assert_called_once_with("id-1", "100_1 High St AB1 2CD_Survey.pdf")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Walking one property's folder tree — driven through the public run()
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
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")]},
|
|
)
|
|
|
|
# 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")
|
|
|
|
|
|
def test_recurses_into_subfolders_and_renames_all_files(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")],
|
|
},
|
|
)
|
|
|
|
# Act
|
|
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)
|
|
|
|
|
|
def test_ignores_package_items(tmp_path: Path) -> None:
|
|
# Arrange
|
|
sp, csv_path = _one_property_site(tmp_path, {"": [_make_package("Notebook")]})
|
|
|
|
# Act
|
|
SharepointRenamerOrchestrator(sp, csv_path).run()
|
|
|
|
# Assert
|
|
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)
|
|
|
|
|
|
def test_run_warns_once_when_property_missing_from_all_roots(
|
|
tmp_path: Path, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
# Arrange
|
|
csv_path = _write_csv(tmp_path, [("400", "4 Elm St", "AA2 2BB")])
|
|
sp = MagicMock()
|
|
|
|
def fake_get(path: str) -> dict[str, Any]:
|
|
if path == BASE_PATH:
|
|
return {"value": [_make_folder("_Sero Batch 2")]}
|
|
raise ValueError("not found")
|
|
|
|
sp.get_folders_in_path.side_effect = fake_get
|
|
|
|
# Act
|
|
SharepointRenamerOrchestrator(sp, csv_path).run()
|
|
|
|
# Assert
|
|
sp.rename_file.assert_not_called()
|
|
warnings = [r for r in caplog.records if "Missing folder" in r.message]
|
|
assert len(warnings) == 1
|
|
assert "400" in warnings[0].message
|
|
|
|
|
|
def test_run_summary_lists_uprns_with_no_folder_under_any_root(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
# Arrange
|
|
csv_path = _write_csv(
|
|
tmp_path, [("400", "4 Elm St", "AA2 2BB"), ("500", "5 Pine Ln", "BB3 3CC")]
|
|
)
|
|
found_assessment = _assessment_path(BASE_PATH, "5 Pine Ln", "BB3 3CC")
|
|
|
|
sp = MagicMock()
|
|
|
|
def fake_get(path: str) -> dict[str, Any]:
|
|
if path == BASE_PATH:
|
|
return {"value": []}
|
|
if path == found_assessment:
|
|
return {"value": []}
|
|
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 == ["400"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Skips, failures and dry runs — driven through the public run()
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_rejected_file_is_recorded_with_its_error_and_siblings_still_rename(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
# Arrange
|
|
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")
|
|
|
|
sp.rename_file.side_effect = fake_rename
|
|
|
|
# Act
|
|
summary = SharepointRenamerOrchestrator(sp, csv_path).run()
|
|
|
|
# Assert
|
|
assert summary == RenameSummary(
|
|
renamed=1,
|
|
failed=[
|
|
RenameFailure(
|
|
uprn="100",
|
|
original_name="Locked.pdf",
|
|
new_name="100_1 High St AB1 2CD_Locked.pdf",
|
|
error="locked by another user",
|
|
)
|
|
],
|
|
)
|
|
|
|
|
|
def test_summary_counts_already_canonical_and_image_skips_apart(tmp_path: Path) -> None:
|
|
# Arrange
|
|
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 = SharepointRenamerOrchestrator(sp, csv_path).run()
|
|
|
|
# Assert
|
|
assert summary == RenameSummary(skipped_already_canonical=1, skipped_images=2)
|
|
sp.rename_file.assert_not_called()
|
|
|
|
|
|
def test_dry_run_logs_intent_and_counts_it_without_calling_the_api(
|
|
tmp_path: Path, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
# Arrange
|
|
sp, csv_path = _one_property_site(
|
|
tmp_path,
|
|
{"": [_make_file("Survey.pdf", "id-1"), _make_file("Report.docx", "id-2")]},
|
|
)
|
|
|
|
# Act
|
|
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)
|