feat: initial implementation taller-wox.fitlabs.dev

Portal FastAPI + 5 endpoints REST para Bootcamp Agentic AI con
watsonx Orchestrate (FactorIT). Single container, Coolify-ready.

- Landing brandeado FIT con formulario de registro (honeypot anti-bot)
- Tokens itsdangerous para descargas (24h expiry)
- 5 endpoints API: historical/available procedures, member-insights,
  schedule, generate-report (Jinja2 + Plotly)
- SQLite con upsert-on-email para leads + log de descargas
- Admin endpoints (HTTP Basic): leads.json, leads.csv, stats
- 23 tests pytest pasando
- Dockerfile listo para Coolify con volúmenes persistentes
  (/app/leads.db, /app/app/data/reports_output, /app/material)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

tests/conftest.py

@@ -0,0 +1,38 @@
+import os
+import tempfile
+from pathlib import Path
+
+import pytest
+
+os.environ["SECRET_KEY"] = "test-secret-key-only-for-tests-not-secure"
+os.environ["ADMIN_USER"] = "testadmin"
+os.environ["ADMIN_PASS"] = "testpass"
+os.environ["TOKEN_EXPIRY_HOURS"] = "24"
+os.environ["BASE_URL"] = "http://testserver"
+
+_tmp = Path(tempfile.mkdtemp(prefix="taller-wox-test-"))
+os.environ["DB_PATH"] = str(_tmp / "test_leads.db")
+os.environ["MATERIAL_DIR"] = str(_tmp / "material")
+os.environ["REPORTS_OUTPUT_DIR"] = str(_tmp / "reports_output")
+(_tmp / "material").mkdir(parents=True, exist_ok=True)
+(_tmp / "reports_output").mkdir(parents=True, exist_ok=True)
+
+
+@pytest.fixture
+def client():
+    from fastapi.testclient import TestClient
+
+    from app.main import app
+    return TestClient(app)
+
+
+@pytest.fixture(autouse=True)
+def _init_db_each_test(tmp_path, monkeypatch):
+    db_file = tmp_path / "test.db"
+    monkeypatch.setenv("DB_PATH", str(db_file))
+    from app import config
+    config.get_settings.cache_clear()
+    from app.db import init_db
+    init_db()
+    yield
+    config.get_settings.cache_clear()

tests/test_admin.py

@@ -0,0 +1,56 @@
+import base64
+
+
+def _auth_header(user: str = "testadmin", password: str = "testpass") -> dict:
+    token = base64.b64encode(f"{user}:{password}".encode()).decode()
+    return {"Authorization": f"Basic {token}"}
+
+
+def test_admin_leads_requires_auth(client):
+    response = client.get("/admin/leads.json")
+    assert response.status_code == 401
+
+
+def test_admin_leads_with_bad_credentials_rejected(client):
+    response = client.get("/admin/leads.json", headers=_auth_header(password="wrong"))
+    assert response.status_code == 401
+
+
+def test_admin_leads_json_returns_list(client):
+    client.post(
+        "/register",
+        data={"nombre": "Ana", "email": "a@a.com", "empresa": "ACo", "consentimiento": "on"},
+    )
+    response = client.get("/admin/leads.json", headers=_auth_header())
+    assert response.status_code == 200
+    body = response.json()
+    assert isinstance(body, list)
+    assert any(l["email"] == "a@a.com" for l in body)
+
+
+def test_admin_leads_csv_returns_csv(client):
+    client.post(
+        "/register",
+        data={"nombre": "Beto", "email": "b@b.com", "empresa": "BCo", "consentimiento": "on"},
+    )
+    response = client.get("/admin/leads.csv", headers=_auth_header())
+    assert response.status_code == 200
+    assert "text/csv" in response.headers["content-type"]
+    body = response.text
+    assert body.startswith("")
+    assert "b@b.com" in body
+
+
+def test_admin_stats_returns_counts(client):
+    client.post(
+        "/register",
+        data={"nombre": "Carla", "email": "c@c.com", "empresa": "CCo", "consentimiento": "on"},
+    )
+    response = client.get("/admin/stats", headers=_auth_header())
+    assert response.status_code == 200
+    body = response.json()
+    assert "total_leads" in body
+    assert "total_downloads" in body
+    assert "downloads_por_archivo" in body
+    assert "top_5_empresas" in body
+    assert body["total_leads"] >= 1

tests/test_benefits_api.py

@@ -0,0 +1,85 @@
+def test_member_insights_returns_static_object(client):
+    response = client.get("/api/member-insights")
+    assert response.status_code == 200
+    body = response.json()
+    assert body["result"]["member"]["name"] == "Charlie Smith"
+    assert body["result"]["member"]["plan"] == "Gold PPO"
+    assert len(body["result"]["overdue_procedures"]) == 4
+
+
+def test_schedule_returns_text_wrapped(client):
+    response = client.get("/api/schedule")
+    assert response.status_code == 200
+    body = response.json()
+    assert "agendar una cita" in body["result"].lower()
+    assert "1-800-FIT-CARE" in body["result"]
+
+
+def test_historical_procedures_no_filter_returns_all(client):
+    response = client.post(
+        "/api/historical-procedures",
+        json={"filters": "[]", "group_by": "[]"},
+    )
+    assert response.status_code == 200
+    assert len(response.json()["result"]) == 50
+
+
+def test_historical_procedures_filter_equals(client):
+    response = client.post(
+        "/api/historical-procedures",
+        json={
+            "filters": '[{"column": "member_name", "operator": "equals", "value": "Charlie Smith"}]',
+            "group_by": "[]",
+        },
+    )
+    rows = response.json()["result"]
+    assert len(rows) == 5
+    assert all(r["member_name"] == "Charlie Smith" for r in rows)
+
+
+def test_historical_procedures_filter_contains(client):
+    response = client.post(
+        "/api/historical-procedures",
+        json={
+            "filters": '[{"column": "procedure", "operator": "contains", "value": "mri"}]',
+            "group_by": "[]",
+        },
+    )
+    rows = response.json()["result"]
+    assert len(rows) > 0
+    assert all("MRI" in r["procedure"] for r in rows)
+
+
+def test_historical_procedures_filter_gt(client):
+    response = client.post(
+        "/api/historical-procedures",
+        json={
+            "filters": '[{"column": "total_cost", "operator": "gt", "value": 5000}]',
+            "group_by": "[]",
+        },
+    )
+    rows = response.json()["result"]
+    assert len(rows) > 0
+    assert all(r["total_cost"] > 5000 for r in rows)
+
+
+def test_available_procedures_no_filter(client):
+    response = client.post(
+        "/api/available-procedures",
+        json={"filters": "[]", "group_by": "[]"},
+    )
+    rows = response.json()["result"]
+    assert len(rows) == 28
+
+
+def test_available_procedures_filter_mri(client):
+    response = client.post(
+        "/api/available-procedures",
+        json={
+            "filters": '[{"column": "procedure", "operator": "contains", "value": "MRI"}]',
+            "group_by": "[]",
+        },
+    )
+    rows = response.json()["result"]
+    assert len(rows) >= 3
+    assert all(r["procedure"] == "MRI" for r in rows)

tests/test_frontend.py

@@ -0,0 +1,122 @@
+import os
+from pathlib import Path
+
+
+def test_landing_renders(client):
+    response = client.get("/")
+    assert response.status_code == 200
+    assert "Bootcamp Agentic AI" in response.text
+    assert 'name="email"' in response.text
+    assert 'name="website"' in response.text
+
+
+def test_register_valid_form_redirects_with_token(client):
+    response = client.post(
+        "/register",
+        data={
+            "nombre": "Felipe Arentsen",
+            "email": "felipe@factorit.com",
+            "empresa": "FactorIT",
+            "consentimiento": "on",
+        },
+        follow_redirects=False,
+    )
+    assert response.status_code == 303
+    assert response.headers["location"].startswith("/descargas?token=")
+
+
+def test_register_honeypot_filled_silently_drops(client):
+    response = client.post(
+        "/register",
+        data={
+            "nombre": "Bot",
+            "email": "bot@spam.com",
+            "empresa": "Spam Inc",
+            "consentimiento": "on",
+            "website": "https://spam.com",
+        },
+        follow_redirects=False,
+    )
+    assert response.status_code == 303
+    from app.db import get_lead_by_email
+    assert get_lead_by_email("bot@spam.com") is None
+
+
+def test_register_duplicate_email_reissues_token(client):
+    for _ in range(2):
+        client.post(
+            "/register",
+            data={
+                "nombre": "Felipe",
+                "email": "dup@factorit.com",
+                "empresa": "FactorIT",
+                "consentimiento": "on",
+            },
+            follow_redirects=False,
+        )
+    from app.db import get_lead_by_email
+    lead = get_lead_by_email("dup@factorit.com")
+    assert lead["times_registered"] == 2
+
+
+def test_descargas_with_invalid_token_redirects_home(client):
+    response = client.get("/descargas?token=invalid-junk", follow_redirects=False)
+    assert response.status_code == 307
+    assert response.headers["location"].startswith("/?error=")
+
+
+def test_descargas_with_valid_token_renders(client):
+    reg = client.post(
+        "/register",
+        data={
+            "nombre": "Maria",
+            "email": "maria@test.com",
+            "empresa": "Test Co",
+            "consentimiento": "on",
+        },
+        follow_redirects=False,
+    )
+    token = reg.headers["location"].split("token=")[1]
+    response = client.get(f"/descargas?token={token}")
+    assert response.status_code == 200
+    assert "Hola Maria" in response.text
+    assert "Material técnico" in response.text
+    assert "Material funcional" in response.text
+
+
+def test_download_with_valid_token_serves_file(client):
+    material_dir = Path(os.environ["MATERIAL_DIR"])
+    material_dir.mkdir(parents=True, exist_ok=True)
+    (material_dir / "taller-wox-tecnico.zip").write_bytes(b"PK\x03\x04 fake zip")
+
+    reg = client.post(
+        "/register",
+        data={
+            "nombre": "Test",
+            "email": "test-dl@test.com",
+            "empresa": "TC",
+            "consentimiento": "on",
+        },
+        follow_redirects=False,
+    )
+    token = reg.headers["location"].split("token=")[1]
+    response = client.get(f"/download/taller-wox-tecnico.zip?token={token}")
+    assert response.status_code == 200
+    assert response.content == b"PK\x03\x04 fake zip"
+    assert "attachment" in response.headers["content-disposition"]
+
+
+def test_download_invalid_filename_returns_404(client):
+    reg = client.post(
+        "/register",
+        data={
+            "nombre": "Test",
+            "email": "bad-fn@test.com",
+            "empresa": "TC",
+            "consentimiento": "on",
+        },
+        follow_redirects=False,
+    )
+    token = reg.headers["location"].split("token=")[1]
+    response = client.get(f"/download/etc-passwd?token={token}")
+    assert response.status_code == 404

tests/test_health.py

@@ -0,0 +1,5 @@
+def test_health(client):
+    response = client.get("/health")
+    assert response.status_code == 200
+    body = response.json()
+    assert body["status"] == "ok"

tests/test_reports_api.py

@@ -0,0 +1,23 @@
+import json
+import os
+from pathlib import Path
+
+
+def test_generate_report_with_care_report_preset(client):
+    response = client.post(
+        "/api/reports/generate-report",
+        json={"layout_config": json.dumps(["care_report"])},
+    )
+    assert response.status_code == 200
+    body = response.json()
+    assert "public_url" in body
+    assert body["public_url"].endswith(".html")
+
+    report_id = body["public_url"].rsplit("/", 1)[-1]
+    output_dir = Path(os.environ["REPORTS_OUTPUT_DIR"])
+    assert (output_dir / report_id).exists()
+    content = (output_dir / report_id).read_text()
+    assert "Care Report" in content
+    assert "Customer Overview" in content
+    assert "Claim Review Summary" in content
+    assert "<table" in content