SlabUploader/backend/tests/test_publish.py

"""S6 publish sequence tests (PUBLISH-5/10/15/25/30/31). Woo transport is mocked."""

import base64
import json
import os
import shutil
import tempfile

import pytest
from fastapi.testclient import TestClient

import backend.db as db_mod


def _full_slab(sku="SLAB-01", **over):
    base = {
        "id": "s1", "sku": sku, "length_in": 114.0, "thickness_in": 1.75, "sqft": 30.0,
        "bdft": 18.4, "width_min_in": 18.0, "width_max_in": 24.3, "width_avg_in": 24.3,
        "species_id": 1, "wood_category_ids": json.dumps([2]), "edge_type_term_id": 1,
        "figure_term_ids": json.dumps([1]), "grade_term_id": 1, "thickness_term_id": 1,
        "moisture_term_id": 1, "fig_tag_ids": json.dumps([]), "feat_tag_ids": json.dumps([]),
        "price": 184.0, "price_source": "recommendation",
        "title": "Figured Black Walnut Live Edge Slab", "short_description": "slab, 9ft 6in",
        "description": "A fine slab.", "content_source": "manual", "status": "ready",
        "client_rev": 1, "server_rev": 1, "deleted_at": None, "created_at": "t", "updated_at": "t",
        "woo_product_id": None,
    }
    base.update(over)
    return base


@pytest.fixture
def env(monkeypatch):
    tmp = tempfile.mkdtemp(prefix="pub_")
    os.environ["SLAB_DB_PATH"] = os.path.join(tmp, "p.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()
    from backend.crypto import encrypt

    conn = db_mod.init_db()
    # taxonomy
    conn.execute("INSERT INTO woo_categories(id,name,slug,parent_id,woo_id) VALUES (1,'Black Walnut','black-walnut',NULL,100)")
    conn.execute("INSERT INTO woo_categories(id,name,slug,parent_id,woo_id) VALUES (2,'Live edge wood','live-edge-wood',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)")
    # creds
    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"),))
    # fresh sync stamp so we skip re-sync
    from datetime import datetime, timezone

    conn.execute("INSERT OR REPLACE INTO settings(key,value,value_enc) VALUES ('taxonomy_last_sync_at',?,NULL)",
                 (datetime.now(timezone.utc).replace(tzinfo=None).isoformat(),))
    conn.commit()
    from backend.app import app

    yield TestClient(app), conn
    db_mod._reset_for_tests()
    shutil.rmtree(tmp, ignore_errors=True)


def _insert_slab(conn, s):
    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 "
        "(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
        (s["id"], s["sku"], s["length_in"], s["thickness_in"], s["sqft"], s["bdft"], s["width_min_in"],
         s["width_max_in"], s["width_avg_in"], s["species_id"], s["wood_category_ids"], s["edge_type_term_id"],
         s["figure_term_ids"], s["grade_term_id"], s["thickness_term_id"], s["moisture_term_id"],
         s["fig_tag_ids"], s["feat_tag_ids"], s["price"], s["price_source"], s["title"], s["short_description"],
         s["description"], s["content_source"], s["status"], 1, 1, "t", "t"),
    )


def _install_mock(monkeypatch, sku_exists=0, product_id=999, media_ids=(42,)):
    from backend.woo import WooClient

    state = {"media_n": 0}

    def fake_request(url, method, data, headers):
        if "/products?sku=" in url:
            body = json.dumps([{"id": 1}] * sku_exists).encode()
            return 200, {}, body
        if url.endswith("/wp-json/wp/v2/media") and method == "POST":
            mid = media_ids[min(state["media_n"], len(media_ids) - 1)]
            state["media_n"] += 1
            return 201, {}, json.dumps({"id": mid}).encode()
        if url.endswith("/wp-json/wc/v3/products") and method == "POST":
            return 201, {}, json.dumps({"id": product_id}).encode()
        return 200, {}, b"[]"

    orig_init = WooClient.__init__

    def patched_init(self, *a, **k):
        orig_init(self, *a, **k)
        self._request = fake_request

    monkeypatch.setattr(WooClient, "__init__", patched_init)


def test_publish_success(env, monkeypatch):
    client, conn = env
    s = _full_slab()
    _insert_slab(conn, s)
    conn.execute("INSERT INTO slab_photos(id,slab_id,seq,is_calibration,processed_blob,created_at) VALUES ('p1','s1',1,1,X'89504E47','t')")
    conn.commit()
    _install_mock(monkeypatch, sku_exists=0, product_id=999, media_ids=(42,))
    r = client.post("/api/v1/slabs/s1/publish")
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["ok"] is True and body["woo_product_id"] == 999 and body["draft"] is False
    row = conn.execute("SELECT status, woo_product_id FROM slabs WHERE id='s1'").fetchone()
    assert row["status"] == "published" and row["woo_product_id"] == 999
    # photos purged
    assert conn.execute("SELECT processed_blob FROM slab_photos WHERE id='p1'").fetchone()["processed_blob"] is None


def test_uat_draft(env, monkeypatch):
    client, conn = env
    s = _full_slab(sku="SLAB-UAT-1")
    _insert_slab(conn, s)
    conn.commit()
    _install_mock(monkeypatch)
    r = client.post("/api/v1/slabs/s1/publish")
    assert r.status_code == 200
    assert r.json()["draft"] is True


def test_not_ready_422(env, monkeypatch):
    client, conn = env
    s = _full_slab(price=None, price_source=None)
    _insert_slab(conn, s)
    conn.commit()
    _install_mock(monkeypatch)
    r = client.post("/api/v1/slabs/s1/publish")
    assert r.status_code == 422
    assert "price" in r.json()["error"]["detail"]["missing"]


def test_duplicate_sku_409(env, monkeypatch):
    client, conn = env
    s = _full_slab(sku="SLAB-DUP")
    _insert_slab(conn, s)
    conn.commit()
    _install_mock(monkeypatch, sku_exists=1)
    r = client.post("/api/v1/slabs/s1/publish")
    assert r.status_code == 409
    assert r.json()["error"]["code"] == "duplicate_sku"


def test_partial_media_failure_502(env, monkeypatch):
    client, conn = env
    s = _full_slab()
    _insert_slab(conn, s)
    conn.execute("INSERT INTO slab_photos(id,slab_id,seq,is_calibration,processed_blob,created_at) VALUES ('p1','s1',1,1,X'89504E47','t')")
    conn.commit()
    from backend.woo import WooClient, WooError

    def fake_request(url, method, data, headers):
        if "/products?sku=" in url:
            return 200, {}, b"[]"
        if url.endswith("/wp-json/wp/v2/media"):
            return 201, {}, json.dumps({"id": 42}).encode()
        if url.endswith("/wp-json/wc/v3/products"):
            raise WooError("product create returned 500")
        return 200, {}, b"[]"

    orig_init = WooClient.__init__

    def patched_init(self, *a, **k):
        orig_init(self, *a, **k)
        self._request = fake_request

    monkeypatch.setattr(WooClient, "__init__", patched_init)
    r = client.post("/api/v1/slabs/s1/publish")
    assert r.status_code == 502
    assert r.json()["error"]["detail"]["partial_media"] is True
    # sync_log written, status failed
    assert conn.execute("SELECT status FROM slabs WHERE id='s1'").fetchone()["status"] == "failed"
    assert conn.execute("SELECT COUNT(*) c FROM sync_log WHERE status='failed'").fetchone()["c"] >= 1