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

[enh] #140 remove asyncio and aiohttp in tests

parent 7e87ac07
No related branches found
No related tags found
No related merge requests found
This commit is part of merge request !118. Comments created here will be created in the context of that merge request.
Show whitespace changes
Inline Side-by-side
Showing
with 54 additions and 596 deletions
tests/api/bma/test_blockchain.py
+ 1
− 209
View file @ 328bb1a7
...@@ -21,18 +21,6 @@ import jsonschema ...@@ -21,18 +21,6 @@ import jsonschema
from jsonschema import SchemaError, ValidationError from jsonschema import SchemaError, ValidationError
from duniterpy.api.bma.blockchain import ( from duniterpy.api.bma.blockchain import (
parameters,
block,
current,
hardship,
memberships,
newcomers,
certifications,
joiners,
actives,
leavers,
ud,
tx,
BLOCK_NUMBERS_SCHEMA, BLOCK_NUMBERS_SCHEMA,
BLOCK_SCHEMA, BLOCK_SCHEMA,
BLOCKS_SCHEMA, BLOCKS_SCHEMA,
...@@ -40,12 +28,9 @@ from duniterpy.api.bma.blockchain import ( ...@@ -40,12 +28,9 @@ from duniterpy.api.bma.blockchain import (
MEMBERSHIPS_SCHEMA, MEMBERSHIPS_SCHEMA,
PARAMETERS_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): def test_parameters(self):
json_sample = { json_sample = {
"currency": "g1", "currency": "g1",
...@@ -75,22 +60,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase): ...@@ -75,22 +60,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e: except (SchemaError, ValidationError) as e:
raise self.failureException from 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): def test_schema_block_0(self):
json_sample = { json_sample = {
"version": 2, "version": 2,
...@@ -242,36 +211,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase): ...@@ -242,36 +211,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e: except (SchemaError, ValidationError) as e:
raise self.failureException from 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): def test_schema_block(self):
json_sample = { json_sample = {
"version": 2, "version": 2,
...@@ -355,24 +294,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase): ...@@ -355,24 +294,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e: except (SchemaError, ValidationError) as e:
raise self.failureException from 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): def test_schema_membership(self):
json_sample = { json_sample = {
"pubkey": "8Fi1VSTbjkXguwThF4v2ZxC5whK7pwG2vcGTkPUPjPGU", "pubkey": "8Fi1VSTbjkXguwThF4v2ZxC5whK7pwG2vcGTkPUPjPGU",
...@@ -402,27 +323,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase): ...@@ -402,27 +323,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e: except (SchemaError, ValidationError) as e:
raise self.failureException from 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): def test_schema_newcomers(self):
json_sample = {"result": {"blocks": [223, 813]}} json_sample = {"result": {"blocks": [223, 813]}}
try: try:
...@@ -430,22 +330,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase): ...@@ -430,22 +330,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e: except (SchemaError, ValidationError) as e:
raise self.failureException from 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): def test_schema_certifications(self):
json_sample = {"result": {"blocks": [223, 813]}} json_sample = {"result": {"blocks": [223, 813]}}
try: try:
...@@ -453,22 +337,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase): ...@@ -453,22 +337,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e: except (SchemaError, ValidationError) as e:
raise self.failureException from 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): def test_schema_joiners(self):
json_sample = {"result": {"blocks": [223, 813]}} json_sample = {"result": {"blocks": [223, 813]}}
try: try:
...@@ -476,22 +344,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase): ...@@ -476,22 +344,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e: except (SchemaError, ValidationError) as e:
raise self.failureException from 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): def test_schema_actives(self):
json_sample = {"result": {"blocks": [223, 813]}} json_sample = {"result": {"blocks": [223, 813]}}
try: try:
...@@ -499,22 +351,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase): ...@@ -499,22 +351,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e: except (SchemaError, ValidationError) as e:
raise self.failureException from 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): def test_schema_leavers(self):
json_sample = {"result": {"blocks": [223, 813]}} json_sample = {"result": {"blocks": [223, 813]}}
try: try:
...@@ -522,22 +358,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase): ...@@ -522,22 +358,6 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e: except (SchemaError, ValidationError) as e:
raise self.failureException from 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): def test_schema_ud(self):
json_sample = {"result": {"blocks": [223, 813]}} json_sample = {"result": {"blocks": [223, 813]}}
try: try:
...@@ -664,37 +484,9 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase): ...@@ -664,37 +484,9 @@ class TestBmaBlockchain(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e: except (SchemaError, ValidationError) as e:
raise self.failureException from 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): def test_schema_tx(self):
json_sample = {"result": {"blocks": [223, 813]}} json_sample = {"result": {"blocks": [223, 813]}}
try: try:
jsonschema.validate(json_sample, BLOCK_NUMBERS_SCHEMA) jsonschema.validate(json_sample, BLOCK_NUMBERS_SCHEMA)
except (SchemaError, ValidationError) as e: except (SchemaError, ValidationError) as e:
raise self.failureException from 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 @ 328bb1a7
...@@ -15,46 +15,26 @@ You should have received a copy of the GNU General Public License ...@@ -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/>. along with this program. If not, see <http://www.gnu.org/licenses/>.
""" """
import asyncio
import unittest import unittest
import aiohttp
from duniterpy.api.client import API, parse_error from duniterpy.api.client import API, parse_error
from duniterpy.api.endpoint import BMAEndpoint, SecuredBMAEndpoint, GVAEndpoint from duniterpy.api.endpoint import BMAEndpoint, SecuredBMAEndpoint, GVAEndpoint
class TestBmaApi(unittest.TestCase): 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): def test_reverse_url_complete(self):
async def go():
endpoint = BMAEndpoint( endpoint = BMAEndpoint(
"test.com", "test.com",
"124.2.2.1", "124.2.2.1",
"2001:0db8:0000:85a3:0000:0000:ac1f:8001 ", "2001:0db8:0000:85a3:0000:0000:ac1f:8001 ",
9092, 9092,
) )
session = aiohttp.ClientSession() api = API(endpoint.conn_handler())
api = API(endpoint.conn_handler(session))
self.assertEqual( self.assertEqual(
api.reverse_url("http", "/test/url"), "http://test.com:9092/test/url" 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): def test_reverse_url_complete_bmas(self):
async def go():
endpoint = SecuredBMAEndpoint( endpoint = SecuredBMAEndpoint(
"test.com", "test.com",
"124.2.2.1", "124.2.2.1",
...@@ -62,18 +42,13 @@ class TestBmaApi(unittest.TestCase): ...@@ -62,18 +42,13 @@ class TestBmaApi(unittest.TestCase):
9092, 9092,
"api_path", "api_path",
) )
session = aiohttp.ClientSession() api = API(endpoint.conn_handler())
api = API(endpoint.conn_handler(session))
self.assertEqual( self.assertEqual(
api.reverse_url("http", "/test/url"), api.reverse_url("http", "/test/url"),
"http://test.com:9092/api_path/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): def test_reverse_url_complete_gva(self):
async def go():
endpoint = GVAEndpoint( endpoint = GVAEndpoint(
"S", "S",
"test.com", "test.com",
...@@ -82,40 +57,24 @@ class TestBmaApi(unittest.TestCase): ...@@ -82,40 +57,24 @@ class TestBmaApi(unittest.TestCase):
9092, 9092,
"gva", "gva",
) )
session = aiohttp.ClientSession() api = API(endpoint.conn_handler())
api = API(endpoint.conn_handler(session))
self.assertEqual(api.reverse_url("https", ""), "https://test.com:9092/gva") 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): def test_reverse_url_only_ipv4(self):
async def go():
endpoint = BMAEndpoint("", "124.2.2.1", "", 9092) endpoint = BMAEndpoint("", "124.2.2.1", "", 9092)
session = aiohttp.ClientSession()
api = API(endpoint.conn_handler(session)) api = API(endpoint.conn_handler())
self.assertEqual( self.assertEqual(
api.reverse_url("http", "/test/url"), "http://124.2.2.1:9092/test/url" 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): def test_reverse_url_only_ipv6(self):
async def go(): endpoint = BMAEndpoint("", "", "2001:0db8:0000:85a3:0000:0000:ac1f:8001", 9092)
endpoint = BMAEndpoint( api = API(endpoint.conn_handler())
"", "", "2001:0db8:0000:85a3:0000:0000:ac1f:8001", 9092
)
session = aiohttp.ClientSession()
api = API(endpoint.conn_handler(session))
self.assertEqual( self.assertEqual(
api.reverse_url("http", "/test/url"), api.reverse_url("http", "/test/url"),
"http://[2001:0db8:0000:85a3:0000:0000:ac1f:8001]:9092/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): def test_parse_error(self):
error = parse_error( error = parse_error(
......
tests/api/bma/test_network.py
+ 0
− 46
View file @ 328bb1a7
...@@ -21,8 +21,6 @@ import jsonschema ...@@ -21,8 +21,6 @@ import jsonschema
from jsonschema import ValidationError, SchemaError from jsonschema import ValidationError, SchemaError
from duniterpy.api.bma import network 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 from tests.api.webserver import WebFunctionalSetupMixin, web
...@@ -70,20 +68,6 @@ class TestBmaNetwork(WebFunctionalSetupMixin, unittest.TestCase): ...@@ -70,20 +68,6 @@ class TestBmaNetwork(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e: except (SchemaError, ValidationError) as e:
raise self.failureException from 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): def test_peering_peers_root(self):
json_sample = { json_sample = {
"depth": 3, "depth": 3,
...@@ -116,22 +100,6 @@ class TestBmaNetwork(WebFunctionalSetupMixin, unittest.TestCase): ...@@ -116,22 +100,6 @@ class TestBmaNetwork(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e: except (SchemaError, ValidationError) as e:
raise self.failureException from 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): def test_ws2p_heads(self):
json_sample = { json_sample = {
"heads": [ "heads": [
...@@ -163,17 +131,3 @@ class TestBmaNetwork(WebFunctionalSetupMixin, unittest.TestCase): ...@@ -163,17 +131,3 @@ class TestBmaNetwork(WebFunctionalSetupMixin, unittest.TestCase):
jsonschema.validate(json_sample, network.WS2P_HEADS_SCHEMA) jsonschema.validate(json_sample, network.WS2P_HEADS_SCHEMA)
except (SchemaError, ValidationError) as e: except (SchemaError, ValidationError) as e:
raise self.failureException from 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 @ 328bb1a7
...@@ -21,18 +21,12 @@ import jsonschema ...@@ -21,18 +21,12 @@ import jsonschema
from jsonschema import SchemaError, ValidationError from jsonschema import SchemaError, ValidationError
from duniterpy.api.bma.tx import ( from duniterpy.api.bma.tx import (
history,
sources,
blocks,
HISTORY_SCHEMA, HISTORY_SCHEMA,
SOURCES_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): def test_bma_tx_history(self):
json_sample = { json_sample = {
"currency": "meta_brouzouf", "currency": "meta_brouzouf",
...@@ -157,36 +151,6 @@ class TestBmaTx(WebFunctionalSetupMixin, unittest.TestCase): ...@@ -157,36 +151,6 @@ class TestBmaTx(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e: except (SchemaError, ValidationError) as e:
raise self.failureException from 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): def test_bma_tx_sources(self):
json_sample = { json_sample = {
"currency": "meta_brouzouf", "currency": "meta_brouzouf",
...@@ -214,17 +178,3 @@ class TestBmaTx(WebFunctionalSetupMixin, unittest.TestCase): ...@@ -214,17 +178,3 @@ class TestBmaTx(WebFunctionalSetupMixin, unittest.TestCase):
jsonschema.validate(json_sample, SOURCES_SCHEMA) jsonschema.validate(json_sample, SOURCES_SCHEMA)
except (SchemaError, ValidationError) as e: except (SchemaError, ValidationError) as e:
raise self.failureException from 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 @ 328bb1a7
...@@ -15,28 +15,20 @@ You should have received a copy of the GNU General Public License ...@@ -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/>. along with this program. If not, see <http://www.gnu.org/licenses/>.
""" """
import json
import unittest import unittest
import jsonschema import jsonschema
from jsonschema import SchemaError, ValidationError from jsonschema import SchemaError, ValidationError
from duniterpy.api.bma.wot import ( from duniterpy.api.bma.wot import (
lookup,
members,
certified_by,
certifiers_of,
REQUIREMENTS_SCHEMA, REQUIREMENTS_SCHEMA,
CERTIFICATIONS_SCHEMA, CERTIFICATIONS_SCHEMA,
LOOKUP_SCHEMA, LOOKUP_SCHEMA,
MEMBERS_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): def test_bma_wot_lookup(self):
json_sample = { json_sample = {
"partial": False, "partial": False,
...@@ -174,20 +166,6 @@ class TestBmaWot(WebFunctionalSetupMixin, unittest.TestCase): ...@@ -174,20 +166,6 @@ class TestBmaWot(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e: except (SchemaError, ValidationError) as e:
raise self.failureException from 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): def test_bma_wot_members(self):
json_sample = { json_sample = {
"results": [ "results": [
...@@ -210,20 +188,6 @@ class TestBmaWot(WebFunctionalSetupMixin, unittest.TestCase): ...@@ -210,20 +188,6 @@ class TestBmaWot(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e: except (SchemaError, ValidationError) as e:
raise self.failureException from 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): def test_bma_wot_cert(self):
json_sample = { json_sample = {
"pubkey": "HsLShAtzXTVxeUtQd7yi5Z5Zh4zNvbu8sTEZ53nfKcqY", "pubkey": "HsLShAtzXTVxeUtQd7yi5Z5Zh4zNvbu8sTEZ53nfKcqY",
...@@ -260,76 +224,6 @@ class TestBmaWot(WebFunctionalSetupMixin, unittest.TestCase): ...@@ -260,76 +224,6 @@ class TestBmaWot(WebFunctionalSetupMixin, unittest.TestCase):
except (SchemaError, ValidationError) as e: except (SchemaError, ValidationError) as e:
raise self.failureException from 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): def test_bma_wot_requirements(self):
json_sample = { json_sample = {
"identities": [ "identities": [
......
tests/api/bma/test_ws.py
+ 1
− 2
View file @ 328bb1a7
...@@ -21,10 +21,9 @@ from jsonschema import SchemaError, ValidationError ...@@ -21,10 +21,9 @@ from jsonschema import SchemaError, ValidationError
from duniterpy.api.bma.ws import WS_BLOCK_SCHEMA, WS_PEER_SCHEMA from duniterpy.api.bma.ws import WS_BLOCK_SCHEMA, WS_PEER_SCHEMA
from duniterpy.api.client import parse_text 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): def test_block(self):
json_sample = """ json_sample = """
{ {
......
tests/api/webserver.py deleted 100644 → 0
+ 0
− 87
View file @ 7e87ac07
"""
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
tests/api/ws2p/test_ws2p.py
+ 1
− 2
View file @ 328bb1a7
...@@ -30,10 +30,9 @@ from duniterpy.api.ws2p.requests import ( ...@@ -30,10 +30,9 @@ from duniterpy.api.ws2p.requests import (
) )
from duniterpy.documents import Identity, BlockUID from duniterpy.documents import Identity, BlockUID
from duniterpy.documents.ws2p.messages import DocumentMessage 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): def test_block(self):
json_sample = { json_sample = {
"heads": [ "heads": [
......
tests/helpers/ws2p.py
+ 6
− 8
View file @ 328bb1a7
...@@ -21,7 +21,7 @@ from duniterpy.helpers.ws2p import generate_ws2p_endpoint ...@@ -21,7 +21,7 @@ from duniterpy.helpers.ws2p import generate_ws2p_endpoint
from duniterpy.api.endpoint import WS2PEndpoint, SecuredBMAEndpoint, BMAEndpoint from duniterpy.api.endpoint import WS2PEndpoint, SecuredBMAEndpoint, BMAEndpoint
async def peering_no_ws2p(self): def peering_no_ws2p(self):
return { return {
"version": 10, "version": 10,
"currency": "g1-test", "currency": "g1-test",
...@@ -35,7 +35,7 @@ async def peering_no_ws2p(self): ...@@ -35,7 +35,7 @@ async def peering_no_ws2p(self):
} }
async def peering_ws2p(self): def peering_ws2p(self):
return { return {
"version": 10, "version": 10,
"currency": "g1-test", "currency": "g1-test",
...@@ -61,20 +61,18 @@ async def peering_ws2p(self): ...@@ -61,20 +61,18 @@ async def peering_ws2p(self):
SecuredBMAEndpoint.from_inline("BMAS g1-test.duniter.org 443"), SecuredBMAEndpoint.from_inline("BMAS g1-test.duniter.org 443"),
], ],
) )
@pytest.mark.asyncio def test_generate_ws2p_endpoint(bma_endpoint, monkeypatch):
async def test_generate_ws2p_endpoint(bma_endpoint, monkeypatch):
monkeypatch.setattr("duniterpy.api.bma.network.peering", peering_ws2p) monkeypatch.setattr("duniterpy.api.bma.network.peering", peering_ws2p)
reference_ep = WS2PEndpoint.from_inline( reference_ep = WS2PEndpoint.from_inline(
"WS2P 96675302 g1-test.duniter.org 443 ws2p" "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 assert reference_ep == generated_ep
@pytest.mark.parametrize("bma_endpoint", ["BMAS g1-test.duniter.org 443"]) @pytest.mark.parametrize("bma_endpoint", ["BMAS g1-test.duniter.org 443"])
@pytest.mark.asyncio def test_generate_ws2p_endpoint_no_ws2p(bma_endpoint, monkeypatch):
async def test_generate_ws2p_endpoint_no_ws2p(bma_endpoint, monkeypatch):
monkeypatch.setattr("duniterpy.api.bma.network.peering", peering_no_ws2p) monkeypatch.setattr("duniterpy.api.bma.network.peering", peering_no_ws2p)
with pytest.raises(ValueError) as excinfo: 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) 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