signal-bot/tests/analytics_seed.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Building a command history to ask questions of.

Everything is arithmetic -- no randomness, no clock -- so a failure names a
row rather than a seed. The people are invented (ala, basia, celina, darek)
and so are the group ids: ``check_personal_data`` learns real names from the
live database, so a real one here would fail only on the maintainer's
machine and pass in CI.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import TYPE_CHECKING

from signal_bot.analytics.backfill import midnight_ms
from signal_bot.analytics.probe import UpstreamCall
from signal_bot.analytics.uses import (
    OUTCOME_OK,
    SOURCE_DAILY,
    SOURCE_JOURNAL,
    SOURCE_LIVE,
    CommandEvent,
)
from signal_bot.stats.backfill import UNKNOWN_HOUR
from signal_bot.stats.periods import weekday_of

if TYPE_CHECKING:
    from collections.abc import Sequence

    from signal_bot.store import Store

GROUP = "group-a=="
OTHER = "group-b=="

ALA = "uuid-ala"
BASIA = "uuid-basia"
CELINA = "uuid-celina"

HOUR_MS = 3_600_000


def at(day: str, hour: int, minute: int = 0) -> int:
    """A Warsaw wall-clock moment on one day, in epoch milliseconds."""
    return midnight_ms(day) + hour * HOUR_MS + minute * 60_000


@dataclass(frozen=True, slots=True)
class Shape:
    """Everything about one attempt except who ran what, and when.

    Grouped rather than passed as arguments, for the reason the repository
    caps arguments at seven: these are the columns a given test usually
    leaves entirely alone, and a call site that names three of them reads
    better than one that names all eleven.
    """

    group_id: str = GROUP
    outcome: str = OUTCOME_OK
    source: str = SOURCE_LIVE
    item_id: str | None = None
    handler_ms: int | None = None
    send_ms: int | None = None
    error_class: str = ""
    args: str = ""
    calls: Sequence[UpstreamCall] = ()


# A module-level singleton, because Shape is frozen and a default argument
# that is a call is evaluated once anyway -- naming it says so.
PLAIN = Shape()


def ran(
    store: Store,
    *,
    day: str,
    hour: int = 12,
    minute: int = 0,
    who: str = ALA,
    command: str = "cat",
    shape: Shape = PLAIN,
) -> int:
    """Record one attempt and return the timestamp it was keyed on."""
    invoked_at_ms = at(day, hour, minute)
    store.commands.record(
        CommandEvent(
            group_id=shape.group_id,
            user_uuid=who,
            command=command,
            args=shape.args,
            item_id=shape.item_id,
            outcome=shape.outcome,
            error_class=shape.error_class,
            invoked_at_ms=invoked_at_ms,
            day=day,
            local_hour=UNKNOWN_HOUR if shape.source == SOURCE_DAILY else hour,
            weekday=weekday_of(day),
            handler_ms=shape.handler_ms,
            send_ms=shape.send_ms,
            source=shape.source,
            calls=shape.calls,
        )
    )
    return invoked_at_ms


def backfilled(
    store: Store, *, day: str, command: str = "cat", who: str = ALA, group_id: str = GROUP
) -> None:
    """One row from the source that knows a day and nothing finer."""
    ran(
        store,
        day=day,
        command=command,
        who=who,
        hour=0,
        shape=Shape(group_id=group_id, source=SOURCE_DAILY),
    )


def reconstructed(
    store: Store,
    *,
    day: str,
    hour: int,
    command: str = "cat",
    who: str = ALA,
    group_id: str = GROUP,
) -> None:
    """One row recovered from the journal: a real hour, but no durations."""
    ran(
        store,
        day=day,
        hour=hour,
        command=command,
        who=who,
        shape=Shape(group_id=group_id, source=SOURCE_JOURNAL),
    )


def a_call(host: str = "api.example.test", status: int = 200, ms: int = 40) -> UpstreamCall:
    """One outbound request made while a command was running."""
    return UpstreamCall(host=host, status=status, ms=ms)