# unit_tests

id: merge_graphs_py_datascience_t3
function_id: merge_graphs_py_datascience
name: test_merge_log_registra_cada_merge
file_path: python/functions/datascience/merge_graphs_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_merge_log_registra_cada_merge():
      g1 = {
          "entities": [{"id": "1", "name": "OpenAI"}],
          "relations": [],
      }
      g2 = {
          "entities": [{"id": "2", "name": "OpenAI"}],
          "relations": [],
      }
      result = merge_graphs([g1, g2], similarity_threshold=0.9)
      assert len(result["merge_log"]) == 1
      log = result["merge_log"][0]
      assert "merged" in log
      assert "into" in log
      assert "similarity" in log
      assert log["similarity"] == 1.0


id: merge_graphs_py_datascience_t4
function_id: merge_graphs_py_datascience
name: test_tres_grafos_merge_transitivo
file_path: python/functions/datascience/merge_graphs_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_tres_grafos_merge_transitivo():
      # A~B y B~C -> A, B, C deben mergearse en uno
      g1 = {"entities": [{"id": "1", "name": "Acme Corp"}], "relations": []}
      g2 = {"entities": [{"id": "2", "name": "Acme Corp"}], "relations": []}
      g3 = {"entities": [{"id": "3", "name": "Acme Corp"}], "relations": []}
      result = merge_graphs([g1, g2, g3], similarity_threshold=0.9)
      assert len(result["entities"]) == 1


id: merge_graphs_py_datascience_t5
function_id: merge_graphs_py_datascience
name: test_grafos_sin_overlap_concatenacion_simple
file_path: python/functions/datascience/merge_graphs_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_grafos_sin_overlap_concatenacion_simple():
      g1 = {
          "entities": [{"id": "1", "name": "Alice"}, {"id": "2", "name": "Bob"}],
          "relations": [{"source_id": "1", "target_id": "2", "relation_type": "knows"}],
      }
      g2 = {
          "entities": [{"id": "3", "name": "Carol"}, {"id": "4", "name": "Dave"}],
          "relations": [{"source_id": "3", "target_id": "4", "relation_type": "knows"}],
      }
      result = merge_graphs([g1, g2], similarity_threshold=0.85)
      # Ninguna entity similar -> concatenacion directa
      assert len(result["entities"]) == 4
      assert len(result["relations"]) == 2
      assert len(result["merge_log"]) == 0


  if __name__ == "__main__":
      test_dos_grafos_con_entity_duplicada_merge()
      test_entities_similares_pero_bajo_threshold_no_merge()
      test_relaciones_re_apuntadas_correctamente()
      test_merge_log_registra_cada_merge()
      test_tres_grafos_merge_transitivo()
      test_grafos_sin_overlap_concatenacion_simple()
      print("All tests passed.")


id: pivot_py_datascience_t0
function_id: pivot_py_datascience
name: test_pivot_basico_con_sum
file_path: python/functions/datascience/pivot_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_pivot_basico_con_sum():
      """Pivot basico con sum."""
      rows = [
          {"region": "US", "product": "A", "sales": 10},
          {"region": "US", "product": "B", "sales": 20},
          {"region": "EU", "product": "A", "sales": 15},
      ]
      result = pivot(rows, index="region", columns="product", values="sales")
      assert len(result) == 2
      us = next(r for r in result if r["region"] == "US")
      eu = next(r for r in result if r["region"] == "EU")
      assert us["A"] == 10
      assert us["B"] == 20
      assert eu["A"] == 15
      assert eu["B"] == 0


id: pivot_py_datascience_t1
function_id: pivot_py_datascience
name: test_pivot_con_count_y_mean
file_path: python/functions/datascience/pivot_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_pivot_con_count_y_mean():
      """Pivot con count y mean."""
      rows = [
          {"region": "US", "product": "A", "sales": 10},
          {"region": "US", "product": "A", "sales": 20},
          {"region": "EU", "product": "A", "sales": 15},
      ]
      result_count = pivot(rows, index="region", columns="product", values="sales", agg="count")
      us_count = next(r for r in result_count if r["region"] == "US")
      assert us_count["A"] == 2

      result_mean = pivot(rows, index="region", columns="product", values="sales", agg="mean")
      us_mean = next(r for r in result_mean if r["region"] == "US")
      assert us_mean["A"] == 15.0


id: pivot_py_datascience_t2
function_id: pivot_py_datascience
name: test_pivot_valores_faltantes_rellenados_con_0
file_path: python/functions/datascience/pivot_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_pivot_valores_faltantes_rellenados_con_0():
      """Valores faltantes rellenados con 0."""
      rows = [
          {"region": "US", "product": "A", "sales": 5},
          {"region": "EU", "product": "B", "sales": 8},
      ]
      result = pivot(rows, index="region", columns="product", values="sales")
      us = next(r for r in result if r["region"] == "US")
      eu = next(r for r in result if r["region"] == "EU")
      assert us["B"] == 0
      assert eu["A"] == 0


id: pivot_py_datascience_t3
function_id: pivot_py_datascience
name: test_pivot_una_sola_fila
file_path: python/functions/datascience/pivot_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_pivot_una_sola_fila():
      """Una sola fila."""
      rows = [{"region": "US", "product": "A", "sales": 42}]
      result = pivot(rows, index="region", columns="product", values="sales")
      assert len(result) == 1
      assert result[0]["region"] == "US"
      assert result[0]["A"] == 42


id: pivot_py_datascience_t4
function_id: pivot_py_datascience
name: test_pivot_multiples_valores_por_celda_requieren_agregacion
file_path: python/functions/datascience/pivot_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_pivot_multiples_valores_por_celda_requieren_agregacion():
      """Multiples valores por celda requieren agregacion."""
      rows = [
          {"region": "US", "product": "A", "sales": 10},
          {"region": "US", "product": "A", "sales": 30},
      ]
      result_sum = pivot(rows, index="region", columns="product", values="sales", agg="sum")
      assert result_sum[0]["A"] == 40

      result_min = pivot(rows, index="region", columns="product", values="sales", agg="min")
      assert result_min[0]["A"] == 10

      result_max = pivot(rows, index="region", columns="product", values="sales", agg="max")
      assert result_max[0]["A"] == 30


id: cache_to_file_py_infra_t0
function_id: cache_to_file_py_infra
name: test_set_y_get_basico
file_path: python/functions/infra/cache_to_file_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_set_y_get_basico(store):
      store.set("hello", {"x": 42})
      assert store.get("hello") == {"x": 42}


id: cache_to_file_py_infra_t1
function_id: cache_to_file_py_infra
name: test_ttl_expirado_retorna_none
file_path: python/functions/infra/cache_to_file_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_ttl_expirado_retorna_none(store):
      store.set("temp", "val", ttl=0.05)
      time.sleep(0.1)
      assert store.get("temp") is None


id: cache_to_file_py_infra_t2
function_id: cache_to_file_py_infra
name: test_archivo_meta_con_metadata_correcta
file_path: python/functions/infra/cache_to_file_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_archivo_meta_con_metadata_correcta(tmp_path):
      s = cache_to_file(str(tmp_path), "ns")
      s.set("mykey", "myval", ttl=60)
      ns_dir = os.path.join(str(tmp_path), "ns")
      meta_files = [f for f in os.listdir(ns_dir) if f.endswith(".meta")]
      assert len(meta_files) == 1
      with open(os.path.join(ns_dir, meta_files[0])) as f:
          meta = json.load(f)
      assert meta["original_key"] == "mykey"
      assert meta["expires_at"] is not None
      assert meta["created_at"] > 0


id: cache_to_file_py_infra_t3
function_id: cache_to_file_py_infra
name: test_clear_elimina_directorio_del_namespace
file_path: python/functions/infra/cache_to_file_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_clear_elimina_directorio_del_namespace(tmp_path):
      s = cache_to_file(str(tmp_path), "mynamespace")
      s.set("a", 1)
      s.set("b", 2)
      removed = s.clear()
      assert removed == 2
      assert s.get("a") is None
      assert s.get("b") is None


id: cache_to_file_py_infra_t4
function_id: cache_to_file_py_infra
name: test_key_con_caracteres_especiales_hash_seguro
file_path: python/functions/infra/cache_to_file_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_key_con_caracteres_especiales_hash_seguro(store):
      key = "https://example.com/path?q=1&r=2 <special>#hash"
      store.set(key, "safe")
      assert store.get(key) == "safe"


id: cache_to_sqlite_py_infra_t0
function_id: cache_to_sqlite_py_infra
name: test_set_y_get_basico
file_path: python/functions/infra/cache_to_sqlite_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_set_y_get_basico(store):
      store.set("foo", {"x": 1})
      assert store.get("foo") == {"x": 1}


id: cache_to_sqlite_py_infra_t1
function_id: cache_to_sqlite_py_infra
name: test_ttl_expirado_retorna_none
file_path: python/functions/infra/cache_to_sqlite_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_ttl_expirado_retorna_none(store):
      store.set("expiring", "hello", ttl=0.05)
      time.sleep(0.1)
      assert store.get("expiring") is None


id: cache_to_sqlite_py_infra_t2
function_id: cache_to_sqlite_py_infra
name: test_ttl_cero_nunca_expira
file_path: python/functions/infra/cache_to_sqlite_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_ttl_cero_nunca_expira(store):
      store.set("forever", 42, ttl=0)
      time.sleep(0.05)
      assert store.get("forever") == 42


id: cache_to_sqlite_py_infra_t3
function_id: cache_to_sqlite_py_infra
name: test_get_or_set_factory_solo_se_llama_en_miss
file_path: python/functions/infra/cache_to_sqlite_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_get_or_set_factory_solo_se_llama_en_miss(store):
      calls = []

      def factory():
          calls.append(1)
          return "computed"

      result1 = store.get_or_set("key", factory, ttl=10)
      result2 = store.get_or_set("key", factory, ttl=10)
      assert result1 == "computed"
      assert result2 == "computed"
      assert len(calls) == 1


id: cache_to_sqlite_py_infra_t4
function_id: cache_to_sqlite_py_infra
name: test_namespaces_independientes
file_path: python/functions/infra/cache_to_sqlite_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_namespaces_independientes(store_and_other):
      s1, s2 = store_and_other
      s1.set("k", "from_ns1")
      assert s2.get("k") is None
      s2.set("k", "from_ns2")
      assert s1.get("k") == "from_ns1"
      assert s2.get("k") == "from_ns2"


id: cache_to_sqlite_py_infra_t5
function_id: cache_to_sqlite_py_infra
name: test_clear_elimina_solo_el_namespace
file_path: python/functions/infra/cache_to_sqlite_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_clear_elimina_solo_el_namespace(store_and_other):
      s1, s2 = store_and_other
      s1.set("a", 1)
      s2.set("b", 2)
      removed = s1.clear()
      assert removed == 1
      assert s1.get("a") is None
      assert s2.get("b") == 2


id: cache_to_sqlite_py_infra_t6
function_id: cache_to_sqlite_py_infra
name: test_stats_contadores_correctos
file_path: python/functions/infra/cache_to_sqlite_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_stats_contadores_correctos(store):
      store.set("x", 10)
      store.get("x")   # hit
      store.get("x")   # hit
      store.get("z")   # miss
      s = store.stats()
      assert s["hits"] == 2
      assert s["misses"] == 1
      assert s["size"] == 1


id: cache_to_sqlite_py_infra_t7
function_id: cache_to_sqlite_py_infra
name: test_concurrencia
file_path: python/functions/infra/cache_to_sqlite_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_concurrencia(tmp_path):
      db = str(tmp_path / "concurrent.db")
      s = cache_to_sqlite(db, "parallel")
      errors = []

      def worker(i):
          try:
              s.set(f"key_{i}", i)
              val = s.get(f"key_{i}")
              assert val == i
          except Exception as e:
              errors.append(e)

      threads = [threading.Thread(target=worker, args=(i,)) for i in range(20)]
      for t in threads:
          t.start()
      for t in threads:
          t.join()

      assert errors == [], f"Errors in threads: {errors}"


id: get_logger_py_infra_t0
function_id: get_logger_py_infra
name: test_logger_tiene_dos_handlers
file_path: python/functions/infra/setup_logger_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_logger_tiene_dos_handlers():
      with tempfile.TemporaryDirectory() as log_dir:
          logger = setup_logger(name="test_two_handlers", log_dir=log_dir)
          assert len(logger.handlers) == 2
          # limpiar para no contaminar otros tests
          logger.handlers.clear()


id: get_logger_py_infra_t1
function_id: get_logger_py_infra
name: test_segundo_call_no_duplica_handlers
file_path: python/functions/infra/setup_logger_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_segundo_call_no_duplica_handlers():
      with tempfile.TemporaryDirectory() as log_dir:
          logger1 = setup_logger(name="test_idempotent", log_dir=log_dir)
          handler_count_after_first = len(logger1.handlers)
          logger2 = setup_logger(name="test_idempotent", log_dir=log_dir)
          assert logger1 is logger2
          assert len(logger2.handlers) == handler_count_after_first
          logger1.handlers.clear()


id: get_logger_py_infra_t2
function_id: get_logger_py_infra
name: test_archivo_se_crea_en_log_dir
file_path: python/functions/infra/setup_logger_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_archivo_se_crea_en_log_dir():
      with tempfile.TemporaryDirectory() as log_dir:
          logger = setup_logger(name="test_file_created", log_dir=log_dir)
          log_files = [f for f in os.listdir(log_dir) if f.endswith(".log")]
          assert len(log_files) == 1
          logger.handlers.clear()


id: get_logger_py_infra_t3
function_id: get_logger_py_infra
name: test_get_logger_retorna_logger_configurado
file_path: python/functions/infra/setup_logger_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_get_logger_retorna_logger_configurado():
      with tempfile.TemporaryDirectory() as log_dir:
          # Primero configurar para que get_logger encuentre handlers
          setup_logger(name="test_get_logger", log_dir=log_dir)
          logger = get_logger(name="test_get_logger")
          assert len(logger.handlers) == 2
          logger.handlers.clear()


id: get_logger_py_infra_t4
function_id: get_logger_py_infra
name: test_logger_level_es_debug
file_path: python/functions/infra/setup_logger_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_logger_level_es_debug():
      with tempfile.TemporaryDirectory() as log_dir:
          logger = setup_logger(name="test_level_debug", log_dir=log_dir, level=logging.DEBUG)
          assert logger.level == logging.DEBUG
          logger.handlers.clear()


id: normalize_zip_filenames_py_infra_t0
function_id: normalize_zip_filenames_py_infra
name: test_zip_normal
file_path: python/functions/infra/safe_extract_zip_test.py
lang: py
created_at: 2026-04-13T00:00:26Z
updated_at: 2026-04-13T00:00:26Z
code:
  def test_zip_normal():
      """ZIP normal extrae correctamente dentro del destino."""
      zip_path = _make_zip({"hello.txt": b"hello world", "subdir/file.py": b"# code"})
      try:
          with tempfile.TemporaryDirectory() as dest:
              safe_extract_zip(zip_path, dest)
              assert os.path.isfile(os.path.join(dest, "hello.txt"))
              assert os.path.isfile(os.path.join(dest, "subdir", "file.py"))
              with open(os.path.join(dest, "hello.txt"), "rb") as f:
                  assert f.read() == b"hello world"
      finally:
          os.unlink(zip_path)


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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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: 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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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:00:26Z
updated_at: 2026-04-13T00:00:26Z
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
  ```


