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

[enh] #137 add GVAEndpoint and GVASUBEndpoint classes

add unit tests
example now use real GVA Endpoint inline format
parent 9a3d2e4f
No related branches found
No related tags found
2 merge requests!119Release 0.61.0,!90Add GraphQL GVA API support
...@@ -515,7 +515,7 @@ class Client: ...@@ -515,7 +515,7 @@ class Client:
try: try:
result = await response.json() result = await response.json()
except aiohttp.client_exceptions.ContentTypeError as exception: 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 response to debug...
return result return result
......
...@@ -640,6 +640,151 @@ class ESSubscribtionEndpoint(Endpoint): ...@@ -640,6 +640,151 @@ class ESSubscribtionEndpoint(Endpoint):
return hash((ESSubscribtionEndpoint.API, self.server, self.port)) 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"
endpoint_format = f"^GVA(?: ({constants.ENDPOINT_FLAGS_REGEX}))?(?: ({constants.HOST_REGEX}))?(?: ({constants.IPV4_REGEX}))?(?: ({constants.IPV6_REGEX}))? ([0-9]+)(?: ({constants.PATH_REGEX}))?$"
re_inline = re.compile(endpoint_format)
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"
endpoint_format = f"^GVASUB(?: ({constants.ENDPOINT_FLAGS_REGEX}))?(?: ({constants.HOST_REGEX}))?(?: ({constants.IPV4_REGEX}))?(?: ({constants.IPV6_REGEX}))? ([0-9]+)(?: ({constants.PATH_REGEX}))?$"
re_inline = re.compile(endpoint_format)
MANAGED_API = { MANAGED_API = {
BMAEndpoint.API: BMAEndpoint, BMAEndpoint.API: BMAEndpoint,
SecuredBMAEndpoint.API: SecuredBMAEndpoint, SecuredBMAEndpoint.API: SecuredBMAEndpoint,
...@@ -647,6 +792,8 @@ MANAGED_API = { ...@@ -647,6 +792,8 @@ MANAGED_API = {
ESCoreEndpoint.API: ESCoreEndpoint, ESCoreEndpoint.API: ESCoreEndpoint,
ESUserEndpoint.API: ESUserEndpoint, ESUserEndpoint.API: ESUserEndpoint,
ESSubscribtionEndpoint.API: ESSubscribtionEndpoint, ESSubscribtionEndpoint.API: ESSubscribtionEndpoint,
GVAEndpoint.API: GVAEndpoint,
GVASUBEndpoint.API: GVASUBEndpoint,
} # type: Dict[str, Any] } # type: Dict[str, Any]
......
...@@ -59,3 +59,4 @@ WS2P_PRIVATE_PREFIX_REGEX = "O[CT][SAM]" ...@@ -59,3 +59,4 @@ WS2P_PRIVATE_PREFIX_REGEX = "O[CT][SAM]"
WS2P_PUBLIC_PREFIX_REGEX = "I[CT]" WS2P_PUBLIC_PREFIX_REGEX = "I[CT]"
WS2P_HEAD_REGEX = "HEAD:?(?:[0-9]+)?" WS2P_HEAD_REGEX = "HEAD:?(?:[0-9]+)?"
EMPTY_HASH = "E3B0C44298FC1C149AFBF4C8996FB92427AE41E4649B934CA495991B7852B855" EMPTY_HASH = "E3B0C44298FC1C149AFBF4C8996FB92427AE41E4649B934CA495991B7852B855"
ENDPOINT_FLAGS_REGEX = "[S]"
...@@ -11,7 +11,7 @@ from graphql.error import GraphQLSyntaxError ...@@ -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] # 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] # 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 # 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"
################################################ ################################################
...@@ -39,14 +39,13 @@ async def main(): ...@@ -39,14 +39,13 @@ async def main():
try: try:
ast_document = language.parse(query) ast_document = language.parse(query)
except GraphQLSyntaxError as exception: except GraphQLSyntaxError as exception:
print("Query syntax error: {0}".format(exception.message)) print(f"Query syntax error: {exception.message}")
sys.exit(1) sys.exit(1)
# validate query against schema # validate query against schema
errors = validate(schema, ast_document) errors = validate(schema, ast_document)
if errors: if errors:
print("Schema errors:") print(f"Schema errors: {errors}")
print(errors)
sys.exit(1) sys.exit(1)
# send valid query to api # send valid query to api
......
"""
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
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment