susana/BeyondCXAnalytics-Demo
2
0
Fork 0
Code Issues 2 Pull Requests 1 Actions Packages Projects Releases 6 Wiki Activity

Initial commit - ACME demo version

Browse Source
This commit is contained in:
sujucu70
2026-02-04 11:08:21 +01:00
commit 1bb0765766
180 changed files with 52249 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

168
backend/tests/test_api.sh Normal file
View File

@@ -0,0 +1,168 @@
#!/usr/bin/env bash
set -euo pipefail
# ===========================
# Configuración
# ===========================
HOST="${HOST:-localhost}"
PORT="${PORT:-8000}"
API_URL="http://$HOST:$PORT/analysis"
# Credenciales Basic Auth (ajusta si usas otras)
API_USER="${API_USER:-beyond}"
API_PASS="${API_PASS:-beyond2026}"
# Ruta del CSV en tu máquina para subirlo
LOCAL_CSV_FILE="${LOCAL_CSV_FILE:-data/example/synthetic_interactions.csv}"
# Carpetas de salida
OUT_DIR="${OUT_DIR:-./test_results}"
mkdir -p "$OUT_DIR"
print_header() {
echo
echo "============================================================"
echo "$1"
echo "============================================================"
}
# ===========================
# 1. Health-check simple (sin auth)
# ===========================
print_header "1) Comprobando que el servidor responde (sin auth) - debería devolver 401"
set +e
curl -s -o /dev/null -w "HTTP status: %{http_code}\n" \
-X POST "$API_URL"
set -e
# ===========================
# 2. Test: subir CSV (analysis=premium por defecto)
# ===========================
print_header "2) Subiendo CSV local con análisis 'premium' (default) y guardando JSON"
if [ ! -f "$LOCAL_CSV_FILE" ]; then
echo "⚠️ Aviso: el fichero LOCAL_CSV_FILE='$LOCAL_CSV_FILE' no existe."
echo " Cambia la variable LOCAL_CSV_FILE o copia el CSV a esa ruta."
else
curl -v \
-u "$API_USER:$API_PASS" \
-X POST "$API_URL" \
-F "csv_file=@${LOCAL_CSV_FILE}" \
-o "${OUT_DIR}/resultados_premium.json"
echo "✅ JSON guardado en: ${OUT_DIR}/resultados_premium.json"
echo " Primeras líneas:"
head -n 20 "${OUT_DIR}/resultados_premium.json" || true
fi
# ===========================
# 3. Test: subir CSV con analysis=basic
# ===========================
print_header "3) Subiendo CSV local con análisis 'basic' y guardando JSON"
if [ ! -f "$LOCAL_CSV_FILE" ]; then
echo "⚠️ Saltando este test porque LOCAL_CSV_FILE='$LOCAL_CSV_FILE' no existe."
else
curl -v \
-u "$API_USER:$API_PASS" \
-X POST "$API_URL" \
-F "csv_file=@${LOCAL_CSV_FILE}" \
-F "analysis=basic" \
-o "${OUT_DIR}/resultados_basic.json"
echo "✅ JSON guardado en: ${OUT_DIR}/resultados_basic.json"
echo " Primeras líneas:"
head -n 20 "${OUT_DIR}/resultados_basic.json" || true
fi
# ===========================
# 4. Test: con economy_json personalizado (premium)
# ===========================
print_header "4) Subiendo CSV con configuración económica personalizada (analysis=premium)"
if [ ! -f "$LOCAL_CSV_FILE" ]; then
echo "⚠️ Saltando este test porque LOCAL_CSV_FILE='$LOCAL_CSV_FILE' no existe."
else
curl -v \
-u "$API_USER:$API_PASS" \
-X POST "$API_URL" \
-F "csv_file=@${LOCAL_CSV_FILE}" \
-F 'economy_json={"labor_cost_per_hour":30,"automation_volume_share":0.7,"customer_segments":{"VIP":"high","Basico":"medium"}}' \
-F "analysis=premium" \
-o "${OUT_DIR}/resultados_economy_premium.json"
echo "✅ JSON con economía personalizada guardado en: ${OUT_DIR}/resultados_economy_premium.json"
echo " Primeras líneas:"
head -n 20 "${OUT_DIR}/resultados_economy_premium.json" || true
fi
# ===========================
# 5. Test de error: economy_json inválido
# ===========================
print_header "5) Petición con economy_json inválido - debe devolver 400"
set +e
curl -v \
-u "$API_USER:$API_PASS" \
-X POST "$API_URL" \
-F "csv_file=@${LOCAL_CSV_FILE}" \
-F "economy_json={invalid json" \
-o "${OUT_DIR}/error_economy_invalid.json"
STATUS=$?
set -e
echo "✅ Respuesta guardada en: ${OUT_DIR}/error_economy_invalid.json"
cat "${OUT_DIR}/error_economy_invalid.json" || true
# ===========================
# 6. Test de error: analysis inválido
# ===========================
print_header "6) Petición con analysis inválido - debe devolver 400"
set +e
curl -v \
-u "$API_USER:$API_PASS" \
-X POST "$API_URL" \
-F "csv_file=@${LOCAL_CSV_FILE}" \
-F "analysis=ultra" \
-o "${OUT_DIR}/error_analysis_invalid.json"
set -e
echo "✅ Respuesta guardada en: ${OUT_DIR}/error_analysis_invalid.json"
cat "${OUT_DIR}/error_analysis_invalid.json" || true
# ===========================
# 7. Test de error: sin csv_file (debe devolver 422)
# ===========================
print_header "7) Petición inválida (sin csv_file) - debe devolver 422 (FastAPI validation)"
set +e
curl -v \
-u "$API_USER:$API_PASS" \
-X POST "$API_URL" \
-o "${OUT_DIR}/error_missing_csv.json"
set -e
echo "✅ Respuesta guardada en: ${OUT_DIR}/error_missing_csv.json"
cat "${OUT_DIR}/error_missing_csv.json" || true
# ===========================
# 8. Test de error: credenciales incorrectas
# ===========================
print_header "8) Petición con credenciales incorrectas - debe devolver 401"
set +e
curl -v \
-u "wrong:wrong" \
-X POST "$API_URL" \
-F "csv_file=@${LOCAL_CSV_FILE}" \
-o "${OUT_DIR}/error_auth.json"
set -e
echo "✅ Respuesta de error de auth guardada en: ${OUT_DIR}/error_auth.json"
cat "${OUT_DIR}/error_auth.json" || true
echo
echo "✨ Tests terminados. Revisa la carpeta: ${OUT_DIR}"

128
backend/tests/test_economy_cost.py Normal file
View File

@@ -0,0 +1,128 @@
import math
from datetime import datetime
import matplotlib
import pandas as pd
from beyond_metrics.dimensions.EconomyCost import EconomyCostMetrics, EconomyConfig
matplotlib.use("Agg")
def _sample_df() -> pd.DataFrame:
data = [
{
"interaction_id": "id1",
"datetime_start": datetime(2024, 1, 1, 10, 0),
"queue_skill": "ventas",
"channel": "voz",
"duration_talk": 600,
"hold_time": 60,
"wrap_up_time": 30,
},
{
"interaction_id": "id2",
"datetime_start": datetime(2024, 1, 1, 10, 5),
"queue_skill": "ventas",
"channel": "voz",
"duration_talk": 300,
"hold_time": 30,
"wrap_up_time": 20,
},
{
"interaction_id": "id3",
"datetime_start": datetime(2024, 1, 1, 11, 0),
"queue_skill": "soporte",
"channel": "chat",
"duration_talk": 400,
"hold_time": 20,
"wrap_up_time": 30,
},
]
return pd.DataFrame(data)
def test_init_and_required_columns():
df = _sample_df()
cfg = EconomyConfig(labor_cost_per_hour=20.0, overhead_rate=0.1, tech_costs_annual=10000.0)
em = EconomyCostMetrics(df, cfg)
assert not em.is_empty
# Falta de columna obligatoria -> ValueError
df_missing = df.drop(columns=["duration_talk"])
import pytest
with pytest.raises(ValueError):
EconomyCostMetrics(df_missing, cfg)
def test_metrics_without_config_do_not_crash():
df = _sample_df()
em = EconomyCostMetrics(df, None)
assert em.cpi_by_skill_channel().empty
assert em.annual_cost_by_skill_channel().empty
assert em.cost_breakdown() == {}
assert em.inefficiency_cost_by_skill_channel().empty
assert em.potential_savings() == {}
def test_basic_cpi_and_annual_cost():
df = _sample_df()
cfg = EconomyConfig(labor_cost_per_hour=20.0, overhead_rate=0.1)
em = EconomyCostMetrics(df, cfg)
cpi = em.cpi_by_skill_channel()
assert not cpi.empty
# Debe haber filas para ventas/voz y soporte/chat
assert ("ventas", "voz") in cpi.index
assert ("soporte", "chat") in cpi.index
annual = em.annual_cost_by_skill_channel()
assert "annual_cost" in annual.columns
# costes positivos
assert (annual["annual_cost"] > 0).any()
def test_cost_breakdown_and_potential_savings():
df = _sample_df()
cfg = EconomyConfig(
labor_cost_per_hour=20.0,
overhead_rate=0.1,
tech_costs_annual=5000.0,
automation_cpi=0.2,
automation_volume_share=0.5,
automation_success_rate=0.8,
)
em = EconomyCostMetrics(df, cfg)
breakdown = em.cost_breakdown()
assert "labor_pct" in breakdown
assert "overhead_pct" in breakdown
assert "tech_pct" in breakdown
total_pct = (
breakdown["labor_pct"]
+ breakdown["overhead_pct"]
+ breakdown["tech_pct"]
)
# Permitimos pequeño error por redondeo a 2 decimales
assert abs(total_pct - 100.0) < 0.2
savings = em.potential_savings()
assert "annual_savings" in savings
assert savings["annual_savings"] >= 0.0
def test_plot_methods_return_axes():
from matplotlib.axes import Axes
df = _sample_df()
cfg = EconomyConfig(labor_cost_per_hour=20.0, overhead_rate=0.1)
em = EconomyCostMetrics(df, cfg)
ax1 = em.plot_cost_waterfall()
ax2 = em.plot_cpi_by_channel()
assert isinstance(ax1, Axes)
assert isinstance(ax2, Axes)

238
backend/tests/test_operational_performance.py Normal file
View File

@@ -0,0 +1,238 @@
import math
from datetime import datetime, timedelta
import matplotlib
import numpy as np
import pandas as pd
from beyond_metrics.dimensions.OperationalPerformance import OperationalPerformanceMetrics
matplotlib.use("Agg")
def _sample_df() -> pd.DataFrame:
"""
Dataset sintético pequeño para probar la dimensión de rendimiento operacional.
Incluye:
- varios skills
- FCR, abandonos, transferencias
- reincidencia <7 días
- logged_time para occupancy
"""
base = datetime(2024, 1, 1, 10, 0, 0)
rows = [
# cliente C1, resolved, no abandon, voz, ventas
{
"interaction_id": "id1",
"datetime_start": base,
"queue_skill": "ventas",
"channel": "voz",
"duration_talk": 600,
"hold_time": 60,
"wrap_up_time": 30,
"agent_id": "A1",
"transfer_flag": 0,
"is_resolved": 1,
"abandoned_flag": 0,
"customer_id": "C1",
"logged_time": 900,
},
# C1 vuelve en 3 días mismo canal/skill
{
"interaction_id": "id2",
"datetime_start": base + timedelta(days=3),
"queue_skill": "ventas",
"channel": "voz",
"duration_talk": 700,
"hold_time": 30,
"wrap_up_time": 40,
"agent_id": "A1",
"transfer_flag": 1,
"is_resolved": 1,
"abandoned_flag": 0,
"customer_id": "C1",
"logged_time": 900,
},
# cliente C2, soporte, chat, no resuelto, transferido
{
"interaction_id": "id3",
"datetime_start": base + timedelta(hours=1),
"queue_skill": "soporte",
"channel": "chat",
"duration_talk": 400,
"hold_time": 20,
"wrap_up_time": 30,
"agent_id": "A2",
"transfer_flag": 1,
"is_resolved": 0,
"abandoned_flag": 0,
"customer_id": "C2",
"logged_time": 800,
},
# cliente C3, abandonado
{
"interaction_id": "id4",
"datetime_start": base + timedelta(hours=2),
"queue_skill": "soporte",
"channel": "voz",
"duration_talk": 100,
"hold_time": 50,
"wrap_up_time": 10,
"agent_id": "A2",
"transfer_flag": 0,
"is_resolved": 0,
"abandoned_flag": 1,
"customer_id": "C3",
"logged_time": 600,
},
# cliente C4, una sola interacción, email
{
"interaction_id": "id5",
"datetime_start": base + timedelta(days=10),
"queue_skill": "ventas",
"channel": "email",
"duration_talk": 300,
"hold_time": 0,
"wrap_up_time": 20,
"agent_id": "A1",
"transfer_flag": 0,
"is_resolved": 1,
"abandoned_flag": 0,
"customer_id": "C4",
"logged_time": 700,
},
]
return pd.DataFrame(rows)
# ----------------------------------------------------------------------
# Inicialización y validación básica
# ----------------------------------------------------------------------
def test_init_and_required_columns():
df = _sample_df()
op = OperationalPerformanceMetrics(df)
assert not op.is_empty
# Falta columna obligatoria -> ValueError
df_missing = df.drop(columns=["duration_talk"])
try:
OperationalPerformanceMetrics(df_missing)
assert False, "Debería lanzar ValueError si falta duration_talk"
except ValueError:
pass
# ----------------------------------------------------------------------
# AHT y distribución
# ----------------------------------------------------------------------
def test_aht_distribution_basic():
df = _sample_df()
op = OperationalPerformanceMetrics(df)
dist = op.aht_distribution()
assert "p10" in dist and "p50" in dist and "p90" in dist and "p90_p50_ratio" in dist
# Comprobamos que el ratio P90/P50 es razonable (>1)
assert dist["p90_p50_ratio"] >= 1.0
# ----------------------------------------------------------------------
# FCR, escalación, abandono
# ----------------------------------------------------------------------
def test_fcr_escalation_abandonment_rates():
df = _sample_df()
op = OperationalPerformanceMetrics(df)
fcr = op.fcr_rate()
esc = op.escalation_rate()
aband = op.abandonment_rate()
# FCR: interacciones resueltas / total
# is_resolved=1 en id1, id2, id5 -> 3 de 5
assert math.isclose(fcr, 60.0, rel_tol=1e-6)
# Escalación: transfer_flag=1 en id2, id3 -> 2 de 5
assert math.isclose(esc, 40.0, rel_tol=1e-6)
# Abandono: abandoned_flag=1 en id4 -> 1 de 5
assert math.isclose(aband, 20.0, rel_tol=1e-6)
# ----------------------------------------------------------------------
# Reincidencia y repetición de canal
# ----------------------------------------------------------------------
def test_recurrence_and_repeat_channel():
df = _sample_df()
op = OperationalPerformanceMetrics(df)
rec = op.recurrence_rate_7d()
rep = op.repeat_channel_rate()
# Clientes: C1, C2, C3, C4 -> 4 clientes
# Recurrente: C1 (tiene 2 contactos en 3 días). Solo 1 de 4 -> 25%
assert math.isclose(rec, 25.0, rel_tol=1e-6)
# Reincidencias (<7d):
# Solo el par de C1: voz -> voz, mismo canal => 100%
assert math.isclose(rep, 100.0, rel_tol=1e-6)
# ----------------------------------------------------------------------
# Occupancy
# ----------------------------------------------------------------------
def test_occupancy_rate():
df = _sample_df()
op = OperationalPerformanceMetrics(df)
occ = op.occupancy_rate()
# handle_time = (600+60+30) + (700+30+40) + (400+20+30) + (100+50+10) + (300+0+20)
# = 690 + 770 + 450 + 160 + 320 = 2390
# logged_time total = 900 + 900 + 800 + 600 + 700 = 3900
expected_occ = 2390 / 3900 * 100
assert math.isclose(occ, round(expected_occ, 2), rel_tol=1e-6)
# ----------------------------------------------------------------------
# Performance Score
# ----------------------------------------------------------------------
def test_performance_score_structure_and_range():
df = _sample_df()
op = OperationalPerformanceMetrics(df)
score_info = op.performance_score()
assert "score" in score_info
assert 0.0 <= score_info["score"] <= 10.0
# ----------------------------------------------------------------------
# Plots
# ----------------------------------------------------------------------
def test_plot_methods_return_axes():
df = _sample_df()
op = OperationalPerformanceMetrics(df)
ax1 = op.plot_aht_boxplot_by_skill()
ax2 = op.plot_resolution_funnel_by_skill()
from matplotlib.axes import Axes
assert isinstance(ax1, Axes)
assert isinstance(ax2, Axes)

200
backend/tests/test_satisfaction_experience.py Normal file
View File

@@ -0,0 +1,200 @@
import math
from datetime import datetime, timedelta
import pytest
import matplotlib
import numpy as np
import pandas as pd
from beyond_metrics.dimensions.SatisfactionExperience import SatisfactionExperienceMetrics
matplotlib.use("Agg")
def _sample_df_negative_corr() -> pd.DataFrame:
"""
Dataset sintético donde CSAT decrece claramente cuando AHT aumenta,
para que la correlación sea negativa (< -0.3).
"""
base = datetime(2024, 1, 1, 10, 0, 0)
rows = []
# AHT crece, CSAT baja
aht_values = [200, 300, 400, 500, 600, 700, 800, 900]
csat_values = [5.0, 4.7, 4.3, 3.8, 3.3, 2.8, 2.3, 2.0]
skills = ["ventas", "retencion"]
channels = ["voz", "chat"]
for i, (aht, csat) in enumerate(zip(aht_values, csat_values), start=1):
rows.append(
{
"interaction_id": f"id{i}",
"datetime_start": base + timedelta(minutes=5 * i),
"queue_skill": skills[i % len(skills)],
"channel": channels[i % len(channels)],
"csat_score": csat,
"duration_talk": aht * 0.7,
"hold_time": aht * 0.2,
"wrap_up_time": aht * 0.1,
}
)
return pd.DataFrame(rows)
def _sample_df_full() -> pd.DataFrame:
"""
Dataset más completo con NPS y CES para otras pruebas.
"""
base = datetime(2024, 1, 1, 10, 0, 0)
rows = []
for i in range(1, 11):
aht = 300 + 30 * i
csat = 3.0 + 0.1 * i # ligero incremento
nps = -20 + 5 * i
ces = 4.0 - 0.05 * i
rows.append(
{
"interaction_id": f"id{i}",
"datetime_start": base + timedelta(minutes=10 * i),
"queue_skill": "ventas" if i <= 5 else "retencion",
"channel": "voz" if i % 2 == 0 else "chat",
"csat_score": csat,
"duration_talk": aht * 0.7,
"hold_time": aht * 0.2,
"wrap_up_time": aht * 0.1,
"nps_score": nps,
"ces_score": ces,
}
)
return pd.DataFrame(rows)
# ----------------------------------------------------------------------
# Inicialización y validación
# ----------------------------------------------------------------------
def test_init_and_required_columns():
df = _sample_df_negative_corr()
sm = SatisfactionExperienceMetrics(df)
assert not sm.is_empty
# Quitar una columna REALMENTE obligatoria -> debe lanzar ValueError
df_missing = df.drop(columns=["duration_talk"])
with pytest.raises(ValueError):
SatisfactionExperienceMetrics(df_missing)
# Quitar csat_score ya NO debe romper: es opcional
df_no_csat = df.drop(columns=["csat_score"])
sm2 = SatisfactionExperienceMetrics(df_no_csat)
# simplemente no tendrá métricas de csat
assert sm2.is_empty is False
# ----------------------------------------------------------------------
# CSAT promedio y tablas
# ----------------------------------------------------------------------
def test_csat_avg_by_skill_channel():
df = _sample_df_full()
sm = SatisfactionExperienceMetrics(df)
table = sm.csat_avg_by_skill_channel()
# Debe tener al menos 2 skills y 2 canales
assert "ventas" in table.index
assert "retencion" in table.index
# Algún canal
assert any(col in table.columns for col in ["voz", "chat"])
def test_nps_and_ces_tables():
df = _sample_df_full()
sm = SatisfactionExperienceMetrics(df)
nps = sm.nps_avg_by_skill_channel()
ces = sm.ces_avg_by_skill_channel()
# Deben devolver DataFrame no vacío
assert not nps.empty
assert not ces.empty
assert "ventas" in nps.index
assert "ventas" in ces.index
# ----------------------------------------------------------------------
# Correlación CSAT vs AHT
# ----------------------------------------------------------------------
def test_csat_aht_correlation_negative():
df = _sample_df_negative_corr()
sm = SatisfactionExperienceMetrics(df)
corr = sm.csat_aht_correlation()
r = corr["r"]
code = corr["interpretation_code"]
assert r < -0.3
assert code == "negativo"
# ----------------------------------------------------------------------
# Clasificación por skill (sweet spot)
# ----------------------------------------------------------------------
def test_csat_aht_skill_summary_structure():
df = _sample_df_full()
sm = SatisfactionExperienceMetrics(df)
summary = sm.csat_aht_skill_summary()
assert "csat_avg" in summary.columns
assert "aht_avg" in summary.columns
assert "classification" in summary.columns
assert set(summary.index) == {"ventas", "retencion"}
# ----------------------------------------------------------------------
# Plots
# ----------------------------------------------------------------------
def test_plot_methods_return_axes():
df = _sample_df_full()
sm = SatisfactionExperienceMetrics(df)
ax1 = sm.plot_csat_vs_aht_scatter()
ax2 = sm.plot_csat_distribution()
from matplotlib.axes import Axes
assert isinstance(ax1, Axes)
assert isinstance(ax2, Axes)
def test_dataset_without_csat_does_not_break():
# Dataset “core” sin csat/nps/ces
df = pd.DataFrame(
{
"interaction_id": ["id1", "id2"],
"datetime_start": [datetime(2024, 1, 1, 10), datetime(2024, 1, 1, 11)],
"queue_skill": ["ventas", "soporte"],
"channel": ["voz", "chat"],
"duration_talk": [300, 400],
"hold_time": [30, 20],
"wrap_up_time": [20, 30],
}
)
sm = SatisfactionExperienceMetrics(df)
# No debe petar, simplemente devolver vacío/NaN
assert sm.csat_avg_by_skill_channel().empty
corr = sm.csat_aht_correlation()
assert math.isnan(corr["r"])

221
backend/tests/test_volumetria.py Normal file
View File

@@ -0,0 +1,221 @@
import math
from datetime import datetime
import matplotlib
import pandas as pd
from beyond_metrics.dimensions.Volumetria import VolumetriaMetrics
# Usamos backend "Agg" para que matplotlib no intente abrir ventanas
matplotlib.use("Agg")
def _sample_df() -> pd.DataFrame:
"""
DataFrame de prueba con el nuevo esquema de columnas:
Campos usados por VolumetriaMetrics:
- interaction_id
- datetime_start
- queue_skill
- channel
5 interacciones:
- 3 por canal "voz", 2 por canal "chat"
- 3 en skill "ventas", 2 en skill "soporte"
- 3 en enero, 2 en febrero
"""
data = [
{
"interaction_id": "id1",
"datetime_start": datetime(2024, 1, 1, 9, 0),
"queue_skill": "ventas",
"channel": "voz",
},
{
"interaction_id": "id2",
"datetime_start": datetime(2024, 1, 1, 9, 30),
"queue_skill": "ventas",
"channel": "voz",
},
{
"interaction_id": "id3",
"datetime_start": datetime(2024, 1, 1, 10, 0),
"queue_skill": "soporte",
"channel": "voz",
},
{
"interaction_id": "id4",
"datetime_start": datetime(2024, 2, 1, 10, 0),
"queue_skill": "ventas",
"channel": "chat",
},
{
"interaction_id": "id5",
"datetime_start": datetime(2024, 2, 2, 11, 0),
"queue_skill": "soporte",
"channel": "chat",
},
]
return pd.DataFrame(data)
# ----------------------------------------------------------------------
# VALIDACIÓN BÁSICA
# ----------------------------------------------------------------------
def test_init_validates_required_columns():
df = _sample_df()
# No debe lanzar error con las columnas por defecto
vm = VolumetriaMetrics(df)
assert not vm.is_empty
# Si falta alguna columna requerida, debe lanzar ValueError
for col in ["interaction_id", "datetime_start", "queue_skill", "channel"]:
df_missing = df.drop(columns=[col])
try:
VolumetriaMetrics(df_missing)
assert False, f"Debería fallar al faltar la columna: {col}"
except ValueError:
pass
# ----------------------------------------------------------------------
# VOLUMEN Y DISTRIBUCIONES
# ----------------------------------------------------------------------
def test_volume_by_channel_and_skill():
df = _sample_df()
vm = VolumetriaMetrics(df)
vol_channel = vm.volume_by_channel()
vol_skill = vm.volume_by_skill()
# Canales
assert vol_channel.sum() == len(df)
assert vol_channel["voz"] == 3
assert vol_channel["chat"] == 2
# Skills
assert vol_skill.sum() == len(df)
assert vol_skill["ventas"] == 3
assert vol_skill["soporte"] == 2
def test_channel_and_skill_distribution_pct():
df = _sample_df()
vm = VolumetriaMetrics(df)
dist_channel = vm.channel_distribution_pct()
dist_skill = vm.skill_distribution_pct()
# 3/5 = 60%, 2/5 = 40%
assert math.isclose(dist_channel["voz"], 60.0, rel_tol=1e-6)
assert math.isclose(dist_channel["chat"], 40.0, rel_tol=1e-6)
assert math.isclose(dist_skill["ventas"], 60.0, rel_tol=1e-6)
assert math.isclose(dist_skill["soporte"], 40.0, rel_tol=1e-6)
# ----------------------------------------------------------------------
# HEATMAP Y SAZONALIDAD
# ----------------------------------------------------------------------
def test_heatmap_24x7_shape_and_values():
df = _sample_df()
vm = VolumetriaMetrics(df)
heatmap = vm.heatmap_24x7()
# 7 días x 24 horas
assert heatmap.shape == (7, 24)
# Comprobamos algunas celdas concretas
# 2024-01-01 es lunes (dayofweek=0), llamadas a las 9h (2) y 10h (1)
assert heatmap.loc[0, 9] == 2
assert heatmap.loc[0, 10] == 1
# 2024-02-01 es jueves (dayofweek=3), 10h
assert heatmap.loc[3, 10] == 1
# 2024-02-02 es viernes (dayofweek=4), 11h
assert heatmap.loc[4, 11] == 1
def test_monthly_seasonality_cv():
df = _sample_df()
vm = VolumetriaMetrics(df)
cv = vm.monthly_seasonality_cv()
# Volumen mensual: [3, 2]
# mean = 2.5, std (ddof=1) ≈ 0.7071 -> CV ≈ 28.28%
assert math.isclose(cv, 28.28, rel_tol=1e-2)
def test_peak_offpeak_ratio():
df = _sample_df()
vm = VolumetriaMetrics(df)
ratio = vm.peak_offpeak_ratio()
# Horas pico definidas en la clase: 10-19
# Pico: 10h,10h,11h -> 3 interacciones
# Valle: 9h,9h -> 2 interacciones
# Ratio = 3/2 = 1.5
assert math.isclose(ratio, 1.5, rel_tol=1e-6)
def test_concentration_top20_skills_pct():
df = _sample_df()
vm = VolumetriaMetrics(df)
conc = vm.concentration_top20_skills_pct()
# Skills: ventas=3, soporte=2, total=5
# Top 20% de skills (ceil(0.2 * 2) = 1 skill) -> ventas=3
# 3/5 = 60%
assert math.isclose(conc, 60.0, rel_tol=1e-6)
# ----------------------------------------------------------------------
# CASO DATAFRAME VACÍO
# ----------------------------------------------------------------------
def test_empty_dataframe_behaviour():
df_empty = pd.DataFrame(
columns=["interaction_id", "datetime_start", "queue_skill", "channel"]
)
vm = VolumetriaMetrics(df_empty)
assert vm.is_empty
assert vm.volume_by_channel().empty
assert vm.volume_by_skill().empty
assert math.isnan(vm.monthly_seasonality_cv())
assert math.isnan(vm.peak_offpeak_ratio())
assert math.isnan(vm.concentration_top20_skills_pct())
# ----------------------------------------------------------------------
# PLOTS
# ----------------------------------------------------------------------
def test_plot_methods_return_axes():
df = _sample_df()
vm = VolumetriaMetrics(df)
ax1 = vm.plot_heatmap_24x7()
ax2 = vm.plot_channel_distribution()
ax3 = vm.plot_skill_pareto()
from matplotlib.axes import Axes
assert isinstance(ax1, Axes)
assert isinstance(ax2, Axes)
assert isinstance(ax3, Axes)