Fixed EndDeviceList pollRate Notifications (#307)

Author: joshvote

src/envoy/admin/crud/aggregator.py

@@ -1,4 +1,4 @@
-from typing import Sequence, Iterable
+from typing import Iterable, Sequence
 
 import sqlalchemy as sa
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -14,7 +14,7 @@ async def count_all_aggregators(session: AsyncSession) -> int:
     return resp.scalar_one()
 
 
-async def select_all_aggregators(session: AsyncSession, start: int, limit: int) -> Sequence[Aggregator]:
+async def select_all_aggregators(session: AsyncSession, start: int | None, limit: int | None) -> Sequence[Aggregator]:
     """Admin selecting of aggregators - will include domains relationship (the NULL_AGGREGATOR is not included)"""
 
     stmt = (

src/envoy/admin/manager/config.py

@@ -35,13 +35,18 @@ async def update_current_config(session: AsyncSession, updated_values: RuntimeSe
         else:
             existing_db_config.changed_time = now
 
+        changed_fsal_pollrate = False
+        changed_edevl_pollrate = False
+
         if updated_values.dcap_pollrate_seconds is not None:
             existing_db_config.dcap_pollrate_seconds = updated_values.dcap_pollrate_seconds
 
         if updated_values.edevl_pollrate_seconds is not None:
+            changed_edevl_pollrate = existing_db_config.edevl_pollrate_seconds != updated_values.edevl_pollrate_seconds
             existing_db_config.edevl_pollrate_seconds = updated_values.edevl_pollrate_seconds
 
         if updated_values.fsal_pollrate_seconds is not None:
+            changed_fsal_pollrate = existing_db_config.fsal_pollrate_seconds != updated_values.fsal_pollrate_seconds
             existing_db_config.fsal_pollrate_seconds = updated_values.fsal_pollrate_seconds
 
         if updated_values.derpl_pollrate_seconds is not None:
@@ -61,7 +66,13 @@ async def update_current_config(session: AsyncSession, updated_values: RuntimeSe
 
         await session.commit()
 
-        await NotificationManager.notify_changed_deleted_entities(SubscriptionResource.FUNCTION_SET_ASSIGNMENTS, now)
+        if changed_fsal_pollrate:
+            await NotificationManager.notify_changed_deleted_entities(
+                SubscriptionResource.FUNCTION_SET_ASSIGNMENTS, now
+            )
+
+        if changed_edevl_pollrate:
+            await NotificationManager.notify_changed_deleted_entities(SubscriptionResource.SITE, now)
 
     @staticmethod
     async def fetch_config_response(session: AsyncSession) -> RuntimeServerConfigResponse:

src/envoy/notification/crud/batch.py

@@ -6,6 +6,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
+from envoy.admin.crud.aggregator import select_all_aggregators
 from envoy.notification.crud.archive import (
     fetch_entities_with_archive_by_datetime,
     fetch_entities_with_archive_by_id,
@@ -25,6 +26,7 @@
 from envoy.server.crud.common import localize_start_time_for_entity
 from envoy.server.crud.server import select_server_config
 from envoy.server.manager.der_constants import PUBLIC_SITE_DER_ID
+from envoy.server.model.aggregator import Aggregator
 from envoy.server.model.archive.doe import ArchiveDynamicOperatingEnvelope, ArchiveSiteControlGroup
 from envoy.server.model.archive.site import (
     ArchiveDefaultSiteControl,
@@ -90,6 +92,21 @@ def __init__(
         self.models_by_batch_key = AggregatorBatchedEntities._generate_batch_dict(resource, models)
         self.deleted_by_batch_key = AggregatorBatchedEntities._generate_batch_dict(resource, deleted_models)
 
+    @staticmethod
+    def aggregator_id_instance(
+        timestamp: datetime, resource: SubscriptionResource, aggregators: Sequence[Aggregator]
+    ) -> "AggregatorBatchedEntities":
+        """This will generate an instance with the models_by_batch_key loaded with a key for each aggregator instance
+        (key being a single tuple[aggregator_id: int]). Each of the entries will be an empty list.
+
+        This will be a mechanism for prepping an "empty list" notification for representing things like a pollRate
+        change"""
+
+        batch = AggregatorBatchedEntities(timestamp, resource, [], [])
+        for agg in aggregators:
+            batch.models_by_batch_key[(agg.aggregator_id,)] = []
+        return batch
+
 
 def get_batch_key(resource: SubscriptionResource, entity: TResourceModel) -> tuple:
     """
@@ -101,7 +118,7 @@ def get_batch_key(resource: SubscriptionResource, entity: TResourceModel) -> tup
 
     Given the SubscriptionResource - it's safe to rely on the ordering of the batch key tuple entries:
 
-    SubscriptionResource.SITE: (aggregator_id: int, site_id: int)
+    SubscriptionResource.SITE: (aggregator_id: int, site_id: int) OR (aggregator_id: int)
     SubscriptionResource.DYNAMIC_OPERATING_ENVELOPE: (aggregator_id: int, site_id: int, site_control_group_id: int)
     SubscriptionResource.READING: (aggregator_id: int, site_id: int, group_id: int)
     SubscriptionResource.TARIFF_GENERATED_RATE: (aggregator_id: int, tariff_id: int, site_id: int, day: date)
@@ -247,10 +264,21 @@ async def fetch_sites_by_changed_at(
 ) -> AggregatorBatchedEntities[Site, ArchiveSite]:
     """Fetches all sites matching the specified changed_at and returns them keyed by their aggregator/site id
 
-    Also fetches any site from the archive that was deleted at the specified timestamp"""
+    Also fetches any site from the archive that was deleted at the specified timestamp.
 
-    active_sites, deleted_sites = await fetch_entities_with_archive_by_datetime(session, Site, ArchiveSite, timestamp)
+    This will also consider runtime config for the EndDeviceList.pollRate - If this is altered, it will instead
+    generate a notification for EVERY EndDevice"""
+
+    # If run time config has been marked as a "Site" update - it's because it has had a change in poll rate
+    # In this circumstance - we generate a special kind of update that will result in an "Empty List" Notification
+    # that will just show the pollRate change.
+    runtime_cfg = await select_server_config(session)
+    if runtime_cfg is not None and runtime_cfg.changed_time == timestamp:
+        aggregators = await select_all_aggregators(session, None, None)
+        return AggregatorBatchedEntities.aggregator_id_instance(timestamp, SubscriptionResource.SITE, aggregators)
 
+    # Otherwise - we proceed as if sites are the table that is changing
+    active_sites, deleted_sites = await fetch_entities_with_archive_by_datetime(session, Site, ArchiveSite, timestamp)
     return AggregatorBatchedEntities(timestamp, SubscriptionResource.SITE, active_sites, deleted_sites)
 
 
@@ -691,7 +719,7 @@ async def fetch_default_site_controls_by_changed_at(
 async def fetch_runtime_config_by_changed_at(
     session: AsyncSession, timestamp: datetime
 ) -> AggregatorBatchedEntities[SiteScopedRuntimeServerConfig, ArchiveSiteScopedRuntimeServerConfig]:  # type: ignore # noqa: E501
-    """Fetches all RuntimeServerCOnfig instances matching the specified changed_at and returns them keyed by their
+    """Fetches all RuntimeServerConfig instances matching the specified changed_at and returns them keyed by their
     aggregator/site id"""
 
     runtime_cfg = await select_server_config(session)

src/envoy/notification/task/check.py

@@ -235,7 +235,7 @@ def entities_to_notification(
     scope = scope_for_subscription(sub, href_prefix)
     if resource == SubscriptionResource.SITE:
         return NotificationMapper.map_sites_to_response(
-            cast(Sequence[Site], entities), sub, scope, notification_type, config.disable_edev_registration  # type: ignore # mypy quirk # noqa: E501
+            cast(Sequence[Site], entities), sub, scope, notification_type, config.disable_edev_registration, config.edevl_pollrate_seconds  # type: ignore # mypy quirk # noqa: E501
         )
     elif resource == SubscriptionResource.TARIFF_GENERATED_RATE:
         if pricing_reading_type is None:
@@ -406,10 +406,27 @@ async def check_db_change_or_delete(
             if entity_limit > MAX_NOTIFICATION_PAGE_SIZE:
                 entity_limit = MAX_NOTIFICATION_PAGE_SIZE
 
-            entities_to_notify = entities_serviced_by_subscription(sub, resource, entities)
-            all_notifications.extend(
-                get_entity_pages(resource, sub, batch_key, entity_limit, entities_to_notify, notification_type)
-            )
+            if entities:
+                # Normally we're going to have a batch of entities that should be sent out via notifications
+                entities_to_notify = entities_serviced_by_subscription(sub, resource, entities)
+                all_notifications.extend(
+                    get_entity_pages(resource, sub, batch_key, entity_limit, entities_to_notify, notification_type)
+                )
+            else:
+                # But we can end up in this state if the subscription is at the List and an attribute on the list has
+                # changed (eg pollRate) - i.e. there are no child list items to indicate as changed - JUST the list.
+                if sub.resource_type == resource:
+                    # All we need is a match on the type of subscription to generate the subscription
+                    all_notifications.append(
+                        NotificationEntities(
+                            entities=[],  # No entities - we're just wanting the parent List to notify as empty
+                            subscription=sub,
+                            notification_id=uuid4(),
+                            notification_type=NotificationType.ENTITY_CHANGED,
+                            batch_key=batch_key,
+                            pricing_reading_type=None,
+                        )
+                    )
 
     # Finally time to enqueue the outgoing notifications
     logger.info(

src/envoy/server/mapper/sep2/pub_sub.py

@@ -453,6 +453,7 @@ def map_sites_to_response(
         scope: AggregatorRequestScope,
         notification_type: NotificationType,
         disable_registration: bool,
+        poll_rate_seconds: int,
     ) -> Notification:
         """Turns a list of sites into a notification"""
         return Notification.model_validate(
@@ -462,6 +463,7 @@ def map_sites_to_response(
                 "status": _map_to_notification_status(notification_type),
                 "resource": {
                     "type": XSI_TYPE_END_DEVICE_LIST,
+                    "pollRate": poll_rate_seconds,
                     "all_": len(sites),
                     "results": len(sites),
                     "EndDevice": [EndDeviceMapper.map_to_response(scope, s, disable_registration, 0) for s in sites],

tests/integration/admin/test_notification.py

@@ -1,3 +1,4 @@
+import asyncio
 from datetime import datetime
 from decimal import Decimal
 from http import HTTPStatus
@@ -26,6 +27,7 @@
 from sqlalchemy import delete, insert, select
 
 from envoy.notification.task.transmit import HEADER_NOTIFICATION_ID
+from envoy.server.model.server import RuntimeServerConfig
 from envoy.server.model.subscription import Subscription, SubscriptionResource
 from envoy.server.model.tariff import PRICE_DECIMAL_POWER
 
@@ -610,6 +612,55 @@ async def test_delete_site_with_active_subscription(
     ), "Expected unique notification ids for each request"
 
 
+@pytest.mark.anyio
+async def test_update_server_config_edev_list_notification(
+    admin_client_auth: AsyncClient, notifications_enabled: MockedAsyncClient, pg_base_config
+):
+    """Tests that updating server config generates subscription notifications for EndDeviceList"""
+
+    subscription1_uri = "http://my.example:542/uri"
+
+    async with generate_async_session(pg_base_config) as session:
+        # Clear any other subs first
+        await session.execute(delete(Subscription))
+
+        # Will pickup site notifications
+        await session.execute(
+            insert(Subscription).values(
+                aggregator_id=1,
+                changed_time=datetime.now(),
+                resource_type=SubscriptionResource.SITE,
+                resource_id=None,
+                scoped_site_id=None,
+                notification_uri=subscription1_uri,
+                entity_limit=10,
+            )
+        )
+        await session.commit()
+
+    # Update edev list config
+    edev_list_poll_rate = 131009115
+    config_request = generate_class_instance(
+        RuntimeServerConfigRequest, optional_is_none=True, edevl_pollrate_seconds=edev_list_poll_rate
+    )
+    resp = await admin_client_auth.post(ServerConfigRuntimeUri, content=config_request.model_dump_json())
+    assert resp.status_code == HTTPStatus.NO_CONTENT
+
+    # Give the notifications a chance to propagate
+    assert await notifications_enabled.wait_for_n_requests(n=1, timeout_seconds=30)
+
+    # Sub 1 got one notification, sub 2 got the other
+    assert notifications_enabled.call_count_by_method[HTTPMethod.GET] == 0
+    assert notifications_enabled.call_count_by_method[HTTPMethod.POST] == 1
+    assert notifications_enabled.call_count_by_method_uri[(HTTPMethod.POST, subscription1_uri)] == 1
+
+    assert all([HEADER_NOTIFICATION_ID in r.headers for r in notifications_enabled.logged_requests])
+    assert len(set([r.headers[HEADER_NOTIFICATION_ID] for r in notifications_enabled.logged_requests])) == len(
+        notifications_enabled.logged_requests
+    ), "Expected unique notification ids for each request"
+    assert f'pollRate="{edev_list_poll_rate}"' in notifications_enabled.logged_requests[0].content
+
+
 @pytest.mark.anyio
 async def test_update_server_config_fsa_notification(
     admin_client_auth: AsyncClient, notifications_enabled: MockedAsyncClient, pg_base_config
@@ -671,6 +722,68 @@ async def test_update_server_config_fsa_notification(
     ), "Expected unique notification ids for each request"
 
 
+@pytest.mark.parametrize("none_fsa_value", [True, False])
+@pytest.mark.anyio
+async def test_update_server_config_fsa_notification_no_change(
+    admin_client_auth: AsyncClient, notifications_enabled: MockedAsyncClient, pg_base_config, none_fsa_value: bool
+):
+    """Tests that updating server config (with no changed value for FSA) generates 0 notifications"""
+
+    subscription1_uri = "http://my.example:542/uri"
+    fsa_poll_rate = 123
+
+    async with generate_async_session(pg_base_config) as session:
+        # Clear any other subs first
+        await session.execute(delete(Subscription))
+
+        # Force the server FSA pollrate config to a known value
+        await session.execute(delete(RuntimeServerConfig))
+        await session.execute(
+            insert(RuntimeServerConfig).values(
+                changed_time=datetime.now(),
+                dcap_pollrate_seconds=None,
+                edevl_pollrate_seconds=None,
+                fsal_pollrate_seconds=fsa_poll_rate,
+                derpl_pollrate_seconds=None,
+                derl_pollrate_seconds=None,
+                mup_postrate_seconds=None,
+                site_control_pow10_encoding=None,
+                disable_edev_registration=False,
+            )
+        )
+
+        # Will pickup FSA notifications for site 2
+        await session.execute(
+            insert(Subscription).values(
+                aggregator_id=1,
+                changed_time=datetime.now(),
+                resource_type=SubscriptionResource.FUNCTION_SET_ASSIGNMENTS,
+                resource_id=None,
+                scoped_site_id=2,
+                notification_uri=subscription1_uri,
+                entity_limit=10,
+            )
+        )
+
+        await session.commit()
+
+    # Update config with the subscriptions now live
+    config_request = generate_class_instance(
+        RuntimeServerConfigRequest,
+        optional_is_none=True,
+        disable_edev_registration=True,
+        fsal_pollrate_seconds=None if none_fsa_value else fsa_poll_rate,
+    )
+    resp = await admin_client_auth.post(ServerConfigRuntimeUri, content=config_request.model_dump_json())
+    assert resp.status_code == HTTPStatus.NO_CONTENT
+
+    # Give any notifications a chance to propagate
+    await asyncio.sleep(3)
+
+    # No notifications should've been generated as we aren't actually changing any values associated with the FSA
+    assert len(notifications_enabled.logged_requests) == 0
+
+
 @pytest.mark.anyio
 async def test_update_site_default_config_notification(
     admin_client_auth: AsyncClient, notifications_enabled: MockedAsyncClient, pg_base_config

tests/unit/admin/crud/test_aggregator.py

@@ -2,13 +2,13 @@
 
 import psycopg
 import pytest
-from assertical.fixtures.postgres import generate_async_session
 import sqlalchemy as sa
+from assertical.fixtures.postgres import generate_async_session
 from sqlalchemy.exc import IntegrityError
 
+from envoy.admin import crud
 from envoy.admin.crud.certificate import select_all_certificates_for_aggregator
 from envoy.server import model
-from envoy.admin import crud
 from envoy.server.crud.aggregator import select_aggregator
 
 
@@ -27,8 +27,10 @@ async def test_count_all_aggregators_empty(pg_empty_config: psycopg.Connection)
 @pytest.mark.parametrize(
     "start, limit, expected_aggregator_ids, expected_domain_ids",
     [
+        (None, None, [1, 2, 3], [1, 2, 3, 4]),
         (0, 500, [1, 2, 3], [1, 2, 3, 4]),
         (0, 1, [1], [1, 4]),
+        (None, 1, [1], [1, 4]),
         (1, 1, [2], [2]),
         (2, 1, [3], [3]),
         (3, 1, [], []),

tests/unit/server/mapper/sep2/test_pub_sub.py

@@ -532,7 +532,7 @@ def test_NotificationMapper_map_sites_to_response(notification_type: Notificatio
         DeviceOrAggregatorRequestScope, seed=1001, href_prefix="/custom/prefix"
     )
 
-    notification = NotificationMapper.map_sites_to_response([site1, site2], sub, scope, notification_type, False)
+    notification = NotificationMapper.map_sites_to_response([site1, site2], sub, scope, notification_type, False, 9876)
     assert isinstance(notification, Notification)
     assert notification.subscribedResource.startswith("/custom/prefix")
     assert EndDeviceListUri in notification.subscribedResource
@@ -544,6 +544,7 @@ def test_NotificationMapper_map_sites_to_response(notification_type: Notificatio
         assert notification.status == NotificationStatus.DEFAULT
 
     assert notification.resource.type == XSI_TYPE_END_DEVICE_LIST
+    assert notification.resource.pollRate == 9876
     assert_list_type(EndDeviceResponse, notification.resource.EndDevice, count=2)
     assert_entity_hrefs_contain_entity_id_and_prefix(
         [e.href for e in notification.resource.EndDevice], [site1.site_id, site2.site_id], scope.href_prefix