Coverage for seedboxsync/cli/commands/cmd_health.py: 95%

40 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-07-26 17:46 +0000

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# 

7"""All commands related to health checks in SeedboxSync.""" 

8 

9from datetime import datetime, timedelta 

10from typing import Any 

11from urllib.error import URLError 

12from urllib.request import urlopen 

13import click 

14from seedboxsync.__version__ import __version__ as version 

15from seedboxsync.cli import Context, pass_context 

16from seedboxsync.core import utils 

17from seedboxsync.core.dao import SeedboxSync, TaskStatus, typed_peewee_dict 

18 

19 

20@click.command("health") 

21@pass_context 

22def cli(ctx: Context) -> None: 

23 """ 

24 Show the health status of the SeedboxSync CLI and web service. 

25 

26 Args: 

27 ctx (Context): The SeedboxSync Click context. 

28 """ 

29 exit_code = 0 

30 

31 # CLI part 

32 db_version = SeedboxSync.get_db_version() 

33 click.echo(f"Version: {version}") 

34 click.echo(f"Database version: {db_version}") 

35 click.secho("CLI - OK", fg="green") 

36 

37 # Task manager part 

38 heartbeat: dict[str, Any] | None = None 

39 try: 

40 heartbeat = typed_peewee_dict( 

41 TaskStatus.select( 

42 TaskStatus.key, 

43 TaskStatus.running, 

44 TaskStatus.started, 

45 TaskStatus.finished, 

46 ) 

47 .where(TaskStatus.key == "heartbeat") 

48 .dicts() 

49 .first() 

50 ) 

51 except TaskStatus.DoesNotExist: # type: ignore[attr-defined] 

52 exit_code = 1 

53 

54 if heartbeat is None: 

55 click.secho("Task manager - NOK", fg="red") 

56 exit_code = 2 

57 elif datetime.now() - heartbeat["finished"] > timedelta(minutes=5): 

58 click.secho("Task manager - NOK", fg="red") 

59 exit_code = 3 

60 else: 

61 click.secho("Task manager - OK", fg="green") 

62 

63 # Flask WebUI part 

64 health_url = utils.get_web_healthcheck_url() 

65 try: 

66 with urlopen(health_url, timeout=5) as response: 

67 if response.status == 200: 

68 click.secho("WebUI - OK", fg="green") 

69 else: 

70 click.secho("WebUI - NOK", fg="red") 

71 exit_code = 5 

72 except URLError: 

73 click.secho("WebUI - NOK", fg="red") 

74 exit_code = 6 

75 

76 ctx.exit(exit_code)