From 714a25bfa527f074d8971d2eb7bdd8bc5de6a8cf Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 7 Apr 2026 13:39:52 +0000 Subject: [PATCH 1/2] Initial plan From fd159d1ad013ae7d998c28ae954c5c570520718b Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 7 Apr 2026 13:49:29 +0000 Subject: [PATCH 2/2] Add ShopifySovereignBridge class and tests for Robert biometric cart sync Agent-Logs-Url: https://github.com/Tryonme-com/tryonyou-app/sessions/31d8b757-26c2-4010-8bba-7f37ca6db89b Co-authored-by: LVT-ENG <214667862+LVT-ENG@users.noreply.github.com> --- api/shopify_bridge.py | 97 ++++++++++++++++ tests/test_shopify_bridge.py | 211 +++++++++++++++++++++++++++++++++++ 2 files changed, 308 insertions(+) create mode 100644 tests/test_shopify_bridge.py diff --git a/api/shopify_bridge.py b/api/shopify_bridge.py index 6018567a..de76f505 100644 --- a/api/shopify_bridge.py +++ b/api/shopify_bridge.py @@ -21,6 +21,9 @@ import urllib.error import urllib.parse import urllib.request +from typing import Any + +import requests as _requests SIREN_SELL = "943 610 196" PATENTE = "PCT/EP2025/067317" @@ -130,3 +133,97 @@ def resolve_shopify_checkout_url(lead_id: int, fabric_sensation: str) -> str | N if inv: return inv return build_shopify_perfect_selection_url(lead_id, fabric_sensation) + + +class ShopifySovereignBridge: + """ + Puente soberano Shopify — inyecta métricas biométricas de Robert en el carrito + y sincroniza el stock del Armario Solidario contra el Admin API de Shopify. + + Patente: PCT/EP2025/067317 + """ + + def __init__(self, api_key: str, shop_url: str) -> None: + host = shop_url.replace("https://", "").replace("http://", "").split("/")[0] + ver = os.environ.get("SHOPIFY_ADMIN_API_VERSION", "2026-04").strip() or "2026-04" + self.api_base = f"https://{host}/admin/api/{ver}" + self.headers: dict[str, str] = { + "X-Shopify-Access-Token": api_key, + "Content-Type": "application/json", + } + + def sync_robert_to_shopify( + self, look_id: int, engine_metrics: dict[str, Any] + ) -> str | None: + """ + Inyecta las métricas de Robert (talla recomendada por biometría) + en el carrito de Shopify para evitar devoluciones. + + Returns the draft order invoice_url on success, or None on failure. + """ + fit_score = engine_metrics.get("fitScore", 0) + payload: dict[str, Any] = { + "draft_order": { + "line_items": [ + { + "variant_id": look_id, + "quantity": 1, + "properties": [ + { + "name": "Robert_Fit_Score", + "value": f"{fit_score}%", + }, + { + "name": "Biometric_Validation", + "value": "Sovereign_V10", + }, + ], + } + ], + "note": "Venta realizada a través de Espejo Digital TryOnYou.", + "tags": f"TryOnYou,Robert,PCT_EP2025_067317,FitScore_{fit_score}", + } + } + try: + response = _requests.post( + f"{self.api_base}/draft_orders.json", + json=payload, + headers=self.headers, + timeout=15, + ) + except _requests.RequestException: + return None + if response.status_code not in (200, 201): + return None + inv = response.json().get("draft_order", {}).get("invoice_url") + if isinstance(inv, str) and inv.startswith("http"): + return inv + return f"Checkout de Shopify listo para Look {look_id}" + + def update_inventory_physics(self, fabric_key: str, stock_change: int) -> bool: + """ + Actualiza stock cuando el Armario Solidario retira una prenda. + + Calls Shopify Inventory API (inventory_levels/adjust) for the given + inventory item (fabric_key = inventory_item_id) at the configured + SHOPIFY_LOCATION_ID location. Returns True on success, False on failure. + """ + print(f"Sincronizando stock en Shopify para {fabric_key}: {stock_change}") + location_id_raw = os.environ.get("SHOPIFY_LOCATION_ID", "").strip() + if not location_id_raw.isdigit(): + return False + payload: dict[str, Any] = { + "location_id": int(location_id_raw), + "inventory_item_id": fabric_key, + "available_adjustment": stock_change, + } + try: + response = _requests.post( + f"{self.api_base}/inventory_levels/adjust.json", + json=payload, + headers=self.headers, + timeout=15, + ) + except _requests.RequestException: + return False + return response.status_code in (200, 201) diff --git a/tests/test_shopify_bridge.py b/tests/test_shopify_bridge.py new file mode 100644 index 00000000..5fc48afb --- /dev/null +++ b/tests/test_shopify_bridge.py @@ -0,0 +1,211 @@ +"""Tests for ShopifySovereignBridge — Protocolo Robert / Biometría V10.""" + +from __future__ import annotations + +import os +import sys +import unittest +from unittest.mock import MagicMock, patch + +_ROOT = os.path.normpath(os.path.join(os.path.dirname(__file__), "..")) +_API = os.path.join(_ROOT, "api") +for _p in (_ROOT, _API): + if _p not in sys.path: + sys.path.insert(0, _p) + +from shopify_bridge import ShopifySovereignBridge # noqa: E402 + + +class TestShopifySovereignBridgeInit(unittest.TestCase): + def test_api_base_built_from_shop_url(self) -> None: + bridge = ShopifySovereignBridge("tok", "mystore.myshopify.com") + self.assertIn("mystore.myshopify.com", bridge.api_base) + self.assertTrue(bridge.api_base.startswith("https://")) + + def test_api_base_strips_https_prefix(self) -> None: + bridge = ShopifySovereignBridge("tok", "https://mystore.myshopify.com") + self.assertNotIn("https://https://", bridge.api_base) + self.assertIn("mystore.myshopify.com", bridge.api_base) + + def test_headers_contain_access_token(self) -> None: + bridge = ShopifySovereignBridge("secret-key", "mystore.myshopify.com") + self.assertEqual(bridge.headers["X-Shopify-Access-Token"], "secret-key") + + def test_headers_content_type_json(self) -> None: + bridge = ShopifySovereignBridge("tok", "mystore.myshopify.com") + self.assertEqual(bridge.headers["Content-Type"], "application/json") + + def test_api_base_contains_admin_api(self) -> None: + bridge = ShopifySovereignBridge("tok", "mystore.myshopify.com") + self.assertIn("/admin/api/", bridge.api_base) + + +class TestSyncRobertToShopify(unittest.TestCase): + def _make_bridge(self) -> ShopifySovereignBridge: + return ShopifySovereignBridge("tok", "mystore.myshopify.com") + + def test_returns_invoice_url_on_success(self) -> None: + bridge = self._make_bridge() + mock_resp = MagicMock() + mock_resp.status_code = 201 + mock_resp.json.return_value = { + "draft_order": {"invoice_url": "https://mystore.myshopify.com/invoices/abc"} + } + with patch("shopify_bridge._requests.post", return_value=mock_resp): + result = bridge.sync_robert_to_shopify(42, {"fitScore": 95}) + self.assertEqual(result, "https://mystore.myshopify.com/invoices/abc") + + def test_returns_fallback_string_when_no_invoice_url(self) -> None: + bridge = self._make_bridge() + mock_resp = MagicMock() + mock_resp.status_code = 201 + mock_resp.json.return_value = {"draft_order": {}} + with patch("shopify_bridge._requests.post", return_value=mock_resp): + result = bridge.sync_robert_to_shopify(7, {"fitScore": 88}) + self.assertIsNotNone(result) + self.assertIn("7", str(result)) + + def test_returns_none_on_http_error(self) -> None: + bridge = self._make_bridge() + mock_resp = MagicMock() + mock_resp.status_code = 401 + mock_resp.json.return_value = {} + with patch("shopify_bridge._requests.post", return_value=mock_resp): + result = bridge.sync_robert_to_shopify(1, {"fitScore": 80}) + self.assertIsNone(result) + + def test_returns_none_on_request_exception(self) -> None: + import requests as real_requests + + bridge = self._make_bridge() + with patch( + "shopify_bridge._requests.post", + side_effect=real_requests.RequestException("timeout"), + ): + result = bridge.sync_robert_to_shopify(1, {"fitScore": 80}) + self.assertIsNone(result) + + def test_payload_includes_fit_score(self) -> None: + bridge = self._make_bridge() + mock_resp = MagicMock() + mock_resp.status_code = 201 + mock_resp.json.return_value = {"draft_order": {}} + with patch("shopify_bridge._requests.post", return_value=mock_resp) as mock_post: + bridge.sync_robert_to_shopify(99, {"fitScore": 77}) + _, kwargs = mock_post.call_args + line = kwargs["json"]["draft_order"]["line_items"][0] + props = {p["name"]: p["value"] for p in line["properties"]} + self.assertEqual(props["Robert_Fit_Score"], "77%") + self.assertEqual(props["Biometric_Validation"], "Sovereign_V10") + + def test_payload_includes_look_id_as_variant(self) -> None: + bridge = self._make_bridge() + mock_resp = MagicMock() + mock_resp.status_code = 201 + mock_resp.json.return_value = {"draft_order": {}} + with patch("shopify_bridge._requests.post", return_value=mock_resp) as mock_post: + bridge.sync_robert_to_shopify(55, {"fitScore": 90}) + _, kwargs = mock_post.call_args + line = kwargs["json"]["draft_order"]["line_items"][0] + self.assertEqual(line["variant_id"], 55) + self.assertEqual(line["quantity"], 1) + + def test_note_mentions_tryonyou(self) -> None: + bridge = self._make_bridge() + mock_resp = MagicMock() + mock_resp.status_code = 201 + mock_resp.json.return_value = {"draft_order": {}} + with patch("shopify_bridge._requests.post", return_value=mock_resp) as mock_post: + bridge.sync_robert_to_shopify(1, {"fitScore": 85}) + _, kwargs = mock_post.call_args + note = kwargs["json"]["draft_order"]["note"] + self.assertIn("TryOnYou", note) + + +class TestUpdateInventoryPhysics(unittest.TestCase): + def _make_bridge(self) -> ShopifySovereignBridge: + return ShopifySovereignBridge("tok", "mystore.myshopify.com") + + def _set_location(self, loc: str = "98765") -> None: + os.environ["SHOPIFY_LOCATION_ID"] = loc + + def tearDown(self) -> None: + os.environ.pop("SHOPIFY_LOCATION_ID", None) + + def test_returns_true_on_success(self) -> None: + self._set_location() + bridge = self._make_bridge() + mock_resp = MagicMock() + mock_resp.status_code = 200 + with patch("shopify_bridge._requests.post", return_value=mock_resp): + result = bridge.update_inventory_physics("LOC-123", -1) + self.assertTrue(result) + + def test_returns_false_without_location_id_env(self) -> None: + os.environ.pop("SHOPIFY_LOCATION_ID", None) + bridge = self._make_bridge() + mock_resp = MagicMock() + mock_resp.status_code = 200 + with patch("shopify_bridge._requests.post", return_value=mock_resp): + result = bridge.update_inventory_physics("LOC-123", -1) + self.assertFalse(result) + + def test_returns_false_on_http_error(self) -> None: + self._set_location() + bridge = self._make_bridge() + mock_resp = MagicMock() + mock_resp.status_code = 422 + with patch("shopify_bridge._requests.post", return_value=mock_resp): + result = bridge.update_inventory_physics("LOC-123", -1) + self.assertFalse(result) + + def test_returns_false_on_request_exception(self) -> None: + import requests as real_requests + + self._set_location() + bridge = self._make_bridge() + with patch( + "shopify_bridge._requests.post", + side_effect=real_requests.RequestException("conn error"), + ): + result = bridge.update_inventory_physics("LOC-999", -2) + self.assertFalse(result) + + def test_prints_sync_message(self) -> None: + self._set_location() + bridge = self._make_bridge() + mock_resp = MagicMock() + mock_resp.status_code = 200 + with patch("shopify_bridge._requests.post", return_value=mock_resp): + with patch("builtins.print") as mock_print: + bridge.update_inventory_physics("FABRIC-7", -3) + mock_print.assert_called_once() + args = mock_print.call_args[0][0] + self.assertIn("FABRIC-7", args) + self.assertIn("-3", args) + + def test_calls_inventory_adjust_endpoint(self) -> None: + self._set_location() + bridge = self._make_bridge() + mock_resp = MagicMock() + mock_resp.status_code = 200 + with patch("shopify_bridge._requests.post", return_value=mock_resp) as mock_post: + bridge.update_inventory_physics("LOC-42", 5) + url = mock_post.call_args[0][0] + self.assertIn("inventory_levels/adjust", url) + + def test_payload_uses_fabric_key_as_inventory_item_id(self) -> None: + self._set_location("12345") + bridge = self._make_bridge() + mock_resp = MagicMock() + mock_resp.status_code = 200 + with patch("shopify_bridge._requests.post", return_value=mock_resp) as mock_post: + bridge.update_inventory_physics("ITEM-99", -2) + _, kwargs = mock_post.call_args + self.assertEqual(kwargs["json"]["inventory_item_id"], "ITEM-99") + self.assertEqual(kwargs["json"]["location_id"], 12345) + self.assertEqual(kwargs["json"]["available_adjustment"], -2) + + +if __name__ == "__main__": + unittest.main()