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

issue #56 WIP - Fix session mandatory in Endpoint.conn_handler() parameters

parent b35ccec9
No related branches found
No related tags found
No related merge requests found
Pipeline #2682 passed
......@@ -42,7 +42,7 @@ class Endpoint:
def inline(self) -> str:
raise NotImplementedError("inline() is not implemented")
def conn_handler(self, session: aiohttp.ClientSession = None, proxy: str = None) -> ConnectionHandler:
def conn_handler(self, session: aiohttp.ClientSession, proxy: str = None) -> ConnectionHandler:
raise NotImplementedError("conn_handler is not implemented")
def __str__(self) -> str:
......@@ -85,7 +85,7 @@ class UnknownEndpoint(Endpoint):
doc += " {0}".format(p)
return doc
def conn_handler(self, session: aiohttp.ClientSession = None, proxy: str = None) -> ConnectionHandler:
def conn_handler(self, session: aiohttp.ClientSession, proxy: str = None) -> ConnectionHandler:
return ConnectionHandler("", "", "", 0, "")
def __str__(self) -> str:
......@@ -166,8 +166,7 @@ class BMAEndpoint(Endpoint):
IPv6=(" {0}".format(self.ipv6) if self.ipv6 else ""),
PORT=(" {0}".format(self.port) if self.port else ""))
# fixme: session must be mandatory
def conn_handler(self, session: aiohttp.ClientSession = None, proxy: str = None) -> ConnectionHandler:
def conn_handler(self, session: aiohttp.ClientSession, proxy: str = None) -> ConnectionHandler:
"""
Return connection handler instance for the endpoint
......@@ -247,8 +246,7 @@ class SecuredBMAEndpoint(BMAEndpoint):
inlined = [str(info) for info in (self.server, self.ipv4, self.ipv6, self.port, self.path) if info]
return SecuredBMAEndpoint.API + " " + " ".join(inlined)
# fixme: session must be mandatory
def conn_handler(self, session: aiohttp.ClientSession = None, proxy: str = None) -> ConnectionHandler:
def conn_handler(self, session: aiohttp.ClientSession, proxy: str = None) -> ConnectionHandler:
"""
Return connection handler instance for the endpoint
......@@ -297,7 +295,7 @@ class WS2PEndpoint(Endpoint):
inlined = [str(info) for info in (self.ws2pid, self.server, self.port, self.path) if info]
return WS2PEndpoint.API + " " + " ".join(inlined)
def conn_handler(self, session: aiohttp.ClientSession = None, proxy: str = None) -> ConnectionHandler:
def conn_handler(self, session: aiohttp.ClientSession, proxy: str = None) -> ConnectionHandler:
"""
Return connection handler instance for the endpoint
......@@ -345,7 +343,7 @@ class ESUserEndpoint(Endpoint):
inlined = [str(info) for info in (self.server, self.port) if info]
return ESUserEndpoint.API + " " + " ".join(inlined)
def conn_handler(self, session: aiohttp.ClientSession = None, proxy: str = None) -> ConnectionHandler:
def conn_handler(self, session: aiohttp.ClientSession, proxy: str = None) -> ConnectionHandler:
"""
Return connection handler instance for the endpoint
......@@ -392,7 +390,7 @@ class ESSubscribtionEndpoint(Endpoint):
inlined = [str(info) for info in (self.server, self.port) if info]
return ESSubscribtionEndpoint.API + " " + " ".join(inlined)
def conn_handler(self, session: aiohttp.ClientSession = None, proxy: str = None) -> ConnectionHandler:
def conn_handler(self, session: aiohttp.ClientSession, proxy: str = None) -> ConnectionHandler:
"""
Return connection handler instance for the endpoint
......
import asyncio
import unittest
import aiohttp
from duniterpy.api.client import API, parse_error
from duniterpy.api.endpoint import BMAEndpoint
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)
api = API(endpoint.conn_handler(), "any")
session = aiohttp.ClientSession()
api = API(endpoint.conn_handler(session), "any")
self.assertEqual(api.reverse_url("http", "/test/url"), "http://test.com:9092/any/test/url")
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)
api = API(endpoint.conn_handler(), "any")
session = aiohttp.ClientSession()
api = API(endpoint.conn_handler(session), "any")
self.assertEqual(api.reverse_url("http", "/test/url"), "http://124.2.2.1:9092/any/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)
api = API(endpoint.conn_handler(), "any")
session = aiohttp.ClientSession()
api = API(endpoint.conn_handler(session), "any")
self.assertEqual(api.reverse_url("http", "/test/url"),
"http://[2001:0db8:0000:85a3:0000:0000:ac1f:8001]:9092/any/test/url")
await session.close()
self.loop.run_until_complete(go())
def test_parse_error(self):
error = parse_error("""{
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment