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>
88 lines
3.0 KiB
Python
88 lines
3.0 KiB
Python
"""Tests para http_get_json."""
|
|
|
|
import json
|
|
import sys
|
|
import unittest
|
|
import urllib.error
|
|
import urllib.request
|
|
from io import BytesIO
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
sys.path.insert(0, "/home/lucas/fn_registry/python/functions")
|
|
|
|
from infra.http_get_json import http_get_json
|
|
|
|
|
|
def _make_response(data: bytes, status: int = 200, content_type: str = "application/json"):
|
|
"""Crea un mock de HTTPResponse."""
|
|
resp = MagicMock()
|
|
resp.read.return_value = data
|
|
resp.status = status
|
|
resp.__enter__ = lambda s: s
|
|
resp.__exit__ = MagicMock(return_value=False)
|
|
return resp
|
|
|
|
|
|
class TestHttpGetJson(unittest.TestCase):
|
|
|
|
def test_mock_respuesta_200_con_json(self):
|
|
payload = {"ok": True, "value": 42}
|
|
mock_resp = _make_response(json.dumps(payload).encode())
|
|
with patch("urllib.request.urlopen", return_value=mock_resp):
|
|
result = http_get_json("http://example.com/api")
|
|
self.assertEqual(result, payload)
|
|
|
|
def test_mock_respuesta_404_error_con_status_code(self):
|
|
err = urllib.error.HTTPError(
|
|
url="http://example.com/missing",
|
|
code=404,
|
|
msg="Not Found",
|
|
hdrs=None, # type: ignore[arg-type]
|
|
fp=BytesIO(b"not found"),
|
|
)
|
|
with patch("urllib.request.urlopen", side_effect=err):
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
http_get_json("http://example.com/missing")
|
|
self.assertIn("404", str(ctx.exception))
|
|
|
|
def test_mock_respuesta_json_invalido_error_descriptivo(self):
|
|
mock_resp = _make_response(b"not-json!!!")
|
|
with patch("urllib.request.urlopen", return_value=mock_resp):
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
http_get_json("http://example.com/api")
|
|
self.assertIn("not valid JSON", str(ctx.exception))
|
|
|
|
def test_params_serializados_como_query_string(self):
|
|
captured_url = []
|
|
|
|
def fake_urlopen(req, timeout=None):
|
|
captured_url.append(req.full_url)
|
|
return _make_response(b"{}")
|
|
|
|
with patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
|
http_get_json("http://example.com/api", params={"page": "1", "limit": "10"})
|
|
|
|
url = captured_url[0]
|
|
self.assertIn("page=1", url)
|
|
self.assertIn("limit=10", url)
|
|
|
|
def test_headers_custom_enviados(self):
|
|
captured_headers = []
|
|
|
|
def fake_urlopen(req, timeout=None):
|
|
captured_headers.append(dict(req.headers))
|
|
return _make_response(b'{"x": 1}')
|
|
|
|
with patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
|
http_get_json("http://example.com/api", headers={"X-Api-Key": "secret"})
|
|
|
|
# urllib capitaliza el primer caracter de cada header
|
|
headers_lower = {k.lower(): v for k, v in captured_headers[0].items()}
|
|
self.assertIn("x-api-key", headers_lower)
|
|
self.assertEqual(headers_lower["x-api-key"], "secret")
|
|
self.assertIn("accept", headers_lower)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|