diff --git a/tests/api/bma/test_blockchain.py b/tests/api/bma/test_blockchain.py
index cbde93c74510c3ffc25c9fb95188df3383f88d45..87082e8ab4487386ef1f7db409777d93a19919e4 100644
--- a/tests/api/bma/test_blockchain.py
+++ b/tests/api/bma/test_blockchain.py
@@ -21,18 +21,6 @@ import jsonschema
from jsonschema import SchemaError, ValidationError
from duniterpy.api.bma.blockchain import (
- parameters,
- block,
- current,
- hardship,
- memberships,
- newcomers,
- certifications,
- joiners,
- actives,
- leavers,
- ud,
- tx,
BLOCK_NUMBERS_SCHEMA,
BLOCK_SCHEMA,
BLOCKS_SCHEMA,
@@ -40,12 +28,9 @@ from duniterpy.api.bma.blockchain import (
MEMBERSHIPS_SCHEMA,
PARAMETERS_SCHEMA,
)
-from duniterpy.api.client import Client
-from duniterpy.api.endpoint import BMAEndpoint
-from tests.api.webserver import WebFunctionalSetupMixin, web
-class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
+class TestBmaBlockchain(unittest.TestCase):
def test_parameters(self):
json_sample = {
"currency": "g1",
@@ -75,22 +60,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e:
raise self.failureException from e
- def test_parameters_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server(
- "GET", "/blockchain/parameters", handler
- )
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(parameters)
- await client.close()
-
- self.loop.run_until_complete(go())
-
def test_schema_block_0(self):
json_sample = {
"version": 2,
@@ -242,36 +211,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e:
raise self.failureException from e
- def test_block_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server(
- "GET", "/blockchain/block/100", handler
- )
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(block, 100)
- await client.close()
-
- self.loop.run_until_complete(go())
-
- def test_current_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server("GET", "/blockchain/current", handler)
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(current)
- await client.close()
-
- self.loop.run_until_complete(go())
-
def test_schema_block(self):
json_sample = {
"version": 2,
@@ -355,24 +294,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e:
raise self.failureException from e
- def test_hardship_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server(
- "GET",
- "/blockchain/hardship/8Fi1VSTbjkXguwThF4v2ZxC5whK7pwG2vcGTkPUPjPGU",
- handler,
- )
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(hardship, "8Fi1VSTbjkXguwThF4v2ZxC5whK7pwG2vcGTkPUPjPGU")
- await client.close()
-
- self.loop.run_until_complete(go())
-
def test_schema_membership(self):
json_sample = {
"pubkey": "8Fi1VSTbjkXguwThF4v2ZxC5whK7pwG2vcGTkPUPjPGU",
@@ -402,27 +323,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e:
raise self.failureException from e
- def test_membership_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server(
- "GET",
- "/blockchain/memberships"
- "/8Fi1VSTbjkXguwThF4v2ZxC5whK7pwG2vcGTkPUPjPGU",
- handler,
- )
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(
- memberships, "8Fi1VSTbjkXguwThF4v2ZxC5whK7pwG2vcGTkPUPjPGU"
- )
- await client.close()
-
- self.loop.run_until_complete(go())
-
def test_schema_newcomers(self):
json_sample = {"result": {"blocks": [223, 813]}}
try:
@@ -430,22 +330,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e:
raise self.failureException from e
- def test_newcomers_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server(
- "GET", "/blockchain/with/newcomers", handler
- )
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(newcomers)
- await client.close()
-
- self.loop.run_until_complete(go())
-
def test_schema_certifications(self):
json_sample = {"result": {"blocks": [223, 813]}}
try:
@@ -453,22 +337,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e:
raise self.failureException from e
- def test_certifications_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server(
- "GET", "/blockchain/with/certs", handler
- )
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(certifications)
- await client.close()
-
- self.loop.run_until_complete(go())
-
def test_schema_joiners(self):
json_sample = {"result": {"blocks": [223, 813]}}
try:
@@ -476,22 +344,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e:
raise self.failureException from e
- def test_joiners_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server(
- "GET", "/blockchain/with/joiners", handler
- )
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(joiners)
- await client.close()
-
- self.loop.run_until_complete(go())
-
def test_schema_actives(self):
json_sample = {"result": {"blocks": [223, 813]}}
try:
@@ -499,22 +351,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e:
raise self.failureException from e
- def test_actives_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server(
- "GET", "/blockchain/with/actives", handler
- )
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(actives)
- await client.close()
-
- self.loop.run_until_complete(go())
-
def test_schema_leavers(self):
json_sample = {"result": {"blocks": [223, 813]}}
try:
@@ -522,22 +358,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e:
raise self.failureException from e
- def test_leavers_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server(
- "GET", "/blockchain/with/leavers", handler
- )
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(leavers)
- await client.close()
-
- self.loop.run_until_complete(go())
-
def test_schema_ud(self):
json_sample = {"result": {"blocks": [223, 813]}}
try:
@@ -664,37 +484,9 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e:
raise self.failureException from e
- def test_ud_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server("GET", "/blockchain/with/ud", handler)
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(ud)
- await client.close()
-
- self.loop.run_until_complete(go())
-
def test_schema_tx(self):
json_sample = {"result": {"blocks": [223, 813]}}
try:
jsonschema.validate(json_sample, BLOCK_NUMBERS_SCHEMA)
except (SchemaError, ValidationError) as e:
raise self.failureException from e
-
- def test_tx_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server("GET", "/blockchain/with/tx", handler)
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(tx)
- await client.close()
-
- self.loop.run_until_complete(go())
diff --git a/tests/api/bma/test_bma.py b/tests/api/bma/test_bma.py
index 72994ae24d7784492ba7710838bd7e5dcffe6ec4..8fee431469db5f0aa6f045fdf10c175b75311a57 100644
--- a/tests/api/bma/test_bma.py
+++ b/tests/api/bma/test_bma.py
@@ -15,107 +15,66 @@ You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
-import asyncio
import unittest
-import aiohttp
-
from duniterpy.api.client import API, parse_error
from duniterpy.api.endpoint import BMAEndpoint, SecuredBMAEndpoint, GVAEndpoint
class TestBmaApi(unittest.TestCase):
- def setUp(self):
- self.loop = asyncio.new_event_loop()
- asyncio.set_event_loop(self.loop)
-
- def tearDown(self):
- try:
- self.loop.stop()
- self.loop.close()
- finally:
- asyncio.set_event_loop(None)
-
def test_reverse_url_complete(self):
- async def go():
- endpoint = BMAEndpoint(
- "test.com",
- "124.2.2.1",
- "2001:0db8:0000:85a3:0000:0000:ac1f:8001 ",
- 9092,
- )
- session = aiohttp.ClientSession()
- api = API(endpoint.conn_handler(session))
- self.assertEqual(
- api.reverse_url("http", "/test/url"), "http://test.com:9092/test/url"
- )
- await session.close()
-
- self.loop.run_until_complete(go())
+ endpoint = BMAEndpoint(
+ "test.com",
+ "124.2.2.1",
+ "2001:0db8:0000:85a3:0000:0000:ac1f:8001 ",
+ 9092,
+ )
+ api = API(endpoint.conn_handler())
+ self.assertEqual(
+ api.reverse_url("http", "/test/url"), "http://test.com:9092/test/url"
+ )
def test_reverse_url_complete_bmas(self):
- async def go():
- endpoint = SecuredBMAEndpoint(
- "test.com",
- "124.2.2.1",
- "2001:0db8:0000:85a3:0000:0000:ac1f:8001 ",
- 9092,
- "api_path",
- )
- session = aiohttp.ClientSession()
- api = API(endpoint.conn_handler(session))
- self.assertEqual(
- api.reverse_url("http", "/test/url"),
- "http://test.com:9092/api_path/test/url",
- )
- await session.close()
-
- self.loop.run_until_complete(go())
+ endpoint = SecuredBMAEndpoint(
+ "test.com",
+ "124.2.2.1",
+ "2001:0db8:0000:85a3:0000:0000:ac1f:8001 ",
+ 9092,
+ "api_path",
+ )
+ api = API(endpoint.conn_handler())
+ self.assertEqual(
+ api.reverse_url("http", "/test/url"),
+ "http://test.com:9092/api_path/test/url",
+ )
def test_reverse_url_complete_gva(self):
- async def go():
- endpoint = GVAEndpoint(
- "S",
- "test.com",
- "124.2.2.1",
- "2001:0db8:0000:85a3:0000:0000:ac1f:8001 ",
- 9092,
- "gva",
- )
- session = aiohttp.ClientSession()
- api = API(endpoint.conn_handler(session))
- self.assertEqual(api.reverse_url("https", ""), "https://test.com:9092/gva")
- await session.close()
-
- self.loop.run_until_complete(go())
+ endpoint = GVAEndpoint(
+ "S",
+ "test.com",
+ "124.2.2.1",
+ "2001:0db8:0000:85a3:0000:0000:ac1f:8001 ",
+ 9092,
+ "gva",
+ )
+ api = API(endpoint.conn_handler())
+ self.assertEqual(api.reverse_url("https", ""), "https://test.com:9092/gva")
def test_reverse_url_only_ipv4(self):
- async def go():
- endpoint = BMAEndpoint("", "124.2.2.1", "", 9092)
- session = aiohttp.ClientSession()
+ endpoint = BMAEndpoint("", "124.2.2.1", "", 9092)
- api = API(endpoint.conn_handler(session))
- self.assertEqual(
- api.reverse_url("http", "/test/url"), "http://124.2.2.1:9092/test/url"
- )
- await session.close()
-
- self.loop.run_until_complete(go())
+ api = API(endpoint.conn_handler())
+ self.assertEqual(
+ api.reverse_url("http", "/test/url"), "http://124.2.2.1:9092/test/url"
+ )
def test_reverse_url_only_ipv6(self):
- async def go():
- endpoint = BMAEndpoint(
- "", "", "2001:0db8:0000:85a3:0000:0000:ac1f:8001", 9092
- )
- session = aiohttp.ClientSession()
- api = API(endpoint.conn_handler(session))
- self.assertEqual(
- api.reverse_url("http", "/test/url"),
- "http://[2001:0db8:0000:85a3:0000:0000:ac1f:8001]:9092/test/url",
- )
- await session.close()
-
- self.loop.run_until_complete(go())
+ endpoint = BMAEndpoint("", "", "2001:0db8:0000:85a3:0000:0000:ac1f:8001", 9092)
+ api = API(endpoint.conn_handler())
+ self.assertEqual(
+ api.reverse_url("http", "/test/url"),
+ "http://[2001:0db8:0000:85a3:0000:0000:ac1f:8001]:9092/test/url",
+ )
def test_parse_error(self):
error = parse_error(
diff --git a/tests/api/bma/test_network.py b/tests/api/bma/test_network.py
index b9083ed02af580a36112dea8ff2f1ec6b87f1728..ac7f1543dabd6854c03e05073154326c7f01aa81 100644
--- a/tests/api/bma/test_network.py
+++ b/tests/api/bma/test_network.py
@@ -21,12 +21,9 @@ import jsonschema
from jsonschema import ValidationError, SchemaError
from duniterpy.api.bma import network
-from duniterpy.api.client import Client
-from duniterpy.api.endpoint import BMAEndpoint
-from tests.api.webserver import WebFunctionalSetupMixin, web
-class TestBmaNetwork(WebFunctionalSetupMixin, unittest.TestCase):
+class TestBmaNetwork(unittest.TestCase):
def test_peers(self):
json_sample = {
"peers": [
@@ -70,20 +67,6 @@ class TestBmaNetwork(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e:
raise self.failureException from e
- def test_peering_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server("GET", "/network/peering", handler)
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(network.peering)
- await client.close()
-
- self.loop.run_until_complete(go())
-
def test_peering_peers_root(self):
json_sample = {
"depth": 3,
@@ -116,22 +99,6 @@ class TestBmaNetwork(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e:
raise self.failureException from e
- def test_peering_peers_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server(
- "GET", "/network/peering/peers", handler
- )
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(network.peering_peers)
- await client.close()
-
- self.loop.run_until_complete(go())
-
def test_ws2p_heads(self):
json_sample = {
"heads": [
@@ -163,17 +130,3 @@ class TestBmaNetwork(WebFunctionalSetupMixin, unittest.TestCase):
jsonschema.validate(json_sample, network.WS2P_HEADS_SCHEMA)
except (SchemaError, ValidationError) as e:
raise self.failureException from e
-
- def test_ws2p_heads_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server("GET", "/network/ws2p/heads", handler)
- with self.assertRaises(jsonschema.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(network.ws2p_heads)
- await client.close()
-
- self.loop.run_until_complete(go())
diff --git a/tests/api/bma/test_tx.py b/tests/api/bma/test_tx.py
index 8e5f8a5ed737b5ce8f72c3f074fbbbabf43dce89..6f8c5f13fcfdb2bc97e6281e7ece0e7e1f951f54 100644
--- a/tests/api/bma/test_tx.py
+++ b/tests/api/bma/test_tx.py
@@ -21,18 +21,12 @@ import jsonschema
from jsonschema import SchemaError, ValidationError
from duniterpy.api.bma.tx import (
- history,
- sources,
- blocks,
HISTORY_SCHEMA,
SOURCES_SCHEMA,
)
-from duniterpy.api.client import Client
-from duniterpy.api.endpoint import BMAEndpoint
-from tests.api.webserver import WebFunctionalSetupMixin, web
-class TestBmaTx(WebFunctionalSetupMixin, unittest.TestCase):
+class TestBmaTx(unittest.TestCase):
def test_bma_tx_history(self):
json_sample = {
"currency": "meta_brouzouf",
@@ -157,36 +151,6 @@ class TestBmaTx(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e:
raise self.failureException from e
- def test_bma_tx_history_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server("GET", "/tx/history/pubkey", handler)
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(history, "pubkey")
- await client.close()
-
- self.loop.run_until_complete(go())
-
- def test_bma_tx_history_blocks_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server(
- "GET", "/tx/history/pubkey/blocks/0/100", handler
- )
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(blocks, "pubkey", 0, 100)
- await client.close()
-
- self.loop.run_until_complete(go())
-
def test_bma_tx_sources(self):
json_sample = {
"currency": "meta_brouzouf",
@@ -214,17 +178,3 @@ class TestBmaTx(WebFunctionalSetupMixin, unittest.TestCase):
jsonschema.validate(json_sample, SOURCES_SCHEMA)
except (SchemaError, ValidationError) as e:
raise self.failureException from e
-
- def test_bma_tx_sources_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server("GET", "/tx/sources/pubkey", handler)
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(sources, "pubkey")
- await client.close()
-
- self.loop.run_until_complete(go())
diff --git a/tests/api/bma/test_wot.py b/tests/api/bma/test_wot.py
index 80146f85ed453f32cd88f889d143705d6bd90624..995088ba2088480fc2cc0a28cc28d7006c97b9c2 100644
--- a/tests/api/bma/test_wot.py
+++ b/tests/api/bma/test_wot.py
@@ -15,28 +15,20 @@ You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
-import json
import unittest
import jsonschema
from jsonschema import SchemaError, ValidationError
from duniterpy.api.bma.wot import (
- lookup,
- members,
- certified_by,
- certifiers_of,
REQUIREMENTS_SCHEMA,
CERTIFICATIONS_SCHEMA,
LOOKUP_SCHEMA,
MEMBERS_SCHEMA,
)
-from duniterpy.api.client import Client
-from duniterpy.api.endpoint import BMAEndpoint
-from tests.api.webserver import WebFunctionalSetupMixin, web
-class TestBmaWot(WebFunctionalSetupMixin, unittest.TestCase):
+class TestBmaWot(unittest.TestCase):
def test_bma_wot_lookup(self):
json_sample = {
"partial": False,
@@ -174,20 +166,6 @@ class TestBmaWot(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e:
raise self.failureException from e
- def test_bma_wot_lookup_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server("GET", "/wot/lookup/pubkey", handler)
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(lookup, "pubkey")
- await client.close()
-
- self.loop.run_until_complete(go())
-
def test_bma_wot_members(self):
json_sample = {
"results": [
@@ -210,20 +188,6 @@ class TestBmaWot(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e:
raise self.failureException from e
- def test_bma_wot_members_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server("GET", "/wot/members", handler)
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(members)
- await client.close()
-
- self.loop.run_until_complete(go())
-
def test_bma_wot_cert(self):
json_sample = {
"pubkey": "HsLShAtzXTVxeUtQd7yi5Z5Zh4zNvbu8sTEZ53nfKcqY",
@@ -260,76 +224,6 @@ class TestBmaWot(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e:
raise self.failureException from e
- def test_bma_wot_certifiers_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server(
- "GET", "/wot/certifiers-of/pubkey", handler
- )
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(certifiers_of, "pubkey")
- await client.close()
-
- self.loop.run_until_complete(go())
-
- def test_bma_wot_certifiers_inner_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(
- body=bytes(
- json.dumps(
- {
- "pubkey": "7Aqw6Efa9EzE7gtsc8SveLLrM7gm6NEGoywSv4FJx6pZ",
- "uid": "john",
- "isMember": True,
- "certifications": [
- {
- "pubkey": "FADxcH5LmXGmGFgdixSes6nWnC4Vb4pRUBYT81zQRhjn",
- "meta": {"block_number": 38580},
- "uids": ["doe"],
- "isMember": True,
- "wasMember": True,
- "signature": "8XYmBdElqNkkl4AeFjJnC5oj/ujBrzH9FNgPZvK8Cicp8Du0PQa0yYFG95EQ46MJhdV0fUT2g5xyH8N3/OGhDA==",
- }
- ],
- }
- ),
- "utf-8",
- ),
- content_type="application/json",
- )
-
- async def go():
- _, port, _ = await self.create_server(
- "GET", "/wot/certifiers-of/pubkey", handler
- )
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(certifiers_of, "pubkey")
- await client.close()
-
- self.loop.run_until_complete(go())
-
- def test_bma_wot_certified_bad(self):
- async def handler(request):
- await request.read()
- return web.Response(body=b"{}", content_type="application/json")
-
- async def go():
- _, port, _ = await self.create_server(
- "GET", "/wot/certified-by/pubkey", handler
- )
- with self.assertRaises(jsonschema.exceptions.ValidationError):
- client = Client(BMAEndpoint("127.0.0.1", "", "", port))
- await client(certified_by, "pubkey")
- await client.close()
-
- self.loop.run_until_complete(go())
-
def test_bma_wot_requirements(self):
json_sample = {
"identities": [
diff --git a/tests/api/bma/test_ws.py b/tests/api/bma/test_ws.py
index 6df787988b7e9a767a5c30bd2feb63c72e856456..97a37a9914455b79bd1470e8be4a71232f896a3f 100644
--- a/tests/api/bma/test_ws.py
+++ b/tests/api/bma/test_ws.py
@@ -21,10 +21,9 @@ from jsonschema import SchemaError, ValidationError
from duniterpy.api.bma.ws import WS_BLOCK_SCHEMA, WS_PEER_SCHEMA
from duniterpy.api.client import parse_text
-from tests.api.webserver import WebFunctionalSetupMixin
-class TestBmaWebsocket(WebFunctionalSetupMixin, unittest.TestCase):
+class TestBmaWebsocket(unittest.TestCase):
def test_block(self):
json_sample = """
{
diff --git a/tests/api/webserver.py b/tests/api/webserver.py
deleted file mode 100644
index aafd6b05fb9f4f88dadb2034f80c3f92dec6b662..0000000000000000000000000000000000000000
--- a/tests/api/webserver.py
+++ /dev/null
@@ -1,87 +0,0 @@
-"""
-Copyright 2014-2021 Vincent Texier <vit@free.fr>
-
-DuniterPy is free software: you can redistribute it and/or modify
-it under the terms of the GNU General Public License as published by
-the Free Software Foundation, either version 3 of the License, or
-(at your option) any later version.
-
-DuniterPy is distributed in the hope that it will be useful,
-but WITHOUT ANY WARRANTY; without even the implied warranty of
-MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-GNU General Public License for more details.
-
-You should have received a copy of the GNU General Public License
-along with this program. If not, see <http://www.gnu.org/licenses/>.
-"""
-
-import asyncio
-import socket
-import ssl
-from typing import Tuple, Callable, Optional
-
-from aiohttp import web
-
-
-# Thanks to aiohttp tests courtesy
-# Here is a nice mocking server
-def find_unused_port() -> int:
- """
- Return first free network port
-
- :return:
- """
- s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- s.bind(("127.0.0.1", 0))
- port = s.getsockname()[1]
- s.close()
- return port
-
-
-class WebFunctionalSetupMixin:
- def setUp(self) -> None:
- self.handler = None
- self.app = web.Application()
- self.runner = web.AppRunner(self.app)
- self.loop = asyncio.new_event_loop()
- asyncio.set_event_loop(self.loop)
-
- def tearDown(self) -> None:
- if self.handler:
- self.loop.run_until_complete(self.handler.shutdown())
- if self.runner:
- self.loop.run_until_complete(self.runner.cleanup())
- try:
- self.loop.stop()
- self.loop.close()
- finally:
- asyncio.set_event_loop(None)
-
- async def create_server(
- self,
- method: str,
- path: str,
- handler: Optional[Callable] = None,
- ssl_ctx: Optional[ssl.SSLContext] = None,
- ) -> Tuple[web.Application, int, str]:
- """
- Create a web server for tests
-
- :param method: HTTP method type
- :param path: Url path
- :param handler: Callback function
- :param ssl_ctx: SSL context (https is used if not None)
- :return:
- """
- if handler:
- self.app.router.add_route(method, path, handler)
-
- await self.runner.setup()
-
- port = find_unused_port()
- site = web.TCPSite(self.runner, "127.0.0.1", port)
- await site.start()
-
- protocol = "https" if ssl_ctx else "http"
- url = "{}://127.0.0.1:{}".format(protocol, port) + path
- return self.app, port, url
diff --git a/tests/api/ws2p/test_ws2p.py b/tests/api/ws2p/test_ws2p.py
index 7b191a7ff255ce7bf534cab17dec19dea653cfd7..9a5c39b5b381686b9d714488efa6d59cc8f11f55 100644
--- a/tests/api/ws2p/test_ws2p.py
+++ b/tests/api/ws2p/test_ws2p.py
@@ -30,10 +30,9 @@ from duniterpy.api.ws2p.requests import (
)
from duniterpy.documents import Identity, BlockUID
from duniterpy.documents.ws2p.messages import DocumentMessage
-from tests.api.webserver import WebFunctionalSetupMixin
-class TestWs2p(WebFunctionalSetupMixin, unittest.TestCase):
+class TestWs2p(unittest.TestCase):
def test_block(self):
json_sample = {
"heads": [
diff --git a/tests/helpers/ws2p.py b/tests/helpers/ws2p.py
index 989d1de76c611b9c61e72b2671c65e5ada4c8a20..1d8523ee08ac36d267bfa532f7b86acaa593cdc9 100644
--- a/tests/helpers/ws2p.py
+++ b/tests/helpers/ws2p.py
@@ -21,7 +21,7 @@ from duniterpy.helpers.ws2p import generate_ws2p_endpoint
from duniterpy.api.endpoint import WS2PEndpoint, SecuredBMAEndpoint, BMAEndpoint
-async def peering_no_ws2p(self):
+def peering_no_ws2p(self):
return {
"version": 10,
"currency": "g1-test",
@@ -35,7 +35,7 @@ async def peering_no_ws2p(self):
}
-async def peering_ws2p(self):
+def peering_ws2p(self):
return {
"version": 10,
"currency": "g1-test",
@@ -61,20 +61,18 @@ async def peering_ws2p(self):
SecuredBMAEndpoint.from_inline("BMAS g1-test.duniter.org 443"),
],
)
-@pytest.mark.asyncio
-async def test_generate_ws2p_endpoint(bma_endpoint, monkeypatch):
+def test_generate_ws2p_endpoint(bma_endpoint, monkeypatch):
monkeypatch.setattr("duniterpy.api.bma.network.peering", peering_ws2p)
reference_ep = WS2PEndpoint.from_inline(
"WS2P 96675302 g1-test.duniter.org 443 ws2p"
)
- generated_ep = await generate_ws2p_endpoint(bma_endpoint)
+ generated_ep = generate_ws2p_endpoint(bma_endpoint)
assert reference_ep == generated_ep
@pytest.mark.parametrize("bma_endpoint", ["BMAS g1-test.duniter.org 443"])
-@pytest.mark.asyncio
-async def test_generate_ws2p_endpoint_no_ws2p(bma_endpoint, monkeypatch):
+def test_generate_ws2p_endpoint_no_ws2p(bma_endpoint, monkeypatch):
monkeypatch.setattr("duniterpy.api.bma.network.peering", peering_no_ws2p)
with pytest.raises(ValueError) as excinfo:
- await generate_ws2p_endpoint(bma_endpoint)
+ generate_ws2p_endpoint(bma_endpoint)
assert "No WS2P endpoint found" in str(excinfo.value)