|
# 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_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
|