fix: preserve outpoints list validation errors

api/views/outpoints.py

@@ -9,6 +9,28 @@
 from utils import JSON_DECODE_ERROR
 
 
+def _parse_outpoints_list(data):
+    if not isinstance(data, list):
+        raise APIException(JSON_DECODE_ERROR, "invalid outpoints list")
+    if len(data) > 100:
+        raise APIException(PARAMETER_ERROR, "only 100 outpoints allowed")
+
+    outpoints = []
+    for outpoint in data:
+        try:
+            o, a = outpoint.split(":")
+            tx_id = s2rh(o)
+            if len(tx_id) != 32:
+                raise Exception()
+            index = int(a)
+            if index < 0:
+                raise Exception()
+            outpoints.append((tx_id, index, b"".join((tx_id, int_to_bytes(index)))))
+        except:
+            raise APIException(PARAMETER_ERROR, "invalid outpoint %s" % outpoint)
+    return outpoints
+
+
 async def get_outpoints_info(request):
     log = request.app["log"]
     log.info("POST %s" % str(request.rel_url))
@@ -22,29 +44,9 @@ async def get_outpoints_info(request):
         try:
             await request.post()
             data = await request.json()
-            outpoints = []
-            if len(data) > 100:
-                raise APIException(PARAMETER_ERROR, "only 100 outpoints allowed")
-            """
-            connector_utxo
-            connector_unconfirmed_utxo
-            connector_unconfirmed_stxo
-
-            """
-            for outpoint in data:
-                try:
-                    o, a = outpoint.split(":")
-                    tx_id = s2rh(o)
-                    if len(tx_id) != 32:
-                        raise Exception()
-                    outpoints.append((tx_id, int(a),
-                                      b"".join((tx_id, int_to_bytes(int(a))))))
-                except:
-                    raise APIException(PARAMETER_ERROR, "invalid outpoint %s" % outpoint)
-
-
         except:
             raise APIException(JSON_DECODE_ERROR, "invalid outpoints list")
+        outpoints = _parse_outpoints_list(data)
 
         response = await outpoints_info(outpoints, request.app)
 

test/unit/test_outpoints_list.py

@@ -0,0 +1,74 @@
+import pathlib
+import sys
+import types
+import importlib.util
+
+import pytest
+
+
+ROOT = pathlib.Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT / "api"))
+
+
+class PybglStub(types.ModuleType):
+    def __getattr__(self, name):
+        if name == "s2rh":
+            return lambda value: bytes.fromhex(value)[::-1]
+        if name == "int_to_bytes":
+            return lambda value: value.to_bytes((value.bit_length() + 7) // 8 or 1, "big", signed=False)
+        return lambda *args, **kwargs: None
+
+
+pybgl = PybglStub("pybgl")
+sys.modules.setdefault("pybgl", pybgl)
+
+aiohttp = types.ModuleType("aiohttp")
+aiohttp.web = types.SimpleNamespace(json_response=lambda *args, **kwargs: None)
+sys.modules.setdefault("aiohttp", aiohttp)
+
+model = types.ModuleType("model")
+model.outpoints_info = None
+sys.modules.setdefault("model", model)
+
+from utils import JSON_DECODE_ERROR, PARAMETER_ERROR
+
+spec = importlib.util.spec_from_file_location(
+    "outpoints_under_test",
+    ROOT / "api" / "views" / "outpoints.py",
+)
+outpoints = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(outpoints)
+_parse_outpoints_list = outpoints._parse_outpoints_list
+
+
+def test_outpoints_list_limit_reports_parameter_error():
+    with pytest.raises(Exception) as err:
+        _parse_outpoints_list(["%s:0" % ("00" * 32)] * 101)
+
+    assert err.value.err_code == PARAMETER_ERROR
+    assert err.value.message == "only 100 outpoints allowed"
+
+
+def test_outpoints_list_rejects_non_list_json():
+    with pytest.raises(Exception) as err:
+        _parse_outpoints_list({"outpoint": "%s:0" % ("00" * 32)})
+
+    assert err.value.err_code == JSON_DECODE_ERROR
+    assert err.value.message == "invalid outpoints list"
+
+
+def test_outpoints_list_rejects_invalid_outpoint_as_parameter_error():
+    with pytest.raises(Exception) as err:
+        _parse_outpoints_list(["not-a-transaction:0"])
+
+    assert err.value.err_code == PARAMETER_ERROR
+    assert err.value.message == "invalid outpoint not-a-transaction:0"
+
+
+def test_outpoints_list_parses_transaction_hash_and_index():
+    tx_hash = "01" * 32
+
+    parsed = _parse_outpoints_list(["%s:1" % tx_hash])
+
+    tx_id = bytes.fromhex(tx_hash)[::-1]
+    assert parsed == [(tx_id, 1, tx_id + b"\x01")]