From 406a5cd68f68395d013c608bfa3325da93b8d421 Mon Sep 17 00:00:00 2001
From: Vincent Texier <vit@free.fr>
Date: Fri, 20 Nov 2020 11:02:19 +0100
Subject: [PATCH] [enh] #59 add GVAEndpoint and GVASUBEndpoint classes
add unit tests
example now use real GVA Endpoint inline format
---
duniterpy/api/client.py | 2 +-
duniterpy/api/endpoint.py | 161 +++++++++++++++++++++++++++++++
duniterpy/constants.py | 1 +
examples/request_data_graphql.py | 2 +-
tests/api/test_endpoints.py | 78 +++++++++++++++
5 files changed, 242 insertions(+), 2 deletions(-)
create mode 100644 tests/api/test_endpoints.py
diff --git a/duniterpy/api/client.py b/duniterpy/api/client.py
index f527548a..d3864f9a 100644
--- a/duniterpy/api/client.py
+++ b/duniterpy/api/client.py
@@ -515,7 +515,7 @@ class Client:
try:
result = await response.json()
except aiohttp.client_exceptions.ContentTypeError as exception:
- logging.error("Response is not a json format")
+ logging.error("Response is not a json format: %s", exception)
# return response to debug...
return result
diff --git a/duniterpy/api/endpoint.py b/duniterpy/api/endpoint.py
index cde3647e..34e8139b 100644
--- a/duniterpy/api/endpoint.py
+++ b/duniterpy/api/endpoint.py
@@ -640,6 +640,165 @@ class ESSubscribtionEndpoint(Endpoint):
return hash((ESSubscribtionEndpoint.API, self.server, self.port))
+# required to type hint cls in classmethod
+GVAEndpointType = TypeVar("GVAEndpointType", bound="GVAEndpoint")
+
+
+class GVAEndpoint(Endpoint):
+ API = "GVA"
+ re_inline = re.compile(
+ "^GVA(?: ({endpoint_flags}))?(?: ({host_regex}))?(?: ({ipv4_regex}))?(?: ({ipv6_regex}))? ([0-9]+)(?: ({path_regex}))?$".format(
+ host_regex=constants.HOST_REGEX,
+ ipv4_regex=constants.IPV4_REGEX,
+ ipv6_regex=constants.IPV6_REGEX,
+ path_regex=constants.PATH_REGEX,
+ endpoint_flags=constants.ENDPOINT_FLAGS_REGEX,
+ )
+ )
+
+ def __init__(
+ self,
+ flags: str,
+ server: str,
+ ipv4: str,
+ ipv6: str,
+ port: int,
+ path: str,
+ ) -> None:
+ """
+ Init GVAEndpoint instance
+
+ :param flags: Flags of endpoint
+ :param server: IP or domain name
+ :param ipv4: IP as IPv4 format
+ :param ipv6: IP as IPv6 format
+ :param port: Port number
+ :param path: Url path
+ """
+ self.flags = flags
+ self.server = server
+ self.ipv4 = ipv4
+ self.ipv6 = ipv6
+ self.port = port
+ self.path = path
+
+ @classmethod
+ def from_inline(cls: Type[GVAEndpointType], inline: str) -> GVAEndpointType:
+ """
+ Return GVAEndpoint instance from endpoint string
+
+ :param inline: Endpoint string
+ :return:
+ """
+ m = cls.re_inline.match(inline)
+ if m is None:
+ raise MalformedDocumentError(cls.API)
+ flags = m.group(1)
+ server = m.group(2)
+ ipv4 = m.group(3)
+ ipv6 = m.group(4)
+ port = int(m.group(5))
+ path = m.group(6)
+ if not flags:
+ flags = ""
+ if not path:
+ path = ""
+ return cls(flags, server, ipv4, ipv6, port, path)
+
+ def inline(self) -> str:
+ """
+ Return endpoint string
+
+ :return:
+ """
+ inlined = [
+ str(info)
+ for info in (
+ self.flags,
+ self.server,
+ self.ipv4,
+ self.ipv6,
+ self.port,
+ self.path,
+ )
+ if info
+ ]
+ return self.API + " " + " ".join(inlined)
+
+ def conn_handler(
+ self, session: ClientSession, proxy: str = None
+ ) -> ConnectionHandler:
+ """
+ Return connection handler instance for the endpoint
+
+ :param session: AIOHTTP client session instance
+ :param proxy: Proxy url
+ :return:
+ """
+ scheme_http = "https" if "S" in self.flags else "http"
+ scheme_ws = "wss" if "S" in self.flags else "ws"
+
+ if self.server:
+ conn_handler = ConnectionHandler(
+ scheme_http,
+ scheme_ws,
+ self.server,
+ self.port,
+ self.path,
+ session,
+ proxy,
+ )
+ elif self.ipv6:
+ conn_handler = ConnectionHandler(
+ scheme_http,
+ scheme_ws,
+ "[{0}]".format(self.ipv6),
+ self.port,
+ self.path,
+ session,
+ proxy,
+ )
+ else:
+ conn_handler = ConnectionHandler(
+ scheme_http, scheme_ws, self.ipv4, self.port, self.path, session, proxy
+ )
+
+ return conn_handler
+
+ def __str__(self) -> str:
+ return self.inline()
+
+ def __eq__(self, other: Any) -> bool:
+ if not isinstance(other, self.__class__):
+ return NotImplemented
+ return (
+ self.server == other.server
+ and self.ipv4 == other.ipv4
+ and self.ipv6 == other.ipv6
+ and self.port == other.port
+ )
+
+ def __hash__(self) -> int:
+ return hash((self.server, self.ipv4, self.ipv6, self.port))
+
+
+# required to type hint cls in classmethod
+GVASUBEndpointType = TypeVar("GVASUBEndpointType", bound="GVASUBEndpoint")
+
+
+class GVASUBEndpoint(GVAEndpoint):
+ API = "GVASUB"
+ re_inline = re.compile(
+ "^GVASUB(?: ({endpoint_flags}))?(?: ({host_regex}))?(?: ({ipv4_regex}))?(?: ({ipv6_regex}))? ([0-9]+)(?: ({path_regex}))?$".format(
+ host_regex=constants.HOST_REGEX,
+ ipv4_regex=constants.IPV4_REGEX,
+ ipv6_regex=constants.IPV6_REGEX,
+ path_regex=constants.PATH_REGEX,
+ endpoint_flags=constants.ENDPOINT_FLAGS_REGEX,
+ )
+ )
+
+
MANAGED_API = {
BMAEndpoint.API: BMAEndpoint,
SecuredBMAEndpoint.API: SecuredBMAEndpoint,
@@ -647,6 +806,8 @@ MANAGED_API = {
ESCoreEndpoint.API: ESCoreEndpoint,
ESUserEndpoint.API: ESUserEndpoint,
ESSubscribtionEndpoint.API: ESSubscribtionEndpoint,
+ GVAEndpoint.API: GVAEndpoint,
+ GVASUBEndpoint.API: GVASUBEndpoint,
} # type: Dict[str, Any]
diff --git a/duniterpy/constants.py b/duniterpy/constants.py
index 494d7900..31f458a4 100644
--- a/duniterpy/constants.py
+++ b/duniterpy/constants.py
@@ -59,3 +59,4 @@ WS2P_PRIVATE_PREFIX_REGEX = "O[CT][SAM]"
WS2P_PUBLIC_PREFIX_REGEX = "I[CT]"
WS2P_HEAD_REGEX = "HEAD:?(?:[0-9]+)?"
EMPTY_HASH = "E3B0C44298FC1C149AFBF4C8996FB92427AE41E4649B934CA495991B7852B855"
+ENDPOINT_FLAGS_REGEX = "[S]"
diff --git a/examples/request_data_graphql.py b/examples/request_data_graphql.py
index f2c29bbf..21ed2be7 100644
--- a/examples/request_data_graphql.py
+++ b/examples/request_data_graphql.py
@@ -11,7 +11,7 @@ from graphql.error import GraphQLSyntaxError
# You can either use a complete defined endpoint : [NAME_OF_THE_API] [DOMAIN] [IPv4] [IPv6] [PORT]
# or the simple definition : [NAME_OF_THE_API] [DOMAIN] [PORT]
# Here we use the secure BASIC_MERKLED_API (BMAS) for standard http over ssl requests
-GVA_ENDPOINT = "BMAS g1.librelois.fr 443 gva"
+GVA_ENDPOINT = "GVA S g1.librelois.fr 443 gva"
################################################
diff --git a/tests/api/test_endpoints.py b/tests/api/test_endpoints.py
new file mode 100644
index 00000000..8fc88cf3
--- /dev/null
+++ b/tests/api/test_endpoints.py
@@ -0,0 +1,78 @@
+"""
+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 unittest
+
+import duniterpy.api.endpoint as endpoint
+
+
+class TestEndpoint(unittest.TestCase):
+ def test_gva(self):
+ endpoint_str = "GVA test.domain.com 127.0.0.1 2001:0db8:0000:85a3:0000:0000:ac1f:8001 10902 gva"
+
+ gva_endpoint = endpoint.GVAEndpoint.from_inline(endpoint_str)
+
+ self.assertEqual(gva_endpoint.flags, "")
+ self.assertEqual(gva_endpoint.server, "test.domain.com")
+ self.assertEqual(gva_endpoint.ipv4, "127.0.0.1")
+ self.assertEqual(gva_endpoint.ipv6, "2001:0db8:0000:85a3:0000:0000:ac1f:8001")
+ self.assertEqual(gva_endpoint.port, 10902)
+ self.assertEqual(gva_endpoint.path, "gva")
+
+ self.assertEqual(gva_endpoint.inline(), endpoint_str)
+
+ endpoint_str = "GVA S test.domain.com 10902 gva"
+
+ gva_endpoint = endpoint.GVAEndpoint.from_inline(endpoint_str)
+
+ self.assertEqual(gva_endpoint.flags, "S")
+ self.assertEqual(gva_endpoint.server, "test.domain.com")
+ self.assertEqual(gva_endpoint.ipv4, None)
+ self.assertEqual(gva_endpoint.ipv6, None)
+ self.assertEqual(gva_endpoint.port, 10902)
+ self.assertEqual(gva_endpoint.path, "gva")
+
+ self.assertEqual(gva_endpoint.inline(), endpoint_str)
+
+ def test_gva_subscription(self):
+ endpoint_str = "GVASUB test.domain.com 127.0.0.1 2001:0db8:0000:85a3:0000:0000:ac1f:8001 10902 gva"
+
+ gvasub_endpoint = endpoint.GVASUBEndpoint.from_inline(endpoint_str)
+
+ self.assertEqual(gvasub_endpoint.flags, "")
+ self.assertEqual(gvasub_endpoint.server, "test.domain.com")
+ self.assertEqual(gvasub_endpoint.ipv4, "127.0.0.1")
+ self.assertEqual(
+ gvasub_endpoint.ipv6, "2001:0db8:0000:85a3:0000:0000:ac1f:8001"
+ )
+ self.assertEqual(gvasub_endpoint.port, 10902)
+ self.assertEqual(gvasub_endpoint.path, "gva")
+
+ self.assertEqual(gvasub_endpoint.inline(), endpoint_str)
+
+ endpoint_str = "GVASUB S test.domain.com 10902 gva"
+
+ gvasub_endpoint = endpoint.GVASUBEndpoint.from_inline(endpoint_str)
+
+ self.assertEqual(gvasub_endpoint.flags, "S")
+ self.assertEqual(gvasub_endpoint.server, "test.domain.com")
+ self.assertEqual(gvasub_endpoint.ipv4, None)
+ self.assertEqual(gvasub_endpoint.ipv6, None)
+ self.assertEqual(gvasub_endpoint.port, 10902)
+ self.assertEqual(gvasub_endpoint.path, "gva")
+
+ assert gvasub_endpoint.inline(), endpoint_str
--
GitLab