diff --git a/api/views/outpoints.py b/api/views/outpoints.py index fa95663..f17fc28 100644 --- a/api/views/outpoints.py +++ b/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) diff --git a/test/unit/test_outpoints_list.py b/test/unit/test_outpoints_list.py new file mode 100644 index 0000000..3e73f16 --- /dev/null +++ b/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")]