#!/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)