# unit_tests

id: normalize_zip_filenames_py_infra_t1
function_id: normalize_zip_filenames_py_infra
name: test_zip_con_path_traversal
file_path: python/functions/infra/safe_extract_zip_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_zip_con_path_traversal():
      """ZIP con path traversal lanza ValueError."""
      zip_path = _make_zip_with_traversal("../../etc/passwd")
      try:
          with tempfile.TemporaryDirectory() as dest:
              raised = False
              try:
                  safe_extract_zip(zip_path, dest)
              except ValueError as e:
                  raised = True
                  assert "Zip Slip" in str(e)
              assert raised, "Expected ValueError for path traversal"
      finally:
          os.unlink(zip_path)


id: normalize_zip_filenames_py_infra_t2
function_id: normalize_zip_filenames_py_infra
name: test_zip_con_paths_absolutos
file_path: python/functions/infra/safe_extract_zip_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_zip_con_paths_absolutos():
      """ZIP con paths absolutos lanza ValueError."""
      zip_path = _make_zip_with_traversal("/etc/passwd")
      try:
          with tempfile.TemporaryDirectory() as dest:
              raised = False
              try:
                  safe_extract_zip(zip_path, dest)
              except ValueError as e:
                  raised = True
                  assert "Zip Slip" in str(e)
              assert raised, "Expected ValueError for absolute path"
      finally:
          os.unlink(zip_path)


id: normalize_zip_filenames_py_infra_t3
function_id: normalize_zip_filenames_py_infra
name: test_normalize_utf8_correctos_no_cambian
file_path: python/functions/infra/safe_extract_zip_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_normalize_utf8_correctos_no_cambian():
      """ZIP con nombres UTF-8 correctos (flag 0x800) no se modifican."""
      buf = io.BytesIO()
      with zipfile.ZipFile(buf, "w") as zipf:
          info = zipfile.ZipInfo("archivo_normal.txt")
          info.flag_bits |= 0x800  # marcar como UTF-8
          zipf.writestr(info, b"content")
      buf.seek(0)
      with zipfile.ZipFile(buf, "r") as zipf:
          original_name = zipf.infolist()[0].filename
          normalize_zip_filenames(zipf)
          assert zipf.infolist()[0].filename == original_name


id: normalize_zip_filenames_py_infra_t4
function_id: normalize_zip_filenames_py_infra
name: test_normalize_cjk_mojibake_repara
file_path: python/functions/infra/safe_extract_zip_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_normalize_cjk_mojibake_repara():
      """ZIP con nombres CJK en mojibake (UTF-8 bytes leidos como CP437) se reparan.

      Simula un ZIP donde los bytes del filename son UTF-8 valido de un nombre CJK,
      pero el flag 0x800 no esta seteado, asi que zipfile los decodifica como CP437
      produciendo mojibake. normalize_zip_filenames debe detectarlo y repararlo.
      """
      cjk_name = "\u6587\u4ef6.txt"  # 文件.txt

      # Construir ZIP con bytes UTF-8 crudos en el campo filename, sin flag 0x800.
      # Python no permite esto via ZipInfo (fuerza 0x800 para non-ASCII), por eso
      # construimos el ZIP manualmente con _make_zip_with_raw_filename.
      utf8_bytes = cjk_name.encode("utf-8")
      zip_bytes = _make_zip_with_raw_filename(utf8_bytes, b"cjk content")

      with zipfile.ZipFile(io.BytesIO(zip_bytes), "r") as zipf:
          member = zipf.infolist()[0]
          # Sin el flag, zipfile lee los bytes como CP437: debe ser mojibake
          assert not (member.flag_bits & 0x800), "Flag 0x800 no deberia estar seteado"
          assert member.filename != cjk_name, "El nombre aun no debe estar reparado"

          normalize_zip_filenames(zipf)
          repaired = zipf.infolist()[0].filename
          has_cjk = any(
              "\u4e00" <= c <= "\u9fff" or "\u3400" <= c <= "\u4dbf" for c in repaired
          )
          assert has_cjk, f"Esperaba CJK en nombre reparado, got: {repaired!r}"


  if __name__ == "__main__":
      test_zip_normal()
      print("PASS: ZIP normal extrae correctamente dentro del destino")

      test_zip_con_path_traversal()
      print("PASS: ZIP con path traversal lanza ValueError")

      test_zip_con_paths_absolutos()
      print("PASS: ZIP con paths absolutos lanza ValueError")

      test_normalize_utf8_correctos_no_cambian()
      print("PASS: ZIP con nombres UTF-8 correctos no se modifican")

      test_normalize_cjk_mojibake_repara()
      print("PASS: ZIP con nombres CJK mojibake se reparan")

      print("\nAll tests passed.")


id: pdf_add_page_py_infra_t0
function_id: pdf_add_page_py_infra
name: test_anadir_pagina_portrait_a4
file_path: python/functions/infra/pdf_add_page_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_anadir_pagina_portrait_a4():
      doc = pdf_create()
      doc = pdf_add_page(doc, orientation="portrait")
      assert len(doc.pages) == 1
      assert doc.pages[0]["orientation"] == "portrait"
      assert doc.fpdf.page == 1


id: pdf_add_page_py_infra_t1
function_id: pdf_add_page_py_infra
name: test_anadir_pagina_landscape
file_path: python/functions/infra/pdf_add_page_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_anadir_pagina_landscape():
      doc = pdf_create()
      doc = pdf_add_page(doc, orientation="landscape")
      assert len(doc.pages) == 1
      assert doc.pages[0]["orientation"] == "landscape"


id: pdf_add_page_py_infra_t2
function_id: pdf_add_page_py_infra
name: test_pagina_con_dimensiones_personalizadas
file_path: python/functions/infra/pdf_add_page_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_pagina_con_dimensiones_personalizadas():
      doc = pdf_create()
      doc = pdf_add_page(doc, width=148.0, height=210.0)  # A5
      assert len(doc.pages) == 1
      # Verificar que se añadio la pagina
      assert doc.fpdf.page == 1


id: pdf_add_table_py_infra_t0
function_id: pdf_add_table_py_infra
name: test_tabla_con_headers_y_datos
file_path: python/functions/infra/pdf_add_table_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_tabla_con_headers_y_datos():
      doc = pdf_create()
      doc = pdf_add_page(doc)
      headers = ["Dominio", "Funciones", "Tipos"]
      rows = [
          ["core", "45", "12"],
          ["infra", "38", "8"],
          ["finance", "22", "6"],
      ]
      doc = pdf_add_table(doc, headers, rows)
      pdf_bytes = pdf_save(doc)
      assert pdf_bytes[:4] == b"%PDF"
      assert len(pdf_bytes) > 200


id: pdf_add_table_py_infra_t1
function_id: pdf_add_table_py_infra
name: test_tabla_con_col_widths_personalizados
file_path: python/functions/infra/pdf_add_table_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_tabla_con_col_widths_personalizados():
      doc = pdf_create()
      doc = pdf_add_page(doc)
      headers = ["Nombre", "Valor"]
      rows = [["alpha", "100"], ["beta", "200"]]
      doc = pdf_add_table(doc, headers, rows, col_widths=[100.0, 70.0])
      pdf_bytes = pdf_save(doc)
      assert pdf_bytes[:4] == b"%PDF"


id: pdf_add_table_py_infra_t2
function_id: pdf_add_table_py_infra
name: test_tabla_sin_borde
file_path: python/functions/infra/pdf_add_table_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_tabla_sin_borde():
      doc = pdf_create()
      doc = pdf_add_page(doc)
      headers = ["A", "B"]
      rows = [["1", "2"]]
      doc = pdf_add_table(doc, headers, rows, border=0)
      pdf_bytes = pdf_save(doc)
      assert pdf_bytes[:4] == b"%PDF"


id: pdf_add_text_py_infra_t0
function_id: pdf_add_text_py_infra
name: test_texto_simple_con_estilo_defecto
file_path: python/functions/infra/pdf_add_text_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_texto_simple_con_estilo_defecto():
      doc = pdf_create(title="Test")
      doc = pdf_add_page(doc)
      doc = pdf_add_text(doc, "Hola mundo")
      assert isinstance(doc, PDFDoc)
      # El PDF debe ser valido
      pdf_bytes = pdf_save(doc)
      assert pdf_bytes[:4] == b"%PDF"


id: pdf_add_text_py_infra_t1
function_id: pdf_add_text_py_infra
name: test_texto_con_estilo_personalizado_bold_centrado
file_path: python/functions/infra/pdf_add_text_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_texto_con_estilo_personalizado_bold_centrado():
      doc = pdf_create()
      doc = pdf_add_page(doc)
      style = PDFStyle(font_family="Helvetica-Bold", font_size=18.0, alignment="center", color="#FF0000")
      doc = pdf_add_text(doc, "Titulo en Rojo", style=style)
      pdf_bytes = pdf_save(doc)
      assert pdf_bytes[:4] == b"%PDF"


id: pdf_add_text_py_infra_t2
function_id: pdf_add_text_py_infra
name: test_texto_multilinea_con_saltos
file_path: python/functions/infra/pdf_add_text_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_texto_multilinea_con_saltos():
      doc = pdf_create()
      doc = pdf_add_page(doc)
      doc = pdf_add_text(doc, "Linea 1\nLinea 2\nLinea 3")
      pdf_bytes = pdf_save(doc)
      assert pdf_bytes[:4] == b"%PDF"
      assert len(pdf_bytes) > 100


id: pdf_create_py_infra_t0
function_id: pdf_create_py_infra
name: test_crear_documento_con_titulo_y_autor
file_path: python/functions/infra/pdf_create_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_crear_documento_con_titulo_y_autor():
      doc = pdf_create(title="Test Title", author="Test Author")
      assert isinstance(doc, PDFDoc)
      assert doc.title == "Test Title"
      assert doc.author == "Test Author"
      assert doc.fpdf is not None


id: pdf_create_py_infra_t1
function_id: pdf_create_py_infra
name: test_margenes_personalizados_aplicados
file_path: python/functions/infra/pdf_create_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_margenes_personalizados_aplicados():
      doc = pdf_create(margin_left=15.0, margin_right=15.0, margin_top=25.0, margin_bottom=25.0)
      assert doc.margin_left == 15.0
      assert doc.margin_right == 15.0
      assert doc.margin_top == 25.0
      assert doc.margin_bottom == 25.0
      # Verificar que fpdf tiene los margenes aplicados
      assert abs(doc.fpdf.l_margin - 15.0) < 0.1
      assert abs(doc.fpdf.r_margin - 15.0) < 0.1


id: pdf_merge_py_infra_t0
function_id: pdf_merge_py_infra
name: test_fusionar_dos_pdfs
file_path: python/functions/infra/pdf_merge_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_fusionar_dos_pdfs():
      with tempfile.TemporaryDirectory() as tmpdir:
          p1 = _create_pdf(os.path.join(tmpdir, "a.pdf"), "Documento A")
          p2 = _create_pdf(os.path.join(tmpdir, "b.pdf"), "Documento B")
          out = os.path.join(tmpdir, "merged.pdf")
          result = pdf_merge([p1, p2], out)
          assert result == out
          assert os.path.exists(out)
          # Verificar que el PDF combinado es valido y mas grande que cada parte
          with open(out, "rb") as f:
              content = f.read()
          assert content[:4] == b"%PDF"


id: pdf_merge_py_infra_t1
function_id: pdf_merge_py_infra
name: test_error_si_archivo_no_existe
file_path: python/functions/infra/pdf_merge_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_error_si_archivo_no_existe():
      with tempfile.TemporaryDirectory() as tmpdir:
          out = os.path.join(tmpdir, "merged.pdf")
          with pytest.raises(FileNotFoundError):
              pdf_merge(["/no/existe.pdf"], out)


id: pdf_merge_py_infra_t2
function_id: pdf_merge_py_infra
name: test_error_si_lista_vacia
file_path: python/functions/infra/pdf_merge_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_error_si_lista_vacia():
      with tempfile.TemporaryDirectory() as tmpdir:
          out = os.path.join(tmpdir, "merged.pdf")
          with pytest.raises(ValueError):
              pdf_merge([], out)


id: pdf_save_py_infra_t0
function_id: pdf_save_py_infra
name: test_guardar_pdf_a_archivo
file_path: python/functions/infra/pdf_save_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_guardar_pdf_a_archivo():
      doc = _make_simple_doc()
      with tempfile.TemporaryDirectory() as tmpdir:
          path = os.path.join(tmpdir, "test.pdf")
          result = pdf_save(doc, path)
          assert result == path
          assert os.path.exists(path)
          assert os.path.getsize(path) > 0


id: pdf_save_py_infra_t1
function_id: pdf_save_py_infra
name: test_retornar_bytes_del_pdf
file_path: python/functions/infra/pdf_save_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_retornar_bytes_del_pdf():
      doc = _make_simple_doc()
      result = pdf_save(doc)
      assert isinstance(result, bytes)
      assert len(result) > 0


id: pdf_save_py_infra_t2
function_id: pdf_save_py_infra
name: test_pdf_valido_comienza_con_pdf
file_path: python/functions/infra/pdf_save_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_pdf_valido_comienza_con_pdf():
      doc = _make_simple_doc()
      pdf_bytes = pdf_save(doc)
      assert pdf_bytes[:4] == b"%PDF"


id: read_file_with_encoding_py_infra_t0
function_id: read_file_with_encoding_py_infra
name: test_archivo_utf8
file_path: python/functions/infra/read_file_with_encoding_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_archivo_utf8():
      texto = "Hola mundo con acentos: áéíóú"
      path = _write_bytes(texto.encode("utf-8"))
      try:
          result = read_file_with_encoding(path)
          assert result == texto
      finally:
          os.unlink(path)


id: read_file_with_encoding_py_infra_t1
function_id: read_file_with_encoding_py_infra
name: test_archivo_utf8_con_bom
file_path: python/functions/infra/read_file_with_encoding_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_archivo_utf8_con_bom():
      texto = "Contenido con BOM"
      path = _write_bytes(texto.encode("utf-8-sig"))
      try:
          # Usando utf-8-sig explicitamente para que el BOM sea eliminado
          result = read_file_with_encoding(path, encodings=["utf-8-sig"])
          assert result == texto
      finally:
          os.unlink(path)


id: read_file_with_encoding_py_infra_t2
function_id: read_file_with_encoding_py_infra
name: test_archivo_latin1
file_path: python/functions/infra/read_file_with_encoding_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_archivo_latin1():
      texto = "Texto en Latin-1: café"
      path = _write_bytes(texto.encode("latin-1"))
      try:
          result = read_file_with_encoding(path)
          assert result == texto
      finally:
          os.unlink(path)


id: read_file_with_encoding_py_infra_t3
function_id: read_file_with_encoding_py_infra
name: test_archivo_binario_falla
file_path: python/functions/infra/read_file_with_encoding_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_archivo_binario_falla():
      # Bytes que no son validos en ningun encoding de texto comun
      path = _write_bytes(bytes([0x80, 0x81, 0x82, 0x83, 0xFF, 0xFE, 0x00, 0x01]))
      try:
          with pytest.raises(ValueError, match="Unable to decode file"):
              # Forzar solo encodings estrictos para que falle con binario puro
              read_file_with_encoding(path, encodings=["utf-8", "utf-8-sig"])
      finally:
          os.unlink(path)


id: read_file_with_encoding_py_infra_t4
function_id: read_file_with_encoding_py_infra
name: test_encodings_personalizados
file_path: python/functions/infra/read_file_with_encoding_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_encodings_personalizados():
      texto = "Windows text: Ñoño"
      path = _write_bytes(texto.encode("cp1252"))
      try:
          result = read_file_with_encoding(path, encodings=["cp1252"])
          assert result == texto
      finally:
          os.unlink(path)


id: read_file_with_encoding_py_infra_t5
function_id: read_file_with_encoding_py_infra
name: test_archivo_no_existe
file_path: python/functions/infra/read_file_with_encoding_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_archivo_no_existe():
      with pytest.raises(FileNotFoundError):
          read_file_with_encoding("/tmp/archivo_que_no_existe_12345.txt")


id: scan_directory_py_infra_t0
function_id: scan_directory_py_infra
name: test_directorio_con_mezcla_de_archivos
file_path: python/functions/infra/scan_directory_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_directorio_con_mezcla_de_archivos():
      with tempfile.TemporaryDirectory() as tmp:
          root = Path(tmp)
          _make_tree(root, {
              "report.pdf": "pdf content",
              "notes.md": "# Notes",
              "image.png": "png content",
              "data.csv": "a,b,c",
          })

          result = scan_directory(str(root), supported_extensions={".pdf", ".md"})

          rel_paths = [f.rel_path for f in result.processable]
          assert "notes.md" in rel_paths, f"notes.md no en processable: {rel_paths}"
          assert "report.pdf" in rel_paths, f"report.pdf no en processable: {rel_paths}"

          unsup_paths = [f.rel_path for f in result.unsupported]
          assert "image.png" in unsup_paths, f"image.png no en unsupported: {unsup_paths}"
          assert "data.csv" in unsup_paths, f"data.csv no en unsupported: {unsup_paths}"

          assert all(f.classification == "processable" for f in result.processable)
          assert all(f.classification == "unsupported" for f in result.unsupported)


  # ---------------------------------------------------------------------------
  # Test: directorio con dot files
  # ---------------------------------------------------------------------------


id: scan_directory_py_infra_t1
function_id: scan_directory_py_infra
name: test_directorio_con_dot_files
file_path: python/functions/infra/scan_directory_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_directorio_con_dot_files():
      with tempfile.TemporaryDirectory() as tmp:
          root = Path(tmp)
          _make_tree(root, {
              "visible.txt": "content",
              ".hidden": "hidden content",
              ".env": "SECRET=x",
          })

          result = scan_directory(str(root))

          all_paths = [f.rel_path for f in result.processable + result.unsupported]
          assert ".hidden" not in all_paths, f".hidden no deberia aparecer: {all_paths}"
          assert ".env" not in all_paths, f".env no deberia aparecer: {all_paths}"
          assert "visible.txt" in all_paths, f"visible.txt deberia aparecer: {all_paths}"

          skipped_paths = " ".join(result.skipped)
          assert ".hidden" in skipped_paths or ".env" in skipped_paths


  # ---------------------------------------------------------------------------
  # Test: directorio con subdirs ignorados
  # ---------------------------------------------------------------------------


id: scan_directory_py_infra_t2
function_id: scan_directory_py_infra
name: test_directorio_con_subdirs_ignorados
file_path: python/functions/infra/scan_directory_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_directorio_con_subdirs_ignorados():
      with tempfile.TemporaryDirectory() as tmp:
          root = Path(tmp)
          _make_tree(root, {
              "main.py": "print('hello')",
              "__pycache__/module.pyc": "bytecode",
              "node_modules/lib/index.js": "// js",
              ".git/config": "[core]",
              "src/utils.py": "def f(): pass",
          })

          result = scan_directory(str(root))

          all_rels = [f.rel_path for f in result.processable + result.unsupported]

          # Archivos dentro de dirs ignorados no deben aparecer
          assert not any("__pycache__" in r for r in all_rels), \
              f"__pycache__ no deberia estar en resultados: {all_rels}"
          assert not any("node_modules" in r for r in all_rels), \
              f"node_modules no deberia estar en resultados: {all_rels}"
          assert not any(".git" in r for r in all_rels), \
              f".git no deberia estar en resultados: {all_rels}"

          # Archivos fuera de dirs ignorados si deben aparecer
          assert "main.py" in all_rels, f"main.py deberia estar: {all_rels}"
          assert "src/utils.py" in all_rels, f"src/utils.py deberia estar: {all_rels}"


  # ---------------------------------------------------------------------------
  # Test: filtros include/exclude
  # ---------------------------------------------------------------------------


id: scan_directory_py_infra_t3
function_id: scan_directory_py_infra
name: test_filtros_include_exclude
file_path: python/functions/infra/scan_directory_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_filtros_include_exclude():
      with tempfile.TemporaryDirectory() as tmp:
          root = Path(tmp)
          _make_tree(root, {
              "report.pdf": "content",
              "notes.md": "notes",
              "image.png": "image",
              "drafts/draft.md": "draft",
              "temp.tmp": "tmp",
          })

          # Solo incluir .pdf y .md
          result = scan_directory(str(root), include="*.pdf,*.md")
          all_rels = [f.rel_path for f in result.processable + result.unsupported]
          assert "image.png" not in all_rels, f"image.png no deberia incluirse: {all_rels}"
          assert "temp.tmp" not in all_rels, f"temp.tmp no deberia incluirse: {all_rels}"
          assert "report.pdf" in all_rels
          assert "notes.md" in all_rels

          # Excluir path prefix drafts/ y extension .tmp
          result2 = scan_directory(str(root), exclude="drafts/,*.tmp")
          all_rels2 = [f.rel_path for f in result2.processable + result2.unsupported]
          assert "drafts/draft.md" not in all_rels2, \
              f"drafts/draft.md no deberia incluirse: {all_rels2}"
          assert "temp.tmp" not in all_rels2, f"temp.tmp no deberia incluirse: {all_rels2}"
          assert "report.pdf" in all_rels2


  # ---------------------------------------------------------------------------
  # Test: modo strict
  # ---------------------------------------------------------------------------


id: scan_directory_py_infra_t4
function_id: scan_directory_py_infra
name: test_modo_strict
file_path: python/functions/infra/scan_directory_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_modo_strict():
      with tempfile.TemporaryDirectory() as tmp:
          root = Path(tmp)
          _make_tree(root, {
              "doc.pdf": "content",
              "image.png": "image",
          })

          # strict=False no lanza error aunque haya unsupported
          result = scan_directory(str(root), supported_extensions={".pdf"}, strict=False)
          assert len(result.unsupported) == 1

          # strict=True lanza ValueError
          raised = False
          try:
              scan_directory(str(root), supported_extensions={".pdf"}, strict=True)
          except ValueError:
              raised = True
          assert raised, "strict=True deberia lanzar ValueError cuando hay unsupported"


  if __name__ == "__main__":
      test_directorio_con_mezcla_de_archivos()
      print("PASS: directorio con mezcla de archivos")

      test_directorio_con_dot_files()
      print("PASS: directorio con dot files")

      test_directorio_con_subdirs_ignorados()
      print("PASS: directorio con subdirs ignorados")

      test_filtros_include_exclude()
      print("PASS: filtros include/exclude")

      test_modo_strict()
      print("PASS: modo strict")

      print("\nAll tests passed.")


id: extraction_pipeline_py_pipelines_t0
function_id: extraction_pipeline_py_pipelines
name: test_documento_con_entidades_y_relaciones
file_path: python/functions/pipelines/extraction_pipeline_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_documento_con_entidades_y_relaciones():
      """documento con entidades y relaciones retorna ExtractionResult completo"""
      text = (
          "John Smith, a US national, operates the domain evil-corp.com. "
          "He was identified as the main administrator of the infrastructure."
      )
      with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False, encoding="utf-8") as f:
          f.write(text)
          tmp_path = f.name

      try:
          result = extraction_pipeline(
              file_path=tmp_path,
              entity_presets=ENTITY_PRESETS,
              relation_types=RELATION_TYPES,
              llm_chat_json=_llm_with_entities,
              chunk_size=500,
              chunk_overlap=50,
              confidence_threshold=0.5,
              dedup_threshold=0.85,
          )
          assert result is not None
          assert len(result.entities) >= 1
          assert result.stats.total_chunks >= 1
          assert result.stats.total_chars > 0
      finally:
          os.unlink(tmp_path)


id: extraction_pipeline_py_pipelines_t1
function_id: extraction_pipeline_py_pipelines
name: test_documento_vacio
file_path: python/functions/pipelines/extraction_pipeline_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_documento_vacio():
      """documento vacio retorna ExtractionResult con listas vacias"""
      with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False, encoding="utf-8") as f:
          f.write("")
          tmp_path = f.name

      try:
          result = extraction_pipeline(
              file_path=tmp_path,
              entity_presets=ENTITY_PRESETS,
              relation_types=RELATION_TYPES,
              llm_chat_json=_llm_empty,
          )
          assert result is not None
          assert result.entities == []
          assert result.relations == []
          assert result.stats.total_chunks == 0
      finally:
          os.unlink(tmp_path)


id: extraction_pipeline_py_pipelines_t2
function_id: extraction_pipeline_py_pipelines
name: test_documento_sin_entidades_detectables
file_path: python/functions/pipelines/extraction_pipeline_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_documento_sin_entidades_detectables():
      """documento sin entidades detectables retorna listas vacias"""
      text = "The weather is nice today. The sun shines brightly over the mountains."
      with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False, encoding="utf-8") as f:
          f.write(text)
          tmp_path = f.name

      try:
          result = extraction_pipeline(
              file_path=tmp_path,
              entity_presets=ENTITY_PRESETS,
              relation_types=RELATION_TYPES,
              llm_chat_json=_llm_empty,
              confidence_threshold=0.5,
          )
          assert result is not None
          assert result.entities == []
          assert result.relations == []
          assert result.stats.raw_entities_count == 0
      finally:
          os.unlink(tmp_path)


id: extraction_pipeline_py_pipelines_t3
function_id: extraction_pipeline_py_pipelines
name: test_archivo_no_encontrado_lanza_filenotfounderror
file_path: python/functions/pipelines/extraction_pipeline_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_archivo_no_encontrado_lanza_filenotfounderror():
      """archivo no encontrado lanza FileNotFoundError"""
      import pytest
      with pytest.raises(FileNotFoundError):
          extraction_pipeline(
              file_path="/tmp/no_existe_para_test_extraccion_pipeline.txt",
              entity_presets=ENTITY_PRESETS,
              relation_types=RELATION_TYPES,
              llm_chat_json=_llm_empty,
          )


id: extraction_pipeline_py_pipelines_t4
function_id: extraction_pipeline_py_pipelines
name: test_entity_presets_vacio_lanza_valueerror
file_path: python/functions/pipelines/extraction_pipeline_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_entity_presets_vacio_lanza_valueerror():
      """entity presets vacio lanza ValueError"""
      import pytest
      with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False, encoding="utf-8") as f:
          f.write("some text")
          tmp_path = f.name

      try:
          with pytest.raises(ValueError):
              extraction_pipeline(
                  file_path=tmp_path,
                  entity_presets=[],
                  relation_types=RELATION_TYPES,
                  llm_chat_json=_llm_empty,
              )
      finally:
          os.unlink(tmp_path)


id: extraction_pipeline_py_pipelines_t5
function_id: extraction_pipeline_py_pipelines
name: test_progress_callback_se_invoca
file_path: python/functions/pipelines/extraction_pipeline_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_progress_callback_se_invoca():
      """progress callback se invoca durante la ejecucion"""
      calls: list[tuple[str, float]] = []

      def _on_progress(msg: str, pct: float) -> None:
          calls.append((msg, pct))

      text = "John Smith operates evil-corp.com."
      with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False, encoding="utf-8") as f:
          f.write(text)
          tmp_path = f.name

      try:
          extraction_pipeline(
              file_path=tmp_path,
              entity_presets=ENTITY_PRESETS,
              relation_types=RELATION_TYPES,
              llm_chat_json=_llm_with_entities,
              on_progress=_on_progress,
          )
          assert len(calls) > 0
          messages = [c[0] for c in calls]
          assert any("Extracting" in m or "Done" in m or "Dedup" in m for m in messages)
      finally:
          os.unlink(tmp_path)


id: extraction_pipeline_py_pipelines_t6
function_id: extraction_pipeline_py_pipelines
name: test_stats_se_rellenan_correctamente
file_path: python/functions/pipelines/extraction_pipeline_test.py
lang: py
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
code:
  def test_stats_se_rellenan_correctamente():
      """stats se rellenan correctamente con conteos y tiempo"""
      text = "John Smith, a US national, operates the domain evil-corp.com."
      with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False, encoding="utf-8") as f:
          f.write(text)
          tmp_path = f.name

      try:
          result = extraction_pipeline(
              file_path=tmp_path,
              entity_presets=ENTITY_PRESETS,
              relation_types=RELATION_TYPES,
              llm_chat_json=_llm_with_entities,
          )
          assert result.stats.total_chars > 0
          assert result.stats.total_chunks >= 1
          assert result.stats.processing_time_seconds >= 0.0
      finally:
          os.unlink(tmp_path)


# projects

id: fn_monitoring
name: fn_monitoring
tags: ["monitoring","api","dashboard","sqlite","visualization"]
dir_path: projects/fn_monitoring
content_hash: 124049f04d0c7e4f6cce62983c5ffdbf7394bcf6740e573b2a256f4986eebe2c
created_at: 2026-04-13T00:02:28Z
updated_at: 2026-04-13T00:02:28Z
description:
  Monitoreo y visualizacion del estado del fn_registry. API HTTP read-only sobre las bases de datos SQLite y dashboard ImGui que consume la API.

documentation:
  ## Apps

  | App | Descripcion |
  |-----|-------------|
  | sqlite_api | API REST HTTP read-only para registry.db y operations.db (Go, net/http, puerto 8484) |
  | registry_dashboard | Dashboard ImGui con KPIs, charts y tablas del registry (C++, consume sqlite_api) |

  ## Arquitectura

  ```
  sqlite_api (Go, :8484)
    └── registry.db, apps/*/operations.db (read-only)
         ↑
  registry_dashboard (C++/ImGui)
    └── HTTP GET/POST → sqlite_api
    └── Fallback: SQLite directo si API no disponible
  ```


