citas-getter/test_book_final.py

#!/usr/bin/env python3
"""
Unit + E2E tests for /tmp/book_final.py — locks in current functionality,
prevents regressions. Run: python3 /tmp/test_book_final.py

Covers:
  - motivo cleanliness (no timestamp, no IDs, exact value)
  - API parameter order (15 params, matches frontend services.js)
  - Parameter values (servicio=S, cola=I, tipo=2, impuestos=N, procedencia=NULO)
  - Headers (6 browser-like headers)
  - log() function (timestamp prefix, flush)
  - make_request() reconnection logic (mock conn)
  - build_booking_body / build_check_body structure
  - cancel_cita() is NEVER called in execution path
  - Phase timing constants (TARGET_TS, preconnect_ts = TARGET_TS - 5)
  - Sequential booking (no parallel requests)
  - Exit codes (0 on success, 1 on failure)
  - E2E: live API getHorasCitas returns 9 slots for cola I / fecha 19-10-2026
  - E2E: live API insertarCitaPrevia pre-midnight returns "Fecha posterior"
"""

import unittest
import json
import sys
import os
import re
from unittest.mock import patch, MagicMock, mock_open
from datetime import datetime

# ── Import the script as a module ───────────────────────────
SCRIPT_PATH = "/tmp/book_final.py"

# Read & exec only the importable parts (functions + constants, NOT the phase execution)
with open(SCRIPT_PATH, "r") as f:
    _source = f.read()

# Split at the "# ── INITIAL LOG" marker to get just definitions
_parts = _source.split("# ── INITIAL LOG")
_defs = _parts[0]

# Execute the definitions in a namespace
_ns = {}
exec(_defs, _ns)

# Extract constants/functions from the namespace
HOST            = _ns["HOST"]
PORT            = _ns["PORT"]
API             = _ns["API"]
ENDPOINT        = _ns["ENDPOINT"]
CT              = _ns["CT"]
SERVICIO        = _ns["SERVICIO"]
COLA            = _ns["COLA"]
FECHA           = _ns["FECHA"]
DOC             = _ns["DOC"]
NOMBRE          = _ns["NOMBRE"]
APE1            = _ns["APE1"]
APE2            = _ns["APE2"]
TIPO            = _ns["TIPO"]
TEL             = _ns["TEL"]
EMAIL           = _ns["EMAIL"]
COMENTARIO      = _ns["COMENTARIO"]
IMPUESTOS       = _ns["IMPUESTOS"]
IP              = _ns["IP"]
ALL_HORAS       = _ns["ALL_HORAS"]
PRIORITY        = _ns["PRIORITY"]
HEADERS         = _ns["HEADERS"]
TARGET_TS       = _ns["TARGET_TS"]
build_booking_body = _ns["build_booking_body"]
build_check_body    = _ns["build_check_body"]
log                 = _ns["log"]
bench               = _ns["bench"]
BENCH               = _ns["BENCH"]


class TestMotivoCleanliness(unittest.TestCase):
    """The motivo field must be clean — no timestamps, no tracking IDs."""

    def test_motivo_is_natural_spanish(self):
        """motivo must equal exactly 'Solicitud de Alta en el Padrón'."""
        self.assertEqual(COMENTARIO, "Solicitud de Alta en el Padrón")

    def test_motivo_has_no_timestamp_prefix(self):
        """motivo must not start with HH:MM:SS or any time pattern."""
        # Check no HH:MM at the start
        self.assertNotRegex(COMENTARIO, r"^\d{2}:\d{2}")
        # Check no timestamp anywhere
        self.assertNotRegex(COMENTARIO, r"\d{2}:\d{2}:\d{2}")

    def test_motivo_has_no_uuid_or_tracking_id(self):
        """motivo must not contain UUIDs or tracking-like strings."""
        # No UUID pattern
        self.assertNotRegex(COMENTARIO, r"[0-9a-f]{8}-[0-9a-f]{4}")
        # No hex strings of length >= 8
        self.assertNotRegex(COMENTARIO, r"\b[0-9a-f]{8,}\b")

    def test_motivo_in_booking_body(self):
        """The motivo parameter in build_booking_body must be the clean value."""
        body = json.loads(build_booking_body("16:00"))
        params = body["parameters"]
        motivo_param = [p for p in params if "motivo" in p][0]
        self.assertEqual(motivo_param["motivo"], COMENTARIO)
        self.assertEqual(motivo_param["motivo"], "Solicitud de Alta en el Padrón")

    def test_motivo_not_prefixed_with_ts(self):
        """Verify the old bug (f'{ts} {COMENTARIO}') is gone."""
        body_text = build_booking_body("16:00")
        # Decode to check the actual value (json.dumps escapes Unicode ó → \u00f3)
        decoded = json.loads(body_text)
        params = {k: v for p in decoded["parameters"] for k, v in p.items()}
        self.assertEqual(params["motivo"], "Solicitud de Alta en el Padrón")
        # Raw body should NOT contain a timestamp before the motivo
        self.assertNotRegex(body_text, r'"motivo":"\d{2}:\d{2}:\d{2}')


class TestParameterOrder(unittest.TestCase):
    """Parameters must be in exact order matching services.js insertarCitaPrevia."""

    EXPECTED_ORDER = [
        "servicio", "cola", "fecha", "hora", "documento",
        "ip", "nombre", "apellido1", "apellido2",
        "tipo", "telefono", "email", "impuestos", "motivo", "procedencia"
    ]

    def test_booking_body_has_15_params(self):
        """insertarCitaPrevia must have exactly 15 parameters."""
        body = json.loads(build_booking_body("16:00"))
        self.assertEqual(len(body["parameters"]), 15)

    def test_parameter_order_matches_frontend(self):
        """Parameter order must match services.js lines 105-120 exactly."""
        body = json.loads(build_booking_body("16:00"))
        param_names = [list(p.keys())[0] for p in body["parameters"]]
        self.assertEqual(param_names, self.EXPECTED_ORDER)

    def test_parameter_values(self):
        """All parameter values must match the user's data."""
        body = json.loads(build_booking_body("16:00"))
        params = {k: v for p in body["parameters"] for k, v in p.items()}
        self.assertEqual(params["servicio"], "S")
        self.assertEqual(params["cola"], "I")
        self.assertEqual(params["fecha"], "19-10-2026")
        self.assertEqual(params["hora"], "16:00")
        self.assertEqual(params["documento"], "X2491454W")
        self.assertEqual(params["tipo"], "2")  # NIE
        self.assertEqual(params["nombre"], "Humberto")
        self.assertEqual(params["apellido1"], "Normey")
        self.assertEqual(params["apellido2"], "Gomez")
        self.assertEqual(params["telefono"], "687915599")
        self.assertEqual(params["email"], "hj@normey.net")
        self.assertEqual(params["impuestos"], "N")
        self.assertEqual(params["procedencia"], "NULO")
        self.assertEqual(params["ip"], "0.0.0.0")

    def test_check_body_has_3_params(self):
        """getHorasCitas must have exactly 3 parameters."""
        body = json.loads(build_check_body())
        self.assertEqual(len(body["parameters"]), 3)
        param_names = [list(p.keys())[0] for p in body["parameters"]]
        self.assertEqual(param_names, ["servicio", "cola", "fecha"])


class TestHeaders(unittest.TestCase):
    """Browser-like headers must be present (not just Content-Type)."""

    REQUIRED_HEADERS = [
        "Content-Type", "Accept", "X-Requested-With",
        "Origin", "Referer", "User-Agent"
    ]

    def test_all_browser_headers_present(self):
        """All 6 browser-like headers must be present."""
        for h in self.REQUIRED_HEADERS:
            self.assertIn(h, HEADERS, f"Missing header: {h}")

    def test_user_agent_is_browser_like(self):
        """User-Agent should look like Chrome on Windows."""
        ua = HEADERS["User-Agent"]
        self.assertIn("Mozilla/5.0", ua)
        self.assertIn("Chrome", ua)
        self.assertIn("Windows", ua)

    def test_x_requested_with(self):
        """X-Requested-With should be XMLHttpRequest (AJAX)."""
        self.assertEqual(HEADERS["X-Requested-With"], "XMLHttpRequest")

    def test_origin_matches_frontend(self):
        """Origin should be the citaprevia frontend domain."""
        self.assertEqual(HEADERS["Origin"], "https://citaprevia.alicante.es")

    def test_content_type_is_adf(self):
        """Content-Type must be the Oracle ADF format (server rejects other types with 415)."""
        self.assertEqual(HEADERS["Content-Type"], "application/vnd.oracle.adf.action+json")


class TestTimingConstants(unittest.TestCase):
    """Timing constants must be correct."""

    def test_target_ts_is_oct4_midnight_cest(self):
        """TARGET_TS must be Oct 4, 2026 00:00:00 CEST."""
        target_dt = datetime.fromtimestamp(TARGET_TS)
        self.assertEqual(target_dt.year, 2026)
        self.assertEqual(target_dt.month, 10)
        self.assertEqual(target_dt.day, 4)
        self.assertEqual(target_dt.hour, 0)
        self.assertEqual(target_dt.minute, 0)
        self.assertEqual(target_dt.second, 0)

    def test_preconnect_is_3s_before_midnight(self):
        """Pre-connect phase should be 3 seconds before midnight (T-3).

        Changed from T-5 to T-3 because the server closes idle TLS connections
        after ~5 seconds — a T-5 preconnect risks the connection being dropped
        right at midnight, the critical moment.
        """
        with open(SCRIPT_PATH) as f:
            source = f.read()
        self.assertIn("TARGET_TS - 3", source)
        self.assertNotIn("TARGET_TS - 5", source)

    def test_priority_starts_with_contrarian_slot(self):
        """Priority must start with a post-15h middle slot (contrarian vs bots).

        Bots target 16:00 (preferred) and 14:30 (earliest), leaving
        15:15-15:45 under-contested during the midnight rush.
        """
        self.assertEqual(PRIORITY[0], "15:15")

    def test_priority_contrarian_ordering(self):
        """Post-15h slots must come before pre-15h slots in priority.

        Strategy: most bots target 16:00 and 14:30, so we try
        15:15-15:45 + 16:15-16:30 first, then 16:00, then 14:30-15:00.
        """
        # First 5 slots must be post-15h (15:15 onwards)
        for h in PRIORITY[:5]:
            hr = int(h.split(":")[0])
            self.assertGreaterEqual(hr, 15, f"Slot {h} should be post-15h, but got {hr}:{h}")

        # Pre-15h slots (14:30-15:00) must be at the end
        pre_15 = [h for h in ALL_HORAS if int(h.split(":")[0]) < 15]
        for h in pre_15:
            self.assertGreater(PRIORITY.index(h), PRIORITY.index("16:00"),
                f"Pre-15h slot {h} should come AFTER 16:00 in priority")

    def test_all_9_slots_present(self):
        """All 9 confirmed slots must be in the priority list."""
        for h in ALL_HORAS:
            self.assertIn(h, PRIORITY)


class TestLogFunction(unittest.TestCase):
    """log() must produce timestamped output with flush."""

    def test_log_outputs_timestamp_prefix(self):
        """log() output should have a [HH:MM:SS.mmm] prefix."""
        import io
        from contextlib import redirect_stdout
        f = io.StringIO()
        with redirect_stdout(f):
            log("test message")
        output = f.getvalue()
        # Should have a timestamp prefix like [12:34:56.789]
        self.assertRegex(output, r"\[\d{2}:\d{2}:\d{2}\.\d{3}\]")
        self.assertIn("test message", output)

    def test_log_flushes(self):
        """log() should use flush=True (we check via source)."""
        with open(SCRIPT_PATH) as f:
            source = f.read()
        # Verify the log function definition has flush=True
        log_def = source[source.index("def log("):source.index("def log(")+200]
        self.assertIn("flush=True", log_def)


class TestMakeRequestReconnection(unittest.TestCase):
    """make_request() must handle connection drops."""

    def test_make_request_uses_global_conn(self):
        """make_request must reference 'global conn'."""
        with open(SCRIPT_PATH) as f:
            source = f.read()
        # Find the make_request function
        idx = source.index("def make_request(")
        func_body = source[idx:idx+500]
        self.assertIn("global conn", func_body)

    def test_make_request_catches_remote_disconnected(self):
        """make_request must catch RemoteDisconnected and other connection errors."""
        with open(SCRIPT_PATH) as f:
            source = f.read()
        self.assertIn("RemoteDisconnected", source)
        self.assertIn("ConnectionResetError", source)
        self.assertIn("BrokenPipeError", source)

    def test_make_request_reconnects_after_drop(self):
        """After a connection drop, make_request should reconnect and retry."""
        # We can't easily test the actual reconnection (needs a real conn),
        # but we can verify the logic exists in the source
        with open(SCRIPT_PATH) as f:
            source = f.read()
        self.assertIn("reconnecting", source.lower())
        self.assertIn("Re-connected", source)


class TestBookingLoopSafety(unittest.TestCase):
    """The booking loop must prevent duplicates and handle errors."""

    def test_loop_breaks_on_success(self):
        """The loop must break immediately on 'Cita insertada'."""
        with open(SCRIPT_PATH) as f:
            source = f.read()
        # Find the booking loop section
        loop_start = source.index("for hora in PRIORITY:")
        loop_section = source[loop_start:loop_start+5000]
        # After success check, there must be a break
        success_idx = loop_section.index("Cita insertada")
        # Look in the next 400 chars (past all the log lines) for 'break'
        success_section = loop_section[success_idx:success_idx+700]
        self.assertIn("break", success_section)

    def test_loop_breaks_on_duplicate(self):
        """The loop must break on duplicate detection."""
        with open(SCRIPT_PATH) as f:
            source = f.read()
        loop_start = source.index("for hora in PRIORITY:")
        loop_section = source[loop_start:loop_start+5000]
        dup_section = loop_section[loop_section.index("ya existente"):loop_section.index("ya existente")+300]
        self.assertIn("break", dup_section)

    def test_loop_does_not_parallelize(self):
        """No threading, async, or parallel execution."""
        with open(SCRIPT_PATH) as f:
            source = f.read()
        self.assertNotIn("threading", source)
        self.assertNotIn("asyncio", source)
        self.assertNotIn("concurrent", source)
        self.assertNotIn("ThreadPool", source)
        self.assertNotIn("ProcessPool", source)
        self.assertNotIn("multiprocessing", source)

    def test_skew_retry_logic_exists(self):
        """The 5×43ms clock-skew retry must exist (reduced from 200ms for faster recovery)."""
        with open(SCRIPT_PATH) as f:
            source = f.read()
        self.assertIn("range(5)", source)
        self.assertIn("0.043", source)
        self.assertIn("skew_retries_exhausted", source)

    def test_exception_branch_exists(self):
        """The EXCEPTION handling branch must exist."""
        with open(SCRIPT_PATH) as f:
            source = f.read()
        self.assertIn('startswith("EXCEPTION:")', source)

    def test_exception_branch_handles_conn_drop(self):
        """The EXCEPTION branch must force-reconnect."""
        with open(SCRIPT_PATH) as f:
            source = f.read()
        self.assertIn("Force-reconnected", source)
        self.assertIn("HTTPSConnection", source)

    def test_exit_codes(self):
        """Must use sys.exit(0) for success and sys.exit(1) for failure."""
        with open(SCRIPT_PATH) as f:
            source = f.read()
        self.assertIn("sys.exit(0)", source)
        self.assertIn("sys.exit(1)", source)


class TestCancelNotCalled(unittest.TestCase):
    """cancel_cita() must be defined but NEVER called in execution path."""

    def test_cancel_cita_defined(self):
        """cancel_cita function must be defined."""
        with open(SCRIPT_PATH) as f:
            source = f.read()
        self.assertIn("def cancel_cita(", source)

    def test_cancel_cita_not_called(self):
        """cancel_cita must NOT be called anywhere (only in log strings)."""
        with open(SCRIPT_PATH) as f:
            source = f.read()
        # Remove the function definition
        no_def = source.replace("def cancel_cita(", "")
        # Remove log message strings that mention cancel_cita()
        # (these are in f-strings, like log(f"...cancel_cita()..."))
        # Count actual calls: cancel_cita( not inside a string
        lines = no_def.split("\n")
        actual_calls = 0
        for line in lines:
            stripped = line.strip()
            # Skip lines that are log messages (f"...cancel_cita()...")
            if stripped.startswith("log(") or stripped.startswith("print(") or stripped.startswith("#"):
                continue
            if "cancel_cita(" in stripped:
                actual_calls += 1
        self.assertEqual(actual_calls, 0, f"cancel_cita() is called {actual_calls} times!")


class TestJsonStructure(unittest.TestCase):
    """JSON bodies must be compact and correctly structured."""

    def test_booking_body_uses_compact_json(self):
        """JSON should use compact separators (no spaces)."""
        body_text = build_booking_body("16:00")
        # Compact JSON has no spaces after , or :
        self.assertNotIn(", ", body_text)
        self.assertNotIn(": ", body_text)

    def test_check_body_uses_compact_json(self):
        """getHorasCitas body should also be compact."""
        body_text = build_check_body()
        self.assertNotIn(", ", body_text)
        self.assertNotIn(": ", body_text)

    def test_booking_body_name_is_insertarCitaPrevia(self):
        """The API method name must be insertarCitaPrevia."""
        body = json.loads(build_booking_body("16:00"))
        self.assertEqual(body["name"], "insertarCitaPrevia")

    def test_check_body_name_is_getHorasCitas(self):
        """The check method name must be getHorasCitas."""
        body = json.loads(build_check_body())
        self.assertEqual(body["name"], "getHorasCitas")


class TestE2ELiveAPI(unittest.TestCase):
    """End-to-end tests against the LIVE backend API."""

    def setUp(self):
        self.api_base = "https://webservices.alicante.es/WsCitaPreviaPublic-RESTWebService-context-root/rest/1"
        self.endpoint = self.api_base + "/ScCitaVo1"

    def test_live_getHorasCitas_returns_9_slots(self):
        """getHorasCitas for cola I / fecha 19-10-2026 should return 9 slots."""
        import http.client, ssl
        ctx = ssl.create_default_context()
        conn = http.client.HTTPSConnection("webservices.alicante.es", 443, context=ctx, timeout=15)
        conn.connect()
        body = json.dumps({"name": "getHorasCitas", "parameters": [
            {"servicio": "S"}, {"cola": "I"}, {"fecha": "19-10-2026"}
        ]}, separators=(",", ":"))
        headers = {
            "Content-Type": "application/vnd.oracle.adf.action+json",
            "Accept": "application/json, text/plain, */*",
        }
        conn.request("POST", self.endpoint, body=body, headers=headers)
        resp = conn.getresponse()
        raw = resp.read().decode()
        dj = json.loads(raw)
        self.assertEqual(resp.status, 200)
        slots = json.loads(dj["result"])
        self.assertEqual(len(slots), 9)
        horas = [s["hora"] for s in slots]
        self.assertIn("16:00", horas)
        self.assertIn("14:30", horas)
        self.assertIn("16:30", horas)
        conn.close()

    def test_live_insertar_returns_date_limit_error_pre_midnight(self):
        """insertarCitaPrevia pre-midnight should return 'Fecha posterior'."""
        import http.client, ssl
        ctx = ssl.create_default_context()
        conn = http.client.HTTPSConnection("webservices.alicante.es", 443, context=ctx, timeout=15)
        conn.connect()
        body = json.dumps({"name": "insertarCitaPrevia", "parameters": [
            {"servicio": "S"}, {"cola": "I"}, {"fecha": "19-10-2026"}, {"hora": "16:00"},
            {"documento": "X2491454W"}, {"ip": "0.0.0.0"}, {"nombre": "Humberto"},
            {"apellido1": "Normey"}, {"apellido2": "Gomez"}, {"tipo": "2"},
            {"telefono": "687915599"}, {"email": "hj@normey.net"},
            {"impuestos": "N"}, {"motivo": "Solicitud de Alta en el Padrón"},
            {"procedencia": "NULO"}
        ]}, separators=(",", ":"))
        headers = {
            "Content-Type": "application/vnd.oracle.adf.action+json",
            "Accept": "application/json, text/plain, */*",
        }
        conn.request("POST", self.endpoint, body=body, headers=headers)
        resp = conn.getresponse()
        raw = resp.read().decode()
        dj = json.loads(raw)
        self.assertEqual(resp.status, 200)
        result = dj.get("result", "")
        self.assertIn("Fecha posterior", result)
        conn.close()


class TestMotivoNoTimestamp(unittest.TestCase):
    """Specifically verify the timestamp-prefix bug is gone."""

    def test_build_booking_body_motivo_has_no_timestamp(self):
        """The OLD bug: motivo was f'{ts} {COMENTARIO}' — verify it's gone."""
        body_text = build_booking_body("16:00")
        # Decode and check the actual value (Unicode ó is escaped in raw JSON)
        decoded = json.loads(body_text)
        params = {k: v for p in decoded["parameters"] for k, v in p.items()}
        self.assertEqual(params["motivo"], "Solicitud de Alta en el Padrón")
        # Raw body should NOT match pattern: "motivo":"HH:MM:SS ...
        self.assertNotRegex(body_text, r'"motivo":"\d{2}:\d{2}:\d{2}')


if __name__ == "__main__":
    print("=" * 60)
    print("  BOOKING ASSISTANT — TEST SUITE v1.0")
    print("  Locking in functionality. Preventing regressions.")
    print("=" * 60)
    print()
    unittest.main(verbosity=2)