"""Prototype: X-Woo-* headers required on HTTP Woo routes; never clobber Settings.""" import base64 import os import shutil import tempfile import pytest from starlette.datastructures import Headers import backend.db as db_mod from backend.crypto import encrypt, decrypt from backend.woo import ( WooClient, WooDeviceCredsRequired, client_for_request, request_woo_override, settings_woo_creds, HEADER_WOO_USERNAME, HEADER_WOO_APP_PASSWORD, ) @pytest.fixture def db_dir(): tmp = tempfile.mkdtemp(prefix="per_user_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() os.environ.pop("SLAB_WOO_ALLOW_SETTINGS_FALLBACK", None) db_mod._reset_for_tests() conn = db_mod.init_db() conn.execute( "INSERT OR REPLACE INTO settings(key, value, value_enc) VALUES ('woo_wp_username','shop-default',NULL)" ) conn.execute( "INSERT OR REPLACE INTO settings(key, value, value_enc) VALUES ('woo_app_password',NULL,?)", (encrypt("shop-secret"),), ) conn.commit() yield conn db_mod._reset_for_tests() shutil.rmtree(tmp, ignore_errors=True) def _auth_pair(client: WooClient) -> tuple[str, str]: raw = base64.b64decode(client._auth).decode("utf-8") user, password = raw.split(":", 1) return user, password def test_override_headers_produce_distinct_auth_without_clobbering_settings(db_dir): conn = db_dir before = settings_woo_creds(conn) assert before[1] == "shop-default" assert decrypt(before[2]) == "shop-secret" h_a = Headers({HEADER_WOO_USERNAME: "alice", HEADER_WOO_APP_PASSWORD: "alice-app-pw"}) h_b = Headers({HEADER_WOO_USERNAME: "bob", HEADER_WOO_APP_PASSWORD: "bob-app-pw"}) client_a = client_for_request(conn, h_a) client_b = client_for_request(conn, h_b) assert _auth_pair(client_a) == ("alice", "alice-app-pw") assert _auth_pair(client_b) == ("bob", "bob-app-pw") assert client_a._auth != client_b._auth after = settings_woo_creds(conn) assert after == before assert decrypt(after[2]) == "shop-secret" row_u = conn.execute("SELECT value FROM settings WHERE key='woo_wp_username'").fetchone()[0] row_p = conn.execute("SELECT value_enc FROM settings WHERE key='woo_app_password'").fetchone()[0] assert row_u == "shop-default" assert decrypt(row_p) == "shop-secret" def test_missing_headers_raise_device_creds_required(db_dir): conn = db_dir with pytest.raises(WooDeviceCredsRequired): client_for_request(conn, Headers({})) with pytest.raises(WooDeviceCredsRequired): client_for_request(conn, Headers({HEADER_WOO_USERNAME: "alice"})) assert request_woo_override(Headers({HEADER_WOO_USERNAME: "alice"})) is None def test_env_escape_hatch_allows_settings_fallback(db_dir, monkeypatch): conn = db_dir monkeypatch.setenv("SLAB_WOO_ALLOW_SETTINGS_FALLBACK", "1") client = client_for_request(conn, Headers({})) assert _auth_pair(client) == ("shop-default", "shop-secret") def test_test_woo_endpoint_requires_headers(db_dir, monkeypatch): """API: POST /settings/test-woo with X-Woo-* builds client; missing → 401.""" from fastapi.testclient import TestClient from backend import app as app_mod seen: list[str] = [] def fake_sync(client, conn): seen.append(_auth_pair(client)[0]) return {"taxonomy_anchor": "x", "taxonomy_last_sync_at": "2026-01-01T00:00:00"} monkeypatch.setattr(app_mod.taxonomy_mod, "sync", fake_sync) tc = TestClient(app_mod.app) r1 = tc.post( "/api/v1/settings/test-woo", headers={HEADER_WOO_USERNAME: "alice", HEADER_WOO_APP_PASSWORD: "pw-a"}, ) r2 = tc.post( "/api/v1/settings/test-woo", headers={HEADER_WOO_USERNAME: "bob", HEADER_WOO_APP_PASSWORD: "pw-b"}, ) r3 = tc.post("/api/v1/settings/test-woo") assert r1.status_code == 200 and r1.json()["ok"] is True assert r2.status_code == 200 and r2.json()["ok"] is True assert r3.status_code == 401 assert r3.json()["error"]["code"] == "woo_device_creds_required" assert seen == ["alice", "bob"] assert settings_woo_creds(db_dir)[1] == "shop-default" def test_taxonomy_sync_requires_headers(db_dir, monkeypatch): from fastapi.testclient import TestClient from backend import app as app_mod monkeypatch.setattr( app_mod.taxonomy_mod, "sync", lambda client, conn: { "taxonomy_anchor": "x", "taxonomy_last_sync_at": "2026-01-01T00:00:00", }, ) tc = TestClient(app_mod.app) missing = tc.post("/api/v1/admin/taxonomy/sync") assert missing.status_code == 401 assert missing.json()["error"]["code"] == "woo_device_creds_required" ok = tc.post( "/api/v1/admin/taxonomy/sync", headers={HEADER_WOO_USERNAME: "alice", HEADER_WOO_APP_PASSWORD: "pw"}, ) assert ok.status_code == 200 and ok.json()["ok"] is True def test_publish_requires_headers(db_dir): """POST publish without X-Woo-* → 401 woo_device_creds_required.""" from fastapi.testclient import TestClient from backend import app as app_mod import json conn = db_dir conn.execute( "INSERT INTO woo_categories(id,name,slug,parent_id,woo_id) VALUES (1,'Walnut','walnut',NULL,100)" ) conn.execute( "INSERT INTO woo_categories(id,name,slug,parent_id,woo_id) VALUES (2,'Live edge','le',NULL,101)" ) conn.execute( "INSERT INTO woo_attributes(id,name,slug,woo_id) VALUES (8,'Edge Type','edge-type',200)" ) conn.execute( "INSERT INTO woo_attribute_terms(id,attribute_id,name,slug,woo_id) VALUES (1,8,'Flat','flat',300)" ) conn.execute( """INSERT INTO slabs(id,sku,length_in,thickness_in,sqft,bdft,width_min_in,width_max_in,width_avg_in, species_id,wood_category_ids,edge_type_term_id,figure_term_ids,grade_term_id,thickness_term_id, moisture_term_id,fig_tag_ids,feat_tag_ids,price,price_source,title,short_description,description, content_source,status,server_rev,client_rev,created_at,updated_at) VALUES ('s1','SLAB-01',114,1.75,30,18.4,18,24.3,24.3,1,'[2]',1,'[1]',1,1,1,'[]','[]',184, 'recommendation','T','S','D','manual','ready',1,1,'t','t')""" ) conn.commit() tc = TestClient(app_mod.app) r = tc.post("/api/v1/slabs/s1/publish") assert r.status_code == 401, r.text assert r.json()["error"]["code"] == "woo_device_creds_required"