clickstream-data-platform/generator/tests/test_day.py

"""День-функция: чистота, форма волны, правила резки визитов, шов с торговлей.

Числа мира тесты сторожат вилками спеки, а не точными значениями: менти
крутит конфигурацию, и падать тесты должны там, где сдвинулся вывод («средний
день ~50 тыс. событий», «ночью провал, вечером пик»), а не при каждой правке.
"""

import inspect
import re
from dataclasses import replace
from typing import Any

import numpy as np
import pytest
from numpy.typing import NDArray

from clickstream_generator import (
    catalog,
    commerce,
    day,
    ids,
    plan,
    reference,
    schema,
    world,
)
from clickstream_generator.reference import Page
from clickstream_generator.seeds import CANONICAL_SEED

WEEKDAY = 2
WEEKEND = 5


@pytest.fixture(autouse=True)
def fresh_memo():
    """Когорты запоминаются; тесты сравнивают вычисления, а не ссылки."""
    plan.cohort.cache_clear()


@pytest.fixture(scope="module")
def weekday() -> day.Day:
    """Один буднийный день на весь модуль: пересчитывать его тестам незачем."""
    return day.stream(CANONICAL_SEED, WEEKDAY)


def readable(column: NDArray[Any]) -> list[Any]:
    """Колонка в сравнимом виде: массив внутри ячейки — в список."""
    if column.dtype == object:
        return [
            cell.tolist() if isinstance(cell, np.ndarray) else cell for cell in column
        ]
    return column.tolist()


def snapshot(events: day.Day) -> dict[str, Any]:
    """Слепок дня для сравнений: всё, что день отдал наружу.

    Порядок колонок в слепке живёт отдельным списком: словари сравниваются
    без оглядки на него, а порядок — часть обещания (он же порядок
    контракта схемы). Швы `page` и `product` — тоже часть отдаваемого, и
    сторожить их надо тем же слепком, а не отдельной памятью.
    """
    return {
        "order": list(events.columns),
        "values": [readable(column) for column in events.columns.values()],
        "page": events.page.tolist(),
        "product": events.product.tolist(),
    }


def local_seconds(events: day.Day) -> NDArray[np.int64]:
    """Секунда события внутри модельных суток — в поясе счётчика."""
    midnight = np.datetime64(world.ORIGIN, "s") + np.timedelta64(events.day, "D")
    away = np.timedelta64(world.COUNTER_TIMEZONE_MINUTES, "m")
    return (events.columns["UTCEventTime"] - midnight + away).astype("int64")


def hourly(events: day.Day) -> NDArray[np.int64]:
    return np.bincount(local_seconds(events) // 3600, minlength=24)


def test_the_snapshot_notices_everything_the_day_hands_out(weekday: day.Day):
    """Сторож сторожей: слепок обязан замечать порядок колонок и швы.

    Слепок слепой к чему-нибудь — это три зелёных теста детерминизма при
    поехавшем выводе, а не одна пропущенная мелочь.
    """
    original = snapshot(weekday)
    reordered = day.Day(
        day=weekday.day,
        columns=dict(reversed(list(weekday.columns.items()))),
        page=weekday.page,
        product=weekday.product,
    )
    assert snapshot(reordered) != original

    moved = weekday.product.copy()
    moved[0] += 1
    assert snapshot(replace(weekday, product=moved)) != original
    shifted = weekday.page.copy()
    shifted[0] += 1
    assert snapshot(replace(weekday, page=shifted)) != original


def test_a_day_is_a_pure_function_of_the_seed_and_the_day():
    first = snapshot(day.stream(CANONICAL_SEED, 3))
    plan.cohort.cache_clear()
    assert snapshot(day.stream(CANONICAL_SEED, 3)) == first


def test_a_day_generated_alone_is_the_same_day():
    """День D не зависит от того, прожиты ли дни до него."""
    alone = snapshot(day.stream(CANONICAL_SEED, 3))
    plan.cohort.cache_clear()
    for earlier in range(3):
        day.stream(CANONICAL_SEED, earlier)
    assert snapshot(day.stream(CANONICAL_SEED, 3)) == alone


def test_the_next_day_does_not_move_the_days_before_it():
    before = [snapshot(day.stream(CANONICAL_SEED, number)) for number in range(2)]
    day.stream(CANONICAL_SEED, 2)
    assert [snapshot(day.stream(CANONICAL_SEED, n)) for n in range(2)] == before


def test_another_seed_is_another_day():
    ours = snapshot(day.stream(CANONICAL_SEED, 1))
    assert snapshot(day.stream(CANONICAL_SEED + 1, 1)) != ours


def test_events_do_not_start_before_the_origin():
    with pytest.raises(ValueError):
        day.stream(CANONICAL_SEED, -1)


def test_every_column_of_the_contract_is_present_and_typed(weekday: day.Day):
    """Состав колонок решает контракт схемы; день обязан отдать их все."""
    assert list(weekday.columns) == [column.name for column in schema.COLUMNS]
    for column in schema.COLUMNS:
        values = weekday.columns[column.name]
        assert values.size == len(weekday), column.name
        if column.clickhouse_type.startswith("Array("):
            cells = {cell.dtype for cell in values[:1000]}
            assert cells == {np.dtype(column.numpy_dtype)}, column.name
        else:
            assert values.dtype == np.dtype(column.numpy_dtype), column.name


def test_a_pageview_carries_the_trade_columns_empty_not_missing(weekday: day.Day):
    """Пусто — пустой массив и пустая строка, а ключ есть у каждого события.

    Проверяются все торговые колонки, а не выбранные: у просмотра страницы
    пусты они все до одной, иначе строгий приём хранилища не уживётся с
    полями, пустыми по смыслу (мастер-спека, раздел 6).
    """
    pageview = weekday.columns["EventType"] == "pageview"
    assert pageview.mean() > 0.9
    trade = [
        column
        for column in schema.COLUMNS
        if column.group in (schema.ColumnGroup.ECOMMERCE, schema.ColumnGroup.PARAMS)
    ]
    assert len(trade) > 10
    for column in trade:
        cells = weekday.columns[column.name][pageview][:1000]
        if column.clickhouse_type.startswith("Array("):
            assert all(cell.size == 0 for cell in cells), column.name
        else:
            assert set(cells) == {""}, column.name


def test_the_taxonomy_is_three_event_types(weekday: day.Day):
    """`EventType` — добавка стенда: таксономия нужна явно (мастер-спека, 1.1)."""
    assert set(weekday.columns["EventType"].tolist()) == {
        "pageview",
        commerce.ADD_TO_CART,
        commerce.PURCHASE,
    }


def test_no_column_hides_a_hole(weekday: day.Day):
    """`None` в колонке — та же пропажа ключа, только позже и незаметнее."""
    for column in schema.COLUMNS:
        values = weekday.columns[column.name]
        if values.dtype != object:
            continue
        kind = np.ndarray if column.clickhouse_type.startswith("Array(") else str
        assert all(isinstance(cell, kind) for cell in values[:1000]), column.name


def test_identifiers_survive_json(weekday: day.Day):
    """Числа выше 2^53 в JSON округляются — идентификаторам столько не нужно."""
    for name in ("WatchID", "VisitID", "ClientID"):
        assert weekday.columns[name].max() < ids.LIMIT
        assert weekday.columns[name].dtype == np.uint64


def test_the_event_id_is_unique_because_it_deduplicates(weekday: day.Day):
    """`WatchID` — ключ склейки при переигровке дня (спека, раздел 4)."""
    watch = weekday.columns["WatchID"]
    assert len(set(watch.tolist())) == watch.size


def test_a_visit_belongs_to_one_cookie(weekday: day.Day):
    visit = weekday.columns["VisitID"]
    cookie = weekday.columns["ClientID"]
    pairs = set(zip(visit.tolist(), cookie.tolist(), strict=True))
    assert len(pairs) == len(set(visit.tolist()))


def test_the_visit_cutting_rules_rebuild_the_generator_visits(weekday: day.Day):
    """Правило лабы: та же кука, пауза не длиннее таймаута — тот же визит.

    Сборка сессий по правилам из докстринга модуля обязана совпасть с
    `VisitID` — на этой сверке стоит лаба сессий.
    """
    second = local_seconds(weekday)
    cookie = weekday.columns["ClientID"]
    order = np.lexsort((second, cookie))
    cookie, second = cookie[order], second[order]
    visit = weekday.columns["VisitID"][order]

    started = np.ones(visit.size, dtype=bool)
    started[1:] = (cookie[1:] != cookie[:-1]) | (
        second[1:] - second[:-1] > world.VISIT_TIMEOUT_SECONDS
    )
    rebuilt = np.cumsum(started)
    # Совпадение обоюдное: сколько пар «собранный визит — `VisitID`», столько
    # же и тех, и других. Одного равенства мало — оно ловит только склейку
    # двух визитов в один, а разрыв одного визита надвое проходит мимо.
    matched = set(zip(rebuilt.tolist(), visit.tolist(), strict=True))
    assert len(matched) == int(rebuilt.max())
    assert len(matched) == len(set(visit.tolist()))


def test_the_day_boundary_cuts_the_visits(weekday: day.Day):
    """Событий за границей модельных суток в дне нет — партиция дня целая."""
    second = local_seconds(weekday)
    assert second.min() >= 0
    assert second.max() < day.DAY_SECONDS
    date = np.datetime64(world.ORIGIN, "D") + np.timedelta64(weekday.day, "D")
    assert set(weekday.columns["EventDate"].tolist()) == {date.astype("O")}


def test_the_counter_timezone_moves_the_date_apart_from_utc(weekday: day.Day):
    """`toDate(UTCEventTime)` ≠ `EventDate`: сутки считаются в поясе счётчика."""
    utc_date = weekday.columns["UTCEventTime"].astype("datetime64[D]")
    assert np.any(utc_date != weekday.columns["EventDate"])


def test_the_scale_of_an_average_day_is_about_fifty_thousand(weekday: day.Day):
    """Порядок величины: ~50 тыс. событий в средний день (спека, раздел 5)."""
    assert 30_000 < len(weekday) < 70_000
    visits = len(set(weekday.columns["VisitID"].tolist()))
    assert 6_000 < visits < 14_000


def test_the_wave_dips_at_night_and_peaks_in_the_evening(weekday: day.Day):
    counters = hourly(weekday)
    average = counters.mean()
    assert counters[2:6].max() < average / 2
    assert counters[18:22].max() > 1.5 * average
    assert 0 <= int(counters.argmin()) <= 6
    assert 11 <= int(counters.argmax()) <= 22


def test_the_weekend_is_shaped_unlike_a_weekday(weekday: day.Day):
    """Форма, а не объём: выходной раскачивается позже буднего."""
    weekend = hourly(day.stream(CANONICAL_SEED, WEEKEND))
    workday = hourly(weekday)
    morning = slice(7, 10)
    assert weekend[morning].sum() / weekend.sum() < (
        workday[morning].sum() / workday.sum()
    )


def test_the_funnel_converts_about_two_percent_of_visits(weekday: day.Day):
    visits = len(set(weekday.columns["VisitID"].tolist()))
    ordered = int((weekday.columns["EventType"] == commerce.PURCHASE).sum())
    assert 0.01 < ordered / visits < 0.04


def test_the_promised_orders_reach_the_confirmation(weekday: day.Day):
    """Гарантия двухкуковых пар: назначенный планом заказ обязан состояться."""
    audience = plan.audience(CANONICAL_SEED, weekday.day)
    promised = set(audience.client_id[audience.assigned_order].tolist())
    assert promised
    confirmed = weekday.columns["ClientID"][weekday.page == Page.CONFIRMATION]
    assert promised <= set(confirmed.tolist())


def test_the_cart_always_follows_a_product_card(weekday: day.Day):
    """Шов с торговлей: товар в корзине посетитель до того открывал."""
    order = np.lexsort((local_seconds(weekday), weekday.columns["VisitID"]))
    by_visit = weekday.page[order]
    carts = np.flatnonzero(by_visit == Page.CART)
    assert carts.size > 0
    assert np.all(by_visit[carts - 1] == Page.PRODUCT)


def test_the_product_of_a_card_is_the_seam_for_trade_events(weekday: day.Day):
    """`product` — товар карточки и −1 у прочих страниц; больше ничего."""
    goods = catalog.catalog()
    card = weekday.page == Page.PRODUCT
    assert np.all(weekday.product[~card] == -1)
    assert np.all(weekday.product[card] >= 0)
    assert np.all(weekday.product[card] < goods.sku.size)
    shown = zip(weekday.columns["URL"][card], weekday.product[card], strict=True)
    for url, number in list(shown)[:200]:
        # У страницы входа в адресе ещё метки перехода — путь до знака «?».
        assert url.split("?")[0].endswith(f"/product/{goods.sku[number]}")


def test_the_referer_is_the_page_before(weekday: day.Day):
    """Внутри визита реферер — предыдущий адрес; на входе — адрес источника.

    Цепочка считается по просмотрам страниц: торговое событие не открывает
    страницу, а живёт на уже открытой, и реферер у него тот же, что у неё.
    """
    pageview = weekday.columns["EventType"] == "pageview"
    columns = {name: value[pageview] for name, value in weekday.columns.items()}
    order = np.lexsort((columns["UTCEventTime"], columns["VisitID"]))
    url = columns["URL"][order]
    referer = columns["Referer"][order]
    visit = columns["VisitID"][order]
    inside = np.flatnonzero(visit[1:] == visit[:-1]) + 1
    assert np.all(referer[inside] == url[inside - 1])

    entry = np.flatnonzero(visit[1:] != visit[:-1]) + 1
    known = {source.referer for source in reference.TRAFFIC_SOURCES}
    assert set(referer[entry].tolist()) <= known


def test_the_passport_of_a_cookie_does_not_change_between_days():
    """Кука — браузер на устройстве: во всех её днях профиль и город одни."""
    passports: dict[int, tuple[Any, ...]] = {}
    for number in (0, 1, 2):
        events = day.stream(CANONICAL_SEED, number)
        seen = zip(
            events.columns["ClientID"].tolist(),
            events.columns["Browser"].tolist(),
            events.columns["ScreenWidth"].tolist(),
            events.columns["RegionCity"].tolist(),
            strict=True,
        )
        for cookie, *passport in seen:
            assert passports.setdefault(cookie, tuple(passport)) == tuple(passport)


def test_geography_is_one_region_of_presence(weekday: day.Day):
    """Магазин с одним складом: свой миллионник, свой регион, тонкий хвост."""
    cities = weekday.columns["RegionCity"]
    assert (cities == "Samara").mean() > 0.25
    assert set(weekday.columns["RegionCountry"].tolist()) == {reference.COUNTRY_NAME}
    assert set(weekday.columns["RegionCountryID"].tolist()) == {
        reference.COUNTRY_REGION_ID
    }


def test_addresses_are_not_routable(weekday: day.Day):
    """Правдоподобные публичные адреса принадлежат живым организациям."""
    fixed = ("192.0.2.", "198.51.100.", "203.0.113.", "198.18.")

    def cgnat(address: str) -> bool:
        first, second, *_ = (int(byte) for byte in address.split("."))
        return first == reference.MOBILE_IP_FIRST_BYTE and 64 <= second < 128

    for address in set(weekday.columns["IPAddress"].tolist()):
        assert address.startswith(fixed) or cgnat(address), address


def test_phones_carry_a_model_and_desktops_do_not(weekday: day.Day):
    phone = weekday.columns["DeviceCategory"] == reference.PHONE_CATEGORY
    assert phone.mean() > 0.5
    assert np.all(weekday.columns["MobilePhoneModel"][~phone] == "")
    assert np.all(weekday.columns["MobilePhoneModel"][phone] != "")


def test_randomness_is_drawn_in_whole_numbers():
    """Дисциплина спеки (раздел 2): целые числа, никакой системной математики."""
    source = inspect.getsource(day)
    assert set(re.findall(r"\brng\.(\w+)", source)) <= {"integers"}
    assert not re.search(r"^\s*import (random|math)\b", source, re.MULTILINE)