Coverage for seedboxsync/core/ping/client/healthchecks.py: 100%

1# 

2# Copyright (C) 2015-2026 Guillaume Kulakowski <guillaume@kulakowski.fr> 

3# 

4# For the full copyright and license information, please view the LICENSE 

5# file that was distributed with this source code. 

6"""Healthchecks management for SeedboxSync.""" 

7 

8import urllib.request 

9from seedboxsync.core import current_app 

10from seedboxsync.core.ping import AbstractPingClient 

11 

12 

13class Healthchecks(AbstractPingClient): 

14 """Class to manage Healthchecks's pings.""" 

15 

16 def __init__(self) -> None: 

17 """Constructor for Healthchecks.""" 

18 self.app = current_app 

19 

20 def start(self, sub_command: str) -> None: 

21 """ 

22 Send a start ping to Healthchecks for a given subcommand. 

23 

24 Args: 

25 sub_command (str): The SeedboxSync subcommand to ping. 

26 """ 

27 enabled = self.app.seedboxsync_config.get("healthchecks_" + sub_command + "_enabled") 

28 if enabled is False: 

29 self.app.logger.info(f'Healthchecks for "{sub_command}" disabled by configuration') 

30 else: 

31 base_url = self.app.seedboxsync_config.get("healthchecks_" + sub_command + "_ping_url", "") 

32 ping_url = f"{base_url.rstrip('/')}/start" 

33 self.app.logger.debug(f"Ping url: {ping_url}") 

34 

35 try: 

36 urllib.request.urlopen(ping_url, timeout=10) 

37 except OSError: 

38 self.app.logger.exception("Healthchecks, ping failed") 

39 

40 def success(self, sub_command: str) -> None: 

41 """ 

42 Send a success ping to Healthchecks for a given subcommand. 

43 

44 Args: 

45 sub_command (str): The SeedboxSync subcommand to ping. 

46 """ 

47 enabled = self.app.seedboxsync_config.get("healthchecks_" + sub_command + "_enabled") 

48 if enabled is False: 

49 self.app.logger.info(f'Healthchecks for "{sub_command}" disabled by configuration') 

50 else: 

51 ping_url = self.app.seedboxsync_config.get("healthchecks_" + sub_command + "_ping_url", "") 

52 self.app.logger.debug(f"Ping url: {ping_url}") 

53 

54 try: 

55 urllib.request.urlopen(ping_url, timeout=10) 

56 except OSError: 

57 self.app.logger.exception("Healthchecks, ping failed.")