SlabUploader/backend/tests/test_per_user_woo_creds.py

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