Ty
18c262b5b4
Add Settings species pricing UI (children of Species grouper by slug, never hardcoded ids). Map Woo REST parent→parent_id on taxonomy sync; upsert categories so slab FKs survive re-sync. Tests: parent map, FK-safe sync, TV-8 0.96×12.50→12.00, Playwright mock e2e. UAT verified on 5060.
233 lines
8.9 KiB
Python
233 lines
8.9 KiB
Python
"""S4 Woo client + taxonomy sync tests (mocked transport; never hit the real store)."""
|
|
|
|
import base64
|
|
import json
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
|
|
import pytest
|
|
|
|
import backend.db as db_mod
|
|
from backend.woo import (
|
|
WooAuthError,
|
|
WooClient,
|
|
WooConnectionError,
|
|
WooCredentialsMissing,
|
|
WooForbiddenError,
|
|
WooServerError,
|
|
)
|
|
from backend import taxonomy as taxonomy_mod
|
|
|
|
|
|
@pytest.fixture
|
|
def db_dir():
|
|
tmp = tempfile.mkdtemp(prefix="woo_")
|
|
os.environ["SLAB_DB_PATH"] = os.path.join(tmp, "t.db")
|
|
os.environ["SLAB_IMAGES_DIR"] = tmp
|
|
os.environ["WOO_BASE_URL"] = "https://store.example.com"
|
|
os.environ["SLAB_AES_KEY"] = base64.b64encode(b"0123456789abcdef0123456789abcdef").decode()
|
|
db_mod._reset_for_tests()
|
|
conn = db_mod.init_db()
|
|
from backend.crypto import encrypt
|
|
|
|
conn.execute("INSERT OR REPLACE INTO settings(key, value, value_enc) VALUES ('woo_wp_username','admin',NULL)")
|
|
conn.execute("INSERT OR REPLACE INTO settings(key, value, value_enc) VALUES ('woo_app_password',NULL,?)", (encrypt("secret"),))
|
|
conn.commit()
|
|
yield conn
|
|
db_mod._reset_for_tests()
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
|
|
def _client(responses):
|
|
"""Build a WooClient whose transport returns pre-canned (code, headers, body) per call."""
|
|
calls = {"n": 0}
|
|
|
|
def fake_request(url, method, data, headers):
|
|
i = min(calls["n"], len(responses) - 1)
|
|
calls["n"] += 1
|
|
code, hdr, body = responses[i]
|
|
return code, hdr, body
|
|
|
|
return WooClient("admin", "secret", "https://store.example.com", request=fake_request), calls
|
|
|
|
|
|
def test_refuses_non_https(db_dir):
|
|
with pytest.raises(WooCredentialsMissing):
|
|
WooClient("admin", "secret", "http://insecure.example.com")
|
|
|
|
|
|
def test_basic_auth_header_and_url(db_dir):
|
|
seen = {}
|
|
|
|
def fake_request(url, method, data, headers):
|
|
seen["url"] = url
|
|
seen["auth"] = headers["Authorization"]
|
|
return 200, {"X-WP-TotalPages": "1"}, b"[]"
|
|
|
|
c = WooClient("admin", "secret", "https://store.example.com", request=fake_request)
|
|
c.get_categories()
|
|
assert seen["url"] == "https://store.example.com/wp-json/wc/v3/products/categories?per_page=100&page=1"
|
|
expected = "Basic " + base64.b64encode(b"admin:secret").decode()
|
|
assert seen["auth"] == expected
|
|
|
|
|
|
def test_pagination_follows_total_pages(db_dir):
|
|
p1 = (200, {"X-WP-TotalPages": "2"}, b'[{"id":1}]')
|
|
p2 = (200, {"X-WP-TotalPages": "2"}, b'[{"id":2}]')
|
|
c, calls = _client([p1, p2])
|
|
data = c.get_tags()
|
|
assert [d["id"] for d in data] == [1, 2]
|
|
assert calls["n"] == 2
|
|
|
|
|
|
def test_missing_password_fails_closed(db_dir):
|
|
with pytest.raises(WooCredentialsMissing):
|
|
WooClient.from_settings("https://store.example.com", "admin", None)
|
|
|
|
|
|
def test_401_raises_auth_error(db_dir):
|
|
def fake_request(url, method, data, headers):
|
|
raise WooAuthError("authentication rejected (401)")
|
|
|
|
c = WooClient("admin", "secret", "https://store.example.com", request=fake_request)
|
|
with pytest.raises(WooAuthError):
|
|
c.get_categories()
|
|
|
|
|
|
def test_sync_populates_scoped(db_dir):
|
|
cats = (200, {"X-WP-TotalPages": "1"}, json.dumps([{"id": 1, "name": "Cedar", "slug": "cedar", "parent_id": None, "woo_id": 10}]).encode())
|
|
tags = (200, {"X-WP-TotalPages": "1"}, json.dumps([
|
|
{"id": 2, "name": "fig-curl", "slug": "fig-curl", "woo_id": 20},
|
|
{"id": 3, "name": "feat-knot", "slug": "feat-knot", "woo_id": 21},
|
|
{"id": 4, "name": "other-tag", "slug": "other-tag", "woo_id": 22}, # must be dropped
|
|
]).encode())
|
|
attrs = (200, {"X-WP-TotalPages": "1"}, json.dumps([{"id": 5, "name": "Edge Type", "slug": "edge", "woo_id": 30}]).encode())
|
|
terms = (200, {"X-WP-TotalPages": "1"}, json.dumps([{"id": 6, "name": "Flat", "slug": "flat", "woo_id": 40}]).encode())
|
|
c, _ = _client([cats, tags, attrs, terms])
|
|
result = taxonomy_mod.sync(c, db_mod.get_connection())
|
|
conn = db_mod.get_connection()
|
|
assert [r[0] for r in conn.execute("SELECT id FROM woo_categories")] == [1]
|
|
tag_ids = {r[0] for r in conn.execute("SELECT id FROM woo_tags")}
|
|
assert tag_ids == {2, 3} # fig- + feat- kept, other-tag dropped
|
|
assert [r[0] for r in conn.execute("SELECT id FROM woo_attribute_terms")] == [6]
|
|
assert result["taxonomy_anchor"]
|
|
assert conn.execute("SELECT value FROM settings WHERE key='taxonomy_last_sync_at'").fetchone()[0]
|
|
|
|
|
|
def test_sync_failure_keeps_previous_cache(db_dir):
|
|
conn = db_mod.get_connection()
|
|
conn.execute("INSERT INTO woo_categories(id,name,slug,parent_id,woo_id) VALUES (99,'Old','old',NULL,1)")
|
|
conn.commit()
|
|
|
|
def bad_request(url, method, data, headers):
|
|
raise WooServerError("boom")
|
|
|
|
c = WooClient("admin", "secret", "https://store.example.com", request=bad_request)
|
|
with pytest.raises(WooServerError):
|
|
taxonomy_mod.sync(c, conn)
|
|
# previous cache retained
|
|
assert [r[0] for r in conn.execute("SELECT id FROM woo_categories")] == [99]
|
|
|
|
|
|
def test_anchor_stable_across_identical_syncs(db_dir):
|
|
cats = (200, {"X-WP-TotalPages": "1"}, json.dumps([{"id": 1, "name": "Cedar", "slug": "cedar", "parent_id": None, "woo_id": 10}]).encode())
|
|
tags = (200, {"X-WP-TotalPages": "1"}, json.dumps([{"id": 2, "name": "fig-curl", "slug": "fig-curl", "woo_id": 20}]).encode())
|
|
attrs = (200, {"X-WP-TotalPages": "1"}, b"[]")
|
|
c1, _ = _client([cats, tags, attrs])
|
|
r1 = taxonomy_mod.sync(c1, db_mod.get_connection())
|
|
c2, _ = _client([cats, tags, attrs])
|
|
r2 = taxonomy_mod.sync(c2, db_mod.get_connection())
|
|
assert r1["taxonomy_anchor"] == r2["taxonomy_anchor"]
|
|
|
|
|
|
def test_sync_maps_woo_parent_field_to_parent_id(db_dir):
|
|
"""Woo REST categories use `parent` (0 = top-level), not `parent_id`.
|
|
|
|
Live UAT lost hierarchy when sync only read parent_id (SETTINGS-35 / METADATA-10).
|
|
Fixture ids are local to this test — not production store ids.
|
|
"""
|
|
cats = (
|
|
200,
|
|
{"X-WP-TotalPages": "1"},
|
|
json.dumps(
|
|
[
|
|
{"id": 10, "name": "Species", "slug": "species", "parent": 0},
|
|
{"id": 11, "name": "Cedar", "slug": "cedar", "parent": 10},
|
|
{"id": 20, "name": "Wood category", "slug": "wcat", "parent": 0},
|
|
{"id": 21, "name": "Live edge wood", "slug": "liveedge", "parent": 20},
|
|
]
|
|
).encode(),
|
|
)
|
|
tags = (200, {"X-WP-TotalPages": "1"}, b"[]")
|
|
attrs = (200, {"X-WP-TotalPages": "1"}, b"[]")
|
|
c, _ = _client([cats, tags, attrs])
|
|
taxonomy_mod.sync(c, db_mod.get_connection())
|
|
conn = db_mod.get_connection()
|
|
rows = {
|
|
r[0]: {"name": r[1], "slug": r[2], "parent_id": r[3], "woo_id": r[4]}
|
|
for r in conn.execute("SELECT id, name, slug, parent_id, woo_id FROM woo_categories")
|
|
}
|
|
assert rows[10]["parent_id"] is None
|
|
assert rows[11]["parent_id"] == 10
|
|
assert rows[20]["parent_id"] is None
|
|
assert rows[21]["parent_id"] == 20
|
|
# woo_id falls back to Woo category id when the payload has no woo_id field
|
|
assert rows[11]["woo_id"] == 11
|
|
|
|
|
|
def test_sync_with_slab_species_fk_does_not_fail(db_dir):
|
|
"""Live UAT: DELETE woo_categories fails when a slab still points at species_id."""
|
|
conn = db_mod.get_connection()
|
|
conn.execute(
|
|
"INSERT INTO woo_categories(id,name,slug,parent_id,woo_id) VALUES (11,'Old Cedar','old-cedar',NULL,11)"
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO slabs(id, status, species_id, server_rev, created_at, updated_at) "
|
|
"VALUES ('slab-fk-1', 'draft', 11, 1, 't', 't')"
|
|
)
|
|
conn.commit()
|
|
|
|
cats = (
|
|
200,
|
|
{"X-WP-TotalPages": "1"},
|
|
json.dumps(
|
|
[
|
|
{"id": 10, "name": "Species", "slug": "species", "parent": 0},
|
|
{"id": 11, "name": "Cedar", "slug": "cedar", "parent": 10},
|
|
]
|
|
).encode(),
|
|
)
|
|
tags = (200, {"X-WP-TotalPages": "1"}, b"[]")
|
|
attrs = (200, {"X-WP-TotalPages": "1"}, b"[]")
|
|
c, _ = _client([cats, tags, attrs])
|
|
taxonomy_mod.sync(c, conn)
|
|
rows = {
|
|
r[0]: r[3]
|
|
for r in conn.execute("SELECT id, name, slug, parent_id FROM woo_categories")
|
|
}
|
|
assert rows[11] == 10 # parent mapped; row kept for FK
|
|
assert conn.execute("SELECT species_id FROM slabs WHERE id='slab-fk-1'").fetchone()[0] == 11
|
|
|
|
|
|
def test_403_raises_forbidden_error(db_dir):
|
|
def fake_request(url, method, data, headers):
|
|
raise WooForbiddenError("request forbidden (403)")
|
|
|
|
c = WooClient("admin", "secret", "https://store.example.com", request=fake_request)
|
|
with pytest.raises(WooForbiddenError):
|
|
c.get_categories()
|
|
|
|
|
|
def test_requests_include_user_agent(db_dir):
|
|
seen_headers = {}
|
|
|
|
def fake_request(url, method, data, headers):
|
|
seen_headers.update(headers)
|
|
return 200, {"X-WP-TotalPages": "1"}, b"[]"
|
|
|
|
c = WooClient("admin", "secret", "https://store.example.com", request=fake_request)
|
|
c.get_categories()
|
|
assert "User-Agent" in seen_headers
|
|
assert "SlabUploader" in seen_headers["User-Agent"]
|