[enh] #59 add GVAEndpoint and GVASUBEndpoint classes

add unit tests
example now use real GVA Endpoint inline format

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
 

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]
 
 

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]"

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"
 
 
 ################################################

tests/api/test_endpoints.py

+"""
+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