diff --git a/docs/developer/utils.rst b/docs/developer/utils.rst index b5813c67..aaa24fcc 100644 --- a/docs/developer/utils.rst +++ b/docs/developer/utils.rst @@ -27,6 +27,33 @@ completes successfully, just before the response is returned. The ``view`` argument can also be used to access the ``request`` object i.e. ``view.request``. +``radius_accounting_closed`` +~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +**Path**: ``openwisp_radius.signals.radius_accounting_closed`` + +**Arguments**: + +- ``sender``: ``RadiusAccounting`` model class +- ``instance``: closed ``RadiusAccounting`` instance + +This signal is emitted when a ``RadiusAccounting`` session is closed. For +regular ``save()`` paths, it is emitted only when ``stop_time`` changes +from ``None`` to any value, or when a closed session is created directly. +Editing an already closed session does not emit this signal again. + +The signal is emitted after the database transaction is committed. + +Integrations which need to react to closed accounting sessions should +listen to this signal. For example, the monitoring integration uses this +signal to write RADIUS traffic snapshots for sessions closed by regular +``post_save`` paths, ``Accounting-On`` packets and automatic stale-session +cleanup. + +When closing multiple sessions with ``bulk_update()``, use +``RadiusAccounting.emit_radius_accounting_closed()`` after the database +update to emit this signal once for each closed session. + .. _radius_captive_portal_mock_views: Captive portal mock views diff --git a/openwisp_radius/base/models.py b/openwisp_radius/base/models.py index 5cbbf710..54f6cdfb 100644 --- a/openwisp_radius/base/models.py +++ b/openwisp_radius/base/models.py @@ -5,6 +5,7 @@ import string from datetime import timedelta from io import StringIO +from typing import Iterable import django import phonenumbers @@ -52,6 +53,7 @@ BATCH_MAIL_SUBJECT, DEFAULT_PASSWORD_RESET_URL, ) +from ..signals import radius_accounting_closed from ..utils import ( SmsMessage, decode_byte_data, @@ -529,20 +531,66 @@ class AbstractRadiusAccounting(OrgMixin, models.Model): blank=True, ) - def save(self, *args, **kwargs): - if not self.start_time: - self.start_time = now() - super(AbstractRadiusAccounting, self).save(*args, **kwargs) - class Meta: db_table = "radacct" verbose_name = _("accounting") verbose_name_plural = _("accountings") abstract = True + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + # used for radius_accounting_closed signal + self._set_initial_stop_time() + + def refresh_from_db(self, *args, **kwargs): + super().refresh_from_db(*args, **kwargs) + fields = kwargs.get("fields") + self._set_initial_stop_time(fields=fields) + + def save(self, *args, **kwargs): + created = self._state.adding + update_fields = kwargs.get("update_fields") + if not self.start_time: + self.start_time = now() + if update_fields is not None: + update_fields = set(update_fields) | {"start_time"} + kwargs["update_fields"] = update_fields + super(AbstractRadiusAccounting, self).save(*args, **kwargs) + self._emit_radius_accounting_closed( + created=created, update_fields=update_fields + ) + # reset after save + self._set_initial_stop_time(update_fields) + + def _set_initial_stop_time(self, fields=None): + if fields is None or "stop_time" in fields: + self._initial_stop_time = self.stop_time + + def _emit_radius_accounting_closed(self, created, update_fields=None): + """Detect whether this save closed the session and emit the signal.""" + if update_fields is not None and "stop_time" not in update_fields: + return + being_closed = self.stop_time is not None and ( + created or self._initial_stop_time is None + ) + if being_closed: + self.emit_radius_accounting_closed([self]) + def __str__(self): return self.unique_id + @classmethod + def emit_radius_accounting_closed( + cls, sessions: Iterable["AbstractRadiusAccounting"] + ) -> None: + """Emit radius_accounting_closed after commit for closed sessions.""" + for session in sessions: + transaction.on_commit( + lambda session=session: radius_accounting_closed.send( + sender=session.__class__, instance=session + ) + ) + @classmethod def close_stale_sessions(cls, days=None, hours=None): if hours: @@ -579,13 +627,39 @@ def _close_stale_sessions_on_nas_boot(cls, called_station_id): """ if not called_station_id: return 0 - stale_sessions = cls.objects.filter( - called_station_id=called_station_id, - stop_time__isnull=True, - ) - closed_count = stale_sessions.update( - stop_time=now(), terminate_cause="NAS-Reboot" - ) + stop_time = now() + closed_count = 0 + batch_size = 1000 + has_more_sessions = True + while has_more_sessions: + with transaction.atomic(): + closed_sessions = list( + cls.objects.select_for_update() + .filter( + called_station_id=called_station_id, + stop_time__isnull=True, + ) + .only( + "unique_id", + "username", + "organization_id", + "input_octets", + "output_octets", + "calling_station_id", + "called_station_id", + "stop_time", + )[:batch_size] + ) + has_more_sessions = len(closed_sessions) == batch_size + if not closed_sessions: + continue + for session in closed_sessions: + session.stop_time = stop_time + session.terminate_cause = "NAS-Reboot" + closed_count += cls.objects.bulk_update( + closed_sessions, fields=["stop_time", "terminate_cause"] + ) + cls.emit_radius_accounting_closed(closed_sessions) return closed_count diff --git a/openwisp_radius/integrations/monitoring/apps.py b/openwisp_radius/integrations/monitoring/apps.py index 80fcb896..58b9d56d 100644 --- a/openwisp_radius/integrations/monitoring/apps.py +++ b/openwisp_radius/integrations/monitoring/apps.py @@ -4,7 +4,6 @@ from django.db import models from django.db.models import Count, Sum from django.db.models.functions import Cast, Round -from django.db.models.signals import post_save from django.utils.translation import gettext_lazy as _ from openwisp_monitoring.monitoring.configuration import ( _register_chart_configuration_choice, @@ -12,6 +11,7 @@ ) from swapper import load_model +from openwisp_radius.signals import radius_accounting_closed from openwisp_utils.admin_theme import register_dashboard_chart from .utils import ( @@ -121,12 +121,12 @@ def register_radius_metrics(self): _register_chart_configuration_choice(chart_key, chart_config) def connect_signal_receivers(self): - from .receivers import post_save_radiusaccounting + from .receivers import radius_accounting_closed_handler RadiusAccounting = load_model("openwisp_radius", "RadiusAccounting") - post_save.connect( - post_save_radiusaccounting, + radius_accounting_closed.connect( + radius_accounting_closed_handler, sender=RadiusAccounting, - dispatch_uid="post_save_radiusaccounting_radius_acc_metric", + dispatch_uid="radius_accounting_closed_radius_acc_metric", ) diff --git a/openwisp_radius/integrations/monitoring/management/__init__.py b/openwisp_radius/integrations/monitoring/management/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/openwisp_radius/integrations/monitoring/management/commands/__init__.py b/openwisp_radius/integrations/monitoring/management/commands/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/openwisp_radius/integrations/monitoring/management/commands/rebuild_radius_accounting_metrics.py b/openwisp_radius/integrations/monitoring/management/commands/rebuild_radius_accounting_metrics.py new file mode 100644 index 00000000..a9165551 --- /dev/null +++ b/openwisp_radius/integrations/monitoring/management/commands/rebuild_radius_accounting_metrics.py @@ -0,0 +1,122 @@ +from django.core.management.base import BaseCommand, CommandError +from django.utils import timezone +from django.utils.dateparse import parse_datetime +from openwisp_monitoring.db import timeseries_db +from swapper import load_model + +from openwisp_radius.integrations.monitoring import tasks +from openwisp_radius.integrations.monitoring.utils import sha1_hash + +RadiusAccounting = load_model("openwisp_radius", "RadiusAccounting") + + +class Command(BaseCommand): + """ + Rebuild RADIUS accounting metrics missed by the NAS-Reboot bulk update path. + + See https://github.com/openwisp/openwisp-radius/issues/734. + TODO: remove in version 1.4. + """ + + help = "Private command to rebuild monitoring metrics for closed RADIUS sessions." + + def add_arguments(self, parser): + parser.add_argument( + "--commit", + action="store_true", + help="Write metrics. Without this flag the command only reports the count.", + ) + parser.add_argument( + "--start", + help=( + "Only process sessions with stop_time greater than or equal " + "to this date." + ), + ) + parser.add_argument( + "--end", + help=( + "Only process sessions with stop_time lower than or equal " + "to this date." + ), + ) + parser.add_argument( + "--chunk-size", + type=int, + default=1000, + help="Number of sessions fetched per database batch.", + ) + + def handle(self, *args, **options): + queryset = RadiusAccounting.objects.filter( + stop_time__isnull=False, + terminate_cause="NAS-Reboot", + ) + if options["start"]: + queryset = queryset.filter( + stop_time__gte=self._parse_datetime(options["start"], "start") + ) + if options["end"]: + queryset = queryset.filter( + stop_time__lte=self._parse_datetime(options["end"], "end") + ) + queryset = queryset.order_by("stop_time", "unique_id").only( + "unique_id", + "username", + "organization_id", + "input_octets", + "output_octets", + "calling_station_id", + "called_station_id", + "stop_time", + "terminate_cause", + ) + count = queryset.count() + if not options["commit"]: + self.stdout.write(f"Dry run: {count} closed sessions would be processed.") + return + processed = 0 + chunk_size = options["chunk_size"] + self.stdout.write(f"Starting to rebuild {count} accounting metrics.") + for session in queryset.iterator(chunk_size=chunk_size): + self._delete_radius_accounting_metric(session) + tasks.post_save_radiusaccounting( + username=session.username, + organization_id=str(session.organization_id), + input_octets=session.input_octets, + output_octets=session.output_octets, + calling_station_id=session.calling_station_id, + called_station_id=session.called_station_id, + time=session.stop_time, + ) + processed += 1 + if processed % chunk_size == 0 or processed == count: + self.stdout.write( + f"Processed {processed} of {count} accounting metrics." + ) + + def _parse_datetime(self, value, option): + parsed = parse_datetime(value) + if parsed is None: + raise CommandError(f"Invalid --{option} datetime: {value}") + if timezone.is_naive(parsed): + parsed = timezone.make_aware(parsed) + return parsed + + def _delete_radius_accounting_metric(self, session): + tags = { + "organization_id": str(session.organization_id), + "calling_station_id": sha1_hash(session.calling_station_id), + "called_station_id": session.called_station_id, + } + where = " AND ".join( + f"\"{key}\" = '{self._escape_tag_value(value)}'" + for key, value in tags.items() + ) + timeseries_db.query( + "DELETE FROM radius_acc " + f"WHERE time = '{session.stop_time.isoformat()}' AND {where}" + ) + + def _escape_tag_value(self, value): + return str(value).replace("'", r"\'") diff --git a/openwisp_radius/integrations/monitoring/management/commands/test_rebuild_radius_accounting_metrics.py b/openwisp_radius/integrations/monitoring/management/commands/test_rebuild_radius_accounting_metrics.py new file mode 100644 index 00000000..8ffa5cdd --- /dev/null +++ b/openwisp_radius/integrations/monitoring/management/commands/test_rebuild_radius_accounting_metrics.py @@ -0,0 +1,150 @@ +from io import StringIO +from unittest.mock import patch + +from django.contrib.contenttypes.models import ContentType +from django.core.management import call_command +from django.test import tag +from django.utils import timezone +from swapper import load_model + +from openwisp_radius.integrations.monitoring.tests.mixins import ( + CreateDeviceMonitoringMixin, +) +from openwisp_radius.integrations.monitoring.utils import sha1_hash +from openwisp_radius.tests import _RADACCT +from openwisp_radius.tests.mixins import BaseTransactionTestCase + +RegisteredUser = load_model("openwisp_radius", "RegisteredUser") +RadiusAccounting = load_model("openwisp_radius", "RadiusAccounting") + + +@tag("radius_monitoring", "rebuild_radius_accounting_metrics") +class TestRebuildRadiusAccountingMetrics( + CreateDeviceMonitoringMixin, BaseTransactionTestCase +): + def _create_registered_user(self, **kwargs): + options = { + "is_verified": False, + "method": "mobile_phone", + "organization": self.default_org, + } + options.update(**kwargs) + if "user" not in options: + options["user"] = self._create_user() + reg_user = RegisteredUser(**options) + reg_user.full_clean() + reg_user.save() + return reg_user + + def _create_closed_accounting_without_metric(self, **kwargs): + options = _RADACCT.copy() + options.update( + { + "unique_id": "closed-without-metric", + "calling_station_id": "00:00:00:00:00:00", + "input_octets": 8000000000, + "output_octets": 9000000000, + } + ) + options.update(kwargs) + stop_time = options.pop("stop_time", timezone.now()) + terminate_cause = options.pop("terminate_cause", "NAS-Reboot") + session = self._create_radius_accounting(**options) + RadiusAccounting.objects.filter(pk=session.pk).update( + stop_time=stop_time, + terminate_cause=terminate_cause, + ) + session.refresh_from_db() + return session + + def test_rebuild_radius_accounting_metrics_dry_run(self): + user = self._create_user() + device = self._create_device() + self._create_registered_user(user=user) + self._create_closed_accounting_without_metric( + username=user.username, + called_station_id=device.mac_address.replace("-", ":").upper(), + ) + out = StringIO() + call_command("rebuild_radius_accounting_metrics", stdout=out) + self.assertIn("Dry run: 1 closed sessions would be processed.", out.getvalue()) + self.assertEqual( + self.metric_model.objects.filter(configuration="radius_acc").count(), 0 + ) + + @patch( + "openwisp_radius.integrations.monitoring.management.commands." + "rebuild_radius_accounting_metrics.timeseries_db.query" + ) + @patch("logging.Logger.warning") + def test_rebuild_radius_accounting_metrics_commit( + self, mocked_warning, mocked_query + ): + user = self._create_user() + reg_user = self._create_registered_user(user=user) + device = self._create_device() + device_loc = self._create_device_location( + content_object=device, + location=self._create_location(organization=device.organization), + ) + session = self._create_closed_accounting_without_metric( + username=user.username, + called_station_id=device.mac_address.replace("-", ":").upper(), + ) + out = StringIO() + call_command("rebuild_radius_accounting_metrics", commit=True, stdout=out) + output = out.getvalue() + self.assertIn("Starting to rebuild 1 accounting metrics.", output) + self.assertIn("Processed 1 of 1 accounting metrics.", output) + delete_queries = [ + call.args[0] + for call in mocked_query.call_args_list + if call.args[0].startswith("DELETE FROM radius_acc") + ] + self.assertEqual(len(delete_queries), 1) + self.assertEqual( + self.metric_model.objects.filter( + configuration="radius_acc", + name="RADIUS Accounting", + key="radius_acc", + object_id=str(device.id), + content_type=ContentType.objects.get_for_model(self.device_model), + extra_tags={ + "called_station_id": device.mac_address, + "calling_station_id": sha1_hash(session.calling_station_id), + "location_id": str(device_loc.location.id), + "method": reg_user.method, + "organization_id": str(self.default_org.id), + }, + ).count(), + 1, + ) + + @patch("logging.Logger.warning") + def test_rebuild_radius_accounting_metrics_nas_reboot_filter(self, *args): + user = self._create_user() + device = self._create_device() + self._create_registered_user(user=user) + self._create_closed_accounting_without_metric( + unique_id="matching-session", + username=user.username, + called_station_id=device.mac_address.replace("-", ":").upper(), + ) + self._create_closed_accounting_without_metric( + unique_id="ignored-session", + username=user.username, + called_station_id=device.mac_address.replace("-", ":").upper(), + terminate_cause="Session-Timeout", + ) + out = StringIO() + call_command( + "rebuild_radius_accounting_metrics", + commit=True, + stdout=out, + ) + output = out.getvalue() + self.assertIn("Starting to rebuild 1 accounting metrics.", output) + self.assertIn("Processed 1 of 1 accounting metrics.", output) + metric = self.metric_model.objects.get(configuration="radius_acc") + points = metric.chart_set.get(configuration="radius_traffic").read() + self.assertEqual(points["summary"], {"upload": 9, "download": 8}) diff --git a/openwisp_radius/integrations/monitoring/receivers.py b/openwisp_radius/integrations/monitoring/receivers.py index 674b659e..f3b7cae3 100644 --- a/openwisp_radius/integrations/monitoring/receivers.py +++ b/openwisp_radius/integrations/monitoring/receivers.py @@ -1,11 +1,15 @@ from django.db import transaction +from django.utils import timezone from . import tasks -def post_save_radiusaccounting(instance, *args, **kwargs): +def _enqueue_radiusaccounting_metric(instance): if instance.stop_time is None: return + metric_time = instance.stop_time + if timezone.is_naive(metric_time): + metric_time = timezone.make_aware(metric_time) transaction.on_commit( lambda: tasks.post_save_radiusaccounting.delay( username=instance.username, @@ -14,5 +18,10 @@ def post_save_radiusaccounting(instance, *args, **kwargs): output_octets=instance.output_octets, calling_station_id=instance.calling_station_id, called_station_id=instance.called_station_id, + time=metric_time, ) ) + + +def radius_accounting_closed_handler(instance, *args, **kwargs): + _enqueue_radiusaccounting_metric(instance) diff --git a/openwisp_radius/integrations/monitoring/tests/test_metrics.py b/openwisp_radius/integrations/monitoring/tests/test_metrics.py index 4c4450ca..a2beaecd 100644 --- a/openwisp_radius/integrations/monitoring/tests/test_metrics.py +++ b/openwisp_radius/integrations/monitoring/tests/test_metrics.py @@ -18,6 +18,7 @@ RegisteredUser = load_model("openwisp_radius", "RegisteredUser") OrganizationUser = load_model("openwisp_users", "OrganizationUser") +RadiusAccounting = load_model("openwisp_radius", "RadiusAccounting") User = get_user_model() @@ -89,7 +90,7 @@ def test_post_save_radiusaccounting(self, *args): "output_octets": "9000000000", } ) - options["stop_time"] = options["start_time"] + options["stop_time"] = timezone.now() self._create_radius_accounting(**options) self.assertEqual( @@ -140,7 +141,7 @@ def test_post_save_radiusaccounting_device_without_location(self, *args): "output_octets": "9000000000", } ) - options["stop_time"] = options["start_time"] + options["stop_time"] = timezone.now() self._create_radius_accounting(**options) with self.subTest("location_id should not be set"): self.assertEqual( @@ -169,13 +170,210 @@ def test_post_save_radiusaccounting_device_without_location(self, *args): 0, ) - @patch("openwisp_radius.integrations.monitoring.tasks.post_save_radiusaccounting") - def test_post_save_radiusaccouting_open_session(self, mocked_task): + @patch( + "openwisp_radius.integrations.monitoring.tasks.post_save_radiusaccounting.delay" + ) + def test_post_save_radiusaccouting_open_session(self, mocked_delay): + """ + Regression test ensuring open or interim accounting sessions do not + write monitoring metrics before their cumulative octet values are final. + """ radius_options = _RADACCT.copy() radius_options["unique_id"] = "117" session = self._create_radius_accounting(**radius_options) self.assertEqual(session.stop_time, None) - mocked_task.assert_not_called() + mocked_delay.assert_not_called() + + @patch( + "openwisp_radius.integrations.monitoring.tasks.post_save_radiusaccounting.delay" + ) + def test_radius_accounting_closed_handler_open_session(self, mocked_delay): + # Import locally because the standard test suite runs without monitoring. + from ..receivers import radius_accounting_closed_handler + + radius_options = _RADACCT.copy() + radius_options["unique_id"] = "open-session-handler" + session = RadiusAccounting(**radius_options) + self.assertEqual(session.stop_time, None) + radius_accounting_closed_handler(session) + mocked_delay.assert_not_called() + + @patch( + "openwisp_radius.integrations.monitoring.tasks.post_save_radiusaccounting.delay" + ) + def test_radius_accounting_closed_handler_naive_stop_time(self, mocked_delay): + # Import locally because the standard test suite runs without monitoring. + from ..receivers import radius_accounting_closed_handler + + naive_stop_time = timezone.datetime(2026, 1, 1, 12, 0, 0) + radius_options = _RADACCT.copy() + radius_options.update( + { + "unique_id": "naive-stop-time-handler", + "organization": self.default_org, + "called_station_id": "AA-BB-CC-DD-EE-FF", + "calling_station_id": "00:00:00:00:00:00", + "stop_time": naive_stop_time, + } + ) + session = RadiusAccounting(**radius_options) + radius_accounting_closed_handler(session) + mocked_delay.assert_called_once_with( + username=radius_options["username"], + organization_id=str(self.default_org.id), + input_octets=radius_options["input_octets"], + output_octets=radius_options["output_octets"], + calling_station_id=radius_options["calling_station_id"], + called_station_id=radius_options["called_station_id"], + time=timezone.make_aware(naive_stop_time), + ) + + @patch( + "openwisp_radius.integrations.monitoring.tasks.post_save_radiusaccounting.delay" + ) + def test_accounting_on_nas_reboot_writes_monitoring_metric(self, mocked_delay): + """ + Regression test for https://github.com/openwisp/openwisp-radius/issues/734. + + Sessions closed by Accounting-On must write monitoring metrics even + though the closure path uses bulk_update instead of model save(). + """ + radius_options = _RADACCT.copy() + radius_options.update( + { + "unique_id": "nas-reboot-session", + "called_station_id": "AA-BB-CC-DD-EE-FF", + "calling_station_id": "00:00:00:00:00:00", + "input_octets": 8000000000, + "output_octets": 9000000000, + "stop_time": None, + } + ) + self._create_radius_accounting(**radius_options) + mocked_delay.assert_not_called() + closed_count = RadiusAccounting._close_stale_sessions_on_nas_boot( + called_station_id=radius_options["called_station_id"] + ) + self.assertEqual(closed_count, 1) + session = RadiusAccounting.objects.get(unique_id=radius_options["unique_id"]) + mocked_delay.assert_called_once_with( + username=radius_options["username"], + organization_id=str(self.default_org.id), + input_octets=radius_options["input_octets"], + output_octets=radius_options["output_octets"], + calling_station_id=radius_options["calling_station_id"], + called_station_id=radius_options["called_station_id"], + time=session.stop_time, + ) + mocked_delay.reset_mock() + closed_count = RadiusAccounting._close_stale_sessions_on_nas_boot( + called_station_id=radius_options["called_station_id"] + ) + self.assertEqual(closed_count, 0) + mocked_delay.assert_not_called() + + @patch( + "openwisp_radius.integrations.monitoring.tasks.post_save_radiusaccounting.delay" + ) + def test_close_previous_radius_accounting_writes_monitoring_metric( + self, mocked_delay + ): + """ + Regression test ensuring sessions closed by the previous-session + cleanup path write monitoring metrics despite using bulk_update. + """ + radius_options = _RADACCT.copy() + radius_options.update( + { + "unique_id": "previous-session", + "called_station_id": "AA-BB-CC-DD-EE-FF", + "calling_station_id": "00:00:00:00:00:00", + "input_octets": 8000000000, + "output_octets": 9000000000, + "stop_time": None, + } + ) + self._create_radius_accounting(**radius_options) + mocked_delay.assert_not_called() + new_session_options = radius_options.copy() + new_session_options.update( + { + "unique_id": "new-session", + "input_octets": 1000000000, + "output_octets": 2000000000, + } + ) + self._create_radius_accounting(**new_session_options) + session = RadiusAccounting.objects.get(unique_id=radius_options["unique_id"]) + mocked_delay.assert_called_once_with( + username=radius_options["username"], + organization_id=str(self.default_org.id), + input_octets=radius_options["input_octets"], + output_octets=radius_options["output_octets"], + calling_station_id=radius_options["calling_station_id"], + called_station_id=radius_options["called_station_id"], + time=session.stop_time, + ) + + @patch( + "openwisp_radius.integrations.monitoring.tasks.post_save_radiusaccounting.delay" + ) + def test_closed_radius_accounting_metric_uses_stop_time(self, mocked_delay): + """ + Regression test ensuring sessions created already closed are + snapshotted at stop_time rather than task execution time. + """ + stop_time = timezone.now() - timezone.timedelta(days=1) + radius_options = _RADACCT.copy() + radius_options.update( + { + "unique_id": "closed-session-stop-time", + "called_station_id": "AA-BB-CC-DD-EE-FF", + "calling_station_id": "00:00:00:00:00:00", + "input_octets": 8000000000, + "output_octets": 9000000000, + "stop_time": stop_time, + } + ) + self._create_radius_accounting(**radius_options) + mocked_delay.assert_called_once_with( + username=radius_options["username"], + organization_id=str(self.default_org.id), + input_octets=radius_options["input_octets"], + output_octets=radius_options["output_octets"], + calling_station_id=radius_options["calling_station_id"], + called_station_id=radius_options["called_station_id"], + time=stop_time, + ) + + @patch( + "openwisp_radius.integrations.monitoring.tasks.post_save_radiusaccounting.delay" + ) + def test_editing_closed_radius_accounting_does_not_write_metric_again( + self, mocked_delay + ): + """ + Regression test ensuring edits to already closed sessions do not write + duplicate monitoring metrics after the first close snapshot. + """ + stop_time = timezone.now() + radius_options = _RADACCT.copy() + radius_options.update( + { + "unique_id": "closed-session-edit", + "called_station_id": "AA-BB-CC-DD-EE-FF", + "calling_station_id": "00:00:00:00:00:00", + "input_octets": 8000000000, + "output_octets": 9000000000, + "stop_time": stop_time, + } + ) + session = self._create_radius_accounting(**radius_options) + mocked_delay.assert_called_once() + mocked_delay.reset_mock() + session.terminate_cause = "User-Request" + session.save(update_fields=["terminate_cause"]) + mocked_delay.assert_not_called() @patch("logging.Logger.warning") def test_post_save_radius_accounting_shared_accounting(self, mocked_logger): @@ -206,7 +404,7 @@ def test_post_save_radius_accounting_shared_accounting(self, mocked_logger): "output_octets": "9000000000", } ) - options["stop_time"] = options["start_time"] + options["stop_time"] = timezone.now() device_metric_qs = self.metric_model.objects.filter( configuration="radius_acc", name="RADIUS Accounting", @@ -290,7 +488,7 @@ def test_post_save_radius_accounting_device_not_found(self, mocked_logger): "output_octets": "9000000000", } ) - options["stop_time"] = options["start_time"] + options["stop_time"] = timezone.now() # Remove calls for user registration from mocked logger mocked_logger.reset_mock() self._create_radius_accounting(**options) @@ -367,7 +565,7 @@ def test_post_save_radius_accounting_registereduser_not_found(self, mocked_logge "output_octets": "9000000000", } ) - options["stop_time"] = options["start_time"] + options["stop_time"] = timezone.now() self._create_radius_accounting(**options) self.assertEqual( @@ -429,7 +627,7 @@ def test_post_save_radiusaccounting_pending_verification(self): "output_octets": "9000000000", } ) - options["stop_time"] = options["start_time"] + options["stop_time"] = timezone.now() self._create_radius_accounting(**options) self.assertEqual( self.metric_model.objects.filter( @@ -479,7 +677,7 @@ def test_post_save_radiusaccounting_does_not_fallback_to_other_org( "output_octets": "9000000000", } ) - options["stop_time"] = options["start_time"] + options["stop_time"] = timezone.now() self._create_radius_accounting(**options) self.assertEqual( self.metric_model.objects.filter( diff --git a/openwisp_radius/receivers.py b/openwisp_radius/receivers.py index 4d5155b2..39a4d256 100644 --- a/openwisp_radius/receivers.py +++ b/openwisp_radius/receivers.py @@ -126,6 +126,7 @@ def close_previous_radius_accounting_sessions(instance, created, **kwargs): RadiusAccounting.objects.bulk_update( closed_sessions, fields=["stop_time", "terminate_cause"] ) + RadiusAccounting.emit_radius_accounting_closed(closed_sessions) def radius_user_group_change(sender, instance, **kwargs): diff --git a/openwisp_radius/signals.py b/openwisp_radius/signals.py index 8d31e0f1..2a2ba356 100644 --- a/openwisp_radius/signals.py +++ b/openwisp_radius/signals.py @@ -1,3 +1,4 @@ from django.dispatch import Signal radius_accounting_success = Signal() # providing_args=['accounting_data', 'view'] +radius_accounting_closed = Signal() # providing_args=['instance'] diff --git a/openwisp_radius/tests/test_models.py b/openwisp_radius/tests/test_models.py index d9edd23e..8520d802 100644 --- a/openwisp_radius/tests/test_models.py +++ b/openwisp_radius/tests/test_models.py @@ -14,11 +14,12 @@ from netaddr import EUI, mac_unix from openwisp_users.tests.utils import TestMultitenantAdminMixin -from openwisp_utils.tests import capture_any_output, capture_stderr +from openwisp_utils.tests import capture_any_output, capture_stderr, catch_signal from .. import settings as app_settings from ..counters.exceptions import MaxQuotaReached from ..radclient.client import RadClient +from ..signals import radius_accounting_closed from ..tasks import perform_change_of_authorization from ..utils import ( DEFAULT_SESSION_TIME_LIMIT, @@ -155,6 +156,113 @@ def test_multiple_accounting_sessions(self): self.assertEqual(radiusaccounting1.stop_time, radiusaccounting1.update_time) self.assertEqual(radiusaccounting2.stop_time, None) + def test_radius_accounting_closed_signal_on_commit(self): + radiusaccounting_options = _RADACCT.copy() + radiusaccounting_options.update( + { + "organization": self.default_org, + "nas_ip_address": "192.168.182.3", + } + ) + + def assert_signal_not_emitted_on_save(session, **save_kwargs): + with catch_signal(radius_accounting_closed) as handler: + with self.captureOnCommitCallbacks(execute=True): + session.full_clean() + session.save(**save_kwargs) + handler.assert_not_called() + handler.assert_not_called() + + def assert_signal_emitted_on_save(session, **save_kwargs): + with catch_signal(radius_accounting_closed) as handler: + with self.captureOnCommitCallbacks(execute=True): + session.full_clean() + session.save(**save_kwargs) + handler.assert_not_called() + handler.assert_called_once() + self.assertEqual(handler.call_args.kwargs["sender"], RadiusAccounting) + self.assertEqual(handler.call_args.kwargs["instance"], session) + + with self.subTest("Open session creation does not emit"): + session = RadiusAccounting(unique_id="closed-signal-1") + for key, value in radiusaccounting_options.items(): + setattr(session, key, value) + assert_signal_not_emitted_on_save(session) + + with self.subTest("Closed session creation emits on commit"): + session = RadiusAccounting(unique_id="closed-signal-2") + for key, value in radiusaccounting_options.items(): + setattr(session, key, value) + session.stop_time = timezone.now() + assert_signal_emitted_on_save(session) + + with self.subTest("Open session updated as closed emits on commit"): + session = self._create_radius_accounting( + unique_id="closed-signal-3", **radiusaccounting_options + ) + session.stop_time = timezone.now() + assert_signal_emitted_on_save(session) + + with self.subTest("Already closed session saved again does not emit"): + session = self._create_radius_accounting( + unique_id="closed-signal-4", + stop_time=timezone.now(), + **radiusaccounting_options, + ) + assert_signal_not_emitted_on_save(session) + + with self.subTest("Unsaved stop_time change does not emit"): + session = self._create_radius_accounting( + unique_id="closed-signal-5", **radiusaccounting_options + ) + session.stop_time = timezone.now() + session.terminate_cause = "User-Request" + assert_signal_not_emitted_on_save( + session, update_fields=["terminate_cause"] + ) + self.assertIsNone(session._initial_stop_time) + assert_signal_emitted_on_save(session, update_fields=["stop_time"]) + + def test_close_stale_sessions_on_nas_boot_query_count(self): + radiusaccounting_options = _RADACCT.copy() + radiusaccounting_options.update( + { + "organization": self.default_org, + "nas_ip_address": "192.168.182.3", + "called_station_id": "AA-BB-CC-DD-EE-FF", + } + ) + session_options = radiusaccounting_options.copy() + session_options.update( + unique_id="nas-reboot-query-count-1", + username="nas-reboot-query-count-1", + ) + self._create_radius_accounting(**session_options) + session_options = radiusaccounting_options.copy() + session_options.update( + unique_id="nas-reboot-query-count-2", + username="nas-reboot-query-count-2", + ) + self._create_radius_accounting(**session_options) + with self.assertNumQueries(4): + closed_count = RadiusAccounting._close_stale_sessions_on_nas_boot( + called_station_id=radiusaccounting_options["called_station_id"] + ) + self.assertEqual(closed_count, 2) + + def test_save_update_fields_persists_backfilled_start_time(self): + options = _RADACCT.copy() + session = self._create_radius_accounting( + unique_id="start-time-update-fields", **options + ) + RadiusAccounting.objects.filter(pk=session.pk).update(start_time=None) + session.refresh_from_db() + session.terminate_cause = "User-Request" + session.save(update_fields=["terminate_cause"]) + session.refresh_from_db() + self.assertIsNotNone(session.start_time) + self.assertEqual(session.terminate_cause, "User-Request") + @capture_any_output() @mock.patch.object(app_settings, "OPENVPN_DATETIME_FORMAT", "%Y-%m-%d %H:%M:%S") @mock.patch.object(app_settings, "CONVERT_CALLED_STATION_ON_CREATE", True)