From 8392098a89ab9d5ca948f7298827c11fd2a2de69 Mon Sep 17 00:00:00 2001 From: Federico Capoano Date: Mon, 29 Jun 2026 18:54:46 -0300 Subject: [PATCH 1/3] [feature] Added ``radius_accounting_closed`` signal --- docs/developer/utils.rst | 24 ++++++++ openwisp_radius/base/models.py | 58 +++++++++++++++++-- openwisp_radius/signals.py | 1 + openwisp_radius/tests/test_models.py | 83 +++++++++++++++++++++++++++- 4 files changed, 160 insertions(+), 6 deletions(-) diff --git a/docs/developer/utils.rst b/docs/developer/utils.rst index b5813c67..a90cca15 100644 --- a/docs/developer/utils.rst +++ b/docs/developer/utils.rst @@ -27,6 +27,30 @@ 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. + +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..2c06c3a2 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: 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..17ca2a7a 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,86 @@ 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_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) From 56f86086565ae833ba5681ee9554561c3b5705e0 Mon Sep 17 00:00:00 2001 From: Federico Capoano Date: Tue, 30 Jun 2026 10:31:45 -0300 Subject: [PATCH 2/3] [change] Monitoring: trigger TSDB save on sessions close The integration with OpenWISP Monitoring now is triggered exclusively when RADIUS accounting sessions close, instead of the previous approach which used the generic post_save signal and used custom logic to determine if the RADIUS session was being closed or not. --- .../integrations/monitoring/apps.py | 10 +-- .../integrations/monitoring/receivers.py | 11 ++- .../monitoring/tests/test_metrics.py | 86 ++++++++++++++++--- 3 files changed, 91 insertions(+), 16 deletions(-) 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/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..a4f5a788 100644 --- a/openwisp_radius/integrations/monitoring/tests/test_metrics.py +++ b/openwisp_radius/integrations/monitoring/tests/test_metrics.py @@ -89,7 +89,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 +140,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 +169,79 @@ 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_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 +272,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 +356,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 +433,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 +495,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 +545,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( From 14b333b2a48d0d1dea4df7c00fe528c030c54b98 Mon Sep 17 00:00:00 2001 From: Federico Capoano Date: Tue, 30 Jun 2026 18:14:01 -0300 Subject: [PATCH 3/3] [fix] Log metrics for bulk-closed RADIUS sessions #734 The RADIUS monitoring integration previously relied only on Django's ``post_save`` signal to write traffic snapshots. However, sessions closed via ``QuerySet.update()`` (``Accounting-On`` handling) and ``bulk_update()`` (``close_previous_radius_accounting_sessions``) bypassed this signal, causing traffic to be missing from RADIUS traffic charts. Sessions closed via ``QuerySet.update()`` and ``bulk_update()`` now emit the ``radius_accounting_closed`` signal, ensuring that traffic metrics are correctly snapshotted by the RADIUS monitoring integration. Closes #734 --- docs/developer/utils.rst | 5 +- openwisp_radius/base/models.py | 40 ++++- .../monitoring/management/__init__.py | 0 .../management/commands/__init__.py | 0 .../rebuild_radius_accounting_metrics.py | 122 ++++++++++++++ .../test_rebuild_radius_accounting_metrics.py | 150 ++++++++++++++++++ .../monitoring/tests/test_metrics.py | 132 +++++++++++++++ openwisp_radius/receivers.py | 1 + openwisp_radius/tests/test_models.py | 27 ++++ 9 files changed, 469 insertions(+), 8 deletions(-) create mode 100644 openwisp_radius/integrations/monitoring/management/__init__.py create mode 100644 openwisp_radius/integrations/monitoring/management/commands/__init__.py create mode 100644 openwisp_radius/integrations/monitoring/management/commands/rebuild_radius_accounting_metrics.py create mode 100644 openwisp_radius/integrations/monitoring/management/commands/test_rebuild_radius_accounting_metrics.py diff --git a/docs/developer/utils.rst b/docs/developer/utils.rst index a90cca15..aaa24fcc 100644 --- a/docs/developer/utils.rst +++ b/docs/developer/utils.rst @@ -45,7 +45,10 @@ 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. +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 diff --git a/openwisp_radius/base/models.py b/openwisp_radius/base/models.py index 2c06c3a2..54f6cdfb 100644 --- a/openwisp_radius/base/models.py +++ b/openwisp_radius/base/models.py @@ -627,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/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/tests/test_metrics.py b/openwisp_radius/integrations/monitoring/tests/test_metrics.py index a4f5a788..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() @@ -183,6 +184,137 @@ def test_post_save_radiusaccouting_open_session(self, mocked_delay): self.assertEqual(session.stop_time, None) 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" ) 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/tests/test_models.py b/openwisp_radius/tests/test_models.py index 17ca2a7a..8520d802 100644 --- a/openwisp_radius/tests/test_models.py +++ b/openwisp_radius/tests/test_models.py @@ -223,6 +223,33 @@ def assert_signal_emitted_on_save(session, **save_kwargs): 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(