SHA256
292 lines
10 KiB
Python
292 lines
10 KiB
Python
import os
|
|
import sys
|
|
import shutil
|
|
|
|
# Use an isolated DB per test run
|
|
TEST_DB = "instance/test_integrity.sqlite"
|
|
|
|
os.environ["WHO2_DB_PATH"] = TEST_DB
|
|
|
|
import backend.storage as storage
|
|
|
|
|
|
def _clean():
|
|
if os.path.exists(TEST_DB):
|
|
os.remove(TEST_DB)
|
|
if os.path.exists(TEST_DB + "-journal"):
|
|
os.remove(TEST_DB + "-journal")
|
|
if os.path.exists(TEST_DB + "-wal"):
|
|
os.remove(TEST_DB + "-wal")
|
|
if os.path.exists(TEST_DB + "-shm"):
|
|
os.remove(TEST_DB + "-shm")
|
|
|
|
|
|
def _reset():
|
|
_clean()
|
|
storage.init_db()
|
|
|
|
|
|
def test_create_and_authenticate():
|
|
_reset()
|
|
email = "alice@example.com"
|
|
password = "Sup3rS3cret"
|
|
acc = storage.create_account(email, password)
|
|
assert acc["email"] == email
|
|
# first account becomes the bootstrap admin
|
|
assert acc["is_admin"] is True
|
|
assert storage.authenticate(email, password) is not None
|
|
assert storage.authenticate(email, "WRONG") is None
|
|
assert storage.authenticate("noone@example.com", password) is None
|
|
|
|
|
|
def test_only_admins_create_persons():
|
|
_reset()
|
|
admin = storage.create_account("admin@x.com", "pw")
|
|
user = storage.create_account("user@x.com", "pw")
|
|
assert storage.is_admin(admin)
|
|
assert not storage.is_admin(user)
|
|
|
|
try:
|
|
storage.create_person(user, "Nope")
|
|
assert False, "expected PermissionError"
|
|
except PermissionError:
|
|
pass
|
|
|
|
# admin creates a person and gives it the user's email address
|
|
person = storage.create_person_with_account(admin, "User Person", "user@x.com")
|
|
refreshed = storage.get_account_by_id(user["id"])
|
|
assert refreshed["person_id"] == person["id"]
|
|
assert person["emails"] == ["user@x.com"]
|
|
|
|
# a non-admin cannot link or unlink emails either
|
|
try:
|
|
storage.attach_email_to_person(user, person["id"], "admin@x.com")
|
|
assert False, "expected PermissionError"
|
|
except PermissionError:
|
|
pass
|
|
|
|
|
|
def test_person_account_linking():
|
|
_reset()
|
|
admin = storage.create_account("admin@x.com", "pw")
|
|
# separate account so linking can be exercised explicitly
|
|
a = storage.create_account("a@x.com", "pw")
|
|
p = storage.create_person(admin, "Alice")
|
|
storage.link_account_to_person(admin, p["id"], a["id"])
|
|
person = storage.get_person(p["id"])
|
|
assert person["emails"] == ["a@x.com"]
|
|
|
|
# one account can only ever be linked to one person
|
|
p2 = storage.create_person(admin, "Bob")
|
|
try:
|
|
storage.attach_email_to_person(admin, p2["id"], "a@x.com")
|
|
assert False, "expected ValueError"
|
|
except ValueError:
|
|
pass
|
|
|
|
storage.attach_email_to_person(admin, p2["id"], "admin@x.com")
|
|
assert "admin@x.com" in storage.get_person(p2["id"])["emails"]
|
|
# detaching works and is admin-only
|
|
storage.detach_email_from_person(admin, p2["id"], "admin@x.com")
|
|
assert storage.get_person(p2["id"])["emails"] == []
|
|
try:
|
|
storage.detach_email_from_person(a, p["id"], "a@x.com")
|
|
assert False, "expected PermissionError"
|
|
except PermissionError:
|
|
pass
|
|
|
|
|
|
def test_article_visibility():
|
|
_reset()
|
|
owner = storage.create_account("owner@x.com", "pw")
|
|
storage.bootstrap_admin(owner["id"])
|
|
owner["rights"] = list(storage.RIGHTS.keys())
|
|
admin_person = storage.create_person(owner, "Owner")
|
|
owner["person_id"] = admin_person["id"]
|
|
other = storage.create_account("other@x.com", "pw")
|
|
storage.grant_right(owner, other["id"], "view_private")
|
|
other["person_id"] = storage.create_person_with_account(owner, "Other", "other@x.com")["id"]
|
|
other["rights"] = ["view_private"]
|
|
|
|
storage.create_article(
|
|
owner, owner["person_id"], "notes", "Secret",
|
|
"hidden content", visibility="private",
|
|
)
|
|
articles = storage.list_articles(other, owner["person_id"])
|
|
assert len(articles) == 0 or articles[0].get("content") == ""
|
|
storage.grant_right(owner, owner["id"], "view_private")
|
|
articles = storage.list_articles(owner, owner["person_id"])
|
|
assert any(a["title"] == "Secret" and a.get("content") == "hidden content" for a in articles)
|
|
|
|
|
|
def test_group_based_visibility():
|
|
_reset()
|
|
owner = storage.create_account("owner@x.com", "pw")
|
|
storage.bootstrap_admin(owner["id"])
|
|
owner["rights"] = list(storage.RIGHTS.keys())
|
|
owner["person_id"] = storage.create_person(owner, "Owner")["id"]
|
|
reader = storage.create_account("reader@x.com", "pw")
|
|
storage.grant_right(owner, reader["id"], "view_private")
|
|
reader["person_id"] = storage.create_person_with_account(owner, "Reader", "reader@x.com")["id"]
|
|
reader["rights"] = ["view_private"]
|
|
|
|
group = storage.create_group(owner, owner["person_id"], "Team")
|
|
storage.add_group_member(owner, group["id"], reader["person_id"])
|
|
|
|
storage.create_article(
|
|
owner, owner["person_id"], "notes", "Shared",
|
|
"shared content", visibility="group",
|
|
group_ids=[group["id"]],
|
|
)
|
|
noaccess = storage.list_articles(None, owner["person_id"])
|
|
assert not any(a["title"] == "Shared" and a.get("content") == "shared content" for a in noaccess)
|
|
|
|
view = storage.list_articles(reader, owner["person_id"])
|
|
assert any(a["title"] == "Shared" and a.get("content") == "shared content" for a in view)
|
|
|
|
|
|
def test_normal_user_only_notes_and_own_groups():
|
|
_reset()
|
|
admin = storage.create_account("admin@x.com", "pw")
|
|
me = storage.create_person_with_account(admin, "Me", "me@x.com", "pw")
|
|
friend = storage.create_person_with_account(admin, "Friend", "friend@x.com", "pw")
|
|
user = storage.authenticate("me@x.com", "pw")
|
|
assert user["person_id"] == me["id"]
|
|
assert not user["is_admin"]
|
|
|
|
# a normal user may add notes to their own person only
|
|
storage.create_article(user, me["id"], "notes", "Note", "hello", visibility="private")
|
|
try:
|
|
storage.create_article(user, me["id"], "links", "Link", "http://x")
|
|
assert False, "expected PermissionError"
|
|
except PermissionError:
|
|
pass
|
|
try:
|
|
storage.create_article(user, friend["id"], "notes", "Peek", "nope")
|
|
assert False, "expected PermissionError"
|
|
except PermissionError:
|
|
pass
|
|
|
|
# own group + invite a friend into it
|
|
g = storage.create_group(user, me["id"], "My Team")
|
|
storage.add_group_member(user, g["id"], friend["id"])
|
|
assert any(m["id"] == friend["id"] for m in storage.get_group(g["id"])["members"])
|
|
|
|
friend_user = storage.authenticate("friend@x.com", "pw")
|
|
admin_person = storage.create_person(admin, "Admin")
|
|
foreign = storage.create_group(admin, admin_person["id"], "Admins")
|
|
|
|
# the friend is a member, so they see the group but not foreign groups
|
|
assert {x["id"] for x in storage.list_groups(friend_user)} == {g["id"]}
|
|
assert foreign["id"] not in {x["id"] for x in storage.list_groups(friend_user)}
|
|
assert not storage.group_is_visible_to(friend_user, foreign["id"])
|
|
|
|
# a member may invite, a bystander may not
|
|
try:
|
|
storage.add_group_member(friend_user, foreign["id"], friend_user["person_id"])
|
|
assert False, "expected PermissionError"
|
|
except PermissionError:
|
|
pass
|
|
|
|
# only admins may read the rights list
|
|
try:
|
|
storage.list_rights(user)
|
|
assert False, "expected PermissionError"
|
|
except PermissionError:
|
|
pass
|
|
|
|
|
|
def test_ensure_admin_promotes_oldest_account():
|
|
_reset()
|
|
storage.create_account("first@x.com", "pw")
|
|
storage.create_account("second@x.com", "pw")
|
|
conn = storage._conn()
|
|
conn.execute("DELETE FROM account_rights") # simulate a database from before roles existed
|
|
conn.commit()
|
|
conn.close()
|
|
assert storage.authenticate("first@x.com", "pw")["is_admin"] is False
|
|
storage.ensure_admin()
|
|
assert storage.authenticate("first@x.com", "pw")["is_admin"] is True
|
|
assert storage.authenticate("second@x.com", "pw")["is_admin"] is False
|
|
|
|
|
|
def test_article_type_constraint():
|
|
_reset()
|
|
acc = storage.create_account("x@x.com", "pw")
|
|
acc["person_id"] = storage.create_person(acc, "X")["id"]
|
|
try:
|
|
storage.create_article(acc, acc["person_id"], "bogus", "T", "C")
|
|
assert False, "expected ValueError"
|
|
except ValueError as e:
|
|
assert "Invalid article type" in str(e)
|
|
|
|
|
|
def test_right_management():
|
|
_reset()
|
|
admin = storage.create_account("admin@x.com", "pw")
|
|
storage.bootstrap_admin(admin["id"])
|
|
admin["rights"] = list(storage.RIGHTS.keys())
|
|
admin["person_id"] = storage.create_person(admin, "Admin")["id"]
|
|
storage.grant_right(admin, admin["id"], "view_private")
|
|
storage.grant_right(admin, admin["id"], "manage_articles")
|
|
rights = storage.list_rights(admin, admin["id"])
|
|
assert any(r["id"] == admin["id"] and "view_private" in r["rights"] for r in rights)
|
|
storage.revoke_right(admin, admin["id"], "manage_articles")
|
|
rights = storage.list_rights(admin)
|
|
assert any(r["id"] == admin["id"] and "view_private" in r["rights"] for r in rights)
|
|
assert not any(r["id"] == admin["id"] and "manage_articles" in r["rights"] for r in rights)
|
|
|
|
|
|
def test_delete_person_cascades():
|
|
_reset()
|
|
acc = storage.create_account("x@x.com", "pw")
|
|
acc["person_id"] = storage.create_person(acc, "X")["id"]
|
|
g = storage.create_group(acc, acc["person_id"], "G")
|
|
storage.add_group_member(acc, g["id"], acc["person_id"])
|
|
storage.create_article(acc, acc["person_id"], "notes", "T", "C")
|
|
ok = storage.delete_person(acc, acc["person_id"])
|
|
assert ok
|
|
assert storage.get_person(acc["person_id"]) is None
|
|
assert storage.get_group(g["id"]) is None
|
|
|
|
|
|
def run_all():
|
|
tests = [
|
|
test_create_and_authenticate,
|
|
test_only_admins_create_persons,
|
|
test_person_account_linking,
|
|
test_article_visibility,
|
|
test_group_based_visibility,
|
|
test_normal_user_only_notes_and_own_groups,
|
|
test_ensure_admin_promotes_oldest_account,
|
|
test_article_type_constraint,
|
|
test_right_management,
|
|
test_delete_person_cascades,
|
|
]
|
|
passed = 0
|
|
failed = 0
|
|
last_err = None
|
|
for t in tests:
|
|
try:
|
|
t()
|
|
passed += 1
|
|
print("PASS", t.__name__)
|
|
except AssertionError as e:
|
|
failed += 1
|
|
last_err = e
|
|
print("FAIL", t.__name__, "-", str(e))
|
|
except Exception as e:
|
|
failed += 1
|
|
last_err = e
|
|
print("ERROR", t.__name__, "-", str(e))
|
|
_clean()
|
|
if failed:
|
|
print(f"\n{passed} passed, {failed} failed")
|
|
raise SystemExit(1)
|
|
print(f"\nAll {passed} tests passed")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(run_all())
|