diff --git a/tests/unit/test_NWCServiceProvider.py b/tests/unit/test_NWCServiceProvider.py new file mode 100644 index 0000000..71f7975 --- /dev/null +++ b/tests/unit/test_NWCServiceProvider.py @@ -0,0 +1,144 @@ +import json +from loguru import logger +import sys +import os +sys.path.insert(0, os.path.abspath( + os.path.join(os.path.dirname(__file__), '..', '..'))) +#### + +import string +import random +from NWCServiceProvider import NWCServiceProvider +from unittest.mock import AsyncMock, patch +import asyncio +import pytest + + +@pytest.fixture +def nwc_service_provider(): + return NWCServiceProvider("d7b5232fba0e02e32cfe26f20cdf2c803b27ecd81052c2dd5d17e5e1a333fe58", "") + + +@pytest.fixture +def nwc_service_provider2(): + return NWCServiceProvider("ce40821040275f72f3074a89770db3e2744b189f204807c867840eb58565de51", "") + + +@pytest.mark.asyncio +async def test_supported_methods(nwc_service_provider): + def make_invoice(provider, pubkey, content): + return "invoice" + nwc_service_provider.addRequestListener("make_invoice", make_invoice) + s = nwc_service_provider.getSupportedMethods() + assert s == ["make_invoice"] + + +@pytest.mark.asyncio +async def test_encrytdecrypt(nwc_service_provider, nwc_service_provider2): + content = "Hello World" + expected_enc = "qVurNVISSl/9CfREIhk5Lg==?iv=QpCo5dI9gUcoLsSMLA7o7Q==" + enc_a = nwc_service_provider._encrypt_content( + content, nwc_service_provider2.public_key_hex, 21) + enc_b = nwc_service_provider2._encrypt_content( + content, nwc_service_provider.public_key_hex, 21) + + dec_a = nwc_service_provider2._decrypt_content( + enc_a, nwc_service_provider.public_key_hex) + dec_b = nwc_service_provider._decrypt_content( + enc_b, nwc_service_provider2.public_key_hex) + + assert dec_a == content + assert dec_b == content + assert enc_a == expected_enc + assert enc_b == expected_enc + + +@pytest.mark.asyncio +async def test_signverify(nwc_service_provider, nwc_service_provider2): + # Random content + content = "" + for _ in range(100): + content += random.choice(string.ascii_letters) + + tags = [] + for _ in range(random.choice([1, 2, 3, 4])): + tags.append([random.choice(string.ascii_letters)+"_" + + random.choice(string.ascii_letters), random.choice(string.ascii_letters)]) + + event = { + "kind": 1, + "content": content, + "tags": tags, + "created_at": 1234567890 + } + + signed = nwc_service_provider._sign_event(event) + assert nwc_service_provider2._verify_event(signed) + + +@pytest.mark.asyncio +async def test_handle(nwc_service_provider, nwc_service_provider2): + content = nwc_service_provider._json_dumps({ + "method": "pay_invoice", + "params": { + "invoice": "abc" + } + }) + content = nwc_service_provider._encrypt_content( + content, nwc_service_provider2.public_key_hex, 21) + event = { + "kind": 23194, + "content": content, + "tags": [ + ["p", nwc_service_provider2.public_key_hex] + ], + "created_at": 1234567890 + + } + signed = nwc_service_provider._sign_event(event) + + async def _handle_pay_invoice(provider, pubkey, content): + assert pubkey == nwc_service_provider.public_key_hex + assert content["method"] == "pay_invoice" + assert content["params"]["invoice"] == "abc" + return [ + ( + { + "preimage": "00000" + + }, + None, + [ + ["r1", "v1"] + ] + ) + ] + + async def _send_pass(obj): + pass + nwc_service_provider2._send=_send_pass + nwc_service_provider2.addRequestListener( + "pay_invoice", _handle_pay_invoice) + sent_events = await nwc_service_provider2._handle_request(signed) + assert len(sent_events) == 1 + for revent in sent_events: + assert nwc_service_provider2._verify_event(revent) + content = nwc_service_provider2._decrypt_content( + revent["content"], nwc_service_provider.public_key_hex) + logger.debug(event) + logger.debug(revent) + content = json.loads(content) + assert content["result_type"] == "pay_invoice" + assert content["result"]["preimage"] == "00000" + tags = revent["tags"] + r1_tag = [tag for tag in tags if tag[0] == "r1"] + assert len(r1_tag) == 1 + assert r1_tag[0][1] == "v1" + + e_tag = [tag for tag in tags if tag[0] == "e"] + assert len(e_tag) == 1 + assert e_tag[0][1] == event["id"] + + p_tag = [tag for tag in tags if tag[0] == "p"] + assert len(p_tag) == 1 + assert p_tag[0][1] == nwc_service_provider.public_key_hex