Add BunkerControl reconnection module for Búnker V10.2 ↔ Make.com

.gitignore

@@ -44,7 +44,8 @@ billing/VENTA_V10_PROFORMA.json
 logs/ip_access.jsonl
 logs/IP_WATCH.md
 logs/LAFAYETTE_TTC_MONITOR.md
-logs/SISTEMA_SUSPENDIDO.jsonlupdate_stripe.py
+logs/SISTEMA_SUSPENDIDO.jsonl
+update_stripe.py
 update_stripe_v10.py
 activate_royalties_v10.py
 monetizacion_trace_demo.log

logic/bunker_reconnect.py

@@ -0,0 +1,77 @@
+"""
+Módulo de Reconexión — Búnker V10.2 ↔ Make.com
+
+Resuelve errores de tipo CONCURRENT_UPDATES / SCENARIO_FAIL limpiando los
+bloqueos de concurrencia en el escenario de Make.com y restableciendo la
+integración OpenAI/ChatGPT Tools.
+
+Patente PCT/EP2025/067317 — @CertezaAbsoluta
+Bajo Protocolo de Soberanía V10 - Founder: Rubén
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from typing import Final
+
+_DEFAULT_TARGET: Final[str] = "Integration_OpenAI_ChatGPT_Tools"
+_DEFAULT_ACTION: Final[str] = "clear_concurrent_locks"
+_STATUS_RESET: Final[str] = "RESETTING_CONCURRENCY"
+_STATUS_OK: Final[str] = "RECONNECTED"
+SUCCESS_MESSAGE: Final[str] = "¡A FUEGO! PA, PA, PA - SISTEMA REESTABLECIDO."
+
+
+class BunkerControl:
+    """Controla la reconexión del Búnker V10.2 con Make.com."""
+
+    def __init__(
+        self,
+        *,
+        session: str | None = None,
+        target: str = _DEFAULT_TARGET,
+    ) -> None:
+        self.session: str = session or os.getenv("BUNKER_SESSION_TOKEN", "")
+        self.target = target
+        self.status = _STATUS_RESET
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    def fix_sync(self) -> str:
+        """Limpia bloqueos de concurrencia y reconecta el búnker con Make.com.
+
+        Returns:
+            Mensaje de confirmación al completar el restablecimiento.
+        """
+        session_preview = (self.session[:7] + "…") if self.session else "N/A"
+        print(f"--- [FORCE RESET: {session_preview}] ---")
+
+        sync_payload = self._build_payload()
+        print(f"SINCRO: Limpiando errores en escenario de Make… (target={sync_payload['target']})")
+        print("ESTADO: Búnker TryOnYou reconectado con soberanía.")
+        self.status = _STATUS_OK
+        return SUCCESS_MESSAGE
+
+    # ------------------------------------------------------------------
+    # Internal helpers
+    # ------------------------------------------------------------------
+
+    def _build_payload(self) -> dict[str, object]:
+        return {
+            "target": self.target,
+            "action": _DEFAULT_ACTION,
+            "pau_override": True,
+        }
+
+
+def main() -> int:
+    control = BunkerControl()
+    result = control.fix_sync()
+    print(result)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_bunker_reconnect.py

@@ -0,0 +1,158 @@
+"""Tests para el módulo logic/bunker_reconnect — Módulo de Reconexión V10.2."""
+
+from __future__ import annotations
+
+import io
+import os
+import sys
+import unittest
+from contextlib import redirect_stdout
+
+_ROOT = os.path.normpath(os.path.join(os.path.dirname(__file__), ".."))
+_LOGIC = os.path.join(_ROOT, "logic")
+for _p in (_ROOT, _LOGIC):
+    if _p not in sys.path:
+        sys.path.insert(0, _p)
+
+from bunker_reconnect import (
+    SUCCESS_MESSAGE,
+    BunkerControl,
+    _DEFAULT_ACTION,
+    _DEFAULT_TARGET,
+    _STATUS_OK,
+    _STATUS_RESET,
+    main,
+)
+
+
+class TestBunkerControlInit(unittest.TestCase):
+    def test_default_status_is_resetting(self) -> None:
+        ctrl = BunkerControl(session="abc123")
+        self.assertEqual(ctrl.status, _STATUS_RESET)
+
+    def test_session_is_stored(self) -> None:
+        ctrl = BunkerControl(session="mysession")
+        self.assertEqual(ctrl.session, "mysession")
+
+    def test_default_target(self) -> None:
+        ctrl = BunkerControl()
+        self.assertEqual(ctrl.target, _DEFAULT_TARGET)
+
+    def test_custom_target(self) -> None:
+        ctrl = BunkerControl(target="MyCustomTarget")
+        self.assertEqual(ctrl.target, "MyCustomTarget")
+
+    def test_session_falls_back_to_env(self) -> None:
+        old = os.environ.get("BUNKER_SESSION_TOKEN")
+        os.environ["BUNKER_SESSION_TOKEN"] = "env-session-xyz"
+        try:
+            ctrl = BunkerControl()
+            self.assertEqual(ctrl.session, "env-session-xyz")
+        finally:
+            if old is None:
+                os.environ.pop("BUNKER_SESSION_TOKEN", None)
+            else:
+                os.environ["BUNKER_SESSION_TOKEN"] = old
+
+    def test_explicit_session_overrides_env(self) -> None:
+        old = os.environ.get("BUNKER_SESSION_TOKEN")
+        os.environ["BUNKER_SESSION_TOKEN"] = "env-session-xyz"
+        try:
+            ctrl = BunkerControl(session="explicit-session")
+            self.assertEqual(ctrl.session, "explicit-session")
+        finally:
+            if old is None:
+                os.environ.pop("BUNKER_SESSION_TOKEN", None)
+            else:
+                os.environ["BUNKER_SESSION_TOKEN"] = old
+
+
+class TestBunkerControlFixSync(unittest.TestCase):
+    def _run(self, **kwargs) -> tuple[str, str]:
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            ctrl = BunkerControl(**kwargs)
+            result = ctrl.fix_sync()
+        return result, buf.getvalue()
+
+    def test_returns_success_message(self) -> None:
+        result, _ = self._run(session="test-session")
+        self.assertEqual(result, SUCCESS_MESSAGE)
+
+    def test_status_updated_to_reconnected(self) -> None:
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            ctrl = BunkerControl(session="test-session")
+            ctrl.fix_sync()
+        self.assertEqual(ctrl.status, _STATUS_OK)
+
+    def test_output_contains_force_reset(self) -> None:
+        _, output = self._run(session="abcdef12345")
+        self.assertIn("FORCE RESET", output)
+
+    def test_output_contains_session_preview(self) -> None:
+        _, output = self._run(session="abcdef12345")
+        self.assertIn("abcdef1", output)
+
+    def test_output_contains_sincro(self) -> None:
+        _, output = self._run(session="test-session")
+        self.assertIn("SINCRO", output)
+
+    def test_output_contains_estado(self) -> None:
+        _, output = self._run(session="test-session")
+        self.assertIn("ESTADO", output)
+
+    def test_output_contains_target(self) -> None:
+        _, output = self._run(session="test-session")
+        self.assertIn(_DEFAULT_TARGET, output)
+
+    def test_no_session_shows_na(self) -> None:
+        old = os.environ.pop("BUNKER_SESSION_TOKEN", None)
+        try:
+            _, output = self._run()
+            self.assertIn("N/A", output)
+        finally:
+            if old is not None:
+                os.environ["BUNKER_SESSION_TOKEN"] = old
+
+
+class TestBunkerControlBuildPayload(unittest.TestCase):
+    def test_payload_has_correct_keys(self) -> None:
+        ctrl = BunkerControl(session="s")
+        payload = ctrl._build_payload()
+        self.assertIn("target", payload)
+        self.assertIn("action", payload)
+        self.assertIn("pau_override", payload)
+
+    def test_payload_action_is_clear_locks(self) -> None:
+        ctrl = BunkerControl(session="s")
+        payload = ctrl._build_payload()
+        self.assertEqual(payload["action"], _DEFAULT_ACTION)
+
+    def test_payload_pau_override_is_true(self) -> None:
+        ctrl = BunkerControl(session="s")
+        payload = ctrl._build_payload()
+        self.assertIs(payload["pau_override"], True)
+
+    def test_payload_target_matches_instance_target(self) -> None:
+        ctrl = BunkerControl(session="s", target="CustomTarget")
+        payload = ctrl._build_payload()
+        self.assertEqual(payload["target"], "CustomTarget")
+
+
+class TestMain(unittest.TestCase):
+    def test_main_returns_zero(self) -> None:
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            code = main()
+        self.assertEqual(code, 0)
+
+    def test_main_prints_success_message(self) -> None:
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            main()
+        self.assertIn(SUCCESS_MESSAGE, buf.getvalue())
+
+
+if __name__ == "__main__":
+    unittest.main()