peter
ad6ccbac05
Tokens, notifications, quota, debug sessions and the reset-link burn filtered their owner inline. Ten of the fifteen filters had no test that noticed one going missing; each now has a test where another account's rows must not count, show or change.
136 lines
4.9 KiB
Python
136 lines
4.9 KiB
Python
"""The core's owner filters, each held by a second account whose rows must not count, show or change.
|
|
|
|
Every one of these reads or writes a user-owned table through `userscope.owned_by`; the fence holds
|
|
the form (no inline `Model.user_id == …`), these hold that the filter is there at all. Each test gives
|
|
a second account the same kind of rows, so a filter that went missing answers with theirs. DB lane.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import func, select
|
|
|
|
from app import debugsession, notifications, quota
|
|
from app.models import DebugScreenshot, DebugSession, Notification, QuotaEvent
|
|
from app.routes import tokens as tokens_routes
|
|
from app.routes.auth import RESET_TTL, _issue_token
|
|
|
|
|
|
def _notify(db, user, n=1, *, read=False):
|
|
for i in range(n):
|
|
db.add(Notification(user_id=user.id, type="t", title=f"n{i}", read=read))
|
|
db.commit()
|
|
|
|
|
|
def _visible(db, user) -> int:
|
|
return db.scalar(
|
|
select(func.count())
|
|
.select_from(Notification)
|
|
.where(Notification.user_id == user.id, Notification.dismissed.is_(False))
|
|
)
|
|
|
|
|
|
def test_the_notification_list_shows_only_mine(db, make_user, as_user):
|
|
me, other = make_user(), make_user()
|
|
_notify(db, me)
|
|
_notify(db, other, 2)
|
|
body = as_user(me).get("/api/me/notifications").json()
|
|
assert (body["total"], len(body["items"])) == (1, 1)
|
|
|
|
|
|
def test_the_unread_count_counts_only_mine(db, make_user, as_user):
|
|
me, other = make_user(), make_user()
|
|
_notify(db, me)
|
|
_notify(db, other, 2)
|
|
assert as_user(me).get("/api/me/notifications/unread_count").json() == {"count": 1}
|
|
|
|
|
|
def test_reading_all_reads_only_mine(db, make_user, as_user):
|
|
me, other = make_user(), make_user()
|
|
_notify(db, me)
|
|
_notify(db, other, 2)
|
|
assert as_user(me).post("/api/me/notifications/read_all").status_code == 200
|
|
assert as_user(other).get("/api/me/notifications/unread_count").json() == {"count": 2}
|
|
|
|
|
|
def test_clearing_all_dismisses_only_mine(db, make_user, as_user):
|
|
me, other = make_user(), make_user()
|
|
_notify(db, me)
|
|
_notify(db, other, 2)
|
|
assert as_user(me).post("/api/me/notifications/clear").status_code == 200
|
|
db.expire_all()
|
|
assert (_visible(db, me), _visible(db, other)) == (0, 2)
|
|
|
|
|
|
def test_trimming_read_notifications_trims_only_mine(db, make_user):
|
|
me, other = make_user(), make_user()
|
|
_notify(db, me, 3, read=True)
|
|
_notify(db, other, 3, read=True)
|
|
assert notifications.trim_read(db, me.id, cap=1) == 2
|
|
db.commit()
|
|
assert _visible(db, other) == 3
|
|
|
|
|
|
def test_todays_actions_count_only_mine(db, make_user):
|
|
me, other = make_user(), make_user()
|
|
quota.log_action(me.id, quota.QuotaAction.VIDEOS_SEARCH)
|
|
quota.log_action(other.id, quota.QuotaAction.VIDEOS_SEARCH)
|
|
quota.log_action(other.id, quota.QuotaAction.VIDEOS_SEARCH)
|
|
assert quota.actions_today(db, me.id, quota.QuotaAction.VIDEOS_SEARCH) == 1
|
|
|
|
|
|
def test_my_usage_sums_only_mine(db, make_user, as_user):
|
|
me, other = make_user(), make_user()
|
|
db.add_all(
|
|
[
|
|
QuotaEvent(user_id=me.id, action="videos.list", units=5),
|
|
QuotaEvent(user_id=other.id, action="videos.list", units=7),
|
|
]
|
|
)
|
|
db.commit()
|
|
body = as_user(me).get("/api/quota/my-usage").json()
|
|
assert (body["today"], body["all_time"], body["by_action"]) == (5, 5, {"videos.list": 5})
|
|
|
|
|
|
def test_the_token_ceiling_counts_only_mine(db, make_user, as_user, monkeypatch):
|
|
monkeypatch.setattr(tokens_routes, "MAX_TOKENS", 1)
|
|
me, other = make_user(), make_user()
|
|
first = as_user(other).post("/api/tokens", json={"name": "theirs", "preset": "hub"})
|
|
assert first.status_code == 200, first.text
|
|
r = as_user(me).post("/api/tokens", json={"name": "mine", "preset": "hub"})
|
|
assert r.status_code == 200, r.text
|
|
|
|
|
|
def test_a_reset_burns_only_my_other_links(client, db, make_user):
|
|
me, other = make_user(), make_user()
|
|
mine = _issue_token(db, me, "reset", RESET_TTL)
|
|
theirs = _issue_token(db, other, "reset", RESET_TTL)
|
|
db.commit()
|
|
r = client.post(
|
|
"/auth/password-reset/confirm", json={"token": mine, "password": "a brand new passphrase"}
|
|
)
|
|
assert r.json() == {"ok": True}
|
|
assert client.post("/auth/password-reset/check", json={"token": theirs}).json() == {"ok": True}
|
|
|
|
|
|
def test_screenshot_storage_counts_only_mine(db, make_user):
|
|
me, other = make_user(), make_user()
|
|
now = datetime.now(timezone.utc)
|
|
for i, (user, size) in enumerate([(me, 10), (other, 1000)]):
|
|
db.add(
|
|
DebugSession(
|
|
id=f"s{i}",
|
|
user_id=user.id,
|
|
level="DEBUG",
|
|
started_at=now,
|
|
until=now + timedelta(hours=1),
|
|
)
|
|
)
|
|
db.flush()
|
|
db.add(
|
|
DebugScreenshot(
|
|
session_id=f"s{i}", created_at=now, mime="image/png", size=size, data=b"x" * size
|
|
)
|
|
)
|
|
db.commit()
|
|
assert debugsession.stored_screenshot_bytes(db, me.id) == 10
|