Coverage for silkaj/network.py: 85%

1# Copyright 2016-2024 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 )|\ 

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

42 ))?\ 

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

44 

45 try: 

46 from click.globals import get_current_context 

47 

48 ctx = get_current_context() 

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

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

51 except (ModuleNotFoundError, RuntimeError): 

52 endpoint, gtest = None, None 

53 

54 if endpoint: 

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

56 if not m: 

57 sys.exit( 

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

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

60 ) 

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

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

63 

64 if port == 443: 

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

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

67 

68 if gtest: 

69 return ep.endpoint(constants.G1_TEST_DEFAULT_ENDPOINT) 

70 return ep.endpoint(constants.G1_DEFAULT_ENDPOINT) 

71 

72 

73@functools.lru_cache(maxsize=1) 

74def client_instance(): 

75 return Client(determine_endpoint()) 

76 

77 

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

79 client = client_instance() 

80 doc_name = document.__class__.__name__ 

81 try: 

82 client(bma_path, document.signed_raw()) 

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

84 except HTTPError as error: 

85 print(error) 

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

87 

88 

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

90 """ 

91 Nicely displays a message on an expected error code. 

92 Else, displays the HTTP error message. 

93 """ 

94 if error.code == err_code: 

95 sys.exit(message) 

96 print(error) 

97 sys.exit(constants.FAILURE_EXIT_STATUS)