devplacepy/tests/unit/services/presence.py

# retoor <retoor@molodetz.nl>

from datetime import datetime, timedelta, timezone

import uuid_utils

from devplacepy.config import PRESENCE_TIMEOUT_SECONDS, PRESENCE_WRITE_SECONDS
from devplacepy.database import get_table
from devplacepy.services import presence


def _iso(seconds_ago):
    return (datetime.now(timezone.utc) - timedelta(seconds=seconds_ago)).isoformat()


def test_is_online_recent():
    assert presence.is_online({"last_seen": _iso(5)}) is True


def test_is_online_stale():
    assert presence.is_online({"last_seen": _iso(PRESENCE_TIMEOUT_SECONDS + 30)}) is False


def test_is_online_missing_and_empty():
    assert presence.is_online({"last_seen": ""}) is False
    assert presence.is_online({}) is False
    assert presence.is_online(None) is False


def test_is_online_unparseable_is_offline():
    assert presence.is_online({"last_seen": "not-a-timestamp"}) is False


def test_is_online_is_the_first_observation_of_stays_online():
    from devplacepy.config import PRESENCE_ONLINE_MARGIN_SECONDS

    for seconds in (0, 5, PRESENCE_TIMEOUT_SECONDS - 1, PRESENCE_TIMEOUT_SECONDS + 1,
                    PRESENCE_TIMEOUT_SECONDS + PRESENCE_ONLINE_MARGIN_SECONDS + 1):
        row = {"last_seen": _iso(seconds)}
        expected = presence.stays_online(
            presence.seconds_since(row["last_seen"]), was_online=False
        )
        assert presence.is_online(row) is expected


def test_seconds_since_none_on_missing_or_bad():
    assert presence.seconds_since(None) is None
    assert presence.seconds_since("") is None
    assert presence.seconds_since("garbage") is None
    assert presence.seconds_since(_iso(10)) >= 10


def test_touch_writes_once_then_throttles(local_db):
    uid = str(uuid_utils.uuid7())
    users = get_table("users")
    users.insert(
        {
            "uid": uid,
            "username": f"presence_{uid[:8]}",
            "email": f"{uid[:8]}@presence.test",
            "last_seen": None,
        }
    )
    presence._last_write.pop(uid, None)

    presence.touch(uid)
    first = users.find_one(uid=uid)["last_seen"]
    assert first

    presence.touch(uid)
    assert users.find_one(uid=uid)["last_seen"] == first

    presence._last_write[uid] = presence.time.monotonic() - PRESENCE_WRITE_SECONDS - 1
    presence.touch(uid)
    second = users.find_one(uid=uid)["last_seen"]
    assert second >= first
    assert presence.is_online(users.find_one(uid=uid)) is True


def test_get_online_users_filters_by_cutoff(local_db):
    from devplacepy.database import get_online_users

    now = datetime.now(timezone.utc)
    fresh = str(uuid_utils.uuid7())
    stale = str(uuid_utils.uuid7())
    users = get_table("users")
    users.insert(
        {"uid": fresh, "username": f"onl_{fresh[:8]}", "email": f"{fresh[:8]}@o.test", "last_seen": now.isoformat()}
    )
    users.insert(
        {"uid": stale, "username": f"onl_{stale[:8]}", "email": f"{stale[:8]}@o.test", "last_seen": (now - timedelta(seconds=99999)).isoformat()}
    )
    cutoff = (now - timedelta(seconds=60)).isoformat()
    online = {u["uid"] for u in get_online_users(cutoff, limit=1000)}
    assert fresh in online
    assert stale not in online


def test_presence_online_users_includes_recent(local_db):
    fresh = str(uuid_utils.uuid7())
    get_table("users").insert(
        {"uid": fresh, "username": f"po_{fresh[:8]}", "email": f"{fresh[:8]}@o.test", "last_seen": datetime.now(timezone.utc).isoformat()}
    )
    assert fresh in {u["uid"] for u in presence.online_users(limit=1000)}


def test_online_users_sorted_alphabetically(local_db):
    now = datetime.now(timezone.utc).isoformat()
    users = get_table("users")
    made = []
    for label in ("zeta", "alpha", "mike"):
        uid = str(uuid_utils.uuid7())
        name = f"srt{label}{uid[:6]}"
        users.insert({"uid": uid, "username": name, "email": f"{uid[:8]}@o.test", "last_seen": now})
        made.append(name)
    listed = [u["username"] for u in presence.online_users(limit=1000) if u["username"] in made]
    assert listed == sorted(made, key=str.lower)


def test_stays_online_hysteresis():
    from devplacepy.config import (
        PRESENCE_ONLINE_MARGIN_SECONDS,
        PRESENCE_TIMEOUT_SECONDS,
    )

    within = PRESENCE_TIMEOUT_SECONDS - 5
    band = PRESENCE_TIMEOUT_SECONDS + 5
    beyond = PRESENCE_TIMEOUT_SECONDS + PRESENCE_ONLINE_MARGIN_SECONDS + 5
    # fresh activity: online regardless of prior state
    assert presence.stays_online(within, was_online=False) is True
    assert presence.stays_online(within, was_online=True) is True
    # in the grace band: only an already-online user stays online (hysteresis)
    assert presence.stays_online(band, was_online=False) is False
    assert presence.stays_online(band, was_online=True) is True
    # beyond the grace: offline even if previously online
    assert presence.stays_online(beyond, was_online=True) is False
    # no last_seen: always offline
    assert presence.stays_online(None, was_online=True) is False


def test_online_candidates_uses_grace_window(local_db):
    from devplacepy.config import (
        PRESENCE_ONLINE_MARGIN_SECONDS,
        PRESENCE_TIMEOUT_SECONDS,
    )

    now = datetime.now(timezone.utc)
    band = str(uuid_utils.uuid7())
    beyond = str(uuid_utils.uuid7())
    users = get_table("users")
    users.insert(
        {"uid": band, "username": f"cb_{band[:8]}", "email": f"{band[:8]}@o.test",
         "last_seen": (now - timedelta(seconds=PRESENCE_TIMEOUT_SECONDS + 3)).isoformat()}
    )
    users.insert(
        {"uid": beyond, "username": f"cb_{beyond[:8]}", "email": f"{beyond[:8]}@o.test",
         "last_seen": (now - timedelta(seconds=PRESENCE_TIMEOUT_SECONDS + PRESENCE_ONLINE_MARGIN_SECONDS + 30)).isoformat()}
    )
    candidates = {u["uid"] for u in presence.online_candidates(limit=1000)}
    strict = {u["uid"] for u in presence.online_users(limit=1000)}
    # a grace-band user is a candidate (for hysteresis) but NOT strictly online
    assert band in candidates
    assert band not in strict
    # a user past the grace window is in neither
    assert beyond not in candidates


def test_online_candidates_tracks_beyond_the_display_limit(local_db):
    from devplacepy.config import PRESENCE_ONLINE_LIMIT, PRESENCE_TRACK_LIMIT

    # the display roster is capped for the feed panel, the tracked authority set is not
    assert PRESENCE_TRACK_LIMIT > PRESENCE_ONLINE_LIMIT
    now = datetime.now(timezone.utc).isoformat()
    users = get_table("users")
    made = []
    for index in range(PRESENCE_ONLINE_LIMIT + 5):
        uid = str(uuid_utils.uuid7())
        users.insert(
            {"uid": uid, "username": f"trk{index:03d}{uid[:6]}",
             "email": f"{uid[:8]}@o.test", "last_seen": now}
        )
        made.append(uid)
    tracked = {u["uid"] for u in presence.online_candidates()}
    assert set(made) <= tracked