test_main.py

import logging
from copy import deepcopy

from main import merge_user_events


logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
logger = logging.getLogger(__name__)


def test_session_split_and_sort():
    logger.info("test_session_split_and_sort: start")
    events = [
        {"user_id": "u1", "ts": 2201, "type": "click", "meta": {"page": "/late"}},
        {"user_id": "u1", "ts": 1000, "type": "click", "meta": {"page": "/"}},
        {"user_id": "u1", "ts": 1600, "type": "view", "meta": {"page": "/home"}},
        {"user_id": "u2", "ts": 1100, "type": "signup", "meta": {"plan": "pro"}},
    ]

    sessions = merge_user_events(events)

    assert [s["start_ts"] for s in sessions] == [1000, 1100, 2201]
    assert sessions[0]["user_id"] == "u1"
    assert sessions[1]["user_id"] == "u2"
    assert sessions[2]["user_id"] == "u1"
    assert sessions[0]["types"] == [
        {"type": "click", "count": 1},
        {"type": "view", "count": 1},
    ]
    assert sessions[2]["types"] == [{"type": "click", "count": 1}]
    logger.info("test_session_split_and_sort: done")


def test_meta_merge_and_immutability():
    logger.info("test_meta_merge_and_immutability: start")
    events = [
        {
            "user_id": "u1",
            "ts": 100,
            "type": "click",
            "meta": {"page": "/", "nested": {"a": 1}},
        },
        {
            "user_id": "u1",
            "ts": 200,
            "type": "click",
            "meta": {"page": "/home", "nested": {"b": 2}},
        },
        {
            "user_id": "u1",
            "ts": 300,
            "type": "view",
            "meta": {"nested": {"c": 3}},
        },
    ]

    original = deepcopy(events)
    sessions = merge_user_events(events)
    assert events == original, "input events were mutated"

    meta = sessions[0]["meta"]
    assert meta == {"page": "/", "nested": {"a": 1, "b": 2, "c": 3}}

    types = sessions[0]["types"]
    assert types == [
        {"type": "click", "count": 2},
        {"type": "view", "count": 1},
    ]
    logger.info("test_meta_merge_and_immutability: done")


if __name__ == "__main__":
    logger.info("running tests")
    test_session_split_and_sort()
    test_meta_merge_and_immutability()
    logger.info("tests passed")