Versión 2X: refactor y tests (#7)

Refactor completo a “Versión 2X”

Mueve load_dotenv() y la validación de TELEGRAM_TOKEN/TELEGRAM_CHAT_ID dentro de main() y añade if __name__ == "__main__": main().

Extrae el bloque de lectura en parse_sensor_block(lines: list[str]) -> dict[str, float] usando un diccionario de mapeo para etiquetas.

Gestiona SerialException antes y durante la lectura de 6 líneas, cerrando y reconectando tras un breve sleep().

Implementa flush_db(conn, buffer) -> int que escriba el buffer y borre datos de hace >7 días, y una función aparte wal_checkpoint(conn) para el PRAGMA wal_checkpoint(TRUNCATE).

Asegura el cierre en un finally: ser.close(), conn.close(), session.close().

Añade constantes documentadas para DATA_DIR, LOG_DIR, ARDUINO_VID, ARDUINO_PID en el README.

Refuerza el Dockerfile: crea appuser, permisos 700, HEALTHCHECK, cachea requirements.txt.

Crea .github/workflows/ci.yml con cache de pip, instaladores de hooks, pre-commit, pytest-cov, y artefactos de cobertura.

Genera tests unitarios para parser, DB, reconexión serial y un test de integración completo que simule lectura → parseo → flush → checkpoint → notificación.

Author: bryamzxz

CHANGELOG.md

@@ -12,3 +12,9 @@
 - CI con cache de pip, pre-commit, pytest y cobertura.
 - Nuevas pruebas de reconexión y checkpoint.
 
+## v2X
+
+- Refactor de `server.py` con parseo desacoplado y reconexión robusta.
+- Nueva función `wal_checkpoint` y limpieza segura de BD.
+- Tests ampliados y flujo de integración completo.
+

README.md

@@ -1,4 +1,4 @@
-# Sensor App
+# Sensor App v2X
 
 ![CI](https://github.com/tu_usuario/sensor_app/actions/workflows/ci.yml/badge.svg)
 ![Coverage](https://img.shields.io/badge/coverage-unknown-lightgrey.svg)
@@ -98,6 +98,9 @@ De este modo los chequeos se ejecutarán automáticamente antes de cada commit.
 * **Retención**:
 
   * La línea `DELETE FROM lecturas WHERE Tiempo < datetime('now','-7 days')` mantiene sólo 7 días de datos. Modifícala si necesitas otro periodo.
+* **Rutas y VID/PID**:
+  * Las carpetas `data/` y `logs/` se crean con permisos 700 y pueden cambiarse con las constantes `DATA_DIR` y `LOG_DIR` en `server.py`.
+  * Los identificadores USB del Arduino (`ARDUINO_VID`, `ARDUINO_PID`) también son configurables al inicio del script.
 
 ## Contribuciones
 

server.py

@@ -16,7 +16,7 @@
 from typing import Any, Dict
 import pytz
 
-__version__ = "1.0.0"
+__version__ = "2.0.0"
 
 # ------------------------------
 # ⚙️ Configuración general
@@ -121,6 +121,8 @@ def parse_sensor_block(lines: list[str]) -> Dict[str, float]:
                 datos["Presion"] = num
             elif "Gas Resistencia" in key:
                 datos["Gas_Resistencia"] = num
+            else:
+                logging.warning(f"Etiqueta desconocida: {key}")
     except Exception as e:  # catch unexpected errors
         logging.error(f"❌ Error parseando bloque: {e}")
     return datos
@@ -142,24 +144,28 @@ def safe_execute(
 
 
 def flush_db(conn: sqlite3.Connection, buffer: list[Dict[str, Any]]) -> int:
-    """Inserta los registros del buffer en la base de datos de forma segura."""
+    """Inserta el buffer y depura registros antiguos."""
     registros = [tuple(d.get(c) for c in COLUMNAS) for d in buffer]
     try:
         with conn:
             conn.executemany(
                 "INSERT OR REPLACE INTO lecturas VALUES (?,?,?,?,?,?)", registros
             )
+            conn.execute(
+                "DELETE FROM lecturas WHERE Tiempo < datetime('now','-7 days')"
+            )
     except sqlite3.DatabaseError as e:
         logging.error(f"❌ Error al escribir en BD: {e}")
         return 0
 
-    safe_execute(conn, "DELETE FROM lecturas WHERE Tiempo < datetime('now','-7 days')")
-    # separa checkpoint para evitar bloquear si falla
-    safe_execute(conn, "PRAGMA wal_checkpoint(TRUNCATE);")
-
     return len(registros)
 
 
+def wal_checkpoint(conn: sqlite3.Connection) -> None:
+    """Ejecuta un checkpoint WAL sin interrumpir errores."""
+    safe_execute(conn, "PRAGMA wal_checkpoint(TRUNCATE);")
+
+
 # ------------------------------
 # 🔌 Serial con reconexión
 # ------------------------------
@@ -262,19 +268,25 @@ def main() -> None:
     TELEGRAM_TOKEN = token
     TELEGRAM_CHAT_ID = chat_id
 
-    conn = conectar_db(DB_FILE)
-    ser = conectar_serial()
-
-    buffer = []
-    contador_total = conn.execute("SELECT COUNT(*) FROM lecturas").fetchone()[0] or 0
-    logging.info(f"Se reanuda el script. Registros en BD: {contador_total}")
+    conn: sqlite3.Connection | None = None
+    ser: serial.Serial | None = None
+    try:
+        conn = conectar_db(DB_FILE)
+        ser = conectar_serial()
+        assert conn is not None
+        assert ser is not None
+
+        buffer = []
+        contador_total = (
+            conn.execute("SELECT COUNT(*) FROM lecturas").fetchone()[0] or 0
+        )
+        logging.info(f"Se reanuda el script. Registros en BD: {contador_total}")
 
-    ahora = time.time()
-    proximo_flush_db = ahora + random.randint(MIN_ESCRITURA_DB, MAX_ESCRITURA_DB)
+        ahora = time.time()
+        proximo_flush_db = ahora + random.randint(MIN_ESCRITURA_DB, MAX_ESCRITURA_DB)
 
-    logging.info("📡 Iniciando captura de datos...")
+        logging.info("📡 Iniciando captura de datos...")
 
-    try:
         while True:
             try:
                 if not ser.is_open:
@@ -287,6 +299,7 @@ def main() -> None:
             except serial.SerialException:
                 logging.error("Serial perdido. Cerrando y reconectando...")
                 ser.close()
+                time.sleep(2)
                 ser = conectar_serial()
                 continue
 
@@ -310,6 +323,7 @@ def main() -> None:
                             "Serial perdido durante lectura. Cerrando y reconectando..."
                         )
                         ser.close()
+                        time.sleep(2)
                         ser = conectar_serial()
                         lost_serial = True
                         break
@@ -328,6 +342,7 @@ def main() -> None:
                 logging.info(f"Iniciando flush de {len(buffer)} registros a la BD.")
                 t0 = time.time()
                 escritos = flush_db(conn, buffer)
+                wal_checkpoint(conn)
                 t1 = time.time()
                 if escritos:
                     contador_total += escritos
@@ -345,9 +360,10 @@ def main() -> None:
     except KeyboardInterrupt:
         logging.info("🚫 Detenido manualmente.")
     finally:
-        if ser.is_open:
+        if ser and ser.is_open:
             ser.close()
-        conn.close()
+        if conn:
+            conn.close()
         session.close()
         logging.info("Script finalizado. Conexiones cerradas.")
 

tests/test_db.py

@@ -6,7 +6,7 @@
 os.environ.setdefault("TELEGRAM_TOKEN", "x")
 os.environ.setdefault("TELEGRAM_CHAT_ID", "x")
 
-from server import conectar_db, flush_db, safe_execute  # noqa: E402
+from server import conectar_db, flush_db, safe_execute, wal_checkpoint  # noqa: E402
 
 
 def test_safe_execute_error(tmp_path):
@@ -33,6 +33,7 @@ def test_flush_db(tmp_path):
     ]
 
     inserted = flush_db(conn, buffer)
+    wal_checkpoint(conn)
     assert inserted == 1
     cur = conn.execute("SELECT COUNT(*) FROM lecturas")
     assert cur.fetchone()[0] == 1

tests/test_integration.py

@@ -9,7 +9,14 @@
 
 import serial  # noqa: E402
 
-from server import conectar_db, conectar_serial, flush_db  # noqa: E402
+from server import (  # noqa: E402
+    conectar_db,
+    conectar_serial,
+    flush_db,
+    wal_checkpoint,
+    parse_sensor_block,
+    enviar_notificacion,
+)  # noqa: E402
 
 
 class DummySerial:
@@ -53,6 +60,7 @@ def fake_safe_execute(conn, query, params=None):
 
     monkeypatch.setattr("server.safe_execute", fake_safe_execute)
     inserted = flush_db(conn, buffer)
+    wal_checkpoint(conn)
     assert inserted == 1
 
 
@@ -75,3 +83,48 @@ def serial_side_effect(*args, **kwargs):
         with mock.patch("serial.Serial", side_effect=serial_side_effect):
             result = conectar_serial()
             assert result is ser_second
+
+
+def test_full_flow(tmp_path, monkeypatch):
+    db_path = tmp_path / "db.sqlite"
+    conn = conectar_db(db_path)
+    lines = [
+        "------ Lecturas ------",
+        "TMP117 Temp: 20.0 C",
+        "BME680 Temp: 21.0 C",
+        "Humedad: 50 %",
+        "Presion: 1000 hPa",
+        "Gas Resistencia: 200 kOhm",
+        "",
+    ]
+    ser = DummySerial(lines)
+
+    posts = []
+
+    def fake_post(url, data, timeout):
+        posts.append((url, data))
+
+        class Resp:
+            status_code = 200
+
+            def json(self):
+                return {"result": {"message_id": 1}}
+
+            def raise_for_status(self):
+                pass
+
+        return Resp()
+
+    monkeypatch.setattr("server.session.post", fake_post)
+
+    header = ser.readline().decode()
+    assert "Lecturas" in header
+    sensor_lines = [ser.readline().decode().strip() for _ in range(6)]
+    datos = {"Tiempo": "2030-01-01 00:00:00"}
+    datos.update(parse_sensor_block(sensor_lines))
+    inserted = flush_db(conn, [datos])
+    wal_checkpoint(conn)
+    enviar_notificacion(datos)
+
+    assert inserted == 1
+    assert posts

tests/test_parser.py

@@ -31,3 +31,8 @@ def test_parse_sensor_block_valid():
 def test_parse_sensor_block_invalid():
     lines = ["foo", "bar"]
     assert parse_sensor_block(lines) == {}
+
+
+def test_parse_sensor_block_unknown_label():
+    lines = ["Desconocido: 123"]
+    assert parse_sensor_block(lines) == {}