[enh] #59 add GVAEndpoint and GVASUBEndpoint classes

add unit tests
example now use real GVA Endpoint inline format

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]
 
 

duniterpy/constants.py

@@ -25,11 +25,9 @@ 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(
+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]["
     "0-9]|25[0-5])"
@@ -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]"

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

+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