diff --git a/silkaj/constants.py b/silkaj/constants.py
index 04cc3000325232e0beda683f9be83fcee4d203db..b49ffe9000bedbb638d31362e665b2728b8f1f75 100644
--- a/silkaj/constants.py
+++ b/silkaj/constants.py
@@ -16,6 +16,11 @@
SILKAJ_VERSION = "0.10.0dev"
G1_SYMBOL = "Ğ1"
GTEST_SYMBOL = "ĞTest"
+G1_GVA_DEFAULT_ENDPOINT = "GVA S duniter-g1.p2p.legal 163.172.99.239 443 gva"
+# G1_GVA_DEFAULT_ENDPOINT = "GVA S g1.elo.tf 443 gva"
+# G1_TEST_GVA_DEFAULT_ENDPOINT = "GVA S gt.elo.tf 443 gva"
+# G1_TEST_GVA_DEFAULT_ENDPOINT = "GVA S gt.librelois.fr 443 gva"
+G1_TEST_GVA_DEFAULT_ENDPOINT = "GVA S g1-test-dev.pini.fr 443 gva"
G1_DEFAULT_ENDPOINT = "g1.duniter.org", "443"
G1_TEST_DEFAULT_ENDPOINT = "g1-test.duniter.org", "443"
ASYNC_SLEEP = 0.15
diff --git a/silkaj/network_tools.py b/silkaj/network_tools.py
index d7d583665133422e3788a8f08841f02c883ff6b7..89347b5e5ea3cf0c5237571cf4b8f064871c9887 100644
--- a/silkaj/network_tools.py
+++ b/silkaj/network_tools.py
@@ -13,9 +13,17 @@
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
+import re
+import sys
+from ipaddress import ip_address
+
+import click
+import graphql
+from duniterpy import constants as du_const
from duniterpy.api.client import Client
+from duniterpy.api.endpoint import GVAEndpoint
-from silkaj.constants import G1_DEFAULT_ENDPOINT, G1_TEST_DEFAULT_ENDPOINT
+from silkaj import constants
def singleton(class_):
@@ -49,9 +57,10 @@ class EndPoint(object):
else:
ep["domain"], ep["port"] = peer, "443"
else:
- ep["domain"], ep["port"] = (
- G1_TEST_DEFAULT_ENDPOINT if gtest else G1_DEFAULT_ENDPOINT
- )
+ if gtest:
+ ep["domain"], ep["port"] = constants.G1_TEST_DEFAULT_ENDPOINT
+ else:
+ ep["domain"], ep["port"] = constants.G1_DEFAULT_ENDPOINT
if ep["domain"].startswith("[") and ep["domain"].endswith("]"):
ep["domain"] = ep["domain"][1:-1]
self.ep = ep
@@ -63,3 +72,96 @@ class EndPoint(object):
class ClientInstance(object):
def __init__(self):
self.client = Client(EndPoint().BMA_ENDPOINT)
+
+
+@click.pass_context
+def determine_endpoint(ctx):
+ """
+ Pass custom endpoint, parse through a regex
+ {host|ipv4|[ipv6]}:{port}{/path}
+ ^(?:(HOST)|(IPV4|[(IPV6)]))(?::(PORT))?(?:/(PATH))?$
+ If gtest flag passed, return default gtest endpoint
+ Else, return g1 default endpoint
+ """
+ # TODO: Correct HOST_REGEX in DuniterPy since it allows 127.0.0.1 which should go into IPv4
+ # Use ipaddress.ip_address().version to confirm the kind of address
+ # >>> from duniterpy.api.endpoint import GVAEndpoint
+ # >>> GVAEndpoint.from_inline("GVA S 127.0.0.1 443 gva").server
+ # '127.0.0.1'
+ # >>> GVAEndpoint.from_inline("GVA S 127.0.0.1 443 gva").ipv4
+ # >>>
+ # TODO: DuniterPy: ports regex between 0 and 65536
+ # https://stackoverflow.com/a/12968117
+ PORT_REGEX = "[0-9]{1,4}|[1-5][0-9]{4}|6[0-4][0-9]{3}|65[0-4][0-9]{2}|655[0-2][0-9]|6553[0-5]"
+ regex = f"^(?:(?P<host>{du_const.HOST_REGEX})|(?P<ipv4>{du_const.IPV4_REGEX})|\
+(?:\\[(?P<ipv6>{du_const.IPV6_REGEX})\\]))(?::(?P<port>{PORT_REGEX}))?(?:/(?P<path>{du_const.PATH_REGEX}))?$"
+ endpoint = ctx.obj["PEER"]
+ if endpoint:
+ m = re.search(re.compile(regex), endpoint)
+ if not m:
+ sys.exit(
+ "Error: Passed endpoint is of wrong format.\n\
+Expected format: {host|ipv4|[ipv6]}:{port}{/path}"
+ )
+ port = int(m["port"]) if m["port"] else 443
+ flags = "S" if port == 443 else ""
+ host = m["host"]
+ ipv4 = m["ipv4"]
+
+ try:
+ if ip_address(host).version == 4:
+ ipv4 = host
+ host = None
+ except ValueError:
+ pass
+ return GVAEndpoint(flags, host, ipv4, m["ipv6"], port, m["path"])
+
+ elif ctx.obj["GTEST"]:
+ return GVAEndpoint.from_inline(constants.G1_TEST_GVA_DEFAULT_ENDPOINT)
+ else:
+ return GVAEndpoint.from_inline(constants.G1_GVA_DEFAULT_ENDPOINT)
+
+
+async def gva_query(query, endpoint=None):
+ if not endpoint:
+ endpoint = determine_endpoint()
+ client = Client(endpoint)
+ schema = await get_graphql_schema(client)
+ check_query(schema, query)
+ response = await client.query(query)
+ await client.close()
+ return response["data"]
+
+
+async def get_graphql_schema(client):
+ # Get query to get schema from api
+ introspection_query = graphql.get_introspection_query(False)
+
+ # Get schema from api
+ import urllib
+
+ try:
+ response = await client.query(introspection_query)
+ except urllib.error.HTTPError:
+ sys.exit("Error: The specified endpoint is not responding")
+
+ # TODO: handle when the path is incorrect
+ # handle exception when response is empty
+ # In DuniterPy would be greater
+ # when migrated to urllib to use urllib.error.HTTPError exception
+
+ # Convert response dict to schema
+ return graphql.build_client_schema(response["data"])
+
+
+def check_query(schema, query):
+ # Check query syntax
+ try:
+ ast_document = graphql.language.parse(query)
+ except graphql.error.GraphQLSyntaxError as exception:
+ sys.exit(f"Query syntax error: {exception.message}")
+
+ # Validate query against schema
+ errors = graphql.validate(schema, ast_document)
+ if errors:
+ sys.exit(f"Schema errors: {errors}")
diff --git a/tests/helpers.py b/tests/helpers.py
new file mode 100644
index 0000000000000000000000000000000000000000..90b003dbdd09875274b5eba682ab2f5a74b83211
--- /dev/null
+++ b/tests/helpers.py
@@ -0,0 +1,26 @@
+# Copyright 2016-2021 Maël Azimi <m.a@moul.re>
+#
+# Silkaj is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Silkaj 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 Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
+
+import click
+
+from silkaj import cli
+
+
+def define_click_context(endpoint=None, gtest=False):
+ ctx = click.Context(cli.cli)
+ ctx.obj = dict()
+ ctx.obj["PEER"] = endpoint
+ ctx.obj["GTEST"] = gtest
+ click.globals.push_context(ctx)
diff --git a/tests/test_network_tools.py b/tests/test_network_tools.py
index 85d30850019f1b19e41f3218012a421fd2e63b4c..f24a50dec40e179ee0d7a7ffe6bcea5f4aa78170 100644
--- a/tests/test_network_tools.py
+++ b/tests/test_network_tools.py
@@ -14,5 +14,47 @@
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
import pytest
+from duniterpy.api.endpoint import GVAEndpoint
-from silkaj import network_tools
+from silkaj import constants, network_tools
+from tests import helpers
+
+ipv6 = "2001:0db8:85a3:0000:0000:8a2e:0370:7334"
+
+
+@pytest.mark.parametrize(
+ "endpoint, host, ipv4, ipv6, port, path",
+ [
+ ("127.0.0.1", None, "127.0.0.1", None, 443, None),
+ ("127.0.0.1:80", None, "127.0.0.1", None, 80, None),
+ ("127.0.0.1:443", None, "127.0.0.1", None, 443, None),
+ ("127.0.0.1/path", None, "127.0.0.1", None, 443, "path"),
+ ("127.0.0.1:80/path", None, "127.0.0.1", None, 80, "path"),
+ ("domain.tld:80/path", "domain.tld", None, None, 80, "path"),
+ ("localhost:80/path", "localhost", None, None, 80, "path"),
+ (f"[{ipv6}]", None, None, ipv6, 443, None),
+ (f"[{ipv6}]/path", None, None, ipv6, 443, "path"),
+ (f"[{ipv6}]:80/path", None, None, ipv6, 80, "path"),
+ ],
+)
+def test_determine_endpoint_custom(endpoint, host, ipv4, ipv6, port, path):
+ helpers.define_click_context(endpoint)
+ ep = network_tools.determine_endpoint()
+ assert ep.server == host
+ assert ep.ipv4 == ipv4
+ assert ep.ipv6 == ipv6
+ assert ep.port == port
+ assert ep.path == path
+
+
+@pytest.mark.parametrize(
+ "gtest, endpoint",
+ [
+ (True, constants.G1_TEST_GVA_DEFAULT_ENDPOINT),
+ (False, constants.G1_GVA_DEFAULT_ENDPOINT),
+ ],
+)
+def test_determine_endpoint(gtest, endpoint):
+ helpers.define_click_context(gtest=gtest)
+ ep = network_tools.determine_endpoint()
+ assert ep == GVAEndpoint.from_inline(endpoint)