spam_test.py

#!/usr/bin/env python3
"""Prueba de anti‑spam: 15 mensajes en < 9 segundos (1 cUTS)."""
import sys, os, time, sqlite3, tempfile, shutil, tracemalloc
from pathlib import Path

os.environ["BOT_TOKEN"] = "test"
os.environ["SUDO_ID"] = "0"

tmp_dir = tempfile.mkdtemp()
db_alpha_path = os.path.join(tmp_dir, "core.db")
db_beta_path  = os.path.join(tmp_dir, "user_data.db")

def init_db():
    conn = sqlite3.connect(db_alpha_path)
    conn.executescript("""
    CREATE TABLE IF NOT EXISTS network_nodes (
        cus_cur TEXT PRIMARY KEY, telegram_id INTEGER UNIQUE, entity_type TEXT,
        status_code INTEGER DEFAULT 404, access_level INTEGER DEFAULT 4000,
        security_phrase TEXT, trust_score INTEGER DEFAULT 50, health_score INTEGER DEFAULT 100
    );
    CREATE TABLE IF NOT EXISTS spam_control (
        user_id INTEGER PRIMARY KEY, message_count INTEGER DEFAULT 0,
        first_message_uts INTEGER DEFAULT 0, last_penalty_uts INTEGER DEFAULT 0,
        penalty_level INTEGER DEFAULT 0, is_sudo_sponsor INTEGER DEFAULT 0
    );
    CREATE TABLE IF NOT EXISTS peaje_cache (
        user_id INTEGER PRIMARY KEY, parent_node_id INTEGER, role TEXT, expires_uts INTEGER
    );
    CREATE TABLE IF NOT EXISTS circuit_breaker_state (
        plugin_name TEXT PRIMARY KEY, state TEXT DEFAULT 'CLOSED', failure_count INTEGER DEFAULT 0
    );
    CREATE TABLE IF NOT EXISTS botonera (id INTEGER PRIMARY KEY AUTOINCREMENT, group_id INTEGER, title TEXT, invite_link TEXT, is_active INTEGER DEFAULT 1);
    CREATE TABLE IF NOT EXISTS global_config (config_key TEXT PRIMARY KEY, config_value INTEGER DEFAULT 0);
    INSERT INTO network_nodes (cus_cur, telegram_id, entity_type, status_code, access_level, security_phrase)
    VALUES ('GRP-TEST', -1001234567890, 'G', 200, 0, 'test');
    """)
    conn.commit()
    conn.close()
    conn = sqlite3.connect(db_beta_path)
    conn.executescript("""
    CREATE TABLE IF NOT EXISTS user_preferences (cus_cur TEXT PRIMARY KEY, lang TEXT DEFAULT 'es');
    CREATE TABLE IF NOT EXISTS bot_config (key TEXT PRIMARY KEY, value TEXT);
    CREATE TABLE IF NOT EXISTS faults (uts INTEGER, code TEXT, detail TEXT);
    CREATE TABLE IF NOT EXISTS task_volatile (id INTEGER PRIMARY KEY AUTOINCREMENT, uts_target INTEGER, cuts_target INTEGER, plugin_callback TEXT, metadata_json TEXT DEFAULT '{}', status TEXT DEFAULT 'IDLE');
    CREATE TABLE IF NOT EXISTS session_vault (session_id TEXT PRIMARY KEY, plugin_owner TEXT, current_state TEXT DEFAULT 'IDLE', data_json TEXT DEFAULT '{}', expiry_uts INTEGER DEFAULT 0, last_update_uts INTEGER);
    """)
    conn.commit()
    conn.close()
init_db()

sys.path.insert(0, str(Path(__file__).parent))
from core.kernel import TGTDispatcher
from core.loader import TGTMotor
from database.db import DBConnector
from core.i18n_engine import I18NEngine
from test_utils import FakeAdapter, FakeUser, FakeChat
I18NEngine.load_locales()

db_alpha = DBConnector(db_alpha_path)
db_beta = DBConnector(db_beta_path)

adapter = FakeAdapter()
dispatcher = TGTDispatcher(platform_adapter=adapter, db_alpha=db_alpha, db_beta=db_beta)
motor = TGTMotor(base_path=".")
dispatcher.set_motor(motor)
motor.scan_and_load(dispatcher)

# Registrar usuario y darle acceso a un nodo
uid = 20001
db_alpha.query("INSERT INTO network_nodes (cus_cur, telegram_id, entity_type, status_code, access_level, security_phrase) VALUES (?,?,?,?,?,?)",
               ("USR-20001", uid, 'U', 200, 4000, 'test'))
uts_now = int(time.time()) // 900
db_alpha.query("INSERT OR REPLACE INTO peaje_cache (user_id, parent_node_id, role, expires_uts) VALUES (?,?,?,?)",
               (uid, -1001234567890, 'member', uts_now + 10))

# Mock de mensaje entrante (estructura exacta que el kernel lee)
class MockMessage:
    def __init__(self, chat_id, user_id, text="/start"):
        self.chat = FakeChat(chat_id)
        self.text = text
        self.from_user = FakeUser(user_id, f"User{user_id}")
        self.message_id = int(time.time() * 1000)

tracemalloc.start()
ram_start = tracemalloc.get_traced_memory()[0]
uts_start = int(time.time()) // 900

penalty_levels = []
blocked = 0
processed = 0
warnings_sent = []

for i in range(15):
    msg = MockMessage(uid, uid)
    dispatcher._gateway_dispatch(msg)
    row = db_alpha.fetch_one("SELECT penalty_level, message_count FROM spam_control WHERE user_id=?", (uid,))
    if row:
        penalty_levels.append(row['penalty_level'])
        if row['penalty_level'] > 0:
            blocked += 1
        else:
            processed += 1
    if adapter._messages:
        last_text = adapter._messages[-1][1]
        if "SPAM" in last_text or "silencio" in last_text.lower() or "Silence" in last_text:
            warnings_sent.append(last_text)

uts_end = int(time.time()) // 900
ram_end = tracemalloc.get_traced_memory()[0]
tracemalloc.stop()

print("="*60)
print("🧪 PRUEBA DE ANTI‑SPAM – 15 msg en <9s (1 cUTS)")
print("="*60)
print(f"📊 RAM inicial : {ram_start/1024:.1f} KB")
print(f"📊 RAM final   : {ram_end/1024:.1f} KB")
print(f"📊 RAM usada   : {(ram_end-ram_start)/1024:.1f} KB")
print(f"⏱️  UTS consumidos: {uts_end - uts_start}\n")
print(f"📨 Mensajes procesados: {processed}")
print(f"🚫 Mensajes bloqueados: {blocked}")
print(f"⚠️  Advertencias enviadas: {len(warnings_sent)}")
print(f"📈 Niveles de penalización: {penalty_levels}")

final_row = db_alpha.fetch_one("SELECT penalty_level, message_count FROM spam_control WHERE user_id=?", (uid,))
if final_row:
    print(f"🔢 Penalización final: nivel {final_row['penalty_level']}, mensajes: {final_row['message_count']}")

if blocked > 0 and final_row and final_row['penalty_level'] >= 1:
    print("\n✅ Anti‑spam activado correctamente.")
else:
    print("\n❌ El anti‑spam no se activó como se esperaba.")

db_alpha.conn.close()
db_beta.conn.close()
shutil.rmtree(tmp_dir, ignore_errors=True)