"""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_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"]