# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence. """Family B: hours, weekdays, day series, months and dying commands. Every hour test seeds a ``daily_uses`` row and asserts it is absent. That is how the exclusion gets proved rather than remembered: those rows carry UNKNOWN_HOUR because their source knew only a date, and letting one through files it under hour -1 -- or under midnight, if a later edit "tidies" the sentinel away. """ from __future__ import annotations from typing import TYPE_CHECKING from signal_bot.analytics import hours, trend from signal_bot.analytics.scope import Scope, Window, window from tests.analytics_seed import ALA, BASIA, backfilled, ran, reconstructed if TYPE_CHECKING: import sqlite3 from signal_bot.store import Store # A Saturday, so the weekday index is not accidentally right at 0. DAY = "2026-08-29" NEXT = "2026-08-30" SATURDAY = 5 def conn_of(store: Store) -> sqlite3.Connection: return store.commands.connection def test_the_hour_histogram_has_a_bucket_for_every_hour(store: Store) -> None: """A missing 04:00 draws the wrong shape, and the empty hours are data.""" ran(store, day=DAY, hour=9) buckets = hours.by_hour(conn_of(store), Scope()) assert len(buckets) == hours.HOURS_IN_DAY assert buckets[9] == 1 assert sum(buckets) == 1 def test_a_row_with_no_hour_is_left_out(store: Store) -> None: """The exclusion, asserted rather than trusted.""" ran(store, day=DAY, hour=9) backfilled(store, day=DAY) assert sum(hours.by_hour(conn_of(store), Scope())) == 1 def test_a_reconstructed_row_keeps_its_hour(store: Store) -> None: """journald rows were matched to a real message, so their hour is real. Excluding them would throw away most of the history for no gain, which is why scope names two predicates and not one. """ reconstructed(store, day=DAY, hour=21) buckets = hours.by_hour(conn_of(store), Scope()) assert buckets[21] == 1 def test_the_weekday_grid_is_seven_by_twenty_four(store: Store) -> None: ran(store, day=DAY, hour=13) grid = hours.by_weekday_hour(conn_of(store), Scope()) assert len(grid) == hours.DAYS_IN_WEEK assert all(len(row) == hours.HOURS_IN_DAY for row in grid) assert grid[SATURDAY][13] == 1 def test_each_command_gets_its_own_hours(store: Store) -> None: ran(store, day=DAY, hour=8, command="cat") ran(store, day=DAY, hour=22, command="pope") grid = hours.by_command_hour(conn_of(store), Scope()) assert grid["cat"][8] == 1 assert grid["pope"][22] == 1 assert sum(grid["cat"]) == 1 def test_weekdays_carry_the_days_they_were_seen_on(store: Store) -> None: """An average needs a denominator, and one Tuesday is not a weekly rate.""" ran(store, day=DAY, hour=9) ran(store, day=DAY, hour=10, minute=5) assert hours.by_weekday(conn_of(store), Scope()) == [(SATURDAY, 2, 1)] def test_the_peak_ignores_hours_nobody_ever_used(store: Store) -> None: """At three days old most of the night is empty. "The deadest hour is 04:00 with zero" would be true of nine hours and interesting about none of them. """ ran(store, day=DAY, hour=19) ran(store, day=DAY, hour=19, minute=1) ran(store, day=DAY, hour=6) found = hours.peak(hours.by_hour(conn_of(store), Scope())) assert (found.busiest, found.busiest_uses) == (19, 2) assert (found.quietest, found.quietest_uses) == (6, 1) def test_a_silent_history_has_no_peak(store: Store) -> None: assert hours.peak([0] * hours.HOURS_IN_DAY) == hours.Peak() def test_a_silent_day_is_drawn_as_a_zero(store: Store) -> None: """A sparse series draws a line straight over a day nobody used it.""" ran(store, day="2026-08-29", hour=9) ran(store, day="2026-08-31", hour=9) dense = trend.dense_days(trend.by_day(conn_of(store), Scope())) assert dense == [("2026-08-29", 1), ("2026-08-30", 0), ("2026-08-31", 1)] def test_an_empty_series_densifies_to_nothing() -> None: assert trend.dense_days([]) == [] def test_months_carry_how_many_days_they_actually_saw(store: Store) -> None: """A partial month is not a quiet one, and the panel must be able to say.""" ran(store, day=DAY, hour=9) ran(store, day=NEXT, hour=9) assert trend.by_month(conn_of(store), Scope()) == [("2026-08", 2, 2)] def test_per_command_day_series(store: Store) -> None: ran(store, day=DAY, hour=9, command="cat") ran(store, day=NEXT, hour=9, command="cat") ran(store, day=NEXT, hour=10, command="pope") series = trend.by_command_day(conn_of(store), Scope()) assert series["cat"] == {DAY: 1, NEXT: 1} assert series["pope"] == {NEXT: 1} def test_no_direction_is_claimed_from_a_short_history(store: Store) -> None: """Three points of a new bot's novelty curve are not a trend.""" series = [("2026-08-29", 1), ("2026-08-30", 5), ("2026-08-31", 20)] assert trend.direction(series, Window(days=3)) == trend.UNKNOWN def test_a_direction_is_claimed_once_there_is_enough(store: Store) -> None: rising = [(f"2026-08-{day:02d}", day) for day in range(1, 15)] falling = [(f"2026-08-{day:02d}", 15 - day) for day in range(1, 15)] flat = [(f"2026-08-{day:02d}", 10) for day in range(1, 15)] measured = Window(days=14) assert trend.direction(rising, measured) == trend.RISING assert trend.direction(falling, measured) == trend.FALLING assert trend.direction(flat, measured) == trend.FLAT def test_a_direction_needs_two_points() -> None: assert trend.direction([("2026-08-29", 3)], Window(days=20)) == trend.UNKNOWN assert trend.direction([], Window(days=20)) == trend.UNKNOWN def test_growth_from_nothing_is_rising_and_nothing_is_flat() -> None: """A first half of zero cannot be divided by, and both cases are real.""" measured = Window(days=20) from_zero = [("a", 0), ("b", 0), ("c", 5), ("d", 5)] stayed_zero = [("a", 0), ("b", 0), ("c", 0), ("d", 0)] assert trend.direction(from_zero, measured) == trend.RISING assert trend.direction(stayed_zero, measured) == trend.FLAT def test_a_command_that_stopped_being_used_is_named(store: Store) -> None: ran(store, day="2026-08-20", hour=9, command="gecko") ran(store, day="2026-08-31", hour=9, command="cat") dying = trend.dying(conn_of(store), Scope(), "2026-08-31") assert [(entry.command, entry.idle_days) for entry in dying] == [("gecko", 11)] def test_a_command_used_today_is_not_dying(store: Store) -> None: ran(store, day="2026-08-31", hour=9, command="cat") assert trend.dying(conn_of(store), Scope(), "2026-08-31") == [] def test_two_people_on_one_day_are_counted_once_each(store: Store) -> None: ran(store, day=DAY, hour=9, who=ALA) ran(store, day=DAY, hour=10, who=BASIA) assert trend.by_day(conn_of(store), Scope()) == [(DAY, 2, 2)] assert window(conn_of(store), Scope()).days == 1