feat(backend): add operational smoke tooling and safe pricing cleanup endpoints

- add backend README and refresh API map and smoke regression docs
- add full backend smoke regression script
- add admin pricing cleanup preview and dry-run endpoints
- add helper script for test pricing cleanup
- verify typed error contracts, draft flow, publish readiness and preview flows
- verify publish preview retention and clean backend startup behavior
This commit is contained in:
greebo
2026-03-19 22:54:12 +03:00
parent 127c5bff71
commit 0f9c2a1cbd
16 changed files with 551 additions and 235 deletions

View File

@@ -0,0 +1,127 @@
from __future__ import annotations
from app.repositories.pricing_cleanup import (
delete_pricing_categories_by_ids,
list_pricing_categories_with_rule_counts,
)
def _matches_any_prefix(value: str | None, prefixes: list[str]) -> list[str]:
if not value:
return []
matches: list[str] = []
lower_value = value.lower()
for prefix in prefixes:
if lower_value.startswith(prefix.lower()):
matches.append(prefix)
return matches
async def build_pricing_cleanup_preview(
*,
scheme_id: str,
code_prefixes: list[str],
name_prefixes: list[str],
pricing_category_ids: list[str],
delete_only_without_rules: bool,
) -> dict:
rows = await list_pricing_categories_with_rule_counts(scheme_id=scheme_id)
requested_ids = set(pricing_category_ids)
items: list[dict] = []
safe_to_delete_count = 0
for row in rows:
matched_by: list[str] = []
for prefix in _matches_any_prefix(row["code"], code_prefixes):
matched_by.append(f"code_prefix:{prefix}")
for prefix in _matches_any_prefix(row["name"], name_prefixes):
matched_by.append(f"name_prefix:{prefix}")
if row["pricing_category_id"] in requested_ids:
matched_by.append("pricing_category_id")
if not matched_by:
continue
deletable = True
if delete_only_without_rules and row["rules_count"] > 0:
deletable = False
if deletable:
safe_to_delete_count += 1
items.append(
{
"pricing_category_id": row["pricing_category_id"],
"name": row["name"],
"code": row["code"],
"rules_count": row["rules_count"],
"matched_by": matched_by,
"deletable": deletable,
}
)
return {
"scheme_id": scheme_id,
"code_prefixes": code_prefixes,
"name_prefixes": name_prefixes,
"pricing_category_ids": pricing_category_ids,
"delete_only_without_rules": delete_only_without_rules,
"total_candidates": len(items),
"safe_to_delete_count": safe_to_delete_count,
"items": items,
}
async def execute_pricing_cleanup(
*,
scheme_id: str,
code_prefixes: list[str],
name_prefixes: list[str],
pricing_category_ids: list[str],
delete_only_without_rules: bool,
dry_run: bool,
) -> dict:
preview = await build_pricing_cleanup_preview(
scheme_id=scheme_id,
code_prefixes=code_prefixes,
name_prefixes=name_prefixes,
pricing_category_ids=pricing_category_ids,
delete_only_without_rules=delete_only_without_rules,
)
deletable_items = [item for item in preview["items"] if item["deletable"]]
skipped_items = [item for item in preview["items"] if not item["deletable"]]
would_delete_ids = [item["pricing_category_id"] for item in deletable_items]
deleted_ids: list[str] = []
if not dry_run and would_delete_ids:
await delete_pricing_categories_by_ids(
scheme_id=scheme_id,
pricing_category_ids=would_delete_ids,
)
deleted_ids = list(would_delete_ids)
return {
"scheme_id": scheme_id,
"dry_run": dry_run,
"delete_only_without_rules": delete_only_without_rules,
"requested_total": len(pricing_category_ids) + len(code_prefixes) + len(name_prefixes),
"matched_total": preview["total_candidates"],
"would_delete_count": len(would_delete_ids),
"deleted_count": 0 if dry_run else len(deleted_ids),
"skipped_count": len(skipped_items),
"would_delete_category_ids": would_delete_ids,
"deleted_category_ids": deleted_ids,
"skipped": [
{
"pricing_category_id": item["pricing_category_id"],
"reason": "category_has_rules",
}
for item in skipped_items
],
}