Coverage for silkaj/network.py: 85%

1# Copyright 2016-2025 Maël Azimi <m.a@moul.re> 

2# 

3# Silkaj is free software: you can redistribute it and/or modify 

4# it under the terms of the GNU Affero General Public License as published by 

5# the Free Software Foundation, either version 3 of the License, or 

6# (at your option) any later version. 

7# 

8# Silkaj is distributed in the hope that it will be useful, 

9# but WITHOUT ANY WARRANTY; without even the implied warranty of 

10# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

11# GNU Affero General Public License for more details. 

12# 

13# You should have received a copy of the GNU Affero General Public License 

14# along with Silkaj. If not, see <https://www.gnu.org/licenses/>. 

15 

16import functools 

17import re 

18import sys 

19from typing import Any 

20from urllib.error import HTTPError 

21 

22from duniterpy import constants as du_const 

23from duniterpy.api import endpoint as ep 

24from duniterpy.api.client import Client 

25from duniterpy.documents import Document 

26 

27from silkaj import constants 

28 

29 

30def determine_endpoint() -> ep.Endpoint: 

31 """ 

32 Pass custom endpoint, parse through a regex 

33 {host|ipv4|[ipv6]}:{port}{/path} 

34 ^(?:(HOST)|(IPV4|[(IPV6)]))(?::(PORT))?(?:/(PATH))?$ 

35 If gtest flag passed, return default gtest endpoint 

36 Else, return g1 default endpoint 

37 """ 

38 

39 regex = f"^(?:(?P<host>{du_const.HOST_REGEX})|(?P<ipv4>{du_const.IPV4_REGEX})|\ 

40(?:\\[(?P<ipv6>{du_const.IPV6_REGEX})\\]))(?::(?P<port>{du_const.PORT_REGEX}))?\ 

41(?:/(?P<path>{du_const.PATH_REGEX}))?$" 

42 

43 try: 

44 from click.globals import get_current_context 

45 

46 ctx = get_current_context() 

47 endpoint = ctx.obj.get("ENDPOINT", None) 

48 gtest = ctx.obj.get("GTEST", None) 

49 except (ModuleNotFoundError, RuntimeError): 

50 endpoint, gtest = None, None 

51 

52 if endpoint: 

53 m = re.search(re.compile(regex), endpoint) 

54 if not m: 

55 sys.exit( 

56 "Error: Passed endpoint is of wrong format.\n\ 

57Expected format: {host|ipv4|[ipv6]}:{port}{/path}", 

58 ) 

59 port = int(m["port"]) if m["port"] else 443 

60 host, ipv4 = ep.fix_host_ipv4_mix_up(m["host"], m["ipv4"]) 

61 

62 if port == 443: 

63 return ep.SecuredBMAEndpoint(host, ipv4, m["ipv6"], port, m["path"]) 

64 return ep.BMAEndpoint(host, ipv4, m["ipv6"], port) 

65 

66 if gtest: 

67 return ep.endpoint(constants.G1_TEST_DEFAULT_ENDPOINT) 

68 return ep.endpoint(constants.G1_DEFAULT_ENDPOINT) 

69 

70 

71@functools.lru_cache(maxsize=1) 

72def client_instance(): 

73 return Client(determine_endpoint()) 

74 

75 

76def send_document(bma_path: Any, document: Document) -> None: 

77 client = client_instance() 

78 doc_name = document.__class__.__name__ 

79 try: 

80 client(bma_path, document.signed_raw()) 

81 print(f"{doc_name} successfully sent") 

82 except HTTPError as error: 

83 print(error) 

84 sys.exit(f"Error while publishing {doc_name.lower()}") 

85 

86 

87def exit_on_http_error(error: HTTPError, err_code: int, message: str) -> None: 

88 """ 

89 Nicely displays a message on an expected error code. 

90 Else, displays the HTTP error message. 

91 """ 

92 if error.code == err_code: 

93 sys.exit(message) 

94 print(error) 

95 sys.exit(constants.FAILURE_EXIT_STATUS)