feat(woo): S4 Woo client + taxonomy sync + test-woo + admin taxonomy

- backend/woo.py: stdlib urllib Woo v3 client. Basic-auth, HTTPS-only, X-WP-TotalPages
  pagination, typed errors (CredentialsMissing/Auth/NotFound/Server/Connection),
  injectable transport for tests, WP core media upload.
- backend/taxonomy.py: sync pulls categories/tags/attributes/terms into cache in one
  commit (rollback keeps prior cache on failure); fig-*/feat-* tag scoping; stable
  taxonomy_anchor hash; get_cached grouped admin view.
- backend/app.py: POST /settings/test-woo (fail-closed, resync on success),
  GET /admin/taxonomy, POST /admin/taxonomy/sync (502 on Woo failure).
- frontend: Settings Test Woo button + result display.
- tests: 8 woo/taxonomy tests (auth header, non-https refusal, pagination, missing
  creds, 401, scoped sync, failure-keeps-cache, anchor stability).

Verified: pytest 31 passed; check_requirements + check_coverage exit 0; npm build ok;
live QA test-woo fail-closed + sync 502.
This commit is contained in:
2026-09-19 16:40:37 -07:00
parent 38eebbc8bb
commit d83c9d5c21
5 changed files with 470 additions and 0 deletions
+74
View File
@@ -149,6 +149,10 @@ async def _unwrap_error(request: Request, exc: HTTPException):
return JSONResponse(status_code=exc.status_code, content={"error": {"code": "validation" if exc.status_code == 400 else "internal", "message": str(exc.detail), "detail": None}})
def _error(code: str, message: str, detail: Any = None) -> dict:
return {"error": {"code": code, "message": message, "detail": detail}}
@app.get("/api/health")
async def health() -> dict[str, Any]:
"""Health per APP-30/45: non-200 when db or image volume not writable (probe file)."""
@@ -180,6 +184,76 @@ async def put_settings(payload: dict[str, Any]) -> dict[str, Any]:
return settings_mod.update_settings(payload)
from . import taxonomy as taxonomy_mod
from .woo import (
WooClient,
WooAuthError,
WooConnectionError,
WooCredentialsMissing,
WooError,
)
def _woo_credentials() -> tuple[str, str | None, str | None]:
conn = get_connection()
base = os.environ.get("WOO_BASE_URL", "")
user_row = conn.execute("SELECT value FROM settings WHERE key='woo_wp_username'").fetchone()
pass_row = conn.execute("SELECT value_enc FROM settings WHERE key='woo_app_password'").fetchone()
return base, (user_row[0] if user_row else None), (pass_row[0] if pass_row else None)
@app.post("/api/v1/settings/test-woo")
async def test_woo() -> dict[str, Any]:
base, username, enc_password = _woo_credentials()
try:
client = WooClient.from_settings(base, username, enc_password)
except WooCredentialsMissing as e:
return {
"ok": False,
"detail": str(e),
"taxonomy_anchor": None,
"taxonomy_last_sync_at": None,
}
try:
result = taxonomy_mod.sync(client, get_connection())
except WooAuthError as e:
return {"ok": False, "detail": f"auth rejected by store: {e}", "taxonomy_anchor": None, "taxonomy_last_sync_at": None}
except WooConnectionError as e:
return {"ok": False, "detail": f"store unreachable: {e}", "taxonomy_anchor": None, "taxonomy_last_sync_at": None}
except WooError as e:
return {"ok": False, "detail": str(e), "taxonomy_anchor": None, "taxonomy_last_sync_at": None}
return {
"ok": True,
"detail": "taxonomy synced",
"taxonomy_anchor": result["taxonomy_anchor"],
"taxonomy_last_sync_at": result["taxonomy_last_sync_at"],
}
@app.get("/api/v1/admin/taxonomy")
async def get_taxonomy() -> dict[str, Any]:
return taxonomy_mod.get_cached(get_connection())
@app.post("/api/v1/admin/taxonomy/sync")
async def post_taxonomy_sync() -> dict[str, Any]:
base, username, enc_password = _woo_credentials()
try:
client = WooClient.from_settings(base, username, enc_password)
except WooCredentialsMissing as e:
return JSONResponse(status_code=502, content=_error("woo_unavailable", str(e)))
try:
result = taxonomy_mod.sync(client, get_connection())
except WooError as e:
return JSONResponse(status_code=502, content=_error("woo_unavailable", str(e)))
return {
"ok": True,
"detail": "taxonomy synced",
"taxonomy_anchor": result["taxonomy_anchor"],
"taxonomy_last_sync_at": result["taxonomy_last_sync_at"],
}
# expose for tests
from .slabs import router as slabs_router