02-ago: [proyecto] vigilancia - integra middleware Alexa (alexa_notify.js/VoiceMonkey) + .env con credenciales de Telegram y VoiceMonkey + tests 18/18

vigilancia/test_prueba.py

@@ -142,12 +142,15 @@ def test_webhook_mock():
     real = notificador.requests
     notificador.requests = FakeRequests()
     orig = config.WEBHOOK_ALEXA
+    orig_script = config.ALEXA_NOTIFY_SCRIPT
     config.WEBHOOK_ALEXA = "http://localhost:1880/alexa"
+    config.ALEXA_NOTIFY_SCRIPT = ""
     try:
         disparar_webhook_alexa("PRUEBA")
         check("Webhook disparado cuando configurado", len(llamadas) == 1)
     finally:
         config.WEBHOOK_ALEXA = orig
+        config.ALEXA_NOTIFY_SCRIPT = orig_script
         notificador.requests = real
 
     llamadas.clear()
@@ -172,6 +175,43 @@ def test_yolo():
           res is False)
 
 
+def test_alexa_middleware():
+    print("\n--- RF-08: Middleware alexa_notify.js (mock subprocess) ---")
+    llamadas = {}
+
+    def fake_run(cmd, **kwargs):
+        llamadas["cmd"] = cmd
+        llamadas["env_token"] = kwargs.get("env", {}).get("VOICEMONKEY_TOKEN", "")
+        class R:
+            returncode = 0
+            stderr = ""
+        return R()
+
+    import notificador
+    real_run = notificador.subprocess.run
+    notificador.subprocess.run = fake_run
+    orig_script = config.ALEXA_NOTIFY_SCRIPT
+    orig_token = config.VOICEMONKEY_TOKEN
+    orig_dev = config.VOICEMONKEY_DEVICE
+    orig_node = config.ALEXA_NODE
+    config.ALEXA_NOTIFY_SCRIPT = "alexa_notify.js"
+    config.VOICEMONKEY_TOKEN = "VM_TOKEN_FAKE"
+    config.VOICEMONKEY_DEVICE = "echo-salon"
+    config.ALEXA_NODE = "node"
+    try:
+        disparar_webhook_alexa("PRUEBA movimiento")
+        check("Ejecuta node + script + texto",
+              llamadas.get("cmd") == ["node", "alexa_notify.js", "PRUEBA movimiento"])
+        check("Inyecta VOICEMONKEY_TOKEN al proceso",
+              llamadas.get("env_token") == "VM_TOKEN_FAKE")
+    finally:
+        notificador.subprocess.run = real_run
+        config.ALEXA_NOTIFY_SCRIPT = orig_script
+        config.VOICEMONKEY_TOKEN = orig_token
+        config.VOICEMONKEY_DEVICE = orig_dev
+        config.ALEXA_NODE = orig_node
+
+
 def test_flujo_completo():
     print("\n--- RF-03/RF-04/RF-05: Flujo completo snapshot+cooldown (mock) ---")
     os.makedirs(config.SNAPSHOT_DIR, exist_ok=True)
@@ -202,6 +242,7 @@ if __name__ == "__main__":
     test_franja_horaria()
     test_telegram_mock()
     test_webhook_mock()
+    test_alexa_middleware()
     test_yolo()
     test_flujo_completo()
     print(f"\n=== RESULTADO: {APROBADOS} OK, {FALLIDOS} FALLO ===")