forked from mttu-developers/konabot
106 lines
4.7 KiB
Python
106 lines
4.7 KiB
Python
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
import pytest
|
|
|
|
from konabot.common.database import DatabaseManager
|
|
from konabot.common.permsys import PermManager
|
|
from konabot.common.permsys.entity import PermEntity
|
|
from konabot.common.permsys.migrates import execute_migration, get_current_version
|
|
from konabot.common.permsys.repo import PermRepo
|
|
|
|
|
|
@asynccontextmanager
|
|
async def tempdb():
|
|
with TemporaryDirectory() as _tempdir:
|
|
tempdir = Path(_tempdir)
|
|
db = DatabaseManager(tempdir / "perm.sqlite3")
|
|
yield db
|
|
await db.close_all_connections()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_db_version():
|
|
async with tempdb() as db:
|
|
async with db.get_conn() as conn:
|
|
v = await get_current_version(conn)
|
|
assert v == 0
|
|
v = await get_current_version(conn)
|
|
assert v == 0
|
|
await execute_migration(conn, version=1)
|
|
v = await get_current_version(conn)
|
|
assert v == 1
|
|
await execute_migration(conn, version=0)
|
|
v = await get_current_version(conn)
|
|
assert v == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_perm():
|
|
async with tempdb() as db:
|
|
async with db.get_conn() as conn:
|
|
await execute_migration(conn)
|
|
|
|
service = PermManager(db)
|
|
entity_global = PermEntity("sys", "global", "global")
|
|
entity1 = PermEntity("nonexist-platform", "user", "passthem")
|
|
chain1 = [entity1, entity_global]
|
|
entity2 = PermEntity("nonexist-platform", "user", "jack")
|
|
chain2 = [entity2, entity_global]
|
|
|
|
async with db.get_conn() as conn:
|
|
repo = PermRepo(conn)
|
|
|
|
# 测试使用内置方法会创建 Entity 在数据库
|
|
assert await repo._get_entity_id_or_none(entity1) is None
|
|
assert await repo.get_entity_id(entity1) is not None
|
|
assert await repo._get_entity_id_or_none(entity1) is not None
|
|
|
|
# 测试使用内置方法获得 perm_info
|
|
assert await repo.get_perm_info(entity1, "module1") is None
|
|
|
|
assert not await service.check_has_permission(chain1, "*")
|
|
|
|
await service.update_permission(entity1, "*", True)
|
|
assert await service.check_has_permission(chain1, "*")
|
|
assert await service.check_has_permission(chain1, "module1")
|
|
assert await service.check_has_permission(chain1, "module1.pack1")
|
|
assert not await service.check_has_permission(chain2, "*")
|
|
assert not await service.check_has_permission(chain2, "module1")
|
|
assert not await service.check_has_permission(chain2, "module1.pack1")
|
|
|
|
await service.update_permission(entity2, "module1", True)
|
|
assert not await service.check_has_permission(chain2, "*")
|
|
assert await service.check_has_permission(chain2, "module1")
|
|
assert await service.check_has_permission(chain2, "module1.pack1")
|
|
assert await service.check_has_permission(chain2, "module1.pack2")
|
|
assert not await service.check_has_permission(chain2, "module2")
|
|
assert not await service.check_has_permission(chain2, "module2.pack1")
|
|
assert not await service.check_has_permission(chain2, "module2.pack2")
|
|
|
|
await service.update_permission(entity2, "module1.pack2", False)
|
|
assert not await service.check_has_permission(chain2, "*")
|
|
assert await service.check_has_permission(chain2, "module1")
|
|
assert await service.check_has_permission(chain2, "module1.pack1")
|
|
assert not await service.check_has_permission(chain2, "module1.pack2")
|
|
assert not await service.check_has_permission(chain2, "module2")
|
|
assert not await service.check_has_permission(chain2, "module2.pack1")
|
|
assert not await service.check_has_permission(chain2, "module2.pack2")
|
|
|
|
await service.update_permission(entity_global, "module2", True)
|
|
assert not await service.check_has_permission(chain2, "*")
|
|
assert await service.check_has_permission(chain2, "module1")
|
|
assert await service.check_has_permission(chain2, "module1.pack1")
|
|
assert not await service.check_has_permission(chain2, "module1.pack2")
|
|
assert await service.check_has_permission(chain2, "module2")
|
|
assert await service.check_has_permission(chain2, "module2.pack1")
|
|
assert await service.check_has_permission(chain2, "module2.pack2")
|
|
|
|
assert not await service.check_has_permission(entity2, "module2.pack2")
|
|
assert await service.check_has_permission(entity_global, "module2.pack2")
|
|
|
|
async with db.get_conn() as conn:
|
|
repo = PermRepo(conn)
|
|
assert await repo.get_perm_info(entity2, "module1") is True
|
|
assert await repo.get_perm_info(entity2, "module1.pack2") is False
|