From 7fd1ced259dcb0ffcc433d3015f1249b74801029 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/endpoint.py | 161 +++++++++++++++++++++++++++++++
duniterpy/constants.py | 7 +-
examples/request_data_graphql.py | 2 +-
tests/api/test_endpoints.py | 59 +++++++++++
4 files changed, 224 insertions(+), 5 deletions(-)
create mode 100644 tests/api/test_endpoints.py
diff --git a/duniterpy/api/endpoint.py b/duniterpy/api/endpoint.py
index cde3647e..2afe91d7 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: Optional[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 = GVAEndpoint.re_inline.match(inline)
+ if m is None:
+ raise MalformedDocumentError(GVAEndpoint.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 GVAEndpoint.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, BMAEndpoint):
+ 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..842900ab 100644
--- a/duniterpy/constants.py
+++ b/duniterpy/constants.py
@@ -25,10 +25,8 @@ BLOCK_ID_REGEX = "[0-9]+"
BLOCK_UID_REGEX = "{block_id_regex}-{block_hash_regex}".format(
block_id_regex=BLOCK_ID_REGEX, block_hash_regex=BLOCK_HASH_REGEX
)
-CONDITIONS_REGEX = (
- "(&&|\\|\\|| |[()]|(SIG\\({pubkey_regex}\\)|(XHX\\({hash_regex}\\))))*".format(
- pubkey_regex=PUBKEY_REGEX, hash_regex=HASH_REGEX
- )
+CONDITIONS_REGEX = "(&&|\\|\\|| |[()]|(SIG\\({pubkey_regex}\\)|(XHX\\({hash_regex}\\))))*".format(
+ pubkey_regex=PUBKEY_REGEX, hash_regex=HASH_REGEX
)
IPV4_REGEX = (
"(?:(?:[0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\\.){3}(?:[0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4]["
@@ -59,3 +57,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..56813a10
--- /dev/null
+++ b/tests/api/test_endpoints.py
@@ -0,0 +1,59 @@
+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)
+
+ assert gva_endpoint.flags == ""
+ assert gva_endpoint.server == "test.domain.com"
+ assert gva_endpoint.ipv4 == "127.0.0.1"
+ assert gva_endpoint.ipv6 == "2001:0db8:0000:85a3:0000:0000:ac1f:8001"
+ assert gva_endpoint.port == 10902
+ assert gva_endpoint.path == "gva"
+
+ assert gva_endpoint.inline() == endpoint_str
+
+ endpoint_str = "GVA S test.domain.com 10902 gva"
+
+ gva_endpoint = endpoint.GVAEndpoint.from_inline(endpoint_str)
+
+ assert gva_endpoint.flags == "S"
+ assert gva_endpoint.server == "test.domain.com"
+ assert gva_endpoint.ipv4 is None
+ assert gva_endpoint.ipv6 is None
+ assert gva_endpoint.port == 10902
+ assert gva_endpoint.path == "gva"
+
+ assert 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"
+
+ gva_endpoint = endpoint.GVAEndpoint.from_inline(endpoint_str)
+
+ assert gva_endpoint.flags == ""
+ assert gva_endpoint.server == "test.domain.com"
+ assert gva_endpoint.ipv4 == "127.0.0.1"
+ assert gva_endpoint.ipv6 == "2001:0db8:0000:85a3:0000:0000:ac1f:8001"
+ assert gva_endpoint.port == 10902
+ assert gva_endpoint.path == "gva"
+
+ assert gva_endpoint.inline() == endpoint_str
+
+ endpoint_str = "GVASUB S test.domain.com 10902 gva"
+
+ gva_endpoint = endpoint.GVAEndpoint.from_inline(endpoint_str)
+
+ assert gva_endpoint.flags == "S"
+ assert gva_endpoint.server == "test.domain.com"
+ assert gva_endpoint.ipv4 is None
+ assert gva_endpoint.ipv6 is None
+ assert gva_endpoint.port == 10902
+ assert gva_endpoint.path == "gva"
+
+ assert gva_endpoint.inline() == endpoint_str
--
GitLab