1126 lines
48 KiB
Python
1126 lines
48 KiB
Python
#!/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"]
|
||
_track_rtt = _ns["_track_rtt"]
|
||
_current_load = _ns["_current_load"]
|
||
_compute_load = _ns["_compute_load"]
|
||
_load_trend = _ns["_load_trend"]
|
||
_rtt_history = _ns["_rtt_history"]
|
||
CITA_TYPES = _ns["CITA_TYPES"]
|
||
DOC_TYPES = _ns["DOC_TYPES"]
|
||
REQUIRED_FIELDS = _ns["REQUIRED_FIELDS"]
|
||
parse_args = _ns["parse_args"]
|
||
print_cita_types = _ns["print_cita_types"]
|
||
validate_config = _ns["validate_config"]
|
||
prompt_missing = _ns["prompt_missing"]
|
||
|
||
|
||
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 1000 chars (past all the log lines) for 'break'
|
||
success_section = loop_section[success_idx:success_idx+2000]
|
||
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+10000]
|
||
# Must use _is_duplicate() (centralised duplicate detection) OR the old inline
|
||
# string check — either way, the loop must break on a duplicate
|
||
if "_is_duplicate" in loop_section:
|
||
dup_idx = loop_section.index("_is_duplicate")
|
||
else:
|
||
dup_idx = loop_section.index("ya existente")
|
||
dup_section = loop_section[dup_idx:dup_idx+500]
|
||
self.assertIn("break", dup_section)
|
||
|
||
def test_duplicate_detection_includes_new_string(self):
|
||
"""DUPLICATE_PATTERNS must include 'Usted ya tiene cita previa'."""
|
||
self.assertIn("Usted ya tiene cita previa", _ns.get("DUPLICATE_PATTERNS", ""))
|
||
|
||
def test_duplicate_detected_flag_stops_outer_loop(self):
|
||
"""A duplicate detected during ANY retry must stop the ENTIRE booking
|
||
process — not just the current slot — by setting duplicate_detected=True
|
||
and checking it in the outer for-loop break condition."""
|
||
with open(SCRIPT_PATH) as f:
|
||
source = f.read()
|
||
loop_start = source.index("for hora in PRIORITY:")
|
||
loop_section = source[loop_start:loop_start+10000]
|
||
# Flag must be initialized
|
||
self.assertIn("duplicate_detected", source)
|
||
# Must be set to True when a duplicate is detected
|
||
self.assertIn("duplicate_detected = True", loop_section)
|
||
# The outer-loop break must reference duplicate_detected
|
||
self.assertIn("if booked_hora or duplicate_detected", source)
|
||
# Duplicate detection must exist in ALL three branches:
|
||
# initial attempt, skew retry loop, and EXCEPTION retry
|
||
self.assertIn("ya existente", source)
|
||
# Verify there are at least 2 occurrences of "ya existente" (initial + retry)
|
||
count = source.count("ya existente")
|
||
self.assertGreaterEqual(count, 2, "duplicate check should exist in multiple branches")
|
||
|
||
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):
|
||
"""Three-phase backoff: aggressive (43ms×1.5^n) → gradual (1.15x) → extended (2s×1.2^n).
|
||
120s budget per slot, only gives up on certified 'Hora no permitida'."""
|
||
with open(SCRIPT_PATH) as f:
|
||
source = f.read()
|
||
self.assertIn("SKEW_RETRY_BASE", source)
|
||
self.assertIn("SKEW_AGGRESSIVE_N", source)
|
||
self.assertIn("SKEW_AGGRESSIVE_GROWTH", source)
|
||
self.assertIn("SKEW_GRADUAL_GROWTH", source)
|
||
self.assertIn("SKEW_EXTENDED_BASE", source)
|
||
self.assertIn("SKEW_EXTENDED_GROWTH", source)
|
||
self.assertIn("SKEW_MAX_BUDGET", source)
|
||
self.assertIn("0.043", source) # base delay (also PREFIRE_OFFSET)
|
||
self.assertIn("skew_retries_exhausted", source) # diagnostics var still exists
|
||
self.assertIn("_skew_delay", source) # delay calculation function
|
||
# Budget must be ≥ 60s (not just 2.5s) — catches post-midnight cancellations
|
||
self.assertIn("120", source)
|
||
|
||
def test_each_slot_gets_own_retry_window(self):
|
||
"""No global skip flag — each priority slot must get its own retry window.
|
||
The server date may roll during a later slot's retry window even if earlier
|
||
slots exhausted theirs, so retries must NOT be gated by a global flag.
|
||
Only 'Hora no permitida' certifies a slot as taken."""
|
||
with open(SCRIPT_PATH) as f:
|
||
source = f.read()
|
||
loop_start = source.index("for hora in PRIORITY:")
|
||
loop_section = source[loop_start:loop_start+8000]
|
||
# Must NOT skip retries based on a global flag
|
||
self.assertNotIn("if skew_retries_exhausted:", loop_section)
|
||
# Must use time-based budget (not fixed count)
|
||
self.assertIn("SKEW_MAX_BUDGET", loop_section)
|
||
# Three-phase backoff: aggressive, gradual, extended
|
||
self.assertIn("SKEW_AGGRESSIVE_N", source)
|
||
self.assertIn("SKEW_EXTENDED_BASE", source)
|
||
# "Hora no permitida" must be the certification signal on retry
|
||
self.assertIn("Hora no permitida", loop_section)
|
||
# "Fecha posterior" must NOT be a give-up signal without retry
|
||
self.assertIn("Fecha posterior", loop_section)
|
||
# Must reference the delay function
|
||
self.assertIn("_skew_delay", source)
|
||
# Must have time budget check (not fixed retry count)
|
||
self.assertIn("SKEW_MAX_BUDGET", loop_section)
|
||
|
||
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 TestLoggingSeverity(unittest.TestCase):
|
||
"""log() must support severity levels for structured output."""
|
||
|
||
def test_log_accepts_level_param(self):
|
||
"""log() must accept an optional level parameter."""
|
||
import io
|
||
from contextlib import redirect_stdout
|
||
f = io.StringIO()
|
||
with redirect_stdout(f):
|
||
log("test message", level="WARN")
|
||
output = f.getvalue()
|
||
self.assertIn("WARN", output)
|
||
self.assertIn("test message", output)
|
||
|
||
def test_log_default_level_is_info(self):
|
||
"""log() with no level should default to INFO."""
|
||
import io
|
||
from contextlib import redirect_stdout
|
||
f = io.StringIO()
|
||
with redirect_stdout(f):
|
||
log("default level")
|
||
output = f.getvalue()
|
||
self.assertIn("INFO", output)
|
||
|
||
def test_log_supports_all_levels(self):
|
||
"""All severity levels must appear in source."""
|
||
with open(SCRIPT_PATH) as f:
|
||
source = f.read()
|
||
for level in ["INFO", "WARN", "ERROR", "SUCCESS", "PHASE"]:
|
||
self.assertIn(f'level="{level}"', source, f"level={level} not used in source")
|
||
|
||
def test_log_has_timestamp_prefix(self):
|
||
"""All log output must have [HH:MM:SS.mmm] prefix."""
|
||
import io
|
||
from contextlib import redirect_stdout
|
||
f = io.StringIO()
|
||
with redirect_stdout(f):
|
||
log("structured test", level="SUCCESS")
|
||
output = f.getvalue()
|
||
self.assertRegex(output, r"\[\d{2}:\d{2}:\d{2}\.\d{3}\]")
|
||
|
||
|
||
class TestBenchCategories(unittest.TestCase):
|
||
"""bench() must support category tracking for filtering."""
|
||
|
||
def test_bench_accepts_category_param(self):
|
||
"""bench() must accept an optional category parameter."""
|
||
with open(SCRIPT_PATH) as f:
|
||
source = f.read()
|
||
bench_def = source[source.index("def bench("):source.index("def bench(")+300]
|
||
self.assertIn("category", bench_def)
|
||
|
||
def test_bench_records_category_in_bench_list(self):
|
||
"""BENCH entries must include a 'category' field."""
|
||
with open(SCRIPT_PATH) as f:
|
||
source = f.read()
|
||
bench_def = source[source.index("def bench("):source.index("def bench(")+500]
|
||
self.assertIn("category", bench_def)
|
||
|
||
|
||
class TestServerLoadEstimation(unittest.TestCase):
|
||
"""Server load estimation tracks RTT variance to detect bot contention."""
|
||
|
||
def test_track_rtt_appends_to_history(self):
|
||
"""_track_rtt() must append response_ms to _rtt_history."""
|
||
_rtt_history.clear()
|
||
_track_rtt(150.0)
|
||
_track_rtt(160.0)
|
||
_track_rtt(145.0)
|
||
self.assertEqual(len(_rtt_history), 3)
|
||
self.assertAlmostEqual(_rtt_history[0], 150.0)
|
||
|
||
def test_current_load_calibrating_with_few_samples(self):
|
||
"""_current_load() returns CALIBRATING with < 3 samples."""
|
||
_rtt_history.clear()
|
||
self.assertEqual(_current_load(), "CALIBRATING")
|
||
_track_rtt(150.0)
|
||
self.assertEqual(_current_load(), "CALIBRATING")
|
||
|
||
def test_current_load_low_variance(self):
|
||
"""Low RTT variance (σ < 25ms) → LOW load."""
|
||
_rtt_history.clear()
|
||
for v in [150, 151, 149, 150, 151, 149]:
|
||
_track_rtt(v)
|
||
result = _current_load()
|
||
self.assertIn("LOW", result)
|
||
|
||
def test_current_load_moderate_variance(self):
|
||
"""Moderate RTT variance (25 ≤ σ < 100ms) → MODERATE load."""
|
||
_rtt_history.clear()
|
||
for v in [150, 200, 120, 220, 130, 210]:
|
||
_track_rtt(v)
|
||
result = _current_load()
|
||
self.assertIn("MODERATE", result)
|
||
|
||
def test_current_load_high_variance(self):
|
||
"""High RTT variance (σ ≥ 100ms) → HIGH load."""
|
||
_rtt_history.clear()
|
||
for v in [150, 400, 100, 450, 80, 380]:
|
||
_track_rtt(v)
|
||
result = _current_load()
|
||
self.assertIn("HIGH", result)
|
||
|
||
def test_load_trend_detects_increasing(self):
|
||
"""_load_trend() should detect when RTTs are increasing (load rising)."""
|
||
_rtt_history.clear()
|
||
# First half: low RTT, second half: high RTT
|
||
for v in [100] * 5 + [300] * 5:
|
||
_track_rtt(v)
|
||
trend_label, first_avg, second_avg = _load_trend()
|
||
self.assertIn("INCREASING", trend_label)
|
||
self.assertLess(first_avg, second_avg)
|
||
|
||
def test_load_trend_detects_stable(self):
|
||
"""_load_trend() should detect stable RTTs."""
|
||
_rtt_history.clear()
|
||
for v in [150, 151, 149, 151, 149, 150, 151, 149]:
|
||
_track_rtt(v)
|
||
trend_label, _, _ = _load_trend()
|
||
self.assertIn("STABLE", trend_label)
|
||
|
||
|
||
class TestHotPathZeroOverhead(unittest.TestCase):
|
||
"""The new logging/timing code must introduce ZERO computational delay
|
||
in the booking hot path (make_request retry loop)."""
|
||
|
||
def test_track_rtt_returns_none_no_computation(self):
|
||
"""_track_rtt() must return None — it only appends, no load computation.
|
||
Load computation is deferred to _compute_load() / _current_load()."""
|
||
_rtt_history.clear()
|
||
result = _track_rtt(150.0)
|
||
self.assertIsNone(result, "_track_rtt should not compute load — return None")
|
||
self.assertEqual(len(_rtt_history), 1)
|
||
|
||
def test_compute_load_matches_current_load(self):
|
||
"""_current_load() should delegate to _compute_load()."""
|
||
_rtt_history.clear()
|
||
for v in [150, 151, 149, 150, 151, 149]:
|
||
_track_rtt(v)
|
||
self.assertEqual(_current_load(), _compute_load())
|
||
|
||
def test_track_rtt_no_round_call(self):
|
||
"""_track_rtt must NOT call round() — raw float append only.
|
||
round() adds ~0.005µs per call × 1000 retries = unnecessary overhead."""
|
||
with open(SCRIPT_PATH) as f:
|
||
source = f.read()
|
||
# Isolate JUST _track_rtt (from def to next def = _compute_load)
|
||
track_start = source.index("def _track_rtt(")
|
||
track_end = source.index("def _compute_load(", track_start)
|
||
track_def = source[track_start:track_end]
|
||
# _track_rtt should only append, not compute
|
||
self.assertIn("_rtt_history.append", track_def)
|
||
self.assertNotIn("round(", track_def)
|
||
self.assertNotIn("sum(", track_def)
|
||
self.assertNotIn("variance", track_def)
|
||
self.assertNotIn("std_dev", track_def)
|
||
|
||
def test_make_request_calls_track_rtt_without_round(self):
|
||
"""make_request must call _track_rtt(elapsed) — no round() in the append."""
|
||
with open(SCRIPT_PATH) as f:
|
||
source = f.read()
|
||
mr_start = source.index("def make_request(")
|
||
mr_end = source.index("def build_booking_body(")
|
||
mr_body = source[mr_start:mr_end]
|
||
self.assertIn("_track_rtt(elapsed)", mr_body)
|
||
self.assertNotIn("_track_rtt(round(", mr_body)
|
||
|
||
|
||
class TestSlotTimingTracking(unittest.TestCase):
|
||
"""Booking loop must track per-slot timing for diagnostics."""
|
||
|
||
def test_slot_timings_in_source(self):
|
||
"""Source must define slot_timings list for per-slot tracking."""
|
||
with open(SCRIPT_PATH) as f:
|
||
source = f.read()
|
||
self.assertIn("slot_timings", source)
|
||
self.assertIn("t_booking_start", source)
|
||
self.assertIn("t_prefire", source)
|
||
|
||
def test_slot_timings_entries_have_required_fields(self):
|
||
"""Each slot_timings entry must have required timing fields."""
|
||
with open(SCRIPT_PATH) as f:
|
||
source = f.read()
|
||
# Verify the slot_timings.append structure has required keys
|
||
self.assertIn('"total_ms"', source)
|
||
self.assertIn('"attempts"', source)
|
||
self.assertIn('"avg_rtt_ms"', source)
|
||
self.assertIn('"status"', source)
|
||
|
||
def test_time_to_success_tracked(self):
|
||
"""Source must track time-to-success from pre-fire to cita."""
|
||
with open(SCRIPT_PATH) as f:
|
||
source = f.read()
|
||
self.assertIn("t_prefire_to_success", source)
|
||
self.assertIn("t_success_total", source)
|
||
self.assertIn("Time-to-success", source)
|
||
|
||
def test_per_slot_summary_in_benchmark(self):
|
||
"""Benchmark summary must include per-slot timing table."""
|
||
with open(SCRIPT_PATH) as f:
|
||
source = f.read()
|
||
self.assertIn("PER-SLOT TIMING", source)
|
||
self.assertIn("SERVER LOAD", source)
|
||
self.assertIn("TIME TO SUCCESS", source)
|
||
self.assertIn("_load_trend", source)
|
||
self.assertIn("_current_load", source)
|
||
|
||
|
||
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 28-10-2026 should return 9 slots.
|
||
(Oct 19-27 fully consumed by midnight bot rushes within the 15-day window;
|
||
Oct 28 is the next bookable date beyond the window with 9 available 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": "28-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(self):
|
||
"""insertarCitaPrevia for a date beyond the 15-day window should return
|
||
'Fecha posterior'. Using 28-10-2026 (16 days from Oct 12) which is
|
||
beyond the max_cita=15 window for cola I."""
|
||
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": "28-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}')
|
||
|
||
|
||
class TestCitaTypesCatalog(unittest.TestCase):
|
||
"""CITA_TYPES catalog must cover all active servicio/cola combos."""
|
||
|
||
def test_default_combo_present(self):
|
||
"""(S, I) — Portugal Tardes Padron — must be the default."""
|
||
self.assertIn(("S", "I"), CITA_TYPES)
|
||
meta = CITA_TYPES[("S", "I")]
|
||
self.assertEqual(meta["nombre"], "Portugal TARDE - Altas y modificaciones de Padrón")
|
||
self.assertTrue(meta["motivo_requerido"])
|
||
self.assertTrue(meta["activa"])
|
||
self.assertTrue(meta["cita_previa"])
|
||
|
||
def test_servicio_s_has_gestiones(self):
|
||
"""Servicio S should have multiple gestiones (10 bookable + 1 non-bookable)."""
|
||
s_gestiones = [k for k in CITA_TYPES if k[0] == 'S']
|
||
self.assertGreaterEqual(len(s_gestiones), 10)
|
||
# Count bookable (cita_previa=True) entries
|
||
s_bookable = [k for k in CITA_TYPES if k[0] == 'S' and CITA_TYPES[k].get('cita_previa')]
|
||
self.assertEqual(len(s_bookable), 10)
|
||
|
||
def test_servicio_i_has_gestiones(self):
|
||
"""Servicio I (Inmigración) should have multiple gestiones."""
|
||
i_gestiones = [k for k in CITA_TYPES if k[0] == 'I']
|
||
self.assertGreater(len(i_gestiones), 5)
|
||
|
||
def test_inmigracion_motivo_not_required(self):
|
||
"""Inmigración (servicio I) gestiones should have motivo_requerido=False."""
|
||
for key, meta in CITA_TYPES.items():
|
||
if key[0] == 'I':
|
||
self.assertFalse(meta["motivo_requerido"], f"Servicio I/{key[1]} should not require motivo")
|
||
|
||
def test_sac_saic_motivo_required(self):
|
||
"""Servicio S gestiones should have motivo_requerido=True (clean reason needed)."""
|
||
for key, meta in CITA_TYPES.items():
|
||
if key == ('S', 'D'):
|
||
continue # D has cita_previa=False
|
||
if key[0] == 'S':
|
||
self.assertTrue(meta["motivo_requerido"],
|
||
f"Servicio S/{key[1]} should require motivo")
|
||
|
||
def test_bono_servicio_l_impuestos_required(self):
|
||
"""Servicio L (Bonos) should require impuestos='S'."""
|
||
for key, meta in CITA_TYPES.items():
|
||
if key[0] == 'L':
|
||
self.assertTrue(meta.get("impuestos_requerido"),
|
||
f"Servicio L/{key[1]} should require impuestos")
|
||
|
||
def test_doc_types_has_four_types(self):
|
||
"""DOC_TYPES should have DNI, NIE, PASS, CIF."""
|
||
self.assertEqual(len(DOC_TYPES), 4)
|
||
self.assertEqual(DOC_TYPES['1'], 'DNI')
|
||
self.assertEqual(DOC_TYPES['2'], 'NIE')
|
||
self.assertEqual(DOC_TYPES['3'], 'PASS')
|
||
self.assertEqual(DOC_TYPES['4'], 'CIF')
|
||
|
||
def test_required_fields_list(self):
|
||
"""REQUIRED_FIELDS should contain all mandatory form fields."""
|
||
for field in REQUIRED_FIELDS:
|
||
self.assertIsInstance(field, str)
|
||
self.assertIn("nif", REQUIRED_FIELDS)
|
||
self.assertIn("nombre", REQUIRED_FIELDS)
|
||
self.assertIn("ape1", REQUIRED_FIELDS)
|
||
self.assertIn("tipo", REQUIRED_FIELDS)
|
||
|
||
|
||
class TestArgparseCli(unittest.TestCase):
|
||
"""CLI argument parsing must work correctly for all cita types."""
|
||
|
||
def test_parse_args_default_no_flags(self):
|
||
"""With no args, all form params should be None (defaults used)."""
|
||
args = parse_args([])
|
||
self.assertIsNone(args.servicio)
|
||
self.assertIsNone(args.cola)
|
||
self.assertIsNone(args.nif)
|
||
|
||
def test_parse_args_servicio_cola(self):
|
||
"""--servicio and --cola should be parsed correctly."""
|
||
args = parse_args(["-s", "S", "-c", "I"])
|
||
self.assertEqual(args.servicio, "S")
|
||
self.assertEqual(args.cola, "I")
|
||
|
||
def test_parse_args_form_params(self):
|
||
"""All form params should be parsed."""
|
||
args = parse_args(["-s", "S", "-c", "I", "-n", "X2491454W", "-t", "2",
|
||
"--nombre", "Humberto", "--ape1", "Normey",
|
||
"--ape2", "Gomez", "--tel", "687915599",
|
||
"-e", "hj@normey.net", "-m", "Solicitud de Alta en el Padrón",
|
||
"-d", "19-10-2026"])
|
||
self.assertEqual(args.nif, "X2491454W")
|
||
self.assertEqual(args.tipo, "2")
|
||
self.assertEqual(args.nombre, "Humberto")
|
||
self.assertEqual(args.ape1, "Normey")
|
||
self.assertEqual(args.motivo, "Solicitud de Alta en el Padrón")
|
||
self.assertEqual(args.fecha, "19-10-2026")
|
||
|
||
def test_parse_args_dry_run(self):
|
||
"""--dry-run flag should be set to True."""
|
||
args = parse_args(["--dry-run"])
|
||
self.assertTrue(args.dry_run)
|
||
|
||
def test_parse_args_help_citas(self):
|
||
"""--help-citas flag should be set to True."""
|
||
args = parse_args(["--help-citas"])
|
||
self.assertTrue(args.help_citas)
|
||
|
||
def test_parse_args_prompt(self):
|
||
"""--prompt flag should be set to True."""
|
||
args = parse_args(["--prompt"])
|
||
self.assertTrue(args.prompt)
|
||
|
||
def test_parse_args_impuestos_choices(self):
|
||
"""--impuestos should only accept S or N."""
|
||
args = parse_args(["--impuestos", "S"])
|
||
self.assertEqual(args.impuestos, "S")
|
||
with self.assertRaises(SystemExit):
|
||
parse_args(["--impuestos", "X"])
|
||
|
||
def test_parse_args_priority_override(self):
|
||
"""--priority should split into a list."""
|
||
args = parse_args(["--priority", "16:00,15:30,14:45"])
|
||
self.assertEqual(args.priority, "16:00,15:30,14:45")
|
||
|
||
def test_parse_args_default_ip(self):
|
||
"""--ip should default to 0.0.0.0."""
|
||
args = parse_args([])
|
||
self.assertEqual(args.ip, "0.0.0.0")
|
||
|
||
def test_parse_args_default_ape2_empty(self):
|
||
"""--ape2 should default to empty string."""
|
||
args = parse_args([])
|
||
self.assertEqual(args.ape2, "")
|
||
|
||
def test_help_citas_exits_cleanly(self):
|
||
"""--help-citas should print the catalog without error."""
|
||
self.assertTrue(callable(print_cita_types))
|
||
# Verify it prints without error when called
|
||
import io
|
||
from contextlib import redirect_stdout
|
||
f = io.StringIO()
|
||
with redirect_stdout(f):
|
||
print_cita_types()
|
||
output = f.getvalue()
|
||
self.assertIn("CITA PREVIA", output)
|
||
# Should list servicios
|
||
self.assertIn("Servicio", output)
|
||
# Should mention document types
|
||
self.assertIn("DNI", output)
|
||
self.assertIn("NIE", output)
|
||
# Should show usage examples
|
||
self.assertIn("USAGE EXAMPLES", output)
|
||
# Should contain the default cita type
|
||
self.assertIn("Portugal TARDE", output)
|
||
|
||
|
||
class TestValidateConfig(unittest.TestCase):
|
||
"""validate_config() must correctly identify missing/invalid params."""
|
||
|
||
def test_valid_config_no_custom_args(self):
|
||
"""With no custom args (backward compatible), validation passes."""
|
||
# Simulate 'no args' Namespace
|
||
import argparse as ap
|
||
ns = ap.Namespace(
|
||
servicio=None, cola=None, fecha=None, nif=None, tipo=None,
|
||
nombre=None, ape1=None, ape2="", tel=None, email=None,
|
||
motivo=None, impuestos=None, ip="0.0.0.0", priority=None,
|
||
prompt=False, dry_run=False, help_citas=False
|
||
)
|
||
# When no custom args, validate_config should not be called (or returns ok)
|
||
# Test with all valid params for S/I
|
||
ns2 = ap.Namespace(
|
||
servicio="S", cola="I", fecha="19-10-2026", nif="X2491454W",
|
||
tipo="2", nombre="Humberto", ape1="Normey", ape2="Gomez",
|
||
tel="687915599", email="hj@normey.net", motivo="Solicitud de Alta en el Padrón",
|
||
impuestos="N", ip="0.0.0.0", priority=None,
|
||
prompt=False, dry_run=False, help_citas=False
|
||
)
|
||
ok, missing, errors = validate_config(ns2)
|
||
self.assertTrue(ok, f"Should be valid: missing={missing}, errors={errors}")
|
||
|
||
def test_invalid_servicio_cola_combo(self):
|
||
"""Invalid servicio/cola combo should produce error."""
|
||
import argparse as ap
|
||
ns = ap.Namespace(
|
||
servicio="Z", cola="X", fecha="19-10-2026", nif="X2491454W",
|
||
tipo="2", nombre="Test", ape1="Test", ape2="",
|
||
tel="600", email="t@t.com", motivo="test",
|
||
impuestos="N", ip="0.0.0.0", priority=None,
|
||
prompt=False, dry_run=False, help_citas=False
|
||
)
|
||
ok, missing, errors = validate_config(ns)
|
||
self.assertFalse(ok)
|
||
self.assertTrue(any("Invalid servicio/cola" in e for e in errors))
|
||
|
||
def test_missing_motivo_for_s_i(self):
|
||
"""S/I requires motivo — missing → error."""
|
||
import argparse as ap
|
||
ns = ap.Namespace(
|
||
servicio="S", cola="I", fecha="19-10-2026", nif="X2491454W",
|
||
tipo="2", nombre="Test", ape1="Apellido", ape2="",
|
||
tel="600000000", email="t@t.com", motivo=None,
|
||
impuestos=None, ip="0.0.0.0", priority=None,
|
||
prompt=False, dry_run=False, help_citas=False
|
||
)
|
||
ok, missing, errors = validate_config(ns)
|
||
self.assertFalse(ok)
|
||
self.assertTrue(any("--motivo" in m for m in missing))
|
||
|
||
def test_motivo_not_required_for_i_c(self):
|
||
"""I/C (Inmigración) does NOT require motivo."""
|
||
import argparse as ap
|
||
ns = ap.Namespace(
|
||
servicio="I", cola="C", fecha="19-10-2026", nif="X2491454W",
|
||
tipo="2", nombre="Test", ape1="Apellido", ape2="",
|
||
tel="600000000", email="t@t.com", motivo="No importa",
|
||
impuestos=None, ip="0.0.0.0", priority=None,
|
||
prompt=False, dry_run=False, help_citas=False
|
||
)
|
||
ok, missing, errors = validate_config(ns)
|
||
self.assertTrue(ok, f"Should be valid: missing={missing}, errors={errors}")
|
||
|
||
def test_motivo_not_required_for_t_d(self):
|
||
"""T/D (Certificados) does NOT require motivo."""
|
||
import argparse as ap
|
||
ns = ap.Namespace(
|
||
servicio="T", cola="D", fecha="19-10-2026", nif="X2491454W",
|
||
tipo="2", nombre="Test", ape1="Apellido", ape2="",
|
||
tel="600000000", email="t@t.com", motivo=None,
|
||
impuestos=None, ip="0.0.0.0", priority=None,
|
||
prompt=False, dry_run=False, help_citas=False
|
||
)
|
||
# No motivo provided, but T/D has motivo_requerido=False
|
||
# However, REQUIRED_FIELDS still checks the base fields
|
||
# nif, tipo, nombre, ape1, fecha, tel, email are all provided
|
||
ok, missing, errors = validate_config(ns)
|
||
self.assertTrue(ok, f"Should be valid: missing={missing}, errors={errors}")
|
||
|
||
def test_servicio_l_requires_impuestos_s(self):
|
||
"""Servicio L (Bonos) requires --impuestos S."""
|
||
import argparse as ap
|
||
ns = ap.Namespace(
|
||
servicio="L", cola="A", fecha="19-10-2026", nif="X2491454W",
|
||
tipo="2", nombre="Test", ape1="Apellido", ape2="",
|
||
tel="600000000", email="t@t.com", motivo=None,
|
||
impuestos="N", ip="0.0.0.0", priority=None,
|
||
prompt=False, dry_run=False, help_citas=False
|
||
)
|
||
ok, missing, errors = validate_config(ns)
|
||
self.assertFalse(ok)
|
||
self.assertTrue(any("impuestos" in e.lower() for e in errors))
|
||
|
||
def test_nie_format_validation(self):
|
||
"""NIE (tipo=2) must start with X/Y/Z."""
|
||
import argparse as ap
|
||
ns = ap.Namespace(
|
||
servicio="S", cola="I", fecha="19-10-2026", nif="12345678",
|
||
tipo="2", nombre="Test", ape1="Apellido", ape2="",
|
||
tel="600000000", email="t@t.com", motivo="test",
|
||
impuestos=None, ip="0.0.0.0", priority=None,
|
||
prompt=False, dry_run=False, help_citas=False
|
||
)
|
||
ok, missing, errors = validate_config(ns)
|
||
self.assertFalse(ok)
|
||
self.assertTrue(any("NIE" in e or "X/Y/Z" in e for e in errors))
|
||
|
||
def test_fecha_format_validation(self):
|
||
"""--fecha must be DD-MM-YYYY format."""
|
||
import argparse as ap
|
||
ns = ap.Namespace(
|
||
servicio="S", cola="I", fecha="2026-10-19", nif="X2491454W",
|
||
tipo="2", nombre="Test", ape1="Apellido", ape2="",
|
||
tel="600000000", email="t@t.com", motivo="test",
|
||
impuestos=None, ip="0.0.0.0", priority=None,
|
||
prompt=False, dry_run=False, help_citas=False
|
||
)
|
||
ok, missing, errors = validate_config(ns)
|
||
self.assertFalse(ok)
|
||
self.assertTrue(any("DD-MM-YYYY" in e for e in errors))
|
||
|
||
def test_cola_required_with_servicio(self):
|
||
"""--cola required when --servicio given."""
|
||
import argparse as ap
|
||
ns = ap.Namespace(
|
||
servicio="S", cola=None, fecha="19-10-2026", nif="X2491454W",
|
||
tipo="2", nombre="Test", ape1="Apellido", ape2="",
|
||
tel="600000000", email="t@t.com", motivo="test",
|
||
impuestos=None, ip="0.0.0.0", priority=None,
|
||
prompt=False, dry_run=False, help_citas=False
|
||
)
|
||
ok, missing, errors = validate_config(ns)
|
||
self.assertFalse(ok)
|
||
self.assertTrue(any("--cola" in e for e in errors))
|
||
|
||
|
||
class TestBackwardCompatibility(unittest.TestCase):
|
||
"""Running with no CLI args must use defaults — no breakage."""
|
||
|
||
def test_build_booking_body_with_defaults(self):
|
||
"""build_booking_body should produce correct output with default constants."""
|
||
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")
|
||
|
||
def test_all_horas_constant_unchanged(self):
|
||
"""ALL_HORAS must still be the 9 confirmed slots."""
|
||
self.assertEqual(len(ALL_HORAS), 9)
|
||
self.assertEqual(ALL_HORAS[0], "14:30")
|
||
self.assertEqual(ALL_HORAS[-1], "16:30")
|
||
|
||
def test_priority_unchanged(self):
|
||
"""PRIORITY must still start with contrarian slot."""
|
||
self.assertEqual(PRIORITY[0], "15:15")
|
||
self.assertEqual(len(PRIORITY), 9)
|
||
|
||
def test_headres_intact(self):
|
||
"""HEADERS must still contain the ADF Content-Type."""
|
||
self.assertEqual(HEADERS["Content-Type"], CT)
|
||
|
||
def test_target_ts_unchanged(self):
|
||
"""TARGET_TS must still be Oct 4, 2026 midnight."""
|
||
dt = datetime.fromtimestamp(TARGET_TS)
|
||
self.assertEqual(dt.month, 10)
|
||
self.assertEqual(dt.day, 4)
|
||
self.assertEqual(dt.hour, 0)
|
||
|
||
|
||
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)
|