diff --git a/.env.exemple b/.env.exemple index b397d31..3ddee8c 100644 --- a/.env.exemple +++ b/.env.exemple @@ -30,6 +30,14 @@ STORAGE_ACCESS_KEY=minioadmin STORAGE_SECRET_KEY=minioadmin STORAGE_REGION=us-east-1 +# ── Realtime (Redis pub/sub) ── +# Fire-and-forget SSE hints consumed by crm-backend's realtime hub. Optional +# everywhere, including production: an empty URL makes core.realtime.emit() a +# silent no-op, so a broker outage degrades the UI to polling and never affects +# a business write. db 1 on purpose — db 0 is crm-backend's cache. +REALTIME_ENABLED=false +REALTIME_REDIS_URL=redis://:changeme_redis_password@localhost:6379/1 + # ── CORS ── ALLOW_ORIGIN=* diff --git a/.gitignore b/.gitignore index a96b88b..93ebac6 100644 --- a/.gitignore +++ b/.gitignore @@ -210,7 +210,11 @@ pyrightconfig.json [Ll]ib [Ll]ib64 [Ll]ocal -[Ss]cripts +# [Ss]cripts — REMOVED. This unanchored toptal venv entry matched this +# project's own top-level scripts/ directory, which silently kept every +# file under scripts/sql/migrations/ out of the repository. The virtualenv +# case it was meant to cover is already handled by the `.venv` and `venv/` +# rules above, so the entry was pure downside. pyvenv.cfg pip-selfcheck.json diff --git a/api/simulation/mappers.py b/api/simulation/mappers.py index 7d10040..900e9bf 100644 --- a/api/simulation/mappers.py +++ b/api/simulation/mappers.py @@ -1,10 +1,15 @@ from api.simulation.schemas import ( ConsumerResult, + CrmDataPreview, + IncompleteMeter, IterationResult, KeyResult, + PreviewBlocker, Simulation, SimulationDetail, ) +from core.i18n import translate +from shared.crm_preflight import Preflight from shared.models.local_models import ( SimulationConsumerResultModel, SimulationIterationResultModel, @@ -80,3 +85,44 @@ def to_simulation_detail(simulation: SimulationModel) -> SimulationDetail: has_timeseries=simulation.result_storage_key is not None, key_result=(to_key_result_schema(simulation.key_result) if simulation.key_result else None), ) + + +def to_crm_data_preview(preflight: Preflight, locale: str) -> CrmDataPreview: + """Render a pre-flight verdict for the manager's screen. + + Blocker messages are localised here rather than in the service so that + translation stays at the API edge — the worker reaches the same verdict via + ``Preflight`` and needs no locale at all. + """ + summary = preflight.summary + return CrmDataPreview( + can_simulate=preflight.ok, + matched_participants=preflight.matched, + unmatched_participants=preflight.unmatched, + meter_count=len(summary.eans), + reading_count=summary.total_rows, + first_timestamp=summary.first_timestamp, + last_timestamp=summary.last_timestamp, + total_consumption_kwh=summary.total_consumption_kwh, + total_injection_kwh=summary.total_injection_kwh, + # Only participants are reported: a gap in a meter this key ignores + # changes nothing about the simulation. + incomplete_meters=[ + IncompleteMeter( + ean=e.ean, + readings=e.distinct_ts, + expected=summary.grid_size, + missing=summary.grid_size - e.distinct_ts, + ) + for e in summary.incomplete + if e.ean in set(preflight.matched) + ], + blockers=[ + PreviewBlocker( + error_code=b.error.code, + message=translate(b.error.key, locale=locale), + detail=b.detail, + ) + for b in preflight.blockers + ], + ) diff --git a/api/simulation/routes.py b/api/simulation/routes.py index 2ad8982..86ac2a4 100644 --- a/api/simulation/routes.py +++ b/api/simulation/routes.py @@ -1,9 +1,13 @@ +from datetime import date from typing import Annotated, Any -from fastapi import APIRouter, Depends, File, Form, Query, UploadFile +from fastapi import APIRouter, Body, Depends, File, Form, Query, UploadFile from sqlalchemy.ext.asyncio import AsyncSession +from api.simulation.mappers import to_crm_data_preview from api.simulation.schemas import ( + CrmDataPreview, + SimulateFromCrmRequest, SimulateRequest, SimulateResponse, Simulation, @@ -12,7 +16,7 @@ ) from api.simulation.service import SimulationService from core.api_response import ApiResponse, ApiResponsePaginated -from core.context_vars import current_internal_community_id +from core.context_vars import current_internal_community_id, current_locale from core.database.database import get_crm_session, get_local_session from core.errors.errors import ErrorException from core.errors.with_default_error import with_default_error @@ -45,6 +49,37 @@ async def get_simulations( return ApiResponsePaginated[list[Simulation]](data=data, pagination=pagination) +# GET (/crm-data-preview) : What the CRM holds for an operation + period, and +# whether the key's participants match those meters. +# +# Declared BEFORE `GET /{id}`: FastAPI matches in declaration order, so putting +# this after the integer catch-all would make "/crm-data-preview" try to parse +# as an id and 422. +@simulation_routes.get("/crm-data-preview", response_model=ApiResponse[CrmDataPreview]) +@with_default_error(default_error=errors.simulation.GET_CRM_PREVIEW) +async def get_crm_data_preview( + local_session: Annotated[AsyncSession, Depends(get_local_session)], + crm_session: Annotated[AsyncSession, Depends(get_crm_session)], + id_key: Annotated[int, Query(description="CRM allocation key to be simulated.")], + id_sharing_operation: Annotated[int, Query(description="CRM sharing operation id.")], + period_start: Annotated[date, Query(description="First day of the period (inclusive).")], + period_end: Annotated[date, Query(description="Last day of the period (inclusive).")], +): + internal_community_id = current_internal_community_id.get() + if internal_community_id is None: + raise ErrorException(error=errors.auth.UNAUTHORIZED, status_code=401) + service = SimulationService(local_session, crm_session) + preflight = await service.preview_crm_data( + id_key=id_key, + id_sharing_operation=id_sharing_operation, + period_start=period_start, + period_end=period_end, + community_id=internal_community_id, + ) + locale = current_locale.get().split("_")[0] + return ApiResponse[CrmDataPreview](data=to_crm_data_preview(preflight, locale)) + + # GET (/{id}) : One simulation with its scalar result tree. @simulation_routes.get("/{id}", response_model=ApiResponse[SimulationDetail]) @with_default_error(default_error=errors.simulation.GET_SIMULATION) @@ -95,6 +130,25 @@ async def start_simulation( return ApiResponse[SimulateResponse](data=data) +# POST (/from-crm) Simulate against CRM meter data +# +# Plain JSON, unlike POST / — with no file part nothing forces multipart. It +# also correctly keeps the default 2 MB body cap rather than the upload one. +@simulation_routes.post("/from-crm", response_model=ApiResponse[SimulateResponse]) +@with_default_error(default_error=errors.simulation.START_SIMULATION) +async def start_simulation_from_crm( + body: Annotated[SimulateFromCrmRequest, Body()], + local_session: Annotated[AsyncSession, Depends(get_local_session)], + crm_session: Annotated[AsyncSession, Depends(get_crm_session)], +): + internal_community_id = current_internal_community_id.get() + if internal_community_id is None: + raise ErrorException(error=errors.auth.UNAUTHORIZED, status_code=401) + service = SimulationService(local_session, crm_session) + data = await service.start_simulation_from_crm(body, internal_community_id) + return ApiResponse[SimulateResponse](data=data) + + # DELETE (/{id}) : Delete a simulation and its results. @simulation_routes.delete("/{id}", response_model=ApiResponse[str]) @with_default_error(default_error=errors.simulation.DELETE_SIMULATION) diff --git a/api/simulation/schemas.py b/api/simulation/schemas.py index 08721b9..f98137e 100644 --- a/api/simulation/schemas.py +++ b/api/simulation/schemas.py @@ -1,3 +1,5 @@ +import datetime + from pydantic import BaseModel, Field from shared.const import SimulationStatus @@ -107,3 +109,64 @@ class SimulateRequest(BaseModel): class SimulateResponse(BaseModel): id: int = Field(..., description="ID of the freshly created simulation row.") status: SimulationStatus = Field(..., description="Initial status (PENDING on success).") + + +# --------------------------------------------------------------------------- +# CRM-sourced simulation (source = DataSource.CRM) +# --------------------------------------------------------------------------- + + +class SimulateFromCrmRequest(BaseModel): + """Body of ``POST /from-crm``. + + Unlike ``SimulateRequest`` this *is* the FastAPI body model: with no file + part there is nothing forcing multipart, so the request is plain JSON. + + ``injection_name`` is deliberately absent — the production profile is summed + from the meters themselves, which is the whole reason this path is simpler + for the user than uploading a file. + """ + + name: str = Field(..., min_length=1, description="User-facing label for the simulation.") + id_key: int = Field(..., description="CRM allocation key id to stress-test.") + id_sharing_operation: int = Field(..., description="CRM sharing operation to read meters from.") + period_start: datetime.date = Field(..., description="First day of the period (inclusive).") + period_end: datetime.date = Field(..., description="Last day of the period (inclusive).") + + +class IncompleteMeter(BaseModel): + """A participant's meter missing part of the period. Zero-filled, not fatal.""" + + ean: str + readings: int = Field(..., description="Distinct timestamps this meter actually has.") + expected: int = Field(..., description="Distinct timestamps across the whole operation.") + missing: int = Field(..., description="expected - readings.") + + +class PreviewBlocker(BaseModel): + """A reason the period cannot be used, already localised.""" + + error_code: int = Field(..., description="Matches the error_code of the eventual 4xx.") + message: str = Field(..., description="Localised, manager-facing explanation.") + detail: str = Field(..., description="Which meters/participants triggered it.") + + +class CrmDataPreview(BaseModel): + """What ``GET /crm-data-preview`` shows before the manager commits to a run. + + ``matched`` / ``unmatched`` are the heart of it: the key's participant names + must be meter EANs, and this is where the manager finds out that they are + not, rather than after a failed run. + """ + + can_simulate: bool + matched_participants: list[str] + unmatched_participants: list[str] + meter_count: int = Field(..., description="Meters with readings in the period.") + reading_count: int + first_timestamp: datetime.datetime | None + last_timestamp: datetime.datetime | None + total_consumption_kwh: float + total_injection_kwh: float + incomplete_meters: list[IncompleteMeter] + blockers: list[PreviewBlocker] diff --git a/api/simulation/service.py b/api/simulation/service.py index 04f56a5..d16fafa 100644 --- a/api/simulation/service.py +++ b/api/simulation/service.py @@ -1,3 +1,4 @@ +import datetime import logging from uuid import uuid4 @@ -6,6 +7,7 @@ from api.simulation.mappers import to_simulation_detail, to_simulation_schema from api.simulation.repository import SimulationRepository from api.simulation.schemas import ( + SimulateFromCrmRequest, SimulateRequest, SimulateResponse, Simulation, @@ -21,10 +23,17 @@ from core.middleware.request_limits import UPLOAD_MAX_BODY_BYTES from core.queue.helper import Event, send_event from core.queue.init import get_jetstream -from shared.const import SIMULATION_SUBJECT, SimulationStatus +from shared import crm_preflight +from shared.const import SIMULATION_SUBJECT, DataSource, SimulationStatus +from shared.crm_meter_repository import CrmMeterRepository +from shared.crm_preflight import Preflight from shared.crm_repository import CRMRepository from shared.custom_errors import errors from shared.models.local_models import SimulationModel +from simulation.key_mapping import ( + consumer_names_of, + from_crm_allocation_key, +) logger = logging.getLogger(__name__) @@ -91,6 +100,128 @@ async def get_timeseries(self, id: int) -> SimulationTimeseries: raise ErrorException(error=errors.simulation.RESULT_NOT_FOUND, status_code=404) from exc return SimulationTimeseries.model_validate_json(content) + # ------------------------------------------------------------------ + # CRM-sourced simulation + # ------------------------------------------------------------------ + + async def preview_crm_data( + self, + *, + id_key: int, + id_sharing_operation: int, + period_start: datetime.date, + period_end: datetime.date, + community_id: int, + ) -> Preflight: + """Aggregate the period and check it against the key, running nothing. + + Also the pre-flight for ``start_simulation_from_crm`` — one code path, so + the answer the manager saw and the answer that gates the run cannot drift. + """ + if period_start > period_end: + raise ErrorException(error=errors.simulation.INVALID_PERIOD, status_code=422) + + key = await self.crm_repository.get_allocation_key(id_key, community_id) + if key is None: + raise ErrorException(error=errors.simulation.KEY_NOT_FOUND, status_code=404) + + crm_meters = CrmMeterRepository(self.crm_session) + # Explicit tenant check: without it a foreign operation id is + # indistinguishable from an empty period, which is a confusing 422 for a + # legitimate user and a soft information leak for everyone else. + if not await crm_meters.sharing_operation_exists( + id_community=community_id, id_sharing_operation=id_sharing_operation + ): + raise ErrorException( + error=errors.simulation.SHARING_OPERATION_NOT_FOUND, status_code=404 + ) + + summary = await crm_meters.summarize( + id_community=community_id, + id_sharing_operation=id_sharing_operation, + period_start=period_start, + period_end=period_end, + ) + return crm_preflight.evaluate(summary, consumer_names_of(from_crm_allocation_key(key))) + + async def start_simulation_from_crm( + self, req: SimulateFromCrmRequest, community_id: int + ) -> SimulateResponse: + """Queue a simulation that reads its input from the CRM. + + Same ordering as the file path minus the upload: validate, commit, then + publish. There is no object to roll back, so the ``_best_effort_delete`` + branches have no counterpart here. + """ + key = await self.crm_repository.get_allocation_key(req.id_key, community_id) + if key is None: + raise ErrorException(error=errors.simulation.KEY_NOT_FOUND, status_code=404) + + # Re-run the pre-flight rather than trusting whatever the client saw: + # the preview may be minutes old, and an import can have landed since. + preflight = await self.preview_crm_data( + id_key=req.id_key, + id_sharing_operation=req.id_sharing_operation, + period_start=req.period_start, + period_end=req.period_end, + community_id=community_id, + ) + if preflight.blockers: + first = preflight.blockers[0] + logger.info( + "CRM simulation refused for community %d op %d key %d: %s", + community_id, + req.id_sharing_operation, + req.id_key, + first.detail, + ) + raise ErrorException(error=first.error, status_code=422) + + model = SimulationModel( + name=req.name, + id_community=community_id, + source=DataSource.CRM, + id_sharing_operation=req.id_sharing_operation, + period_start=req.period_start, + period_end=req.period_end, + id_key=req.id_key, + key_name=key.name, + status=SimulationStatus.PENDING, + data_warnings=preflight.warnings, + ) + await self.repository.create_simulation(model) + await self.local_session.commit() + simulation_id = model.id + app_metrics.simulations_created.add(1) + await self.audit_log_service.log( + AuditLogInput( + action=AuditActions.SIMULATION_CREATED, + entity_type="simulation", + entity_id=str(simulation_id), + payload={ + "name": req.name, + "id_key": req.id_key, + "key_name": key.name, + "source": DataSource.CRM.name, + "id_sharing_operation": req.id_sharing_operation, + "period_start": req.period_start.isoformat(), + "period_end": req.period_end.isoformat(), + }, + ) + ) + + event = Event(type="simulation.requested", data={"simulation_id": simulation_id}) + try: + await send_event(get_jetstream(), SIMULATION_SUBJECT, event) + except Exception as exc: + logger.exception( + "Failed to publish simulation %d to %s", simulation_id, SIMULATION_SUBJECT + ) + await self._mark_failed_to_queue(simulation_id, str(exc)) + raise ErrorException(error=errors.simulation.START_SIMULATION, status_code=500) from exc + + return SimulateResponse(id=simulation_id, status=SimulationStatus.PENDING) + async def start_simulation( self, req: SimulateRequest, file: UploadFile, community_id: int ) -> SimulateResponse: diff --git a/core/config.py b/core/config.py index ae650af..a8a556a 100644 --- a/core/config.py +++ b/core/config.py @@ -61,6 +61,14 @@ class Settings(BaseSettings): # MinIO ignores region but botocore still requires it to sign requests. STORAGE_REGION: str = "us-east-1" + # ---- Realtime (Redis pub/sub) ---- + # Fire-and-forget SSE hints. Deliberately NOT in validate_env_config's + # required set (contrast NATS_URL below): realtime is optional by design, and + # making it mandatory would turn a broker outage into a boot failure. An + # empty URL makes core.realtime.emit() a silent no-op. + REALTIME_ENABLED: bool = False + REALTIME_REDIS_URL: str = "" + # ---- CORS ---- ALLOW_ORIGIN: str = "*" diff --git a/core/realtime/__init__.py b/core/realtime/__init__.py new file mode 100644 index 0000000..5606326 --- /dev/null +++ b/core/realtime/__init__.py @@ -0,0 +1,54 @@ +"""Shared realtime publisher — BYTE-IDENTICAL across every producing service. + +Copied verbatim into billing, administrative-document, news-board, +allocation-key-generation and simulation-key, mirroring the ``core/notifications`` +convention. ``scripts/check-realtime-parity.sh`` at the monorepo root is the gate: +make the change in the reference service (news-board) and copy it out, never edit +one copy. + +Consumed by crm-backend's realtime hub (``src/shared/realtime/``) and delivered +to browsers over SSE. Fire-and-forget by contract: if the recipient has no stream +open the event is dropped, which is correct — every event is a hint, and the +client refetches authoritative state through the API gateway. + +Usage, always AFTER the owning transaction commits:: + + from core.realtime import CommunityAudience, Tier, emit + + await session.commit() + await emit( + topic="generation.finished", + audience=CommunityAudience(community_id=cid, tier=Tier.MANAGER), + resource=("generation", generation_id), + scope_community_id=cid, + hint={"status": "success"}, + ) +""" + +from .bus import close, emit, log_realtime_state +from .channels import ( + Audience, + CommunityAudience, + Tier, + UserAudience, + UsersAudience, + community_channel, + user_channel, +) +from .envelope import MAX_ENVELOPE_BYTES, TOPICS, build_envelope + +__all__ = [ + "MAX_ENVELOPE_BYTES", + "TOPICS", + "Audience", + "CommunityAudience", + "Tier", + "UserAudience", + "UsersAudience", + "build_envelope", + "close", + "community_channel", + "emit", + "log_realtime_state", + "user_channel", +] diff --git a/core/realtime/bus.py b/core/realtime/bus.py new file mode 100644 index 0000000..8abecc5 --- /dev/null +++ b/core/realtime/bus.py @@ -0,0 +1,154 @@ +"""Fire-and-forget realtime publisher. + +One lazily-created ``redis.asyncio`` client per process. Every failure mode — +no configuration, unreachable broker, hung broker, malformed envelope — is a +silent no-op, because the *only* thing lost is UI freshness: crm-backend's +clients refetch authoritative state on every reconnect, and every poller in the +SPA keeps running (slower) as a durability backstop. +""" + +import asyncio +import json +import logging +from collections.abc import Mapping + +import redis.asyncio as redis + +from core.config import settings + +from .channels import Audience +from .envelope import build_envelope + +logger = logging.getLogger(__name__) + +_client: redis.Redis | None = None + +#: Publishing must never add latency to a request or a worker tick. A wedged +#: broker is bounded here rather than by the socket, because a *connected but +#: hung* Redis would otherwise await forever. +_PUBLISH_TIMEOUT_SECONDS = 1.0 + + +def _redacted_url() -> str: + """The DSN with its password removed. NEVER log the raw value. + + ``REALTIME_REDIS_URL`` is composed from ``REDIS_PASSWORD`` in + docker-compose, so it carries a live secret in userinfo position. + """ + url = settings.REALTIME_REDIS_URL + scheme, sep, rest = url.partition("://") + if not sep or "@" not in rest: + return url + return f"{scheme}://***@{rest.rpartition('@')[2]}" + + +def log_realtime_state(component: str) -> None: + """Announce this process's realtime publishing state, once, at startup. + + *** CALL THIS AFTER configure_logging(). NEVER at module import time. *** + + This module is imported long before logging is configured: ``worker/main.py`` + imports the dispatcher, which reaches ``worker/persistence.py``, which imports + this file — all in the import block at the top — while ``configure_logging()`` + runs inside ``async def main()``. And ``core/logging.py`` opens with + ``root.handlers.clear()``. So at import time the root logger has no handlers, + Python falls back to ``logging.lastResort`` at WARNING, and an INFO line here + is dropped with no trace whatsoever — reproducing the exact silence this + function exists to break. Do not "simplify" it into the module body. + + Why it exists: four worker containers once ran images built before this + package existed. ``core/realtime/`` was absent and so were the emit call + sites, so three topics were published by nothing at all — for three days, + while their environment variables looked perfectly correct, because + ``--force-recreate`` rebuilds the container from the EXISTING image. The + absence of this line is the cheapest signal that an image predates the + feature. ``scripts/check-realtime-images.sh`` is the automatable form of the + same check, and ``scripts/check-realtime-parity.sh`` cannot see it at all — + it compares source trees, which were green throughout. + """ + if not settings.REALTIME_ENABLED: + logger.info("Realtime disabled — %s publishes nothing", component) + return + if not settings.REALTIME_REDIS_URL: + # Enabled but unconfigured is a misconfiguration, not a deployment choice. + logger.warning( + "Realtime ENABLED but REALTIME_REDIS_URL is empty — %s publishes nothing", component + ) + return + logger.info("Realtime publisher ready — %s publishing to %s", component, _redacted_url()) + + +def _get_client() -> redis.Redis: + global _client + if _client is None: + _client = redis.from_url( + settings.REALTIME_REDIS_URL, + socket_connect_timeout=1, + socket_timeout=1, + health_check_interval=30, + decode_responses=False, + ) + return _client + + +async def emit( + *, + topic: str, + audience: Audience, + resource: tuple[str, str | int], + hint: Mapping[str, str | int | float | bool | None] | None = None, + scope_community_id: int | None = None, +) -> None: + """Publish a realtime hint. NEVER raises. + + *** CALL THIS AFTER THE COMMIT. NEVER inside ``begin_nested()``, and never + inside the ``try`` that owns the business write. *** + + Publishing pre-commit does not merely lose an event — it tells the browser to + refetch and read PRE-COMMIT state, and because the transport is + fire-and-forget there is NO second event, ever. The result is a permanently + stale UI behind a 200, with no error anywhere. That is the same silhouette as + the sweep-commit-ordering and notification-savepoint traps. + + Note the deliberate asymmetry with ``core.notifications.service.publish()``, + which MUST run inside the caller's transaction because it writes rows. These + two have opposite requirements. Do not "unify" them. + + Do not fire this as a bare ``asyncio.create_task`` either: in a worker that + finishes immediately the task is orphaned (the event is lost anyway) and may + log after teardown. + """ + if not settings.REALTIME_ENABLED or not settings.REALTIME_REDIS_URL: + return # no-op: the default everywhere realtime is not deployed + + try: + envelope = build_envelope( + topic=topic, + resource=resource, + hint=hint, + scope_community_id=scope_community_id, + ) + if envelope is None: + logger.warning("realtime: envelope rejected locally topic=%s", topic) + return + + body = json.dumps(envelope, separators=(",", ":")) + client = _get_client() + async with asyncio.timeout(_PUBLISH_TIMEOUT_SECONDS): + for channel in audience.channels(): + await client.publish(channel, body) + # Blanket by contract: nothing this function can hit is worth propagating + # into a caller that has already committed. + except Exception: + logger.warning("realtime: emit failed topic=%s", topic, exc_info=True) + + +async def close() -> None: + """Release the client. For worker shutdown and test teardown.""" + global _client + if _client is not None: + try: + await _client.aclose() + except Exception: + logger.warning("realtime: client close failed", exc_info=True) + _client = None diff --git a/core/realtime/channels.py b/core/realtime/channels.py new file mode 100644 index 0000000..2144039 --- /dev/null +++ b/core/realtime/channels.py @@ -0,0 +1,89 @@ +"""THE ONLY PLACE A REALTIME CHANNEL STRING IS BUILT (Python side). + +This is a security control, not a style rule. A producer that accidentally +publishes a per-user thing onto a community tier is a cross-tenant leak, and +``tier`` below is a required argument with no default precisely so that mistake +cannot be made by omission. A test greps each service for the literal +``notify:v1:`` outside this module and fails on a hit. + +Grammar (fixed arity, so Redis 6 ACLs can later be granted per prefix without a +redesign) — byte-identical to +``crm-backend/src/shared/realtime/realtime.channels.ts``:: + + notify:v1:u:{internal_app_user_id} + notify:v1:c:{internal_community_id}:{MEMBER|MANAGER} + +Ids are the INTERNAL integer keys (``app_user.id``, ``community.id``) — what +every producer here already holds, and what ``notification.id_user`` is. They are +NOT Keycloak subs or org uuids. + +The community family is what lets a worker with NO user attribution at all (the +generation and simulation jobs carry only ``id_community``) address exactly the +right audience with zero database lookups. Its safety comes from the subscribe +side: crm-backend only ever subscribes a connection to tiers the ticket mint +proved the user holds, from gateway-verified claims. +""" + +from collections.abc import Iterable, Sequence +from dataclasses import dataclass +from enum import Enum + +_PREFIX = "notify:v1" + + +class Tier(str, Enum): + """Channel tiers. Deliberately coarser than a role: there is no ADMIN bucket.""" + + #: Everyone in the community, managers included. + MEMBER = "MEMBER" + #: ADMIN and MANAGER only. + MANAGER = "MANAGER" + + +def user_channel(internal_user_id: int) -> str: + """Channel for one user, in every community and outside all of them.""" + return f"{_PREFIX}:u:{internal_user_id}" + + +def community_channel(internal_community_id: int, tier: Tier) -> str: + """Channel for one tier of one community.""" + return f"{_PREFIX}:c:{internal_community_id}:{tier.value}" + + +@dataclass(frozen=True) +class UserAudience: + """One recipient, addressed by internal ``app_user.id``.""" + + user_id: int + + def channels(self) -> Sequence[str]: + return (user_channel(self.user_id),) + + +@dataclass(frozen=True) +class UsersAudience: + """An explicit set of recipients. Duplicates are collapsed.""" + + user_ids: Iterable[int] + + def channels(self) -> Sequence[str]: + return tuple(user_channel(uid) for uid in dict.fromkeys(self.user_ids)) + + +@dataclass(frozen=True) +class CommunityAudience: + """One tier of one community. + + ``Tier.MANAGER`` reaches ADMIN and MANAGER only; ``Tier.MEMBER`` reaches + everyone in the community, managers included — a manager's connection + subscribes to both tiers, so "everyone" is one publish, not two. + """ + + community_id: int + tier: Tier + + def channels(self) -> Sequence[str]: + return (community_channel(self.community_id, self.tier),) + + +Audience = UserAudience | UsersAudience | CommunityAudience diff --git a/core/realtime/envelope.py b/core/realtime/envelope.py new file mode 100644 index 0000000..a2db705 --- /dev/null +++ b/core/realtime/envelope.py @@ -0,0 +1,92 @@ +"""Realtime envelope construction and validation. + +An envelope is a HINT — "something about resource X changed" — never data. The +rules below mirror ``crm-backend/src/shared/realtime/realtime.envelope.ts``, +which re-validates everything on the way out to a browser: + +* **No business data.** No names, emails, EANs, amounts, invoice numbers, + storage keys, error messages. +* **No display strings.** Toast text is chosen client-side from ``topic`` + + ``hint["status"]`` against the i18n bundle. This is a security control: a + compromised publisher gets a nuisance channel, never a text-injection channel + into every open browser. +* **No recipient field.** The channel already says who. A recipient in the body + invites a subscriber-side "is this for me?" check — authorization on the wrong + leg. + +``ref.id`` is permitted: any authorized reader can already see it, and the +client needs it to decide *which* row to refetch. +""" + +import json +import secrets +from collections.abc import Mapping +from datetime import UTC, datetime +from typing import Any, Final + +#: Hard ceiling on a serialized envelope, in BYTES (not characters). +MAX_ENVELOPE_BYTES: Final[int] = 1024 + +#: The topic registry. Mirrors crm-backend's realtime.topics.ts; an unknown topic +#: is dropped by the hub rather than forwarded, so publishing one is a silent +#: no-op that is much easier to find here. +TOPICS: Final[frozenset[str]] = frozenset( + { + "notification.created", + "generation.finished", + "simulation.finished", + "billing_run.finished", + "session.revoked", + } +) + +_SCALARS = (str, int, float, bool) + + +def _is_scalar(value: Any) -> bool: + # bool is a subclass of int, so it is already covered; None is allowed. + return value is None or isinstance(value, _SCALARS) + + +def build_envelope( + *, + topic: str, + resource: tuple[str, str | int], + hint: Mapping[str, str | int | float | bool | None] | None = None, + scope_community_id: int | None = None, +) -> dict[str, Any] | None: + """Build a valid envelope, or ``None`` if the input violates the contract. + + Returns ``None`` rather than raising: every caller is a fire-and-forget side + effect that must never affect a business write, so a malformed hint has to + degrade to "no event", not to an exception travelling up through a commit + path. + """ + if topic not in TOPICS: + return None + + kind, ref_id = resource + if not kind or ref_id is None: + return None + + flat: dict[str, Any] = {} + for key, value in (hint or {}).items(): + if not _is_scalar(value): + return None + flat[str(key)] = value + + envelope: dict[str, Any] = { + "v": 1, + # A client-side dedupe key. Not sortable on purpose: there is no replay, + # so ordering buys nothing. + "id": secrets.token_hex(8), + "topic": topic, + "at": datetime.now(UTC).isoformat().replace("+00:00", "Z"), + "scope": {"community_id": scope_community_id}, + "ref": {"kind": str(kind), "id": str(ref_id)}, + "hint": flat, + } + + if len(json.dumps(envelope, separators=(",", ":")).encode("utf-8")) > MAX_ENVELOPE_BYTES: + return None + return envelope diff --git a/locales/de.json b/locales/de.json index 2a2bb63..74a1e1e 100644 --- a/locales/de.json +++ b/locales/de.json @@ -20,7 +20,15 @@ "DELETE_SIMULATION": "Die Simulation konnte nicht gelöscht werden (Code: 2107)", "RESULT_NOT_FOUND": "Simulationsergebnisse nicht gefunden (Code: 2108)", "GET_TIMESERIES": "Zeitreihen der Simulation konnten nicht abgerufen werden (Code: 2109)", - "FILE_TOO_LARGE": "Die Datei überschreitet die maximal zulässige Größe (Code: 2110)" + "FILE_TOO_LARGE": "Die Datei überschreitet die maximal zulässige Größe (Code: 2110)", + "INVALID_PERIOD": "Das Startdatum muss vor oder auf dem Enddatum liegen (Code: 2111)", + "SHARING_OPERATION_NOT_FOUND": "Sharing-Vorgang nicht gefunden (Code: 2112)", + "CRM_NO_DATA": "Keine Messdaten für diesen Sharing-Vorgang in diesem Zeitraum (Code: 2113)", + "CRM_DUPLICATE_READINGS": "Doppelte Messwerte erkannt: die Daten wurden vermutlich zweimal importiert (Code: 2114)", + "CRM_NO_INJECTION": "In diesem Zeitraum wurde nichts eingespeist, es gibt also nichts zu teilen (Code: 2115)", + "CRM_RANGE_TOO_LARGE": "Der gewählte Zeitraum enthält zu viele Messwerte, bitte kürzen Sie ihn (Code: 2116)", + "KEY_CONSUMERS_NOT_MATCHED": "Einige Teilnehmer des Schlüssels stimmen mit keinem Zähler (EAN) in diesem Zeitraum überein (Code: 2117)", + "GET_CRM_PREVIEW": "Die Messdaten konnten nicht geprüft werden (Code: 2118)" } } } diff --git a/locales/en.json b/locales/en.json index 52c0ab2..7848e7c 100644 --- a/locales/en.json +++ b/locales/en.json @@ -20,7 +20,15 @@ "DELETE_SIMULATION": "Failed to delete the simulation (code: 2107)", "RESULT_NOT_FOUND": "Simulation results not found (code: 2108)", "GET_TIMESERIES": "Failed to retrieve the simulation time series (code: 2109)", - "FILE_TOO_LARGE": "The file exceeds the maximum allowed size (code: 2110)" + "FILE_TOO_LARGE": "The file exceeds the maximum allowed size (code: 2110)", + "INVALID_PERIOD": "The start date must be on or before the end date (code: 2111)", + "SHARING_OPERATION_NOT_FOUND": "Sharing operation not found (code: 2112)", + "CRM_NO_DATA": "No metering data for this sharing operation over this period (code: 2113)", + "CRM_DUPLICATE_READINGS": "Duplicate readings detected: the data was most likely imported twice (code: 2114)", + "CRM_NO_INJECTION": "No energy was injected over this period, so there is nothing to share (code: 2115)", + "CRM_RANGE_TOO_LARGE": "The selected period holds too many readings, please shorten it (code: 2116)", + "KEY_CONSUMERS_NOT_MATCHED": "Some of the key's participants match no meter (EAN) over this period (code: 2117)", + "GET_CRM_PREVIEW": "The metering data could not be checked (code: 2118)" } } } diff --git a/locales/fr.json b/locales/fr.json index 52037fd..3fbba8d 100644 --- a/locales/fr.json +++ b/locales/fr.json @@ -20,7 +20,15 @@ "DELETE_SIMULATION": "Échec de la suppression de la simulation (code: 2107)", "RESULT_NOT_FOUND": "Résultats de simulation introuvables (code: 2108)", "GET_TIMESERIES": "Échec de la récupération des séries temporelles de la simulation (code: 2109)", - "FILE_TOO_LARGE": "Le fichier dépasse la taille maximale autorisée (code: 2110)" + "FILE_TOO_LARGE": "Le fichier dépasse la taille maximale autorisée (code: 2110)", + "INVALID_PERIOD": "La date de début doit précéder la date de fin (code: 2111)", + "SHARING_OPERATION_NOT_FOUND": "Opération de partage introuvable (code: 2112)", + "CRM_NO_DATA": "Aucune donnée de comptage pour cette opération de partage sur cette période (code: 2113)", + "CRM_DUPLICATE_READINGS": "Des relevés en double ont été détectés : les données ont probablement été importées deux fois (code: 2114)", + "CRM_NO_INJECTION": "Aucune injection sur cette période : il n'y a rien à partager (code: 2115)", + "CRM_RANGE_TOO_LARGE": "La période sélectionnée contient trop de relevés, veuillez la raccourcir (code: 2116)", + "KEY_CONSUMERS_NOT_MATCHED": "Certains participants de la clé ne correspondent à aucun compteur (EAN) sur cette période (code: 2117)", + "GET_CRM_PREVIEW": "Impossible de vérifier les données de comptage (code: 2118)" } } } diff --git a/locales/nl.json b/locales/nl.json index 20a34e4..dfa543d 100644 --- a/locales/nl.json +++ b/locales/nl.json @@ -20,7 +20,15 @@ "DELETE_SIMULATION": "Verwijderen van de simulatie mislukt (code: 2107)", "RESULT_NOT_FOUND": "Simulatieresultaten niet gevonden (code: 2108)", "GET_TIMESERIES": "Ophalen van de tijdreeksen van de simulatie mislukt (code: 2109)", - "FILE_TOO_LARGE": "Het bestand overschrijdt de maximaal toegestane grootte (code: 2110)" + "FILE_TOO_LARGE": "Het bestand overschrijdt de maximaal toegestane grootte (code: 2110)", + "INVALID_PERIOD": "De begindatum moet op of vóór de einddatum liggen (code: 2111)", + "SHARING_OPERATION_NOT_FOUND": "Deeloperatie niet gevonden (code: 2112)", + "CRM_NO_DATA": "Geen meetgegevens voor deze deeloperatie in deze periode (code: 2113)", + "CRM_DUPLICATE_READINGS": "Dubbele meetwaarden gevonden: de gegevens zijn waarschijnlijk twee keer geïmporteerd (code: 2114)", + "CRM_NO_INJECTION": "Er is in deze periode niets geïnjecteerd, dus er valt niets te delen (code: 2115)", + "CRM_RANGE_TOO_LARGE": "De geselecteerde periode bevat te veel meetwaarden, kies een kortere periode (code: 2116)", + "KEY_CONSUMERS_NOT_MATCHED": "Sommige deelnemers van de sleutel komen met geen enkele meter (EAN) in deze periode overeen (code: 2117)", + "GET_CRM_PREVIEW": "De meetgegevens konden niet worden gecontroleerd (code: 2118)" } } } diff --git a/main.py b/main.py index d558682..6eb8def 100644 --- a/main.py +++ b/main.py @@ -16,6 +16,7 @@ from core.middleware.request_limits import RequestLimitsMiddleware from core.middleware.set_auth_context import GatewayScopeMiddleware from core.queue.init import close_nats, init_nats +from core.realtime import log_realtime_state from core.tracing import enrich_span, setup_tracer_provider configure_logging() @@ -24,6 +25,9 @@ @asynccontextmanager async def lifespan(app: FastAPI): + # Absence of this line means the image predates the realtime feature — + # see core/realtime/bus.py. Must come after configure_logging(). + log_realtime_state("simulation-key api") setup_tracer_provider() await init_nats() yield diff --git a/requirements/base.txt b/requirements/base.txt index 56b4bad..7714dfa 100644 --- a/requirements/base.txt +++ b/requirements/base.txt @@ -33,4 +33,9 @@ opentelemetry-semantic-conventions==0.60b1 opentelemetry-util-http==0.60b1 reportlab==4.4.10 -aiosmtplib>=3.0.0 \ No newline at end of file +aiosmtplib>=3.0.0 + +# Realtime SSE hints (fire-and-forget pub/sub consumed by crm-backend's hub). +# redis.asyncio, not the unmaintained aioredis — it is aioredis's successor and +# ships in the same package. +redis==5.2.1 diff --git a/scripts/export_openapi.py b/scripts/export_openapi.py index c4fa88f..64392ec 100644 --- a/scripts/export_openapi.py +++ b/scripts/export_openapi.py @@ -2,6 +2,7 @@ Usage: python scripts/export_openapi.py """ + import json import sys from pathlib import Path diff --git a/scripts/sql/migrations/002_crm_source.sql b/scripts/sql/migrations/002_crm_source.sql new file mode 100644 index 0000000..523cfb4 --- /dev/null +++ b/scripts/sql/migrations/002_crm_source.sql @@ -0,0 +1,51 @@ +-- Migration 002: allow a simulation to source its input from the CRM database +-- instead of an uploaded file. +-- +-- Until now every simulation carried an uploaded CSV/XLSX, so file_storage_key, +-- file_name and injection_name were all NOT NULL. A CRM-sourced run has none of +-- them: it names a sharing operation and a date range, and the worker reads +-- meter_consumption directly, matching the simulated key's participant names +-- against meter EANs. +-- +-- The three file columns therefore become nullable, and a CHECK constraint +-- takes over the job they were doing -- each source shape must be fully +-- populated, so a half-specified row is still impossible. +-- +-- data_warnings holds non-blocking findings from the pre-flight (currently: +-- participants whose meter has gaps, which are zero-filled). It is persisted +-- rather than only shown before launch, because a warning the manager sees once +-- and never again is not really a warning. + +BEGIN; + +ALTER TABLE simulation + ALTER COLUMN file_storage_key DROP NOT NULL, + ALTER COLUMN file_name DROP NOT NULL, + ALTER COLUMN injection_name DROP NOT NULL; + +ALTER TABLE simulation + ADD COLUMN IF NOT EXISTS source SMALLINT NOT NULL DEFAULT 1, + ADD COLUMN IF NOT EXISTS id_sharing_operation INTEGER NULL, + ADD COLUMN IF NOT EXISTS period_start DATE NULL, + ADD COLUMN IF NOT EXISTS period_end DATE NULL, + ADD COLUMN IF NOT EXISTS data_warnings JSONB NULL; + +-- 1=FILE, 2=CRM. Existing rows keep the DEFAULT 1 and satisfy the FILE branch. +ALTER TABLE simulation DROP CONSTRAINT IF EXISTS ck_simulation_source; +ALTER TABLE simulation ADD CONSTRAINT ck_simulation_source CHECK ( + (source = 1 + AND file_storage_key IS NOT NULL + AND file_name IS NOT NULL + AND injection_name IS NOT NULL) + OR (source = 2 + AND id_sharing_operation IS NOT NULL + AND period_start IS NOT NULL + AND period_end IS NOT NULL + AND period_start <= period_end) +); + +INSERT INTO schema_version (version, description) +VALUES (2, 'Allow CRM-sourced simulations (source, sharing operation, period)') +ON CONFLICT DO NOTHING; + +COMMIT; diff --git a/scripts/sql/schema.sql b/scripts/sql/schema.sql index edf8bcb..400c2b1 100644 --- a/scripts/sql/schema.sql +++ b/scripts/sql/schema.sql @@ -29,6 +29,10 @@ INSERT into schema_version (version, description) VALUES( 1, 'First version' ) ON CONFLICT DO NOTHING; +INSERT into schema_version (version, description) VALUES( + 2, 'Allow CRM-sourced simulations (source, sharing operation, period)' +) ON CONFLICT DO NOTHING; + -- ---- simulation ------------------------------------------------------------ -- One row per simulation request. Holds the source file reference, a snapshot @@ -41,12 +45,24 @@ CREATE TABLE IF NOT EXISTS simulation ( name VARCHAR(255) NOT NULL, id_community INTEGER NOT NULL, - -- Source data - -- file_storage_key is the object key inside STORAGE_BUCKET (MinIO). The - -- service uploads on creation; the worker deletes on terminal outcomes. - file_storage_key VARCHAR(512) NOT NULL, - file_name VARCHAR(255) NOT NULL, - injection_name VARCHAR(255) NOT NULL, + -- Source data: 1=FILE (uploaded CSV/XLSX), 2=CRM (meter_consumption). + -- Exactly one of the two column groups below is populated; the + -- ck_simulation_source CHECK is what enforces that, now that the file + -- columns can no longer be NOT NULL. + source SMALLINT NOT NULL DEFAULT 1, + + -- FILE only. file_storage_key is the object key inside STORAGE_BUCKET + -- (MinIO). The service uploads on creation; the worker deletes on terminal + -- outcomes. injection_name names the production column inside the file. + file_storage_key VARCHAR(512) NULL, + file_name VARCHAR(255) NULL, + injection_name VARCHAR(255) NULL, + + -- CRM only. The sharing operation and the inclusive local date range read + -- from meter_consumption. No FK: the CRM lives in a separate database. + id_sharing_operation INTEGER NULL, + period_start DATE NULL, + period_end DATE NULL, -- Simulated key snapshot (the CRM allocation_key; no FK — separate DB) id_key INTEGER NOT NULL, @@ -59,8 +75,25 @@ CREATE TABLE IF NOT EXISTS simulation ( -- Object key of the per-timestep time-series JSON written on success. result_storage_key VARCHAR(512) NULL, + -- Non-blocking findings from the CRM pre-flight (participants whose meter + -- has gaps, which are zero-filled). Persisted so the warning outlives the + -- preview screen. + data_warnings JSONB NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, - updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP + updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT ck_simulation_source CHECK ( + (source = 1 + AND file_storage_key IS NOT NULL + AND file_name IS NOT NULL + AND injection_name IS NOT NULL) + OR (source = 2 + AND id_sharing_operation IS NOT NULL + AND period_start IS NOT NULL + AND period_end IS NOT NULL + AND period_start <= period_end) + ) ); CREATE INDEX IF NOT EXISTS idx_simulation_id_community ON simulation (id_community); diff --git a/shared/const.py b/shared/const.py index 7680699..ac28d2a 100644 --- a/shared/const.py +++ b/shared/const.py @@ -15,3 +15,15 @@ class FeatureName(StrEnum): # subscribes to for simulation runs (declared in core/queue/streams.json). SIMULATION_STREAM = "SIMULATIONS" SIMULATION_SUBJECT = "optimce.simulation.run" + + +class DataSource(IntEnum): + """Where a simulation's input timeseries comes from. + + FILE is the historical path (a CSV/XLSX uploaded by the manager). CRM + reads the same numbers straight out of the core database's + ``meter_consumption`` for one sharing operation over a date range. + """ + + FILE = 1 + CRM = 2 diff --git a/shared/crm_meter_repository.py b/shared/crm_meter_repository.py new file mode 100644 index 0000000..0dcbaa9 --- /dev/null +++ b/shared/crm_meter_repository.py @@ -0,0 +1,279 @@ +"""Read-only access to the CRM core's metering tables. + +This is the single place coupled to the CRM ``meter_consumption`` layout, in +the same spirit as ``billing/ports/crm_core_sqlalchemy.py``. Every statement is +SELECT-only and runs on a CRM ``AsyncSession``. + +Two deliberate constraints on this module: + +* **No pandas.** The API imports it for the pre-flight/preview and + ``requirements/api.txt`` carries no numpy/pandas. The pivot into a DataFrame + lives in ``shared/crm_timeseries.py``, which only the worker imports. +* **No fastapi.** The worker imports it too, and ``Dockerfile.worker`` + installs no HTTP stack. + +Community scope is passed **explicitly** on every call rather than read from a +ContextVar: the worker has no request context, and +``core.database.with_community.with_community_scope`` would silently degrade to +``WHERE false`` there. + +Period boundaries are interpreted in Belgian local time, so a month aligns to +local midnights and is DST-safe. Windows are half-open: +``timestamp >= start AND timestamp < end_exclusive``. +""" + +import datetime +from dataclasses import dataclass +from datetime import date +from zoneinfo import ZoneInfo + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession + +# Metering timestamps are absolute instants (timestamptz); a period expressed +# as local dates is bounded at Belgian local midnights. Mirrors +# crm-backend's CONSUMPTION_TIMEZONE. +_METERING_TZ = ZoneInfo("Europe/Brussels") + + +def period_bounds( + period_start: date, period_end: date +) -> tuple[datetime.datetime, datetime.datetime]: + """Half-open instant bounds ``[start, end_exclusive)`` for an inclusive date range.""" + start = datetime.datetime.combine(period_start, datetime.time.min, tzinfo=_METERING_TZ) + end_exclusive = datetime.datetime.combine( + period_end + datetime.timedelta(days=1), datetime.time.min, tzinfo=_METERING_TZ + ) + return start, end_exclusive + + +@dataclass(frozen=True) +class EanCoverage: + """Per-EAN aggregate over the requested period.""" + + ean: str + row_count: int + distinct_ts: int + consumption_kwh: float + injection_kwh: float + + @property + def has_duplicate_rows(self) -> bool: + """True when the same (ean, timestamp) appears more than once. + + ``meter_consumption`` has no unique constraint on (ean, timestamp), so a + workbook imported twice lands twice and silently doubles this meter's + energy. Callers treat this as fatal, not as a warning. + """ + return self.row_count != self.distinct_ts + + @property + def is_consumer(self) -> bool: + """A meter that actually drew energy over the period. + + A pure injection point would otherwise become an all-zero participant + holding a 0% share of the generated key. + """ + return self.consumption_kwh > 0 + + +@dataclass(frozen=True) +class CrmDataSummary: + """Everything the pre-flight and the preview screen need, in two queries.""" + + eans: list[EanCoverage] + # Distinct timestamps across the whole operation — the length of the grid + # every meter is reindexed onto. + grid_size: int + first_timestamp: datetime.datetime | None + last_timestamp: datetime.datetime | None + + @property + def total_rows(self) -> int: + return sum(e.row_count for e in self.eans) + + @property + def total_consumption_kwh(self) -> float: + return sum(e.consumption_kwh for e in self.eans) + + @property + def total_injection_kwh(self) -> float: + return sum(e.injection_kwh for e in self.eans) + + @property + def all_eans(self) -> list[str]: + """Every meter with a reading in the period, consumers and injectors alike.""" + return [e.ean for e in self.eans] + + @property + def consumer_eans(self) -> list[str]: + return [e.ean for e in self.eans if e.is_consumer] + + @property + def duplicate_eans(self) -> list[str]: + return [e.ean for e in self.eans if e.has_duplicate_rows] + + @property + def incomplete(self) -> list[EanCoverage]: + """Meters missing at least one timestamp of the common grid. + + These are allowed through (the gaps are zero-filled) but are reported to + the manager both before the run and on the finished run. + """ + return [e for e in self.eans if e.distinct_ts < self.grid_size] + + +@dataclass(frozen=True) +class ConsumptionRow: + """One metering reading, as consumed by the pivot.""" + + timestamp: datetime.datetime + ean: str + gross: float + inj_gross: float + + +class CrmMeterRepository: + """SELECT-only reader over the CRM metering tables.""" + + def __init__(self, session: AsyncSession) -> None: + self._session = session + + async def sharing_operation_exists( + self, *, id_community: int, id_sharing_operation: int + ) -> bool: + """Tenant gate: does this operation belong to the caller's community? + + Checked explicitly so a cross-tenant id yields a clean "not found" + rather than an indistinguishable "no data in this period". + """ + result = await self._session.execute( + text( + """ + SELECT EXISTS ( + SELECT 1 FROM sharing_operation so + WHERE so.id = :op AND so.id_community = :cid + ) AS present + """ + ), + {"cid": id_community, "op": id_sharing_operation}, + ) + return bool(result.scalar()) + + async def summarize( + self, + *, + id_community: int, + id_sharing_operation: int, + period_start: date, + period_end: date, + ) -> CrmDataSummary: + """Aggregate the period without transferring the readings themselves. + + Deliberately cheap: ``RequestLimitsMiddleware.TIMEOUT_SECONDS`` caps + every API request at 30 s, and this runs on the request path. + """ + start, end_exclusive = period_bounds(period_start, period_end) + params = { + "cid": id_community, + "op": id_sharing_operation, + "start": start, + "end_excl": end_exclusive, + } + + per_ean = await self._session.execute( + text( + """ + SELECT mc.ean AS ean, + COUNT(*) AS row_count, + COUNT(DISTINCT mc.timestamp) AS distinct_ts, + COALESCE(SUM(mc.gross), 0) AS consumption_kwh, + COALESCE(SUM(mc.inj_gross), 0) AS injection_kwh + FROM meter_consumption mc + WHERE mc.id_community = :cid + AND mc.id_sharing_operation = :op + AND mc.timestamp >= :start AND mc.timestamp < :end_excl + GROUP BY mc.ean + ORDER BY mc.ean + """ + ), + params, + ) + eans = [ + EanCoverage( + ean=row["ean"], + row_count=int(row["row_count"]), + distinct_ts=int(row["distinct_ts"]), + consumption_kwh=float(row["consumption_kwh"]), + injection_kwh=float(row["injection_kwh"]), + ) + for row in per_ean.mappings() + ] + + grid = await self._session.execute( + text( + """ + SELECT COUNT(DISTINCT mc.timestamp) AS grid_size, + MIN(mc.timestamp) AS first_ts, + MAX(mc.timestamp) AS last_ts + FROM meter_consumption mc + WHERE mc.id_community = :cid + AND mc.id_sharing_operation = :op + AND mc.timestamp >= :start AND mc.timestamp < :end_excl + """ + ), + params, + ) + grid_row = grid.mappings().one() + + return CrmDataSummary( + eans=eans, + grid_size=int(grid_row["grid_size"] or 0), + first_timestamp=grid_row["first_ts"], + last_timestamp=grid_row["last_ts"], + ) + + async def fetch_rows( + self, + *, + id_community: int, + id_sharing_operation: int, + period_start: date, + period_end: date, + ) -> list[ConsumptionRow]: + """The readings themselves, ordered for a stable pivot. + + Only ever called from the worker: a year of quarter-hours across a few + dozen meters is well past what belongs on a 30-second request path. + """ + start, end_exclusive = period_bounds(period_start, period_end) + result = await self._session.execute( + text( + """ + SELECT mc.timestamp AS ts, + mc.ean AS ean, + COALESCE(mc.gross, 0) AS gross, + COALESCE(mc.inj_gross, 0) AS inj_gross + FROM meter_consumption mc + WHERE mc.id_community = :cid + AND mc.id_sharing_operation = :op + AND mc.timestamp >= :start AND mc.timestamp < :end_excl + ORDER BY mc.timestamp, mc.ean + """ + ), + { + "cid": id_community, + "op": id_sharing_operation, + "start": start, + "end_excl": end_exclusive, + }, + ) + return [ + ConsumptionRow( + timestamp=row["ts"], + ean=row["ean"], + gross=float(row["gross"]), + inj_gross=float(row["inj_gross"]), + ) + for row in result.mappings() + ] diff --git a/shared/crm_preflight.py b/shared/crm_preflight.py new file mode 100644 index 0000000..1420bc4 --- /dev/null +++ b/shared/crm_preflight.py @@ -0,0 +1,173 @@ +"""Decide whether a CRM-sourced period can be simulated against a given key. + +One definition of "blocking", used in three places: the preview endpoint (so the +manager sees the problem before launching), ``POST /from-crm`` (so a stale +preview cannot slip a bad run through), and the worker (which re-reads the data +at execution time and must reach the same verdict). + +The simulation-specific rule lives here: **the key's participant names must be +meter EANs present in the period.** ``allocation_key.consumer.name`` is free +text with no foreign key to ``meter``; the convention that it holds the EAN is +enforced platform-wide only by comparison, and crm-backend compares it with +``TRIM(cons.name) IN (:eans)``. This module matches that exactly, so a key that +resolves in the CRM's own views also resolves here. + +Framework-free and pandas-free on purpose — the API, which has neither pandas +nor a request-scoped community, imports this too. +""" + +from collections.abc import Sequence +from dataclasses import dataclass, field +from typing import Any + +from core.errors.errors import Error +from shared.crm_meter_repository import CrmDataSummary +from shared.custom_errors import errors + +# A year of quarter-hours across ~60 meters is ~2.1 M rows, which the worker +# pivots comfortably. Well past that we would rather refuse than risk the +# worker being OOM-killed mid-run, which reads to the manager as a silent hang. +MAX_READING_ROWS = 5_000_000 + + +def normalize_participant(name: str) -> str: + """Canonical form for comparing a key participant against a meter EAN. + + ``TRIM`` only — EANs are digits, so there is no case to fold. Mirrors + ``crm-backend/src/modules/me/infra/me.repository.ts::getKeyConsumersForEans``. + Note the file-upload path does *not* trim, so a key with a trailing space + works here and fails there; this is the more forgiving of the two. + """ + return name.strip() + + +@dataclass(frozen=True) +class Blocker: + """A reason this period cannot be used, with the detail the manager needs.""" + + error: Error + detail: str + + +@dataclass(frozen=True) +class Preflight: + summary: CrmDataSummary + # The key's participants, trimmed, in key order. The pivot's column order + # follows this so matrix rows line up with the key. + participants: list[str] + matched: list[str] + unmatched: list[str] + blockers: list[Blocker] + warnings: dict[str, Any] | None = field(default=None) + + @property + def ok(self) -> bool: + return not self.blockers + + +def evaluate( + summary: CrmDataSummary, + key_consumer_names: Sequence[str], +) -> Preflight: + """Classify a period's metering data against the key being simulated. + + Gaps are deliberately **not** blocking: missing quarters are zero-filled and + reported. Duplicates are, because there is no unique constraint on + ``meter_consumption(ean, timestamp)`` and a repeated import would inflate a + participant's volume with no visible symptom. + """ + participants = [normalize_participant(n) for n in key_consumer_names] + + if not summary.eans: + # Nothing else can be said about an empty period; return early so the + # manager gets one clear message rather than four derived ones. + return Preflight( + summary=summary, + participants=participants, + matched=[], + unmatched=participants, + blockers=[ + Blocker( + error=errors.simulation.CRM_NO_DATA, + detail="no readings for this sharing operation over this period", + ) + ], + warnings=None, + ) + + # Matching is against every meter with a reading, not just the consuming + # ones: a key may legitimately include a participant that only injected. + available = set(summary.all_eans) + matched = [p for p in participants if p in available] + unmatched = [p for p in participants if p not in available] + + blockers: list[Blocker] = [] + + if summary.total_rows > MAX_READING_ROWS: + blockers.append( + Blocker( + error=errors.simulation.CRM_RANGE_TOO_LARGE, + detail=( + f"{summary.total_rows} readings exceed the {MAX_READING_ROWS} limit; " + "choose a shorter period" + ), + ) + ) + + duplicates = summary.duplicate_eans + if duplicates: + blockers.append( + Blocker( + error=errors.simulation.CRM_DUPLICATE_READINGS, + detail=( + "the same timestamp appears more than once for meter(s) " + f"{', '.join(duplicates)} — the data was most likely imported twice" + ), + ) + ) + + if unmatched: + blockers.append( + Blocker( + error=errors.simulation.KEY_CONSUMERS_NOT_MATCHED, + detail=( + "the key names participant(s) with no matching meter in this period: " + f"{', '.join(unmatched)}" + ), + ) + ) + + if summary.total_injection_kwh <= 0: + blockers.append( + Blocker( + error=errors.simulation.CRM_NO_INJECTION, + detail="no energy was injected over this period — there is nothing to share", + ) + ) + + # Only the participants actually being simulated are worth warning about; + # a gap in a meter the key ignores changes nothing. + participant_set = set(matched) + incomplete = [e for e in summary.incomplete if e.ean in participant_set] + warnings: dict[str, Any] | None = None + if incomplete: + warnings = { + "incomplete_meters": [ + { + "ean": e.ean, + "readings": e.distinct_ts, + "expected": summary.grid_size, + "missing": summary.grid_size - e.distinct_ts, + } + for e in incomplete + ] + } + + return Preflight( + summary=summary, + participants=participants, + matched=matched, + unmatched=unmatched, + blockers=blockers, + warnings=warnings, + ) diff --git a/shared/crm_timeseries.py b/shared/crm_timeseries.py new file mode 100644 index 0000000..b96e212 --- /dev/null +++ b/shared/crm_timeseries.py @@ -0,0 +1,89 @@ +"""Pivot CRM metering rows into the wide frame the algorithms already expect. + +The whole point of this module is that it produces a DataFrame **shaped exactly +like a parsed upload** — one column per participant plus a single injection +column — so the existing converter in ``shared/data_loading`` consumes it +unchanged. Nothing downstream needs to know the data came from the database +rather than from a file. + +Imports pandas, and is therefore **worker-only**: ``requirements/api.txt`` +carries no pandas, exactly as for ``shared/data_loading.py``. +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import pandas as pd + +from shared.crm_meter_repository import ConsumptionRow + +# Column name standing in for the file-based ``injection_name``. EANs are digit +# strings, so a dunder label cannot collide with a participant column. +INJECTION_COLUMN = "__injection__" + + +class CrmPivotError(ValueError): + """Raised when the fetched rows cannot form a rectangular timeseries.""" + + +def build_dataframe( + rows: Sequence[ConsumptionRow], + consumer_eans: Sequence[str], +) -> pd.DataFrame: + """Build ``[timestamp x consumer EAN] + __injection__`` from raw readings. + + ``consumer_eans`` selects (and orders) the participant columns — meters that + actually drew energy over the period. Injection is summed over **every** EAN + in ``rows``, including pure production sites that are not participants, and + is taken from the same rows as the consumption so the two series can never + disagree in length. + + Timestamps present for some meters but not others are zero-filled; the + caller has already reported those gaps to the manager. + """ + if not rows: + raise CrmPivotError("no readings to pivot") + if not consumer_eans: + raise CrmPivotError("no consumer EANs requested") + if INJECTION_COLUMN in consumer_eans: + raise CrmPivotError(f"{INJECTION_COLUMN!r} is reserved and cannot be a participant") + + frame = pd.DataFrame( + { + "timestamp": [r.timestamp for r in rows], + "ean": [r.ean for r in rows], + "gross": [r.gross for r in rows], + "inj_gross": [r.inj_gross for r in rows], + } + ) + + # Duplicate (ean, timestamp) pairs are refused upstream (they would double a + # participant's energy). Pivoting defensively rather than aggregating means a + # regression surfaces as a loud failure instead of silently inflated volumes. + try: + consumption = frame.pivot(index="timestamp", columns="ean", values="gross") + except ValueError as exc: + raise CrmPivotError(f"duplicate (ean, timestamp) readings: {exc}") from exc + + # A requested participant with no reading at all in the period would be + # silently reindexed into an all-zero column — for simulation that means a + # key participant whose EAN does not exist quietly gets a 0 kWh profile + # instead of an error. Refuse instead; the pre-flight normally catches this + # first and reports it far more helpfully, so reaching here is a backstop. + missing = [ean for ean in consumer_eans if ean not in consumption.columns] + if missing: + raise CrmPivotError(f"no readings for requested participant(s): {missing}") + + # Select the requested participants in order, dropping injection-only meters. + consumption = consumption.reindex(columns=list(consumer_eans)) + # Remaining NaNs are per-timestamp gaps in an otherwise present meter. + consumption = consumption.fillna(0.0).astype(float) + + # One shared production profile, summed across every meter at each instant. + injection = frame.groupby("timestamp")["inj_gross"].sum() + consumption[INJECTION_COLUMN] = injection.reindex(consumption.index).fillna(0.0).astype(float) + + # Chronological order is the only ordering guarantee the algorithms have: + # row t of every column must be contemporaneous. + return consumption.sort_index().reset_index(drop=True) diff --git a/shared/custom_errors.py b/shared/custom_errors.py index fa1e398..430a68c 100644 --- a/shared/custom_errors.py +++ b/shared/custom_errors.py @@ -34,6 +34,27 @@ class _SimulationErrors: # middleware can't pre-screen (chunked / no Content-Length). Maps to 413. FILE_TOO_LARGE = Error(code=2110, key="ERRORS.SIMULATION.FILE_TOO_LARGE") + # --- CRM-sourced input (source = DataSource.CRM) ----------------------- + # Raised by the pre-flight, which runs on the preview endpoint, again on + # POST /from-crm, and a third time in the worker against the data as it is + # at execution time. + INVALID_PERIOD = Error(code=2111, key="ERRORS.SIMULATION.INVALID_PERIOD") + SHARING_OPERATION_NOT_FOUND = Error( + code=2112, key="ERRORS.SIMULATION.SHARING_OPERATION_NOT_FOUND" + ) + CRM_NO_DATA = Error(code=2113, key="ERRORS.SIMULATION.CRM_NO_DATA") + # No unique constraint on meter_consumption(ean, timestamp): a repeated + # import silently doubles a participant's energy, so this is fatal rather + # than a warning. + CRM_DUPLICATE_READINGS = Error(code=2114, key="ERRORS.SIMULATION.CRM_DUPLICATE_READINGS") + CRM_NO_INJECTION = Error(code=2115, key="ERRORS.SIMULATION.CRM_NO_INJECTION") + CRM_RANGE_TOO_LARGE = Error(code=2116, key="ERRORS.SIMULATION.CRM_RANGE_TOO_LARGE") + # The simulated key names participants that no meter in the period matches. + # allocation_key.consumer.name has no FK to meter.ean — the platform-wide + # convention is that the name IS the EAN, compared after TRIM. + KEY_CONSUMERS_NOT_MATCHED = Error(code=2117, key="ERRORS.SIMULATION.KEY_CONSUMERS_NOT_MATCHED") + GET_CRM_PREVIEW = Error(code=2118, key="ERRORS.SIMULATION.GET_CRM_PREVIEW") + class _Errors: auth = _AuthErrors() diff --git a/shared/models/local_models.py b/shared/models/local_models.py index f7c7384..be3a3d2 100644 --- a/shared/models/local_models.py +++ b/shared/models/local_models.py @@ -1,10 +1,12 @@ import datetime +from typing import Any -from sqlalchemy import TIMESTAMP, Float, ForeignKey, Integer, String, Text +from sqlalchemy import TIMESTAMP, Date, Float, ForeignKey, Integer, SmallInteger, String, Text +from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, mapped_column, relationship from core.database.database import LocalBase -from shared.const import SimulationStatus +from shared.const import DataSource, SimulationStatus class SimulationModel(LocalBase): @@ -27,14 +29,28 @@ class SimulationModel(LocalBase): id_community: Mapped[int] = mapped_column(Integer, nullable=False) # --- Source data --- - # file_storage_key is the object key inside STORAGE_BUCKET (MinIO). The - # service uploads the user-supplied file at creation time; the worker - # deletes it once the row reaches SUCCESS or FAILED. - file_storage_key: Mapped[str] = mapped_column(String(512), nullable=False) - file_name: Mapped[str] = mapped_column(String(255), nullable=False) + # Which of the two column groups below is populated. The DB-level + # ck_simulation_source CHECK enforces the pairing; these columns cannot be + # NOT NULL any more because a CRM-sourced run has no file. + source: Mapped[DataSource] = mapped_column( + SmallInteger, nullable=False, default=DataSource.FILE + ) + + # FILE only. file_storage_key is the object key inside STORAGE_BUCKET + # (MinIO). The service uploads the user-supplied file at creation time; the + # worker deletes it once the row reaches SUCCESS or FAILED. + file_storage_key: Mapped[str | None] = mapped_column(String(512), nullable=True) + file_name: Mapped[str | None] = mapped_column(String(255), nullable=True) # Name of the column inside the uploaded file that holds the shared # production profile (the "injection"). - injection_name: Mapped[str] = mapped_column(String(255), nullable=False) + injection_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + + # CRM only. The sharing operation and the inclusive Brussels-local date + # range read out of meter_consumption. Plain columns, never FKs -- the CRM + # is a separate database. + id_sharing_operation: Mapped[int | None] = mapped_column(Integer, nullable=True) + period_start: Mapped[datetime.date | None] = mapped_column(Date, nullable=True) + period_end: Mapped[datetime.date | None] = mapped_column(Date, nullable=True) # --- Simulated key snapshot --- # The CRM ``allocation_key`` being stress-tested. Snapshotted by id (no FK: @@ -56,6 +72,12 @@ class SimulationModel(LocalBase): # for charting (GET /simulation/{id}/timeseries). result_storage_key: Mapped[str | None] = mapped_column(String(512), nullable=True) + # Non-blocking findings from the CRM pre-flight -- currently the + # participants whose meter had gaps in the period and were zero-filled. + # Persisted rather than only shown before launch, so the manager can still + # see it on the finished run. + data_warnings: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True) + key_result: Mapped["SimulationKeyResultModel | None"] = relationship( "SimulationKeyResultModel", lazy="select", diff --git a/tests/api/simulation/test_crm_source_routes.py b/tests/api/simulation/test_crm_source_routes.py new file mode 100644 index 0000000..86298e5 --- /dev/null +++ b/tests/api/simulation/test_crm_source_routes.py @@ -0,0 +1,391 @@ +"""Integration tests for the CRM-sourced simulation routes. + +Full ASGI stack against a real Postgres, same conventions as test_routes.py. +Nothing here touches MinIO — that is the point of the CRM source. + +The behaviour these tests are really about is the participant match: the +simulated key names participants, the CRM holds meters, and the two must line +up before a run is allowed. +""" + +import datetime +from unittest.mock import AsyncMock, MagicMock, patch + +from sqlalchemy import select + +from core.database.models import Community +from shared.const import DataSource, SimulationStatus +from shared.custom_errors import errors +from shared.models.local_models import SimulationModel +from tests.factories.meter_factory import ( + create_meter, + create_readings, + create_sharing_operation, +) +from tests.factories.simulation_factory import create_crm_key +from tests.factories.subscription_factory import create_community, create_subscription + +_EAN_A = "541448000000000001" +_EAN_PV = "541448000000000002" +_PERIOD = {"period_start": "2025-02-01", "period_end": "2025-02-28"} + + +def _admin_headers(community: Community) -> dict[str, str]: + return { + "x-user-id": "test|admin", + "x-community-id": community.auth_community_id, + "x-user-role": "ADMIN", + } + + +async def _community_with_subscription(db_session) -> Community: + community = await create_community(db_session) + await create_subscription(db_session, id_community=community.id, is_active=True) + return community + + +async def _operation_with_data( + db_session, + community: Community, + *, + consumer_skip: set[int] | None = None, + duplicate: bool = False, + injection: float = 40.0, +) -> int: + op = await create_sharing_operation(db_session, id_community=community.id) + await create_meter(db_session, ean=_EAN_A, id_community=community.id) + await create_meter(db_session, ean=_EAN_PV, id_community=community.id) + + await create_readings( + db_session, + ean=_EAN_A, + id_community=community.id, + id_sharing_operation=op, + gross=10.0, + inj_gross=0.0, + skip=consumer_skip, + ) + if duplicate: + await create_readings( + db_session, + ean=_EAN_A, + id_community=community.id, + id_sharing_operation=op, + gross=10.0, + inj_gross=0.0, + skip=consumer_skip, + ) + await create_readings( + db_session, + ean=_EAN_PV, + id_community=community.id, + id_sharing_operation=op, + gross=0.0, + inj_gross=injection, + ) + return op + + +def _body(id_key: int, op: int, **overrides) -> dict: + body = { + "name": "february sim", + "id_key": id_key, + "id_sharing_operation": op, + **_PERIOD, + } + body.update(overrides) + return body + + +# --------------------------------------------------------------------------- +# 1. GET /crm-data-preview — the participant match +# --------------------------------------------------------------------------- + + +async def test_preview_reports_matched_participants(client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community) + key = await create_crm_key(db_session, id_community=community.id, consumer_names=[_EAN_A]) + + response = await client.get( + "/crm-data-preview", + params={"id_key": key.id, "id_sharing_operation": op, **_PERIOD}, + headers=_admin_headers(community), + ) + + assert response.status_code == 200 + data = response.json()["data"] + assert data["can_simulate"] is True + assert data["matched_participants"] == [_EAN_A] + assert data["unmatched_participants"] == [] + assert data["reading_count"] == 8 + assert data["blockers"] == [] + + +async def test_preview_is_reachable_and_not_swallowed_by_the_id_route(client, db_session): + # `GET /{id}` is declared after this route; if the order regressed, the path + # would be parsed as an integer id and 422. + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community) + key = await create_crm_key(db_session, id_community=community.id, consumer_names=[_EAN_A]) + + response = await client.get( + "/crm-data-preview", + params={"id_key": key.id, "id_sharing_operation": op, **_PERIOD}, + headers=_admin_headers(community), + ) + + assert response.status_code == 200 + + +async def test_preview_flags_participants_with_no_matching_meter(client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community) + # The default factory names participants C0/C1 — the shape of a key built + # from a spreadsheet whose columns were not EANs. + key = await create_crm_key(db_session, id_community=community.id) + + response = await client.get( + "/crm-data-preview", + params={"id_key": key.id, "id_sharing_operation": op, **_PERIOD}, + headers=_admin_headers(community), + ) + + data = response.json()["data"] + assert data["can_simulate"] is False + assert data["unmatched_participants"] == ["C0", "C1"] + codes = {b["error_code"] for b in data["blockers"]} + assert errors.simulation.KEY_CONSUMERS_NOT_MATCHED.code in codes + + +async def test_preview_matches_participant_names_after_trimming(client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community) + key = await create_crm_key( + db_session, id_community=community.id, consumer_names=[f" {_EAN_A} "] + ) + + response = await client.get( + "/crm-data-preview", + params={"id_key": key.id, "id_sharing_operation": op, **_PERIOD}, + headers=_admin_headers(community), + ) + + data = response.json()["data"] + assert data["can_simulate"] is True + assert data["matched_participants"] == [_EAN_A] + + +async def test_preview_reports_gaps_as_warnings_without_blocking(client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community, consumer_skip={1, 2}) + key = await create_crm_key(db_session, id_community=community.id, consumer_names=[_EAN_A]) + + response = await client.get( + "/crm-data-preview", + params={"id_key": key.id, "id_sharing_operation": op, **_PERIOD}, + headers=_admin_headers(community), + ) + + data = response.json()["data"] + assert data["can_simulate"] is True, "a gap must warn, not block" + assert data["incomplete_meters"] == [ + {"ean": _EAN_A, "readings": 2, "expected": 4, "missing": 2} + ] + + +async def test_preview_blocks_on_duplicate_readings(client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community, duplicate=True) + key = await create_crm_key(db_session, id_community=community.id, consumer_names=[_EAN_A]) + + response = await client.get( + "/crm-data-preview", + params={"id_key": key.id, "id_sharing_operation": op, **_PERIOD}, + headers=_admin_headers(community), + ) + + data = response.json()["data"] + assert data["can_simulate"] is False + codes = {b["error_code"] for b in data["blockers"]} + assert errors.simulation.CRM_DUPLICATE_READINGS.code in codes + + +async def test_preview_rejects_inverted_period(client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community) + key = await create_crm_key(db_session, id_community=community.id, consumer_names=[_EAN_A]) + + response = await client.get( + "/crm-data-preview", + params={ + "id_key": key.id, + "id_sharing_operation": op, + "period_start": "2025-02-28", + "period_end": "2025-02-01", + }, + headers=_admin_headers(community), + ) + + assert response.status_code == 422 + assert response.json()["error_code"] == errors.simulation.INVALID_PERIOD.code + + +async def test_preview_cannot_reach_another_communitys_operation(client, db_session): + owner = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, owner) + intruder = await _community_with_subscription(db_session) + key = await create_crm_key(db_session, id_community=intruder.id, consumer_names=[_EAN_A]) + + response = await client.get( + "/crm-data-preview", + params={"id_key": key.id, "id_sharing_operation": op, **_PERIOD}, + headers=_admin_headers(intruder), + ) + + assert response.status_code == 404 + assert response.json()["error_code"] == errors.simulation.SHARING_OPERATION_NOT_FOUND.code + + +async def test_preview_cannot_reach_another_communitys_key(client, db_session): + owner = await _community_with_subscription(db_session) + key = await create_crm_key(db_session, id_community=owner.id, consumer_names=[_EAN_A]) + intruder = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, intruder) + + response = await client.get( + "/crm-data-preview", + params={"id_key": key.id, "id_sharing_operation": op, **_PERIOD}, + headers=_admin_headers(intruder), + ) + + assert response.status_code == 404 + assert response.json()["error_code"] == errors.simulation.KEY_NOT_FOUND.code + + +# --------------------------------------------------------------------------- +# 2. POST /from-crm +# --------------------------------------------------------------------------- + + +@patch("api.simulation.service.get_jetstream", MagicMock()) +@patch("api.simulation.service.send_event", new_callable=AsyncMock) +async def test_start_from_crm_persists_a_crm_sourced_row(send_event, client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community) + key = await create_crm_key(db_session, id_community=community.id, consumer_names=[_EAN_A]) + + response = await client.post( + "/from-crm", json=_body(key.id, op), headers=_admin_headers(community) + ) + + assert response.status_code == 200, response.text + assert response.json()["data"]["status"] == SimulationStatus.PENDING + + row = ( + await db_session.execute( + select(SimulationModel).where(SimulationModel.id == response.json()["data"]["id"]) + ) + ).scalar_one() + assert row.source == DataSource.CRM + assert row.id_sharing_operation == op + assert row.period_start == datetime.date(2025, 2, 1) + assert row.period_end == datetime.date(2025, 2, 28) + assert row.id_key == key.id + # No file was uploaded, so the file columns stay empty — which the + # ck_simulation_source CHECK only permits for source = CRM. + assert row.file_storage_key is None + assert row.injection_name is None + send_event.assert_awaited_once() + + +@patch("api.simulation.service.get_jetstream", MagicMock()) +@patch("api.simulation.service.send_event", new_callable=AsyncMock) +async def test_start_from_crm_persists_gap_warnings(send_event, client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community, consumer_skip={1}) + key = await create_crm_key(db_session, id_community=community.id, consumer_names=[_EAN_A]) + + response = await client.post( + "/from-crm", json=_body(key.id, op), headers=_admin_headers(community) + ) + + assert response.status_code == 200, response.text + row = ( + await db_session.execute( + select(SimulationModel).where(SimulationModel.id == response.json()["data"]["id"]) + ) + ).scalar_one() + assert row.data_warnings == { + "incomplete_meters": [{"ean": _EAN_A, "readings": 3, "expected": 4, "missing": 1}] + } + + +@patch("api.simulation.service.get_jetstream", MagicMock()) +@patch("api.simulation.service.send_event", new_callable=AsyncMock) +async def test_start_from_crm_refuses_unmatched_participants(send_event, client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community) + key = await create_crm_key(db_session, id_community=community.id) # C0 / C1 + + response = await client.post( + "/from-crm", json=_body(key.id, op), headers=_admin_headers(community) + ) + + assert response.status_code == 422 + assert response.json()["error_code"] == errors.simulation.KEY_CONSUMERS_NOT_MATCHED.code + send_event.assert_not_awaited() + + +@patch("api.simulation.service.get_jetstream", MagicMock()) +@patch("api.simulation.service.send_event", new_callable=AsyncMock) +async def test_start_from_crm_refuses_duplicate_readings(send_event, client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community, duplicate=True) + key = await create_crm_key(db_session, id_community=community.id, consumer_names=[_EAN_A]) + + response = await client.post( + "/from-crm", json=_body(key.id, op), headers=_admin_headers(community) + ) + + assert response.status_code == 422 + assert response.json()["error_code"] == errors.simulation.CRM_DUPLICATE_READINGS.code + send_event.assert_not_awaited() + + +@patch("api.simulation.service.get_jetstream", MagicMock()) +@patch("api.simulation.service.send_event", new_callable=AsyncMock) +async def test_start_from_crm_cannot_use_another_communitys_operation( + send_event, client, db_session +): + owner = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, owner) + intruder = await _community_with_subscription(db_session) + key = await create_crm_key(db_session, id_community=intruder.id, consumer_names=[_EAN_A]) + + response = await client.post( + "/from-crm", json=_body(key.id, op), headers=_admin_headers(intruder) + ) + + assert response.status_code == 404 + assert response.json()["error_code"] == errors.simulation.SHARING_OPERATION_NOT_FOUND.code + send_event.assert_not_awaited() + + +async def test_start_from_crm_body_is_json_not_query_params(client, db_session): + # Guards the with_default_error / `from __future__ import annotations` trap: + # if the route module ever gains that import, the Pydantic body is demoted + # to query params and every well-formed request 422s with loc=[query, body]. + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community) + key = await create_crm_key(db_session, id_community=community.id, consumer_names=[_EAN_A]) + + with ( + patch("api.simulation.service.get_jetstream", MagicMock()), + patch("api.simulation.service.send_event", new_callable=AsyncMock), + ): + response = await client.post( + "/from-crm", json=_body(key.id, op), headers=_admin_headers(community) + ) + + assert response.status_code != 422, response.text diff --git a/tests/core/test_realtime.py b/tests/core/test_realtime.py new file mode 100644 index 0000000..0676146 --- /dev/null +++ b/tests/core/test_realtime.py @@ -0,0 +1,188 @@ +"""Unit tests for the shared ``core/realtime`` package. + +This file is NOT part of the byte-identity contract (only the four +``core/realtime/*.py`` modules are — see ``scripts/check-realtime-parity.sh``), +but the behaviour it pins is shared by all five producers. +""" + +import asyncio +import json + +import pytest + +from core.realtime import ( + MAX_ENVELOPE_BYTES, + CommunityAudience, + Tier, + UserAudience, + UsersAudience, + build_envelope, + bus, + community_channel, + user_channel, +) + + +class FakeRedis: + """Records publishes. ``fail``/``hang`` drive the two failure paths.""" + + def __init__(self, *, fail: bool = False, hang: bool = False) -> None: + self.published: list[tuple[str, str]] = [] + self.fail = fail + self.hang = hang + + async def publish(self, channel: str, body: bytes | str) -> int: + if self.fail: + raise ConnectionError("redis down") + if self.hang: + await asyncio.sleep(30) + self.published.append((channel, body if isinstance(body, str) else body.decode())) + return 1 + + +@pytest.fixture +def fake_bus(monkeypatch): + """Point emit() at a fake client with the feature switched on.""" + + def _make(**kwargs): + client = FakeRedis(**kwargs) + monkeypatch.setattr(bus.settings, "REALTIME_ENABLED", True, raising=False) + monkeypatch.setattr(bus.settings, "REALTIME_REDIS_URL", "redis://fake/1", raising=False) + monkeypatch.setattr(bus, "_get_client", lambda: client) + return client + + return _make + + +# ---- channels ------------------------------------------------------------ + + +def test_channel_strings_match_the_typescript_contract(): + # These literals are the contract with crm-backend's realtime.channels.ts. + # Changing one side without the other stops delivery silently: there is no + # handshake, the events just go to a channel nobody listens on. + assert user_channel(4821) == "notify:v1:u:4821" + assert community_channel(12, Tier.MEMBER) == "notify:v1:c:12:MEMBER" + assert community_channel(12, Tier.MANAGER) == "notify:v1:c:12:MANAGER" + + +def test_audiences_expand_to_the_right_channels(): + assert UserAudience(user_id=7).channels() == ("notify:v1:u:7",) + # Duplicates collapse: publishing twice to one user would double-toast. + assert UsersAudience(user_ids=[7, 8, 7]).channels() == ("notify:v1:u:7", "notify:v1:u:8") + assert CommunityAudience(community_id=3, tier=Tier.MANAGER).channels() == ( + "notify:v1:c:3:MANAGER", + ) + + +def test_tier_has_no_default(): + # Required argument by design: a producer that publishes a per-user thing + # onto a community tier is a cross-tenant leak, and omission must not be a + # way to get there. + with pytest.raises(TypeError): + CommunityAudience(community_id=3) + + +# ---- envelope ------------------------------------------------------------ + + +def test_build_envelope_stamps_the_wire_fields(): + env = build_envelope( + topic="simulation.finished", + resource=("simulation", 418), + hint={"status": "success"}, + scope_community_id=12, + ) + assert env is not None + assert env["v"] == 1 + assert len(env["id"]) == 16 + assert env["ref"] == {"kind": "simulation", "id": "418"} + assert env["scope"] == {"community_id": 12} + assert env["at"].endswith("Z") + + +@pytest.mark.parametrize( + "kwargs", + [ + {"topic": "not.a.topic", "resource": ("x", 1)}, + {"topic": "simulation.finished", "resource": ("", 1)}, + {"topic": "simulation.finished", "resource": ("x", 1), "hint": {"nested": {"a": 1}}}, + {"topic": "simulation.finished", "resource": ("x", 1), "hint": {"list": [1, 2]}}, + ], +) +def test_build_envelope_returns_none_instead_of_raising(kwargs): + # Every caller runs AFTER a commit. An exception here would travel up through + # a commit path, which is what the whole ordering design exists to prevent. + assert build_envelope(**kwargs) is None + + +def test_build_envelope_rejects_oversize_in_bytes(): + env = build_envelope( + topic="simulation.finished", + resource=("simulation", 1), + hint={"blob": "x" * MAX_ENVELOPE_BYTES}, + ) + assert env is None + + +# ---- emit ---------------------------------------------------------------- + + +async def test_emit_is_a_no_op_when_unconfigured(monkeypatch): + # The default everywhere realtime is not deployed — including every test + # suite, which must never open a socket. + created = [] + monkeypatch.setattr(bus.settings, "REALTIME_ENABLED", False, raising=False) + monkeypatch.setattr(bus.settings, "REALTIME_REDIS_URL", "", raising=False) + monkeypatch.setattr(bus, "_get_client", lambda: created.append(1)) + + await bus.emit( + topic="simulation.finished", audience=UserAudience(user_id=1), resource=("simulation", 1) + ) + assert created == [] + + +async def test_emit_publishes_one_message_per_channel(fake_bus): + client = fake_bus() + await bus.emit( + topic="simulation.finished", + audience=UsersAudience(user_ids=[1, 2]), + resource=("simulation", 418), + hint={"status": "success"}, + ) + assert [c for c, _ in client.published] == ["notify:v1:u:1", "notify:v1:u:2"] + assert json.loads(client.published[0][1])["hint"] == {"status": "success"} + + +async def test_emit_swallows_a_broker_failure(fake_bus): + # Fire-and-forget by contract: the caller has already committed, so a broker + # problem must cost freshness and nothing else. + fake_bus(fail=True) + await bus.emit( + topic="simulation.finished", audience=UserAudience(user_id=1), resource=("simulation", 1) + ) + + +async def test_emit_is_bounded_when_the_broker_hangs(fake_bus): + # A *connected but hung* broker is the case a socket timeout does not cover. + # Without the asyncio.timeout this would stall the worker indefinitely. + fake_bus(hang=True) + await asyncio.wait_for( + bus.emit( + topic="simulation.finished", + audience=UserAudience(user_id=1), + resource=("simulation", 1), + ), + timeout=5, + ) + + +async def test_emit_drops_a_malformed_envelope_without_publishing(fake_bus): + client = fake_bus() + await bus.emit( + topic="simulation.finished", + audience=UserAudience(user_id=1), + resource=("simulation", 1), + hint={"nested": {"a": 1}}, + ) + assert client.published == [] diff --git a/tests/factories/meter_factory.py b/tests/factories/meter_factory.py new file mode 100644 index 0000000..43be406 --- /dev/null +++ b/tests/factories/meter_factory.py @@ -0,0 +1,99 @@ +"""Factories for the CRM metering tables. + +These services map no ORM models for ``meter`` / ``meter_consumption`` — the +production code reads them with raw ``text()`` SQL through +``shared/crm_meter_repository.py`` — so the factories insert with raw SQL too. +That is deliberate: a test that went through an ORM model would stop exercising +the column names the real query actually depends on. + +Like the other factories here they flush and never commit; ``conftest`` wraps +each test in a transaction that is rolled back. +""" + +import datetime + +from sqlalchemy import text + +# Brussels local midnight, expressed as the UTC instant Postgres stores. February +# is CET (UTC+1), so 00:00 local is 23:00 UTC the previous day. Hard-coding the +# offset rather than importing ZoneInfo keeps the fixture obvious about which +# instant it means. +_CET = datetime.timezone(datetime.timedelta(hours=1)) + + +async def create_sharing_operation(session, *, id_community: int, name: str = "Test op") -> int: + result = await session.execute( + text( + """ + INSERT INTO sharing_operation (name, type, is_public, id_community) + VALUES (:name, 1, FALSE, :cid) + RETURNING id + """ + ), + {"name": name, "cid": id_community}, + ) + await session.flush() + return int(result.scalar_one()) + + +async def create_meter(session, *, ean: str, id_community: int) -> str: + await session.execute( + text( + """ + INSERT INTO meter (ean, meter_number, id_community) + VALUES (:ean, :num, :cid) + ON CONFLICT (ean) DO NOTHING + """ + ), + {"ean": ean, "num": f"M-{ean}", "cid": id_community}, + ) + await session.flush() + return ean + + +async def create_readings( + session, + *, + ean: str, + id_community: int, + id_sharing_operation: int, + start: datetime.datetime | None = None, + count: int = 4, + gross: float | None = 1.0, + inj_gross: float | None = 0.0, + step_minutes: int = 15, + skip: set[int] | None = None, +) -> list[datetime.datetime]: + """Insert ``count`` quarter-hourly readings, optionally skipping some. + + ``skip`` holds indices to omit, which is how a gap is produced: the meter is + then missing from those timestamps of the operation-wide grid and the + pre-flight reports it as incomplete. + """ + start = start or datetime.datetime(2025, 2, 1, 0, 0, tzinfo=_CET) + skip = skip or set() + written: list[datetime.datetime] = [] + for i in range(count): + if i in skip: + continue + ts = start + datetime.timedelta(minutes=step_minutes * i) + await session.execute( + text( + """ + INSERT INTO meter_consumption + (ean, id_sharing_operation, timestamp, gross, inj_gross, id_community) + VALUES (:ean, :op, :ts, :gross, :inj, :cid) + """ + ), + { + "ean": ean, + "op": id_sharing_operation, + "ts": ts, + "gross": gross, + "inj": inj_gross, + "cid": id_community, + }, + ) + written.append(ts) + await session.flush() + return written diff --git a/tests/factories/simulation_factory.py b/tests/factories/simulation_factory.py index 53761ea..48c3f0f 100644 --- a/tests/factories/simulation_factory.py +++ b/tests/factories/simulation_factory.py @@ -54,12 +54,18 @@ async def create_crm_key( iterations: int = 1, consumers_per_iteration: int = 2, energy_allocated_percentage: float = 0.5, + consumer_names: list[str] | None = None, ) -> AllocationKeyModel: """Seed a CRM allocation_key tree (consumer names stable across iterations). The simulation API validates ``id_key`` against this CRM table, and the worker reads the tree from here. + + ``consumer_names`` overrides the default ``C0``/``C1`` labels. CRM-sourced + runs need them to be meter EANs, since that is the (unenforced) convention + the platform matches ``consumer.name`` against. """ + names = consumer_names or [f"C{j}" for j in range(consumers_per_iteration)] key = AllocationKeyModel(name=name, description=description, id_community=id_community) session.add(key) await session.flush() @@ -72,10 +78,10 @@ async def create_crm_key( ) session.add(iteration) await session.flush() - for j in range(consumers_per_iteration): + for consumer_name in names: session.add( ConsumerModel( - name=f"C{j}", + name=consumer_name, energy_allocated_percentage=energy_allocated_percentage, id_iteration=iteration.id, id_community=id_community, diff --git a/tests/shared/__init__.py b/tests/shared/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/shared/test_crm_preflight.py b/tests/shared/test_crm_preflight.py new file mode 100644 index 0000000..3251f61 --- /dev/null +++ b/tests/shared/test_crm_preflight.py @@ -0,0 +1,184 @@ +"""Unit tests for the CRM pre-flight verdict. + +No database — ``evaluate`` is a pure function of a ``CrmDataSummary`` plus the +key's participant names. The rule this file exists to pin is the one the feature +was asked for: **a simulated key's participant names must match meter EANs.** +""" + +import datetime + +from shared.crm_meter_repository import CrmDataSummary, EanCoverage +from shared.crm_preflight import MAX_READING_ROWS, evaluate, normalize_participant +from shared.custom_errors import errors + +_A = "541448000000000001" +_B = "541448000000000002" +_PV = "541448000000000003" + + +def _coverage( + ean: str, + *, + row_count: int = 4, + distinct_ts: int = 4, + consumption_kwh: float = 10.0, + injection_kwh: float = 0.0, +) -> EanCoverage: + return EanCoverage( + ean=ean, + row_count=row_count, + distinct_ts=distinct_ts, + consumption_kwh=consumption_kwh, + injection_kwh=injection_kwh, + ) + + +def _summary(eans: list[EanCoverage], *, grid_size: int = 4) -> CrmDataSummary: + return CrmDataSummary( + eans=eans, + grid_size=grid_size, + first_timestamp=datetime.datetime(2025, 2, 1, tzinfo=datetime.UTC), + last_timestamp=datetime.datetime(2025, 2, 28, tzinfo=datetime.UTC), + ) + + +def _codes(preflight) -> set[int]: + return {b.error.code for b in preflight.blockers} + + +_DEFAULT = [_coverage(_A), _coverage(_PV, consumption_kwh=0.0, injection_kwh=50.0)] + + +# --------------------------------------------------------------------------- +# 1. Participant matching — the requested behaviour +# --------------------------------------------------------------------------- + + +def test_key_participants_matching_eans_are_accepted(): + result = evaluate(_summary(_DEFAULT), [_A]) + assert result.ok + assert result.matched == [_A] + assert result.unmatched == [] + + +def test_participant_with_no_matching_meter_blocks(): + result = evaluate(_summary(_DEFAULT), [_A, "NOT-AN-EAN"]) + assert not result.ok + assert errors.simulation.KEY_CONSUMERS_NOT_MATCHED.code in _codes(result) + assert result.unmatched == ["NOT-AN-EAN"] + # The manager needs to know *which* participant, not just that one failed. + blocker = next( + b for b in result.blockers if b.error is errors.simulation.KEY_CONSUMERS_NOT_MATCHED + ) + assert "NOT-AN-EAN" in blocker.detail + + +def test_participant_names_are_trimmed_before_matching(): + # crm-backend matches with TRIM(cons.name); a key with a stray space + # resolves in the CRM's own views and must resolve here too. + result = evaluate(_summary(_DEFAULT), [f" {_A} "]) + assert result.ok + assert result.matched == [_A] + assert result.participants == [_A] + + +def test_injection_only_meter_can_be_a_participant(): + # A key may legitimately include a producer that never draws; matching is + # against every meter with a reading, not just the consuming ones. + result = evaluate(_summary(_DEFAULT), [_A, _PV]) + assert result.ok + assert result.matched == [_A, _PV] + + +def test_participant_order_follows_the_key_not_the_database(): + # Matrix rows are built in this order and must line up with the key. + result = evaluate( + _summary( + [_coverage(_B), _coverage(_A), _coverage(_PV, consumption_kwh=0.0, injection_kwh=50.0)] + ), + [_B, _A], + ) + assert result.participants == [_B, _A] + + +def test_normalize_participant_trims_only(): + # EANs are digits, so there is deliberately no case folding. + assert normalize_participant(" 541448 ") == "541448" + + +# --------------------------------------------------------------------------- +# 2. Other blocking findings +# --------------------------------------------------------------------------- + + +def test_empty_period_blocks_with_a_single_message(): + result = evaluate(_summary([], grid_size=0), [_A]) + assert _codes(result) == {errors.simulation.CRM_NO_DATA.code} + assert result.unmatched == [_A] + + +def test_duplicate_readings_block(): + result = evaluate( + _summary( + [ + _coverage(_A, row_count=8, distinct_ts=4), + _coverage(_PV, consumption_kwh=0.0, injection_kwh=50.0), + ] + ), + [_A], + ) + assert errors.simulation.CRM_DUPLICATE_READINGS.code in _codes(result) + + +def test_no_injection_blocks(): + result = evaluate(_summary([_coverage(_A)]), [_A]) + assert errors.simulation.CRM_NO_INJECTION.code in _codes(result) + + +def test_oversized_period_blocks(): + big = MAX_READING_ROWS + 1 + result = evaluate( + _summary( + [_coverage(_A, row_count=big, distinct_ts=big, injection_kwh=1.0)], + grid_size=big, + ), + [_A], + ) + assert errors.simulation.CRM_RANGE_TOO_LARGE.code in _codes(result) + + +# --------------------------------------------------------------------------- +# 3. Gaps warn but do not block +# --------------------------------------------------------------------------- + + +def test_gaps_warn_but_do_not_block(): + result = evaluate( + _summary( + [ + _coverage(_A, row_count=2, distinct_ts=2), + _coverage(_PV, consumption_kwh=0.0, injection_kwh=50.0), + ] + ), + [_A], + ) + assert result.ok, "a gap must warn, not block" + assert result.warnings == { + "incomplete_meters": [{"ean": _A, "readings": 2, "expected": 4, "missing": 2}] + } + + +def test_gaps_in_meters_the_key_ignores_are_not_reported(): + # A gap in a meter this key does not simulate changes nothing. + result = evaluate( + _summary( + [ + _coverage(_A), + _coverage(_B, row_count=1, distinct_ts=1), + _coverage(_PV, consumption_kwh=0.0, injection_kwh=50.0), + ] + ), + [_A], + ) + assert result.ok + assert result.warnings is None diff --git a/tests/shared/test_crm_timeseries.py b/tests/shared/test_crm_timeseries.py new file mode 100644 index 0000000..a94b765 --- /dev/null +++ b/tests/shared/test_crm_timeseries.py @@ -0,0 +1,167 @@ +"""Unit tests for the CRM rows -> wide DataFrame pivot. + +No database: the pivot is a pure function over ``ConsumptionRow`` values, and +these tests pin the three behaviours the rest of the feature relies on — a +common timestamp grid, zero-filled gaps, and a single summed injection series. +""" + +import datetime + +import numpy as np +import pandas as pd +import pytest + +from shared import crm_timeseries, data_loading +from shared.crm_meter_repository import ConsumptionRow + +_TZ = datetime.UTC + + +def _ts(index: int) -> datetime.datetime: + return datetime.datetime(2025, 2, 1, 0, 0, tzinfo=_TZ) + datetime.timedelta(minutes=15 * index) + + +def _rows(spec: dict[str, list[tuple[int, float, float]]]) -> list[ConsumptionRow]: + """Build rows from {ean: [(timestamp_index, gross, inj_gross), ...]}.""" + return [ + ConsumptionRow(timestamp=_ts(i), ean=ean, gross=gross, inj_gross=inj) + for ean, entries in spec.items() + for (i, gross, inj) in entries + ] + + +# --------------------------------------------------------------------------- +# 1. Shape and ordering +# --------------------------------------------------------------------------- + + +def test_columns_are_participants_plus_injection_in_requested_order(): + rows = _rows({"B": [(0, 2.0, 0.0)], "A": [(0, 1.0, 0.0)]}) + frame = crm_timeseries.build_dataframe(rows, ["A", "B"]) + assert list(frame.columns) == ["A", "B", crm_timeseries.INJECTION_COLUMN] + + +def test_rows_are_ordered_chronologically_regardless_of_input_order(): + # Row t of every column must be contemporaneous; ordering is the only + # alignment guarantee the algorithms have. + rows = [ + ConsumptionRow(timestamp=_ts(2), ean="A", gross=30.0, inj_gross=3.0), + ConsumptionRow(timestamp=_ts(0), ean="A", gross=10.0, inj_gross=1.0), + ConsumptionRow(timestamp=_ts(1), ean="A", gross=20.0, inj_gross=2.0), + ] + frame = crm_timeseries.build_dataframe(rows, ["A"]) + assert frame["A"].tolist() == [10.0, 20.0, 30.0] + + +# --------------------------------------------------------------------------- +# 2. Gaps — the "warn but allow" behaviour +# --------------------------------------------------------------------------- + + +def test_missing_timestamps_are_zero_filled_onto_the_common_grid(): + # A is present at 0,1,2; B only at 0 and 2. The grid is the union, so B + # gets a 0.0 at index 1 rather than the frame collapsing to A's shape. + rows = _rows( + { + "A": [(0, 1.0, 0.0), (1, 1.0, 0.0), (2, 1.0, 0.0)], + "B": [(0, 5.0, 0.0), (2, 5.0, 0.0)], + } + ) + frame = crm_timeseries.build_dataframe(rows, ["A", "B"]) + assert len(frame) == 3 + assert frame["B"].tolist() == [5.0, 0.0, 5.0] + + +# --------------------------------------------------------------------------- +# 3. Injection +# --------------------------------------------------------------------------- + + +def test_injection_sums_every_meter_including_non_participants(): + # C injects but never consumes, so it is not a participant — its production + # must still reach the shared profile. + rows = _rows( + { + "A": [(0, 1.0, 2.0)], + "C": [(0, 0.0, 10.0)], + } + ) + frame = crm_timeseries.build_dataframe(rows, ["A"]) + assert list(frame.columns) == ["A", crm_timeseries.INJECTION_COLUMN] + assert frame[crm_timeseries.INJECTION_COLUMN].tolist() == [12.0] + + +def test_non_participant_meters_are_not_consumer_columns(): + rows = _rows({"A": [(0, 1.0, 0.0)], "C": [(0, 0.0, 9.0)]}) + frame = crm_timeseries.build_dataframe(rows, ["A"]) + assert "C" not in frame.columns + + +# --------------------------------------------------------------------------- +# 4. Refusals +# --------------------------------------------------------------------------- + + +def test_requested_participant_with_no_readings_is_refused(): + # Reindexing would otherwise invent an all-zero column, which for a + # simulation means a key participant silently gets a 0 kWh profile. + rows = _rows({"A": [(0, 1.0, 1.0)]}) + with pytest.raises(crm_timeseries.CrmPivotError, match="GHOST"): + crm_timeseries.build_dataframe(rows, ["A", "GHOST"]) + + +def test_duplicate_ean_timestamp_pairs_are_refused(): + rows = [ + ConsumptionRow(timestamp=_ts(0), ean="A", gross=1.0, inj_gross=1.0), + ConsumptionRow(timestamp=_ts(0), ean="A", gross=1.0, inj_gross=1.0), + ] + with pytest.raises(crm_timeseries.CrmPivotError): + crm_timeseries.build_dataframe(rows, ["A"]) + + +def test_empty_inputs_are_refused(): + with pytest.raises(crm_timeseries.CrmPivotError): + crm_timeseries.build_dataframe([], ["A"]) + with pytest.raises(crm_timeseries.CrmPivotError): + crm_timeseries.build_dataframe(_rows({"A": [(0, 1.0, 1.0)]}), []) + + +def test_injection_column_name_cannot_be_a_participant(): + rows = _rows({"A": [(0, 1.0, 1.0)]}) + with pytest.raises(crm_timeseries.CrmPivotError, match="reserved"): + crm_timeseries.build_dataframe(rows, [crm_timeseries.INJECTION_COLUMN]) + + +# --------------------------------------------------------------------------- +# 5. The whole point: the frame drops into the existing file-path converter +# --------------------------------------------------------------------------- + + +def test_frame_feeds_the_existing_simulation_converter_unchanged(): + rows = _rows( + { + "541448000000000001": [(0, 10.0, 0.0), (1, 11.0, 0.0)], + "541448000000000002": [(0, 5.0, 0.0), (1, 6.0, 0.0)], + "541448000000000003": [(0, 0.0, 100.0), (1, 0.0, 200.0)], + } + ) + participants = ["541448000000000001", "541448000000000002"] + + frame = crm_timeseries.build_dataframe(rows, participants) + raw = data_loading.to_simulation_raw_data(frame, crm_timeseries.INJECTION_COLUMN, participants) + + assert raw.consumer_names == participants + assert raw.C.shape == (2, 2) + np.testing.assert_array_equal(raw.C, np.array([[10.0, 11.0], [5.0, 6.0]])) + # VA is the single production series broadcast across every consumer row. + assert raw.VA.shape == (2, 2) + np.testing.assert_array_equal(raw.VA, np.array([[100.0, 200.0], [100.0, 200.0]])) + + +def test_frame_is_shaped_like_a_parsed_upload(): + # The contract that lets the converter stay untouched: a plain RangeIndex + # and float columns, exactly what pd.read_csv would produce. + rows = _rows({"A": [(0, 1.0, 1.0), (1, 2.0, 2.0)]}) + frame = crm_timeseries.build_dataframe(rows, ["A"]) + assert isinstance(frame.index, pd.RangeIndex) + assert all(pd.api.types.is_float_dtype(dtype) for dtype in frame.dtypes) diff --git a/tests/sql/crm_test_schema.sql b/tests/sql/crm_test_schema.sql index dc938c2..8ba0de6 100644 --- a/tests/sql/crm_test_schema.sql +++ b/tests/sql/crm_test_schema.sql @@ -92,3 +92,65 @@ CREATE TABLE IF NOT EXISTS audit_log ( user_email VARCHAR(256), payload JSONB NOT NULL DEFAULT '{}'::jsonb ); + + +-- ---- Metering tables ------------------------------------------------------- +-- Mirrors crm-backend/database_script/init.sql (meter / meter_data / +-- meter_consumption / sharing_operation), trimmed to the columns this service +-- actually SELECTs. Adapted from billing/tests/sql/crm_test_schema.sql, which +-- carries the same block for the same reason. +-- +-- Two production properties are reproduced deliberately, because the code under +-- test depends on both: +-- * meter_consumption has NO unique constraint on (ean, timestamp) -- that is +-- what makes a double import possible and CRM_DUPLICATE_READINGS necessary. +-- * every measure column is nullable, so COALESCE in the queries is load- +-- bearing rather than defensive. + +CREATE TABLE IF NOT EXISTS sharing_operation ( + id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + name VARCHAR(255) NOT NULL, + type INTEGER NOT NULL DEFAULT 1, + is_public BOOLEAN NOT NULL DEFAULT FALSE, + id_community INTEGER NOT NULL REFERENCES community (id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS meter ( + ean VARCHAR(64) PRIMARY KEY, + meter_number VARCHAR(255), + tarif_group INTEGER, + phases_number INTEGER, + reading_frequency INTEGER, + id_community INTEGER NOT NULL REFERENCES community (id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS meter_data ( + id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + ean VARCHAR(64) NOT NULL REFERENCES meter (ean) ON DELETE CASCADE, + id_member INTEGER, + id_sharing_operation INTEGER REFERENCES sharing_operation (id), + status INTEGER, -- 1=ACTIVE + client_type INTEGER, -- 1=Residentiel, 2=Professionnel, 3=Industriel + injection_status INTEGER, + production_chain INTEGER, + start_date DATE, + end_date DATE, + id_community INTEGER NOT NULL REFERENCES community (id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS meter_consumption ( + id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + ean VARCHAR(64) NOT NULL REFERENCES meter (ean) ON DELETE CASCADE, + id_sharing_operation INTEGER REFERENCES sharing_operation (id), + timestamp TIMESTAMPTZ NOT NULL, + gross DOUBLE PRECISION, + net DOUBLE PRECISION, + shared DOUBLE PRECISION, + inj_gross DOUBLE PRECISION, + inj_shared DOUBLE PRECISION, + inj_net DOUBLE PRECISION, + id_community INTEGER NOT NULL REFERENCES community (id) ON DELETE CASCADE +); + +CREATE INDEX IF NOT EXISTS idx_meter_consumption_lookup + ON meter_consumption (id_sharing_operation, timestamp); diff --git a/tests/worker/test_dispatcher.py b/tests/worker/test_dispatcher.py index a570560..7df26f8 100644 --- a/tests/worker/test_dispatcher.py +++ b/tests/worker/test_dispatcher.py @@ -20,6 +20,7 @@ from sqlalchemy.exc import OperationalError from core.queue.helper import Event +from shared.const import DataSource from worker import dispatcher _KEY = "simulations/1/test-uuid/data.csv" @@ -47,9 +48,13 @@ def _event_bytes(simulation_id) -> bytes: def _snapshot(*, status: int = 0, file_storage_key: str = _KEY) -> dispatcher._SimulationSnapshot: return dispatcher._SimulationSnapshot( id=1, + source=DataSource.FILE, file_storage_key=file_storage_key, file_name="data.csv", injection_name="production", + id_sharing_operation=None, + period_start=None, + period_end=None, id_key=10, id_community=1, status=int(status), @@ -194,12 +199,14 @@ async def test_compute_error_marks_failed(monkeypatch, patched_save, stub_pipeli async def test_storage_object_missing_marks_failed_no_delete( - monkeypatch, patched_save, patched_storage + monkeypatch, patched_save, patched_storage, stub_pipeline ): save_success, save_failure = patched_save download, delete = patched_storage download.side_effect = dispatcher.storage.ObjectNotFound("gone") - monkeypatch.setattr(dispatcher, "_snapshot_simulation", AsyncMock(return_value=_snapshot())) + # stub_pipeline is what makes the CRM key read succeed. It runs before the + # source is loaded now (both source paths need the key's participant names), + # so it has to pass for this test to reach the storage branch it is about. msg = FakeMsg(_event_bytes(1)) await dispatcher._make_handler()(msg) assert msg.acked @@ -210,11 +217,12 @@ async def test_storage_object_missing_marks_failed_no_delete( # ---- Transient failures (nak + NO delete) ---------------------------------- -async def test_storage_transient_naks_no_delete(monkeypatch, patched_save, patched_storage): +async def test_storage_transient_naks_no_delete( + monkeypatch, patched_save, patched_storage, stub_pipeline +): save_success, save_failure = patched_save download, delete = patched_storage download.side_effect = dispatcher.storage.TransientStorageError("503") - monkeypatch.setattr(dispatcher, "_snapshot_simulation", AsyncMock(return_value=_snapshot())) msg = FakeMsg(_event_bytes(1)) await dispatcher._make_handler()(msg) assert msg.naked and not msg.acked diff --git a/tests/worker/test_dispatcher_crm_source.py b/tests/worker/test_dispatcher_crm_source.py new file mode 100644 index 0000000..b90f2b2 --- /dev/null +++ b/tests/worker/test_dispatcher_crm_source.py @@ -0,0 +1,296 @@ +"""Worker tests for the CRM-sourced branch of ``_process``. + +Same style as test_dispatcher.py: drive ``_load_from_crm`` directly with +patched collaborators, so no live NATS, Postgres or MinIO is needed. + +Two things matter here beyond the allocation service's equivalent: + +* the failure classification — a CRM **read** error is transient (NAK) while + **rejected data** is deterministic (FAILED, ack); +* the trimmed-participant round trip — the matrix column labels and the key's + own consumer names must be the same strings, or ``run_simulation``'s internal + name -> percentage lookup raises on a key that matched perfectly. +""" + +from __future__ import annotations + +import datetime +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from shared.const import DataSource +from shared.crm_meter_repository import ConsumptionRow, CrmDataSummary, EanCoverage +from simulation.inputs import ( + SimulationConsumerInput, + SimulationIterationInput, + SimulationKeyInput, +) +from simulation.key_mapping import consumer_names_of +from worker import dispatcher + +_TZ = datetime.UTC +_EAN_A = "541448000000000001" +_EAN_PV = "541448000000000002" + + +def _snapshot( + *, + simulation_id: int = 1, + id_sharing_operation: int | None = 7, + period_start: datetime.date | None = datetime.date(2025, 2, 1), + period_end: datetime.date | None = datetime.date(2025, 2, 28), +) -> dispatcher._SimulationSnapshot: + return dispatcher._SimulationSnapshot( + id=simulation_id, + source=DataSource.CRM, + file_storage_key=None, + file_name=None, + injection_name=None, + id_sharing_operation=id_sharing_operation, + period_start=period_start, + period_end=period_end, + id_key=10, + id_community=42, + status=0, + ) + + +def _coverage(ean: str, *, consumption: float, injection: float, dup: bool = False) -> EanCoverage: + return EanCoverage( + ean=ean, + row_count=8 if dup else 4, + distinct_ts=4, + consumption_kwh=consumption, + injection_kwh=injection, + ) + + +def _summary(eans: list[EanCoverage]) -> CrmDataSummary: + return CrmDataSummary( + eans=eans, + grid_size=4, + first_timestamp=datetime.datetime(2025, 2, 1, tzinfo=_TZ), + last_timestamp=datetime.datetime(2025, 2, 1, 0, 45, tzinfo=_TZ), + ) + + +_HEALTHY = [ + _coverage(_EAN_A, consumption=40.0, injection=0.0), + _coverage(_EAN_PV, consumption=0.0, injection=100.0), +] + + +def _rows() -> list[ConsumptionRow]: + base = datetime.datetime(2025, 2, 1, tzinfo=_TZ) + out: list[ConsumptionRow] = [] + for i in range(4): + ts = base + datetime.timedelta(minutes=15 * i) + out.append(ConsumptionRow(timestamp=ts, ean=_EAN_A, gross=10.0, inj_gross=0.0)) + out.append(ConsumptionRow(timestamp=ts, ean=_EAN_PV, gross=0.0, inj_gross=25.0)) + return out + + +def _key(*names: str) -> SimulationKeyInput: + return SimulationKeyInput( + name="K", + description="d", + iterations=[ + SimulationIterationInput( + number=0, + energy_allocated_percentage=1.0, + consumers=[ + SimulationConsumerInput(name=n, energy_allocated_percentage=0.5) for n in names + ], + ) + ], + ) + + +@pytest.fixture +def patched_save(monkeypatch): + save_failure = AsyncMock() + monkeypatch.setattr(dispatcher.persistence, "save_failure", save_failure) + return save_failure + + +def _patch_crm(monkeypatch, *, summary, rows=None, raises: Exception | None = None): + repository = MagicMock() + if raises is not None: + repository.summarize = AsyncMock(side_effect=raises) + else: + repository.summarize = AsyncMock(return_value=summary) + repository.fetch_rows = AsyncMock(return_value=rows or []) + + session_cm = MagicMock() + session_cm.__aenter__ = AsyncMock(return_value=MagicMock()) + session_cm.__aexit__ = AsyncMock(return_value=False) + + monkeypatch.setattr(dispatcher, "AsyncSessionCRMFactory", MagicMock(return_value=session_cm)) + monkeypatch.setattr(dispatcher, "CrmMeterRepository", MagicMock(return_value=repository)) + return repository + + +# --------------------------------------------------------------------------- +# 1. Happy path +# --------------------------------------------------------------------------- + + +async def test_returns_raw_data_aligned_to_the_key(monkeypatch, patched_save): + repository = _patch_crm(monkeypatch, summary=_summary(_HEALTHY), rows=_rows()) + + result = await dispatcher._load_from_crm(_snapshot(), [_EAN_A]) + + assert not isinstance(result, dispatcher._Terminal) + assert result.consumer_names == [_EAN_A] + assert result.C.tolist() == [[10.0, 10.0, 10.0, 10.0]] + # The PV site is not a participant but still supplies the shared profile. + assert result.VA.tolist() == [[25.0, 25.0, 25.0, 25.0]] + patched_save.assert_not_awaited() + assert repository.summarize.await_args.kwargs["id_community"] == 42 + + +async def test_participant_order_follows_the_key(monkeypatch, patched_save): + _patch_crm( + monkeypatch, + summary=_summary( + [ + _coverage(_EAN_A, consumption=40.0, injection=0.0), + _coverage(_EAN_PV, consumption=40.0, injection=100.0), + ] + ), + rows=_rows(), + ) + + result = await dispatcher._load_from_crm(_snapshot(), [_EAN_PV, _EAN_A]) + + # Rows must line up with the key's order, not the database's. + assert result.consumer_names == [_EAN_PV, _EAN_A] + assert result.C[0].tolist() == [0.0, 0.0, 0.0, 0.0] + assert result.C[1].tolist() == [10.0, 10.0, 10.0, 10.0] + + +# --------------------------------------------------------------------------- +# 2. Trimming — the key and the matrix must agree on the strings +# --------------------------------------------------------------------------- + + +def test_trimming_the_key_keeps_it_consistent_with_the_matrix_labels(): + trimmed = dispatcher._with_trimmed_participants(_key(f" {_EAN_A} ")) + assert consumer_names_of(trimmed) == [_EAN_A] + # run_simulation resolves percentages off these names, so they must be the + # same strings the pivot used as column labels. + assert trimmed.iterations[0].consumers[0].energy_allocated_percentage == 0.5 + + +async def test_padded_participant_name_still_loads(monkeypatch, patched_save): + _patch_crm(monkeypatch, summary=_summary(_HEALTHY), rows=_rows()) + trimmed = dispatcher._with_trimmed_participants(_key(f" {_EAN_A} ")) + + result = await dispatcher._load_from_crm(_snapshot(), consumer_names_of(trimmed)) + + assert not isinstance(result, dispatcher._Terminal) + assert result.consumer_names == [_EAN_A] + + +# --------------------------------------------------------------------------- +# 3. Transient: the CRM is unreachable +# --------------------------------------------------------------------------- + + +async def test_crm_read_failure_is_transient(monkeypatch, patched_save): + _patch_crm(monkeypatch, summary=None, raises=OSError("connection reset")) + + with pytest.raises(dispatcher._TransientError, match="crm read"): + await dispatcher._load_from_crm(_snapshot(), [_EAN_A]) + + patched_save.assert_not_awaited() + + +# --------------------------------------------------------------------------- +# 4. Deterministic: the data itself is unusable +# --------------------------------------------------------------------------- + + +async def test_unmatched_participant_fails_deterministically(monkeypatch, patched_save): + repository = _patch_crm(monkeypatch, summary=_summary(_HEALTHY)) + + result = await dispatcher._load_from_crm(_snapshot(), ["C0"]) + + assert isinstance(result, dispatcher._Terminal) + assert result.storage_key is None + detail = patched_save.await_args.args[1] + assert "crm_data_rejected" in detail + assert "C0" in detail + repository.fetch_rows.assert_not_awaited() + + +async def test_empty_period_fails_deterministically(monkeypatch, patched_save): + _patch_crm(monkeypatch, summary=_summary([])) + + result = await dispatcher._load_from_crm(_snapshot(), [_EAN_A]) + + assert isinstance(result, dispatcher._Terminal) + assert "crm_data_rejected" in patched_save.await_args.args[1] + + +async def test_duplicate_readings_fail_deterministically(monkeypatch, patched_save): + _patch_crm( + monkeypatch, + summary=_summary( + [ + _coverage(_EAN_A, consumption=40.0, injection=0.0, dup=True), + _coverage(_EAN_PV, consumption=0.0, injection=100.0), + ] + ), + ) + + result = await dispatcher._load_from_crm(_snapshot(), [_EAN_A]) + + assert isinstance(result, dispatcher._Terminal) + assert "imported twice" in patched_save.await_args.args[1] + + +async def test_no_injection_fails_deterministically(monkeypatch, patched_save): + _patch_crm(monkeypatch, summary=_summary([_coverage(_EAN_A, consumption=40.0, injection=0.0)])) + + result = await dispatcher._load_from_crm(_snapshot(), [_EAN_A]) + + assert isinstance(result, dispatcher._Terminal) + assert "nothing to share" in patched_save.await_args.args[1] + + +async def test_incomplete_crm_columns_fail_deterministically(monkeypatch, patched_save): + _patch_crm(monkeypatch, summary=_summary([])) + + result = await dispatcher._load_from_crm(_snapshot(id_sharing_operation=None), [_EAN_A]) + + assert isinstance(result, dispatcher._Terminal) + patched_save.assert_awaited_once_with(1, "crm_source_incomplete") + + +# --------------------------------------------------------------------------- +# 5. Gaps still run +# --------------------------------------------------------------------------- + + +async def test_gaps_do_not_stop_the_run_and_are_zero_filled(monkeypatch, patched_save): + rows = [r for r in _rows() if not (r.ean == _EAN_A and r.timestamp.minute >= 30)] + _patch_crm( + monkeypatch, + summary=_summary( + [ + EanCoverage( + _EAN_A, row_count=2, distinct_ts=2, consumption_kwh=20.0, injection_kwh=0.0 + ), + _coverage(_EAN_PV, consumption=0.0, injection=100.0), + ] + ), + rows=rows, + ) + + result = await dispatcher._load_from_crm(_snapshot(), [_EAN_A]) + + assert not isinstance(result, dispatcher._Terminal) + assert result.C.tolist() == [[10.0, 10.0, 0.0, 0.0]] + patched_save.assert_not_awaited() diff --git a/tests/worker/test_persistence_realtime.py b/tests/worker/test_persistence_realtime.py new file mode 100644 index 0000000..5925c41 --- /dev/null +++ b/tests/worker/test_persistence_realtime.py @@ -0,0 +1,136 @@ +"""Realtime emission from the worker's terminal transitions. + +The single most important assertion here is ORDER: the CRM commit must appear +before the emit. Publishing pre-commit does not merely lose an event — it tells +the browser to refetch and read PRE-COMMIT state, and because the transport is +fire-and-forget there is no second event, ever. The result is a permanently +stale UI behind a 200, with no error anywhere. + +``patched_sessions`` routes both session factories at the test's +savepoint-joined session, so ``commit()`` is a savepoint release the conftest +rolls back. +""" + +from unittest.mock import AsyncMock + +import pytest_asyncio + +from core.realtime import Tier +from shared.const import SimulationStatus +from tests.factories.simulation_factory import create_simulation +from tests.worker.test_persistence import _result +from worker import persistence + + +@pytest_asyncio.fixture +async def patched_sessions(db_session, monkeypatch): + """Route both session factories at the test's savepoint-joined session. + + Defined here rather than imported from test_persistence: importing a fixture + and then naming a parameter after it shadows the import, which ruff flags as + a redefinition and which breaks the moment the other module renames it. + """ + + class _Ctx: + async def __aenter__(self): + return db_session + + async def __aexit__(self, exc_type, exc, tb): + return False + + class _Factory: + def __call__(self): + return _Ctx() + + monkeypatch.setattr(persistence, "AsyncSessionLocalFactory", _Factory()) + monkeypatch.setattr(persistence, "AsyncSessionCRMFactory", _Factory()) + monkeypatch.setattr(persistence.storage, "upload", AsyncMock()) + yield + + +@pytest_asyncio.fixture +async def trace(db_session, monkeypatch): + """Record commits and emits into ONE ordered list.""" + events: list[str] = [] + + original_commit = db_session.commit + + async def _commit(): + events.append("commit") + await original_commit() + + async def _emit(**kwargs): + events.append("emit") + events.append(kwargs) + + monkeypatch.setattr(db_session, "commit", _commit) + monkeypatch.setattr(persistence, "emit", AsyncMock(side_effect=_emit)) + return events + + +async def test_save_success_emits_after_the_commit(db_session, patched_sessions, trace): + sim = await create_simulation(db_session, id_community=7) + + await persistence.save_success(sim.id, _result()) + + assert "emit" in trace, "no realtime hint was published for a successful simulation" + # THE assertion. Not "did it emit" but "did it emit after the write landed". + assert trace.index("emit") > trace.index("commit") + + +async def test_save_success_addresses_the_community_manager_tier( + db_session, patched_sessions, trace +): + sim = await create_simulation(db_session, id_community=7) + + await persistence.save_success(sim.id, _result()) + + payload = trace[trace.index("emit") + 1] + assert payload["topic"] == "simulation.finished" + assert payload["hint"] == {"status": "success"} + assert payload["resource"] == ("simulation", sim.id) + # The audience is the community's MANAGER tier, not a user: `simulation` + # carries only id_community, and the hub route is manager-gated. This is what + # lets a worker with no request context reach the right people with zero + # lookups — and it is only non-leaking because of that gating. + assert payload["audience"].community_id == 7 + assert payload["audience"].tier is Tier.MANAGER + + +async def test_save_failure_emits_after_the_commit_with_a_failed_status( + db_session, patched_sessions, trace +): + sim = await create_simulation(db_session, id_community=7) + + await persistence.save_failure(sim.id, "boom") + + assert trace.index("emit") > trace.index("commit") + payload = trace[trace.index("emit") + 1] + assert payload["hint"] == {"status": "failed"} + # No error message on the wire: the envelope is a hint, and the client + # refetches through the gateway, which re-authorizes the read. + assert "boom" not in str(payload) + + +async def test_no_emit_when_the_row_is_already_terminal(db_session, patched_sessions, trace): + # JetStream redelivery: the conditional UPDATE ... WHERE status=PENDING guard + # returns early, so a second delivery must not re-toast every open hub. + sim = await create_simulation(db_session, id_community=7, status=SimulationStatus.SUCCESS) + + await persistence.save_success(sim.id, _result()) + + assert "emit" not in trace + + +async def test_double_delivery_emits_exactly_once(db_session, patched_sessions, trace): + sim = await create_simulation(db_session, id_community=7) + + await persistence.save_success(sim.id, _result()) + await persistence.save_success(sim.id, _result()) + + assert trace.count("emit") == 1 + + +async def test_missing_row_emits_nothing(db_session, patched_sessions, trace): + await persistence.save_success(999_999, _result()) + assert "emit" not in trace diff --git a/worker/dispatcher.py b/worker/dispatcher.py index 8fad9b3..a0cea89 100644 --- a/worker/dispatcher.py +++ b/worker/dispatcher.py @@ -24,6 +24,7 @@ import asyncio import dataclasses +import datetime import logging import time from collections.abc import Awaitable, Callable @@ -37,12 +38,17 @@ from core import storage from core.database.database import AsyncSessionCRMFactory, AsyncSessionLocalFactory from core.queue.helper import Event -from shared import data_loading -from shared.const import SIMULATION_SUBJECT, SimulationStatus +from shared import crm_preflight, crm_timeseries, data_loading +from shared.const import SIMULATION_SUBJECT, DataSource, SimulationStatus +from shared.crm_meter_repository import CrmMeterRepository from shared.crm_repository import CRMRepository from shared.models.local_models import SimulationModel from simulation.compute import SimulationInputError, run_simulation -from simulation.inputs import SimulationKeyInput +from simulation.inputs import ( + SimulationConsumerInput, + SimulationIterationInput, + SimulationKeyInput, +) from simulation.key_mapping import consumer_names_of, from_crm_allocation_key from simulation.result import KeySimResult from worker import persistence @@ -67,9 +73,15 @@ class _SimulationSnapshot: """Per-message snapshot of the row, captured before the session closes.""" id: int - file_storage_key: str - file_name: str - injection_name: str + source: DataSource + # FILE only — None on a CRM-sourced row. + file_storage_key: str | None + file_name: str | None + injection_name: str | None + # CRM only — None on a file-sourced row. + id_sharing_operation: int | None + period_start: datetime.date | None + period_end: datetime.date | None id_key: int id_community: int status: int @@ -275,16 +287,10 @@ async def _process( ) return _Terminal(storage_key=snapshot.file_storage_key) - # ---- Step 2: download the source file ------------------------------ - try: - content = await storage.download(snapshot.file_storage_key) - except storage.ObjectNotFound: - await persistence.save_failure(simulation_id, "storage_object_missing") - return _Terminal(storage_key=None) - except storage.TransientStorageError as exc: - raise _TransientError(f"storage download: {exc}") from exc - - # ---- Step 3: read the simulated key from the CRM DB ---------------- + # ---- Step 2: read the simulated key from the CRM DB ---------------- + # Ahead of the source read because both paths need the key's participant + # names: the file path matches columns against them, the CRM path matches + # meter EANs against them. try: key_model = await _read_crm_key(snapshot.id_key, snapshot.id_community) except Exception as exc: @@ -297,23 +303,25 @@ async def _process( return _Terminal(storage_key=snapshot.file_storage_key) key_input = from_crm_allocation_key(key_model) + if snapshot.source is DataSource.CRM: + # Match EANs the way the rest of the platform does, with TRIM. The key + # itself is normalised (not just the comparison) so that run_simulation's + # own name -> percentage lookup still resolves: it keys off the input + # model's names, which must therefore be the same strings as the matrix + # column labels. The file path is deliberately left untrimmed — changing + # its matching would alter existing behaviour. + key_input = _with_trimmed_participants(key_input) consumer_names = consumer_names_of(key_input) - # ---- Step 4: parse the file (consumer columns matched by name) ----- - try: - raw = data_loading.load( - content, snapshot.file_name, snapshot.injection_name, consumer_names - ) - except ( - data_loading.InvalidInjectionColumnError, - data_loading.UnsupportedFileFormatError, - data_loading.ConsumerColumnsError, - ) as exc: - await persistence.save_failure(simulation_id, f"parse_failed: {exc}") - return _Terminal(storage_key=snapshot.file_storage_key) - except Exception as exc: - await persistence.save_failure(simulation_id, f"parse_failed_unexpected: {exc}") - return _Terminal(storage_key=snapshot.file_storage_key) + # ---- Step 3: obtain the (C, VA, names) triple ---------------------- + # Two sources, one output. Everything downstream is identical. + if snapshot.source is DataSource.CRM: + loaded = await _load_from_crm(snapshot, consumer_names) + else: + loaded = await _load_from_file(snapshot, consumer_names) + if isinstance(loaded, _Terminal): + return loaded + raw = loaded # ---- Step 5: run the simulation off the event loop ---------------- start = time.perf_counter() @@ -346,6 +354,141 @@ async def _process( return _Terminal(storage_key=snapshot.file_storage_key) +def _with_trimmed_participants(key: SimulationKeyInput) -> SimulationKeyInput: + """Return the key with every consumer name trimmed. + + ``allocation_key.consumer.name`` is free text with no FK to ``meter``; the + platform-wide convention that it holds the EAN is enforced only by + comparison, and crm-backend compares with ``TRIM``. Normalising the whole + key (rather than only the comparison) keeps one set of strings in play + across matching, the matrix column labels, and the result rows. + """ + return SimulationKeyInput( + name=key.name, + description=key.description, + iterations=[ + SimulationIterationInput( + number=it.number, + energy_allocated_percentage=it.energy_allocated_percentage, + consumers=[ + SimulationConsumerInput( + name=crm_preflight.normalize_participant(c.name), + energy_allocated_percentage=c.energy_allocated_percentage, + ) + for c in it.consumers + ], + ) + for it in key.iterations + ], + ) + + +async def _load_from_file( + snapshot: _SimulationSnapshot, consumer_names: list[str] +) -> data_loading.SimulationRawData | _Terminal: + """Download the uploaded object and parse it. The historical path.""" + if snapshot.file_storage_key is None or snapshot.file_name is None: + # Unreachable through the API (ck_simulation_source enforces it), but a + # row written directly to the DB could get here. + await persistence.save_failure(snapshot.id, "file_source_incomplete") + return _Terminal(storage_key=None) + + try: + content = await storage.download(snapshot.file_storage_key) + except storage.ObjectNotFound: + await persistence.save_failure(snapshot.id, "storage_object_missing") + return _Terminal(storage_key=None) + except storage.TransientStorageError as exc: + raise _TransientError(f"storage download: {exc}") from exc + + try: + return data_loading.load( + content, snapshot.file_name, snapshot.injection_name or "", consumer_names + ) + except ( + data_loading.InvalidInjectionColumnError, + data_loading.UnsupportedFileFormatError, + data_loading.ConsumerColumnsError, + ) as exc: + await persistence.save_failure(snapshot.id, f"parse_failed: {exc}") + return _Terminal(storage_key=snapshot.file_storage_key) + except Exception as exc: + await persistence.save_failure(snapshot.id, f"parse_failed_unexpected: {exc}") + return _Terminal(storage_key=snapshot.file_storage_key) + + +async def _load_from_crm( + snapshot: _SimulationSnapshot, consumer_names: list[str] +) -> data_loading.SimulationRawData | _Terminal: + """Read meter_consumption for this row's sharing operation and period. + + The pre-flight is re-run here rather than trusted from creation time: the + data can have changed since the run was queued, and this is the read that + actually feeds the simulation. In particular the participant-to-EAN match is + re-checked, so a meter deleted in the meantime fails loudly. + + Failure classification follows the module's existing matrix — a CRM read + error is transient (NAK, redeliver), while rejected or unpivotable data is + deterministic (FAILED, ack). There is never an object to delete. + """ + if ( + snapshot.id_sharing_operation is None + or snapshot.period_start is None + or snapshot.period_end is None + ): + await persistence.save_failure(snapshot.id, "crm_source_incomplete") + return _Terminal(storage_key=None) + + try: + async with AsyncSessionCRMFactory() as crm_session: + repository = CrmMeterRepository(crm_session) + # The worker has no request context, so the community is passed + # explicitly; with_community_scope would degrade to WHERE false. + summary = await repository.summarize( + id_community=snapshot.id_community, + id_sharing_operation=snapshot.id_sharing_operation, + period_start=snapshot.period_start, + period_end=snapshot.period_end, + ) + preflight = crm_preflight.evaluate(summary, consumer_names) + # Skip the expensive read when the period is already rejected. + rows = ( + await repository.fetch_rows( + id_community=snapshot.id_community, + id_sharing_operation=snapshot.id_sharing_operation, + period_start=snapshot.period_start, + period_end=snapshot.period_end, + ) + if preflight.ok + else [] + ) + except Exception as exc: + raise _TransientError(f"crm read: {exc}") from exc + + if preflight.blockers: + detail = "; ".join(b.detail for b in preflight.blockers) + await persistence.save_failure(snapshot.id, f"crm_data_rejected: {detail}") + return _Terminal(storage_key=None) + + try: + frame = crm_timeseries.build_dataframe(rows, preflight.participants) + # The same converter the file path uses — the frame is deliberately + # shaped like a parsed upload so nothing below this line differs. + return data_loading.to_simulation_raw_data( + frame, crm_timeseries.INJECTION_COLUMN, preflight.participants + ) + except ( + crm_timeseries.CrmPivotError, + data_loading.InvalidInjectionColumnError, + data_loading.ConsumerColumnsError, + ) as exc: + await persistence.save_failure(snapshot.id, f"crm_pivot_failed: {exc}") + return _Terminal(storage_key=None) + except Exception as exc: + await persistence.save_failure(snapshot.id, f"crm_pivot_failed_unexpected: {exc}") + return _Terminal(storage_key=None) + + async def _read_crm_key(id_key: int, id_community: int): """Load the simulated key from the CRM DB in a short-lived session. @@ -368,9 +511,13 @@ async def _snapshot_simulation(simulation_id: int) -> _SimulationSnapshot | None return None return _SimulationSnapshot( id=row.id, + source=DataSource(row.source), file_storage_key=row.file_storage_key, file_name=row.file_name, injection_name=row.injection_name, + id_sharing_operation=row.id_sharing_operation, + period_start=row.period_start, + period_end=row.period_end, id_key=row.id_key, id_community=row.id_community, status=int(row.status), diff --git a/worker/main.py b/worker/main.py index 03b9a5d..af7225c 100644 --- a/worker/main.py +++ b/worker/main.py @@ -28,6 +28,7 @@ from core.database.database import crm_engine, local_engine from core.logging import configure_logging from core.queue.init import close_nats, get_jetstream, init_nats +from core.realtime import log_realtime_state from core.tracing import setup_tracer_provider from shared.const import SIMULATION_STREAM from worker import dispatcher @@ -202,6 +203,9 @@ async def _poll_queue_depth(js, shutdown_event: asyncio.Event) -> None: async def main() -> None: configure_logging() + # Absence of this line means the image predates the realtime feature — + # see core/realtime/bus.py. Must come after configure_logging(). + log_realtime_state("simulation-key-worker") setup_tracer_provider() await _connect_nats_with_retry() diff --git a/worker/persistence.py b/worker/persistence.py index 14657a1..9022047 100644 --- a/worker/persistence.py +++ b/worker/persistence.py @@ -28,6 +28,7 @@ from core import storage from core.audit_log import AuditActions, AuditLogInput, AuditLogService from core.database.database import AsyncSessionCRMFactory, AsyncSessionLocalFactory +from core.realtime import CommunityAudience, Tier, emit from shared.const import SimulationStatus from shared.models.local_models import ( SimulationConsumerResultModel, @@ -181,6 +182,22 @@ async def save_success(simulation_id: int, result: KeySimResult) -> None: ) await crm_session.commit() + # Realtime hint, AFTER both commits. Fire-and-forget: dropped if nobody has + # the hub open, which is correct — the row is durable and the hub's own + # poller converges regardless. + # + # Audience is the community's MANAGER tier, not a user: `simulation` carries + # only `id_community`, and the hub route is manager-gated + # (annexes-services.routes.ts, minRole GESTIONNAIRE). A worker with no + # request context reaches exactly the right people with zero lookups. + await emit( + topic="simulation.finished", + audience=CommunityAudience(community_id=community_id, tier=Tier.MANAGER), + resource=("simulation", simulation_id), + scope_community_id=community_id, + hint={"status": "success"}, + ) + async def save_failure(simulation_id: int, error_message: str) -> None: """Mark a simulation FAILED with the given message (idempotent).""" @@ -220,3 +237,13 @@ async def save_failure(simulation_id: int, error_message: str) -> None: id_community=id_community, ) await crm_session.commit() + + # Same contract as save_success. No error message in the envelope — it is a + # hint; the client refetches through the gateway, which re-authorizes. + await emit( + topic="simulation.finished", + audience=CommunityAudience(community_id=id_community, tier=Tier.MANAGER), + resource=("simulation", simulation_id), + scope_community_id=id_community, + hint={"status": "failed"}, + )