Skip to content
Snippets Groups Projects
Commit 893f60eb authored by Vincent Texier's avatar Vincent Texier
Browse files

[enh] #140 remove asyncio and aiohttp in tests

parent 2a25d46b
No related branches found
No related tags found
No related merge requests found
Show whitespace changes
Inline Side-by-side
Showing
with 54 additions and 596 deletions
tests/api/bma/test_blockchain.py
+ 1
− 209
View file @ 893f60eb
......@@ -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())
tests/api/bma/test_bma.py
+ 43
− 84
View file @ 893f60eb
......@@ -15,46 +15,26 @@ 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))
api = API(endpoint.conn_handler())
self.assertEqual(
api.reverse_url("http", "/test/url"), "http://test.com:9092/test/url"
)
await session.close()
self.loop.run_until_complete(go())
def test_reverse_url_complete_bmas(self):
async def go():
endpoint = SecuredBMAEndpoint(
"test.com",
"124.2.2.1",
......@@ -62,18 +42,13 @@ class TestBmaApi(unittest.TestCase):
9092,
"api_path",
)
session = aiohttp.ClientSession()
api = API(endpoint.conn_handler(session))
api = API(endpoint.conn_handler())
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())
def test_reverse_url_complete_gva(self):
async def go():
endpoint = GVAEndpoint(
"S",
"test.com",
......@@ -82,40 +57,24 @@ class TestBmaApi(unittest.TestCase):
9092,
"gva",
)
session = aiohttp.ClientSession()
api = API(endpoint.conn_handler(session))
api = API(endpoint.conn_handler())
self.assertEqual(api.reverse_url("https", ""), "https://test.com:9092/gva")
await session.close()
self.loop.run_until_complete(go())
def test_reverse_url_only_ipv4(self):
async def go():
endpoint = BMAEndpoint("", "124.2.2.1", "", 9092)
session = aiohttp.ClientSession()
api = API(endpoint.conn_handler(session))
api = API(endpoint.conn_handler())
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())
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))
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",
)
await session.close()
self.loop.run_until_complete(go())
def test_parse_error(self):
error = parse_error(
......
tests/api/bma/test_network.py
+ 0
− 46
View file @ 893f60eb
......@@ -21,8 +21,6 @@ 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
......@@ -70,20 +68,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 +100,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 +131,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())
tests/api/bma/test_tx.py
+ 1
− 51
View file @ 893f60eb
......@@ -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())
tests/api/bma/test_wot.py
+ 1
− 107
View file @ 893f60eb
......@@ -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": [
......
tests/api/bma/test_ws.py
+ 1
− 2
View file @ 893f60eb
......@@ -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 = """
{
......
tests/api/webserver.py deleted 100644 → 0
+ 0
− 87
View file @ 2a25d46b
"""
Copyright 2014-2020 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
tests/api/ws2p/test_ws2p.py
+ 1
− 2
View file @ 893f60eb
......@@ -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": [
......
tests/helpers/ws2p.py
+ 6
− 8
View file @ 893f60eb
......@@ -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)
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment