10975988831

Committed 21 Sep 2024 10:06PM UTC coverage: 85.67%. Remained the same

Build # 10975988831

Build Type

Pull #1361

github

Committed by

web-flow

Commit Message

Merge 7780b3b80 into fa2e7d9d3

Pull Request Pull Request #1361: add files app, add models, add view to serve files

Run Details

2928 of 3420 branches covered (85.61%)

Branch coverage included in aggregate %.

37 of 43 new or added lines in 7 files covered. (86.05%)

39 existing lines in 5 files now uncovered.

11594 of 13531 relevant lines covered (85.68%)

0.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.92

/ephios/core/services/health/healthchecks.py

import os

from django.conf import settings
from django.contrib.auth.models import Permission
from django.contrib.humanize.templatetags.humanize import naturaltime
from django.dispatch import receiver
from django.template.defaultfilters import floatformat
from django.utils.safestring import mark_safe
from django.utils.translation import gettext_lazy as _

from ephios.core.dynamic_preferences_registry import LastRunPeriodicCall
from ephios.core.signals import register_healthchecks

# health checks are meant to monitor the health of the application while it is running
# in contrast there are django checks which are meant to check the configuration of the application


def run_healthchecks():
    for _, healthchecks in register_healthchecks.send(None):
        for HealthCheck in healthchecks:
            check = HealthCheck()
            status, message = check.check()
            yield check, status, message


class HealthCheckStatus:
    OK = "ok"
    WARNING = "warning"
    ERROR = "error"
    UNKNOWN = "unknown"


class AbstractHealthCheck:
    @property
    def slug(self):
        """
        Return a unique slug for this health check.
        """
        raise NotImplementedError

    @property
    def name(self):
        """
        Return a short name of this health check.
        """
        raise NotImplementedError

    @property
    def description(self):
        """
        Return a short description of this health check.
        """
        raise NotImplementedError

    @property
    def documentation_link(self):
        """
        Return a link to the documentation of this health check.
        """
        return None

    def check(self):
        """
        Return a tuple of (status, message) where status is one of HealthCheckStatus
        """
        raise NotImplementedError


class DBHealthCheck(AbstractHealthCheck):
    slug = "db"
    name = _("Database")
    description = _("The database is the central storage for all data.")
    documentation_link = "https://docs.djangoproject.com/en/stable/ref/databases/"

    def check(self):
        from django.db import connection

        try:
            connection.cursor()
            Permission.objects.exists()
        except Exception as e:
            return HealthCheckStatus.ERROR, str(e)

        if settings.DATABASES["default"]["ENGINE"] == "django.db.backends.sqlite3":
            return HealthCheckStatus.WARNING, _(
                "Using SQLite, which should not be used in production."
            )

        return HealthCheckStatus.OK, _("Database connection established.")


class CacheHealthCheck(AbstractHealthCheck):
    slug = "cache"
    name = _("Cache")
    description = _("The cache is used to store temporary data.")
    documentation_link = "https://docs.djangoproject.com/en/stable/topics/cache/"

    def check(self):
        from django.core import cache

        try:
            cache.cache.set("_healthcheck", "1")
            if not cache.cache.get("_healthcheck") == "1":
                raise Exception("Cache not available")
        except Exception as e:
            return HealthCheckStatus.ERROR, str(e)

        if (
            settings.CACHES.get("default", {}).get("BACKEND")
            == "django.core.cache.backends.locmem.LocMemCache"
        ):
            return HealthCheckStatus.WARNING, _(
                "Using LocMemCache, which should not be used in production."
            )

        return HealthCheckStatus.OK, _("Cache connection established.")


class CronJobHealthCheck(AbstractHealthCheck):
    slug = "cronjob"
    name = _("Cronjob")
    description = _(
        "A cron job must regularly call ephios to do recurring tasks like sending notifications."
    )
    documentation_link = (
        "https://docs.ephios.de/en/stable/admin/deployment/manual/index.html#setup-cron"
    )

    def check(self):
        last_call = LastRunPeriodicCall.get_last_call()
        if LastRunPeriodicCall.is_stuck():
            if last_call:
                return (
                    HealthCheckStatus.WARNING,
                    mark_safe(
                        _("Cronjob stuck, last run {last_call}.").format(
                            last_call=naturaltime(last_call),
                        )
                    ),
                )
            else:
                return (
                    HealthCheckStatus.ERROR,
                    mark_safe(_("Cronjob stuck, no last run.")),
                )
        else:
            return (
                HealthCheckStatus.OK,
                mark_safe(_("Last run {last_call}.").format(last_call=naturaltime(last_call))),
            )


class DiskSpaceHealthCheck(AbstractHealthCheck):
    slug = "disk_space"
    name = _("Disk space")
    description = _("Disk space needs to be available to store data.")
    documentation_link = (
        "https://docs.ephios.de/en/stable/admin/deployment/manual/index.html#data-directory"
    )

    def check(self):
        # if under 100 MB are available, we consider this an error
        # if under 1 GB are available, we consider this a warning
        # otherwise, we consider this ok
        try:
            disk_usage = os.statvfs(settings.MEDIA_ROOT)
        except AttributeError:
            # not available on windows
            return (
                HealthCheckStatus.UNKNOWN,
                mark_safe(_("Disk space check not available.")),
            )
        free_bytes = disk_usage.f_bavail * disk_usage.f_frsize
        MEGA = 1024 * 1024
        if free_bytes < 100 * MEGA:
            return (
                HealthCheckStatus.ERROR,
                mark_safe(
                    _(
                        "Less than 100 MB of disk space available. "
                        "Please free up some disk space."
                    )
                ),
            )
        if free_bytes < 1024 * MEGA:
            return (
                HealthCheckStatus.WARNING,
                mark_safe(
                    _("Less than 1 GB of disk space available. Please free up some disk space.")
                ),
            )
        return (
            HealthCheckStatus.OK,
            mark_safe(
                _("{disk_space} of disk space available.").format(
                    disk_space=f"{floatformat(free_bytes / MEGA / 1024,1)} GB"
                )
            ),
        )


@receiver(register_healthchecks, dispatch_uid="ephios.core.healthchecks.register_core_healthchecks")
def register_core_healthchecks(sender, **kwargs):
    yield DBHealthCheck
    yield CacheHealthCheck
    yield CronJobHealthCheck
    yield DiskSpaceHealthCheck

1	import os	1✔
2
3	from django.conf import settings	1✔
4	from django.contrib.auth.models import Permission	1✔
5	from django.contrib.humanize.templatetags.humanize import naturaltime	1✔
6	from django.dispatch import receiver	1✔
7	from django.template.defaultfilters import floatformat	1✔
8	from django.utils.safestring import mark_safe	1✔
9	from django.utils.translation import gettext_lazy as _	1✔
10
11	from ephios.core.dynamic_preferences_registry import LastRunPeriodicCall	1✔
12	from ephios.core.signals import register_healthchecks	1✔
13
14	# health checks are meant to monitor the health of the application while it is running
15	# in contrast there are django checks which are meant to check the configuration of the application
16
17
18	def run_healthchecks():	1✔
19	for _, healthchecks in register_healthchecks.send(None):	1✔
20	for HealthCheck in healthchecks:	1✔
21	check = HealthCheck()	1✔
22	status, message = check.check()	1✔
23	yield check, status, message	1✔
24
25
26	class HealthCheckStatus:	1✔
27	OK = "ok"	1✔
28	WARNING = "warning"	1✔
29	ERROR = "error"	1✔
30	UNKNOWN = "unknown"	1✔
31
32
33	class AbstractHealthCheck:	1✔
34	@property	1✔
35	def slug(self):	1✔
36	"""
37	Return a unique slug for this health check.
38	"""
39	raise NotImplementedError	×
40
41	@property	1✔
42	def name(self):	1✔
43	"""
44	Return a short name of this health check.
45	"""
46	raise NotImplementedError	×
47
48	@property	1✔
49	def description(self):	1✔
50	"""
51	Return a short description of this health check.
52	"""
53	raise NotImplementedError	×
54
55	@property	1✔
56	def documentation_link(self):	1✔
57	"""
58	Return a link to the documentation of this health check.
59	"""
60	return None	×
61
62	def check(self):	1✔
63	"""
64	Return a tuple of (status, message) where status is one of HealthCheckStatus
65	"""
66	raise NotImplementedError	×
67
68
69	class DBHealthCheck(AbstractHealthCheck):	1✔
70	slug = "db"	1✔
71	name = _("Database")	1✔
72	description = _("The database is the central storage for all data.")	1✔
73	documentation_link = "https://docs.djangoproject.com/en/stable/ref/databases/"	1✔
74
75	def check(self):	1✔
76	from django.db import connection	1✔
77
78	try:	1✔
79	connection.cursor()	1✔
80	Permission.objects.exists()	1✔
81	except Exception as e:	×
82	return HealthCheckStatus.ERROR, str(e)	×
83
84	if settings.DATABASES["default"]["ENGINE"] == "django.db.backends.sqlite3":	1!
85	return HealthCheckStatus.WARNING, _(	1✔
86	"Using SQLite, which should not be used in production."
87	)
88
89	return HealthCheckStatus.OK, _("Database connection established.")	×
90
91
92	class CacheHealthCheck(AbstractHealthCheck):	1✔
93	slug = "cache"	1✔
94	name = _("Cache")	1✔
95	description = _("The cache is used to store temporary data.")	1✔
96	documentation_link = "https://docs.djangoproject.com/en/stable/topics/cache/"	1✔
97
98	def check(self):	1✔
99	from django.core import cache	1✔
100
101	try:	1✔
102	cache.cache.set("_healthcheck", "1")	1✔
103	if not cache.cache.get("_healthcheck") == "1":	1!
104	raise Exception("Cache not available")	×
105	except Exception as e:	×
106	return HealthCheckStatus.ERROR, str(e)	×
107
108	if (	1!
109	settings.CACHES.get("default", {}).get("BACKEND")
110	== "django.core.cache.backends.locmem.LocMemCache"
111	):
112	return HealthCheckStatus.WARNING, _(	1✔
113	"Using LocMemCache, which should not be used in production."
114	)
115
116	return HealthCheckStatus.OK, _("Cache connection established.")	×
117
118
119	class CronJobHealthCheck(AbstractHealthCheck):	1✔
120	slug = "cronjob"	1✔
121	name = _("Cronjob")	1✔
122	description = _(	1✔
123	"A cron job must regularly call ephios to do recurring tasks like sending notifications."
124	)
125	documentation_link = (	1✔
126	"https://docs.ephios.de/en/stable/admin/deployment/manual/index.html#setup-cron"
127	)
128
129	def check(self):	1✔
130	last_call = LastRunPeriodicCall.get_last_call()	1✔
131	if LastRunPeriodicCall.is_stuck():	1!
132	if last_call:	1!
133	return (	×
134	HealthCheckStatus.WARNING,
135	mark_safe(
136	_("Cronjob stuck, last run {last_call}.").format(
137	last_call=naturaltime(last_call),
138	)
139	),
140	)
141	else:
142	return (	1✔
143	HealthCheckStatus.ERROR,
144	mark_safe(_("Cronjob stuck, no last run.")),
145	)
146	else:
147	return (	×
148	HealthCheckStatus.OK,
149	mark_safe(_("Last run {last_call}.").format(last_call=naturaltime(last_call))),
150	)
151
152
153	class DiskSpaceHealthCheck(AbstractHealthCheck):	1✔
154	slug = "disk_space"	1✔
155	name = _("Disk space")	1✔
156	description = _("Disk space needs to be available to store data.")	1✔
157	documentation_link = (	1✔
158	"https://docs.ephios.de/en/stable/admin/deployment/manual/index.html#data-directory"
159	)
160
161	def check(self):	1✔
162	# if under 100 MB are available, we consider this an error
163	# if under 1 GB are available, we consider this a warning
164	# otherwise, we consider this ok
165	try:	1✔
166	disk_usage = os.statvfs(settings.MEDIA_ROOT)	1✔
UNCOV 167	except AttributeError:	×
168	# not available on windows
UNCOV 169	return (	×
170	HealthCheckStatus.UNKNOWN,
171	mark_safe(_("Disk space check not available.")),
172	)
173	free_bytes = disk_usage.f_bavail * disk_usage.f_frsize	1✔
174	MEGA = 1024 * 1024	1✔
175	if free_bytes < 100 * MEGA:	1!
UNCOV 176	return (	×
177	HealthCheckStatus.ERROR,
178	mark_safe(
179	_(
180	"Less than 100 MB of disk space available. "
181	"Please free up some disk space."
182	)
183	),
184	)
185	if free_bytes < 1024 * MEGA:	1!
UNCOV 186	return (	×
187	HealthCheckStatus.WARNING,
188	mark_safe(
189	_("Less than 1 GB of disk space available. Please free up some disk space.")
190	),
191	)
192	return (	1✔
193	HealthCheckStatus.OK,
194	mark_safe(
195	_("{disk_space} of disk space available.").format(
196	disk_space=f"{floatformat(free_bytes / MEGA / 1024,1)} GB"
197	)
198	),
199	)
200
201
202	@receiver(register_healthchecks, dispatch_uid="ephios.core.healthchecks.register_core_healthchecks")	1✔
203	def register_core_healthchecks(sender, **kwargs):	1✔
204	yield DBHealthCheck	1✔
205	yield CacheHealthCheck	1✔
206	yield CronJobHealthCheck	1✔
207	yield DiskSpaceHealthCheck	1✔

ephios-dev / ephios / 10975988831

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous