egutierrez
9fd0ca9cac
Infra: cache_to_file, cache_to_sqlite, http_download_file, http_get_json, http_post_json, read_file_with_encoding, safe_extract_zip, scan_directory, setup_logger, normalize_zip_filenames. Tipos: 30+ tipos core (agent_action, context, task, message, parse_result...), 6 tipos datascience (entity_candidate, extraction_result...), 2 tipos infra. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
85 lines
3.0 KiB
Python
85 lines
3.0 KiB
Python
"""Tests para http_download_file."""
|
|
|
|
import sys
|
|
import tempfile
|
|
import os
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
sys.path.insert(0, "/home/lucas/fn_registry/python/functions")
|
|
|
|
from infra.http_download_file import http_download_file
|
|
|
|
|
|
def _make_response(content: bytes, content_type: str = "application/octet-stream"):
|
|
resp = MagicMock()
|
|
# Simula lectura en chunks
|
|
chunks = [content[i:i+8192] for i in range(0, len(content), 8192)] + [b""]
|
|
resp.read.side_effect = chunks
|
|
resp.headers = {"Content-Type": content_type}
|
|
resp.__enter__ = lambda s: s
|
|
resp.__exit__ = MagicMock(return_value=False)
|
|
return resp
|
|
|
|
|
|
class TestHttpDownloadFile(unittest.TestCase):
|
|
|
|
def test_mock_descarga_con_contenido_binario(self):
|
|
content = b"\x00\x01\x02\x03" * 100
|
|
mock_resp = _make_response(content, "application/octet-stream")
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
dest = os.path.join(tmpdir, "file.bin")
|
|
with patch("urllib.request.urlopen", return_value=mock_resp):
|
|
result = http_download_file("http://example.com/file.bin", dest)
|
|
|
|
self.assertEqual(result["size_bytes"], len(content))
|
|
self.assertEqual(result["path"], dest)
|
|
with open(dest, "rb") as f:
|
|
self.assertEqual(f.read(), content)
|
|
|
|
def test_directorio_destino_creado_automaticamente(self):
|
|
content = b"hello binary"
|
|
mock_resp = _make_response(content)
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
dest = os.path.join(tmpdir, "nested", "deep", "file.bin")
|
|
self.assertFalse(os.path.exists(os.path.dirname(dest)))
|
|
|
|
with patch("urllib.request.urlopen", return_value=mock_resp):
|
|
http_download_file("http://example.com/file.bin", dest)
|
|
|
|
self.assertTrue(os.path.exists(dest))
|
|
|
|
def test_retorno_con_size_correcto(self):
|
|
content = b"x" * 5000
|
|
mock_resp = _make_response(content, "text/plain")
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
dest = os.path.join(tmpdir, "out.txt")
|
|
with patch("urllib.request.urlopen", return_value=mock_resp):
|
|
result = http_download_file("http://example.com/data.txt", dest)
|
|
|
|
self.assertEqual(result["size_bytes"], 5000)
|
|
self.assertEqual(result["content_type"], "text/plain")
|
|
|
|
def test_timeout_configurado_en_el_request(self):
|
|
content = b"data"
|
|
mock_resp = _make_response(content)
|
|
captured_timeout = []
|
|
|
|
def fake_urlopen(req, timeout=None):
|
|
captured_timeout.append(timeout)
|
|
return mock_resp
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
dest = os.path.join(tmpdir, "file.bin")
|
|
with patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
|
http_download_file("http://example.com/file.bin", dest, timeout=60.0)
|
|
|
|
self.assertEqual(captured_timeout[0], 60.0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|