"""Tests for the /api/gestion4sat/enriquecer-ruts endpoints (masiva SII). Covered: - Auth guard on POST /enriquecer-ruts and GET /enriquecer-ruts/estado - GET /enriquecer-ruts/estado returns latest job with expected fields - POST /enriquecer-ruts triggers a background job when there are empresas with empty razon_social (returns {ok, job_id, total, message}) - Background job completes (estado='completado') and empresa's razon_social is populated from cache/SII - POST /enriquecer-ruts returns {ok:true, total:0, message} when there are no empresas pending - Duplicate enrichment while one is running returns {ok:false, message, job_id} RUT 77.155.297-8 is expected to be cached in cache_rut_sii collection (razon_social='SOLUCIONES RGA SPA'). """ import os import time import pytest import requests BASE_URL = os.environ.get("REACT_APP_BACKEND_URL", "https://doc-manager-111.preview.emergentagent.com").rstrip("/") CACHED_RUT_RAW = "77.155.297-8" # cached in cache_rut_sii CACHED_RUT_NORM = "77155297-8" EXPECTED_RAZON = "SOLUCIONES RGA SPA" TEST_NOMBRE_CLIENTE = "TEST_ENRICH_CACHED_RGA" # ---------- fixtures ---------- @pytest.fixture(scope="module") def token_admin4sat(): r = requests.post( f"{BASE_URL}/api/auth/login", json={"email": "admin4sat@demo.com", "password": "Test1234!"}, timeout=15, ) if r.status_code != 200: pytest.skip(f"login admin4sat failed: {r.status_code} {r.text}") data = r.json() return data.get("access_token") or data.get("token") @pytest.fixture(scope="module") def headers(token_admin4sat): return {"Authorization": f"Bearer {token_admin4sat}", "Content-Type": "application/json"} def _find_empresa_by_rut(headers, rut_fmt): # Use ?q= to search server-side by rut r = requests.get( f"{BASE_URL}/api/gestion4sat/empresas", headers=headers, params={"q": rut_fmt, "limit": 50}, timeout=15, ) if r.status_code != 200: return None empresas = r.json().get("empresas", []) for e in empresas: if e.get("rut_formateado") == rut_fmt or e.get("rut") == rut_fmt: return e return None def _wait_for_estado(headers, estado_target, timeout=60): """Poll status endpoint until estado matches (or times out).""" deadline = time.time() + timeout last = None while time.time() < deadline: r = requests.get(f"{BASE_URL}/api/gestion4sat/enriquecer-ruts/estado", headers=headers, timeout=15) if r.status_code == 200: last = r.json() if last.get("estado") == estado_target: return last time.sleep(2) return last @pytest.fixture(scope="module") def test_empresa(headers): """Create (or reuse) an empresa with cached RUT and empty razon_social so the enrichment has something to do. Returns the empresa dict. Post-test: set its razon_social so subsequent tests can assert 'no pendientes'. """ existing = _find_empresa_by_rut(headers, CACHED_RUT_RAW) if existing: emp_id = existing["id"] # Force razon_social empty before test requests.put( f"{BASE_URL}/api/gestion4sat/empresas/{emp_id}", headers=headers, json={"razon_social": ""}, timeout=15, ) empresa = existing empresa["razon_social"] = "" else: payload = { "nombre_cliente": TEST_NOMBRE_CLIENTE, "razon_social": "", # empty -> pending for enrichment "rut": CACHED_RUT_RAW, "tipo_servicio": "gps_solo", "precio_mensualidad": 0, "notas": "created by test_enriquecer_ruts.py", } r = requests.post(f"{BASE_URL}/api/gestion4sat/empresas", headers=headers, json=payload, timeout=15) if r.status_code != 200: pytest.skip(f"cannot create test empresa: {r.status_code} {r.text}") empresa = r.json() yield empresa # cleanup: mark razon_social so it stops being 'pendiente' try: requests.put( f"{BASE_URL}/api/gestion4sat/empresas/{empresa['id']}", headers=headers, json={"razon_social": EXPECTED_RAZON}, timeout=15, ) except Exception: pass # ---------- Tests: auth guards ---------- class TestAuthGuards: def test_post_enriquecer_requires_auth(self): r = requests.post(f"{BASE_URL}/api/gestion4sat/enriquecer-ruts", timeout=10) assert r.status_code in (401, 403), r.status_code def test_get_estado_requires_auth(self): r = requests.get(f"{BASE_URL}/api/gestion4sat/enriquecer-ruts/estado", timeout=10) assert r.status_code in (401, 403), r.status_code # ---------- Tests: estado endpoint ---------- class TestEstadoEndpoint: def test_estado_returns_job_or_sin_datos(self, headers): r = requests.get(f"{BASE_URL}/api/gestion4sat/enriquecer-ruts/estado", headers=headers, timeout=15) assert r.status_code == 200, r.text data = r.json() assert "estado" in data # Either a real job with progress fields OR sin_datos if data["estado"] != "sin_datos": for key in ("total", "procesados", "actualizados", "no_encontrados", "errores", "detalle"): assert key in data, f"missing field {key} in {data}" assert isinstance(data["detalle"], list) # ---------- Tests: enrichment flow ---------- class TestEnriquecimientoFlow: def test_trigger_enrichment_and_verify_completion(self, headers, test_empresa): """Full flow: create pending empresa → trigger → poll → verify DB update.""" # sanity: ensure our empresa has empty razon_social right now emp_before = _find_empresa_by_rut(headers, CACHED_RUT_RAW) assert emp_before is not None, "test empresa missing" assert not emp_before.get("razon_social"), f"expected empty razon_social, got {emp_before.get('razon_social')}" # Trigger enrichment r = requests.post(f"{BASE_URL}/api/gestion4sat/enriquecer-ruts", headers=headers, timeout=15) assert r.status_code == 200, r.text data = r.json() # Either it started a new job (ok:true) OR a previous job is still running assert data.get("ok") in (True, False) if data.get("ok") is False: # Should mean already running assert "en proceso" in data.get("message", "").lower() # wait for it to complete else: assert "job_id" in data assert isinstance(data.get("total"), int) and data["total"] >= 1 assert "message" in data # Poll for completion final = _wait_for_estado(headers, "completado", timeout=90) assert final is not None, "no status returned" assert final.get("estado") == "completado", f"job did not complete: {final}" assert final.get("procesados") == final.get("total") assert isinstance(final.get("detalle"), list) # Verify our empresa now has razon_social emp_after = _find_empresa_by_rut(headers, CACHED_RUT_RAW) assert emp_after is not None assert emp_after.get("razon_social", "").upper() == EXPECTED_RAZON.upper(), ( f"empresa was not updated: {emp_after.get('razon_social')!r}" ) # Verify detalle contains our RUT (may be truncated to last 20) detalle_ruts = {d.get("rut") for d in final.get("detalle", [])} # not strictly guaranteed to be there (truncation), but if we only had # 1-few pending it should be present if final.get("total", 0) <= 20: assert CACHED_RUT_RAW in detalle_ruts, f"RUT missing from detalle: {detalle_ruts}" def test_no_pending_returns_total_zero(self, headers, test_empresa): """After all empresas have razon_social, enrichment should return total=0.""" # Ensure every empresa in DB has some razon_social so total should be 0. # Loop empresas and PUT any missing. r = requests.get( f"{BASE_URL}/api/gestion4sat/empresas", headers=headers, params={"limit": 2000}, timeout=30, ) assert r.status_code == 200 touched = [] for e in r.json().get("empresas", []): if not e.get("razon_social"): requests.put( f"{BASE_URL}/api/gestion4sat/empresas/{e['id']}", headers=headers, json={"razon_social": "PLACEHOLDER TEST"}, timeout=15, ) touched.append(e["id"]) try: # Wait a moment in case any in-flight job exists _wait_for_estado(headers, "completado", timeout=30) r = requests.post(f"{BASE_URL}/api/gestion4sat/enriquecer-ruts", headers=headers, timeout=15) assert r.status_code == 200, r.text data = r.json() # Expected: {ok: true, message: 'Todas las empresas ya tienen razón social', total: 0} assert data.get("ok") is True, f"expected ok:true, got {data}" assert data.get("total", -1) == 0, f"expected total 0, got {data}" assert "message" in data assert "razón social" in data["message"].lower() or "razon social" in data["message"].lower() finally: # Revert placeholders back to empty so we don't pollute for eid in touched: requests.put( f"{BASE_URL}/api/gestion4sat/empresas/{eid}", headers=headers, json={"razon_social": ""}, timeout=15, ) if __name__ == "__main__": pytest.main([__file__, "-v"])