118 lines
3.3 KiB
Python
118 lines
3.3 KiB
Python
"""Tests para query_osint_db.
|
|
|
|
Mockean urllib.request.urlopen para no depender del service osint_db vivo:
|
|
se interceptan el exito (cuerpo {status:ok,...}), el error de dominio del service
|
|
({status:error,...}) y el error de red (conexion rechazada).
|
|
"""
|
|
|
|
import io
|
|
import json
|
|
import urllib.error
|
|
|
|
from query_osint_db import query_osint_db
|
|
|
|
|
|
class _FakeResponse:
|
|
"""Context manager que imita la respuesta de urllib.request.urlopen."""
|
|
|
|
def __init__(self, payload: dict):
|
|
self._raw = json.dumps(payload).encode("utf-8")
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
def read(self):
|
|
return self._raw
|
|
|
|
|
|
def test_query_ok_devuelve_cuerpo_del_service(monkeypatch):
|
|
payload = {
|
|
"status": "ok",
|
|
"columns": ["count_star()"],
|
|
"rows": [{"count_star()": 42}],
|
|
"row_count": 1,
|
|
"truncated": False,
|
|
}
|
|
captured = {}
|
|
|
|
def fake_urlopen(req, timeout=None):
|
|
captured["url"] = req.full_url
|
|
captured["body"] = json.loads(req.data.decode("utf-8"))
|
|
captured["method"] = req.get_method()
|
|
return _FakeResponse(payload)
|
|
|
|
monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)
|
|
|
|
result = query_osint_db("SELECT COUNT(*) FROM personas")
|
|
|
|
assert result == payload
|
|
assert result["status"] == "ok"
|
|
assert result["rows"] == [{"count_star()": 42}]
|
|
assert captured["url"] == "http://127.0.0.1:8771/api/query"
|
|
assert captured["body"] == {"sql": "SELECT COUNT(*) FROM personas"}
|
|
assert captured["method"] == "POST"
|
|
|
|
|
|
def test_query_error_de_dominio_se_reenvia(monkeypatch):
|
|
payload = {"status": "error", "error": "Catalog Error: Table no existe"}
|
|
|
|
def fake_urlopen(req, timeout=None):
|
|
return _FakeResponse(payload)
|
|
|
|
monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)
|
|
|
|
result = query_osint_db("SELECT * FROM tabla_inexistente")
|
|
|
|
assert result["status"] == "error"
|
|
assert "no existe" in result["error"]
|
|
|
|
|
|
def test_service_caido_devuelve_error_claro(monkeypatch):
|
|
def fake_urlopen(req, timeout=None):
|
|
raise urllib.error.URLError("Connection refused")
|
|
|
|
monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)
|
|
|
|
result = query_osint_db("SELECT 1")
|
|
|
|
assert result["status"] == "error"
|
|
assert "osint_db service not reachable" in result["error"]
|
|
assert "http://127.0.0.1:8771/api/query" in result["error"]
|
|
|
|
|
|
def test_base_url_custom_se_respeta(monkeypatch):
|
|
captured = {}
|
|
|
|
def fake_urlopen(req, timeout=None):
|
|
captured["url"] = req.full_url
|
|
return _FakeResponse({"status": "ok", "rows": [], "row_count": 0})
|
|
|
|
monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)
|
|
|
|
query_osint_db("SELECT 1", base_url="http://10.0.0.5:9000/")
|
|
|
|
assert captured["url"] == "http://10.0.0.5:9000/api/query"
|
|
|
|
|
|
def test_http_error_con_cuerpo_json_se_reenvia(monkeypatch):
|
|
body = json.dumps({"status": "error", "error": "boom"}).encode("utf-8")
|
|
|
|
def fake_urlopen(req, timeout=None):
|
|
raise urllib.error.HTTPError(
|
|
url=req.full_url,
|
|
code=500,
|
|
msg="Internal Server Error",
|
|
hdrs=None,
|
|
fp=io.BytesIO(body),
|
|
)
|
|
|
|
monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)
|
|
|
|
result = query_osint_db("SELECT 1")
|
|
|
|
assert result["status"] == "error"
|
|
assert result["error"] == "boom"
|