diff --git a/.flake8 b/.flake8 index 6f42d352..e0b12106 100644 --- a/.flake8 +++ b/.flake8 @@ -16,7 +16,7 @@ per-file-ignores = examples/*:I900,S105 # flake8-import-order -application-import-names = aiohttp_admin, _auth, _auth_helpers, _models, _resources +application-import-names = aiohttp_admin, conftest, _auth, _auth_helpers, _models, _resources import-order-style = pycharm # flake8-quotes diff --git a/aiohttp_admin/__init__.py b/aiohttp_admin/__init__.py index 746cc338..145faba7 100644 --- a/aiohttp_admin/__init__.py +++ b/aiohttp_admin/__init__.py @@ -11,9 +11,9 @@ from .routes import setup_resources, setup_routes from .security import AdminAuthorizationPolicy, Permissions, TokenIdentityPolicy, check -from .types import Schema, UserDetails +from .types import Schema, State, UserDetails, check_credentials_key, permission_re_key, state_key -__all__ = ("Permissions", "Schema", "UserDetails", "setup") +__all__ = ("Permissions", "Schema", "UserDetails", "permission_re_key", "setup") __version__ = "0.1.0a2" @@ -51,7 +51,7 @@ async def on_startup(admin: web.Application) -> None: enclosing scope later. """ storage._cookie_params["path"] = prefixed_subapp.canonical - admin["state"]["urls"] = { + admin[state_key]["urls"] = { "token": str(admin.router["token"].url_for()), "logout": str(admin.router["logout"].url_for()) } @@ -65,7 +65,8 @@ def value(r: web.RouteDef) -> tuple[str, str]: for res in schema["resources"]: m = res["model"] - admin["state"]["resources"][m.name]["urls"] = {key(r): value(r) for r in m.routes} + urls = admin[state_key]["resources"][m.name]["urls"] + urls.update((key(r), value(r)) for r in m.routes) schema = check(Schema, schema) if secret is None: @@ -74,9 +75,9 @@ def value(r: web.RouteDef) -> tuple[str, str]: admin = web.Application() admin.middlewares.append(pydantic_middleware) admin.on_startup.append(on_startup) - admin["check_credentials"] = schema["security"]["check_credentials"] - admin["identity_callback"] = schema["security"].get("identity_callback") - admin["state"] = {"view": schema.get("view", {}), "js_module": schema.get("js_module")} + admin[check_credentials_key] = schema["security"]["check_credentials"] + admin[state_key] = State({"view": schema.get("view", {}), "js_module": schema.get("js_module"), + "urls": {}, "resources": {}}) max_age = schema["security"].get("max_age") secure = schema["security"].get("secure", True) @@ -90,7 +91,7 @@ def value(r: web.RouteDef) -> tuple[str, str]: setup_resources(admin, schema) resource_patterns = [] - for r, state in admin["state"]["resources"].items(): + for r, state in admin[state_key]["resources"].items(): fields = state["fields"].keys() resource_patterns.append( r"(?#Resource name){r}" @@ -102,7 +103,7 @@ def value(r: web.RouteDef) -> tuple[str, str]: p_re = (r"(?#Global admin permission)~?admin\.(view|edit|add|delete|\*)" r"|" r"(?#Resource permission)(~)?admin\.({})").format("|".join(resource_patterns)) - admin["permission_re"] = re.compile(p_re) + admin[permission_re_key] = re.compile(p_re) prefixed_subapp = app.add_subapp(path, admin) return admin diff --git a/aiohttp_admin/backends/sqlalchemy.py b/aiohttp_admin/backends/sqlalchemy.py index 3754e642..bff59e5b 100644 --- a/aiohttp_admin/backends/sqlalchemy.py +++ b/aiohttp_admin/backends/sqlalchemy.py @@ -5,12 +5,12 @@ import sys from collections.abc import Callable, Coroutine, Iterator, Sequence from types import MappingProxyType as MPT -from typing import Any, Literal, Optional, TypeVar, Union +from typing import Any, Literal, Optional, TypeVar, Union, cast import sqlalchemy as sa from aiohttp import web from sqlalchemy.ext.asyncio import AsyncEngine -from sqlalchemy.orm import DeclarativeBase, QueryableAttribute +from sqlalchemy.orm import DeclarativeBase, DeclarativeBaseNoMeta, Mapper, QueryableAttribute from sqlalchemy.sql.roles import ExpressionElementRole from .abc import AbstractAdminResource, GetListParams, Meta, Record @@ -26,6 +26,7 @@ _FValues = Union[bool, int, str] _Filters = dict[Union[sa.Column[object], QueryableAttribute[Any]], Union[_FValues, Sequence[_FValues]]] +_ModelOrTable = Union[sa.Table, type[DeclarativeBase], type[DeclarativeBaseNoMeta]] logger = logging.getLogger(__name__) @@ -155,7 +156,7 @@ def create_filters(columns: sa.ColumnCollection[str, sa.Column[object]], # ID is based on PK, which we can't infer from types, so must use Any here. class SAResource(AbstractAdminResource[Any]): - def __init__(self, db: AsyncEngine, model_or_table: Union[sa.Table, type[DeclarativeBase]]): + def __init__(self, db: AsyncEngine, model_or_table: _ModelOrTable): if isinstance(model_or_table, sa.Table): table = model_or_table else: @@ -221,7 +222,9 @@ def __init__(self, db: AsyncEngine, model_or_table: Union[sa.Table, type[Declara if not isinstance(model_or_table, sa.Table): # Append fields to represent ORM relationships. - mapper = sa.inspect(model_or_table) + # Mypy doesn't handle union well here. + mapper = cast(Union[Mapper[DeclarativeBase], Mapper[DeclarativeBaseNoMeta]], + sa.inspect(model_or_table)) assert mapper is not None # noqa: S101 for name, relationship in mapper.relationships.items(): # https://github.com/sqlalchemy/sqlalchemy/discussions/10161#discussioncomment-6583442 diff --git a/aiohttp_admin/routes.py b/aiohttp_admin/routes.py index d164971c..32306f1e 100644 --- a/aiohttp_admin/routes.py +++ b/aiohttp_admin/routes.py @@ -6,16 +6,15 @@ from aiohttp import web from . import views -from .types import Schema +from .types import Schema, _ResourceState, resources_key, state_key def setup_resources(admin: web.Application, schema: Schema) -> None: - admin["resources"] = [] - admin["state"]["resources"] = {} + admin[resources_key] = [] for r in schema["resources"]: m = r["model"] - admin["resources"].append(m) + admin[resources_key].append(m) admin.router.add_routes(m.routes) try: @@ -47,11 +46,12 @@ def setup_resources(admin: web.Application, schema: Schema) -> None: for name, props in r.get("field_props", {}).items(): fields[name]["props"].update(props) - state = {"fields": fields, "inputs": inputs, "list_omit": tuple(omit_fields), - "repr": repr_field, "label": r.get("label"), "icon": r.get("icon"), - "bulk_update": r.get("bulk_update", {}), - "show_actions": r.get("show_actions", ())} - admin["state"]["resources"][m.name] = state + state: _ResourceState = { + "fields": fields, "inputs": inputs, "list_omit": tuple(omit_fields), + "repr": repr_field, "label": r.get("label"), "icon": r.get("icon"), + "bulk_update": r.get("bulk_update", {}), "urls": {}, + "show_actions": r.get("show_actions", ())} + admin[state_key]["resources"][m.name] = state def setup_routes(admin: web.Application) -> None: diff --git a/aiohttp_admin/security.py b/aiohttp_admin/security.py index 51fe3b0a..9648cc03 100644 --- a/aiohttp_admin/security.py +++ b/aiohttp_admin/security.py @@ -71,7 +71,7 @@ def permissions_as_dict(permissions: Collection[str]) -> dict[str, dict[str, lis return p_dict -class AdminAuthorizationPolicy(AbstractAuthorizationPolicy): # type: ignore[misc,no-any-unimported] +class AdminAuthorizationPolicy(AbstractAuthorizationPolicy): def __init__(self, schema: Schema): super().__init__() self._identity_callback = schema["security"].get("identity_callback") @@ -79,8 +79,12 @@ def __init__(self, schema: Schema): async def authorized_userid(self, identity: str) -> str: return identity - async def permits(self, identity: Optional[str], permission: Union[str, Enum], - context: tuple[web.Request, Optional[Mapping[str, object]]]) -> bool: + async def permits( + self, identity: Optional[str], permission: Union[str, Enum], + context: Optional[tuple[web.Request, Optional[Mapping[str, object]]]] = None + ) -> bool: + # TODO: https://github.com/aio-libs/aiohttp-security/issues/677 + assert context is not None # noqa: S101 if identity is None: return False @@ -101,7 +105,7 @@ async def permits(self, identity: Optional[str], permission: Union[str, Enum], return has_permission(permission, permissions_as_dict(permissions), record) -class TokenIdentityPolicy(SessionIdentityPolicy): # type: ignore[misc,no-any-unimported] +class TokenIdentityPolicy(SessionIdentityPolicy): def __init__(self, fernet: Fernet, schema: Schema): super().__init__() self._fernet = fernet @@ -130,7 +134,7 @@ async def identify(self, request: web.Request) -> Optional[str]: # Both identites must match. return token_identity if token_identity == cookie_identity else None - async def remember(self, request: web.Request, response: web.Response, + async def remember(self, request: web.Request, response: web.StreamResponse, identity: str, **kwargs: object) -> None: """Send auth tokens to client for authentication.""" # For proper security we send a token for JS to store and an HTTP only cookie: @@ -138,9 +142,9 @@ async def remember(self, request: web.Request, response: web.Response, # Send token that will be saved in local storage by the JS client. response.headers["X-Token"] = json.dumps(await self.user_identity_dict(request, identity)) # Send httponly cookie, which will be invisible to JS. - await super().remember(request, response, identity, **kwargs) + await super().remember(request, response, identity, **kwargs) # type: ignore[arg-type] - async def forget(self, request: web.Request, response: web.Response) -> None: + async def forget(self, request: web.Request, response: web.StreamResponse) -> None: """Delete session cookie (JS client should choose to delete its token).""" await super().forget(request, response) diff --git a/aiohttp_admin/types.py b/aiohttp_admin/types.py index 6c2c64a6..e750111e 100644 --- a/aiohttp_admin/types.py +++ b/aiohttp_admin/types.py @@ -1,7 +1,10 @@ +import re import sys from collections.abc import Callable, Collection, Sequence from typing import Any, Awaitable, Literal, Mapping, Optional +from aiohttp.web import AppKey + if sys.version_info >= (3, 12): from typing import TypedDict else: @@ -110,7 +113,6 @@ class Schema(_Schema): class _ResourceState(TypedDict): - display: Sequence[str] fields: dict[str, ComponentState] inputs: dict[str, InputState] show_actions: Sequence[ComponentState] @@ -118,6 +120,8 @@ class _ResourceState(TypedDict): icon: Optional[str] urls: dict[str, tuple[str, str]] # (method, url) bulk_update: dict[str, dict[str, Any]] + list_omit: tuple[str, ...] + label: Optional[str] class State(TypedDict): @@ -146,3 +150,9 @@ def func(name: str, args: Optional[Sequence[object]] = None) -> FunctionState: def regex(value: str) -> RegexState: """Convert value to a RegExp object on the frontend.""" return {"__type__": "regexp", "value": value} + + +check_credentials_key = AppKey[Callable[[str, str], Awaitable[bool]]]("check_credentials") +permission_re_key = AppKey("permission_re", re.Pattern[str]) +resources_key = AppKey("resources", list[Any]) # TODO(pydantic): AbstractAdminResource +state_key = AppKey("state", State) diff --git a/aiohttp_admin/views.py b/aiohttp_admin/views.py index 750bc36d..1c23a51e 100644 --- a/aiohttp_admin/views.py +++ b/aiohttp_admin/views.py @@ -7,6 +7,7 @@ from pydantic import Json from .security import check +from .types import check_credentials_key, state_key if sys.version_info >= (3, 12): from typing import TypedDict @@ -38,15 +39,15 @@ async def index(request: web.Request) -> web.Response: """Root page which loads react-admin.""" static = request.app.router["static"] js = static.url_for(filename="admin.js") - state = json.dumps(request.app["state"]) + state = json.dumps(request.app[state_key]) # __package__ can be None, despite what the documentation claims. package_name = __main__.__package__ or "My" # Common convention is to have _app suffix for package name, so try and strip that. package_name = package_name.removesuffix("_app").replace("_", " ").title() - name = request.app["state"]["view"].get("name", package_name) + name = request.app[state_key]["view"].get("name", package_name) - icon = request.app["state"]["view"].get("icon", static.url_for(filename="favicon.svg")) + icon = request.app[state_key]["view"].get("icon", static.url_for(filename="favicon.svg")) output = INDEX_TEMPLATE.format(name=name, icon=icon, js=js, state=state) return web.Response(text=output, content_type="text/html") @@ -56,7 +57,7 @@ async def token(request: web.Request) -> web.Response: """Validate user credentials and log the user in.""" data = check(Json[_Login], await request.read()) - check_credentials = request.app["check_credentials"] + check_credentials = request.app[check_credentials_key] if not await check_credentials(data["username"], data["password"]): raise web.HTTPUnauthorized(text="Wrong username or password") diff --git a/examples/permissions.py b/examples/permissions.py index dced00b0..a07c784f 100644 --- a/examples/permissions.py +++ b/examples/permissions.py @@ -11,13 +11,15 @@ import sqlalchemy as sa from aiohttp import web -from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column import aiohttp_admin -from aiohttp_admin import Permissions, UserDetails +from aiohttp_admin import Permissions, UserDetails, permission_re_key from aiohttp_admin.backends.sqlalchemy import SAResource, permission_for as p +db = web.AppKey("db", async_sessionmaker[AsyncSession]) + class Base(DeclarativeBase): """Base model.""" @@ -49,14 +51,16 @@ class User(Base): async def check_credentials(app: web.Application, username: str, password: str) -> bool: """Allow login to any user account regardless of password.""" - async with app["db"]() as sess: + async with app[db]() as sess: user = await sess.get(User, username.lower()) return user is not None async def identity_callback(app: web.Application, identity: str) -> UserDetails: - async with app["db"]() as sess: + async with app[db]() as sess: user = await sess.get(User, identity) + if not user: + raise ValueError("No user found for given identity") return {"permissions": json.loads(user.permissions), "fullName": user.username.title()} @@ -79,7 +83,7 @@ async def create_app() -> web.Application: sess.add(SimpleParent(id=p_simple.id, date=datetime(2023, 2, 13, 19, 4))) app = web.Application() - app["db"] = session + app[db] = session # This is the setup required for aiohttp-admin. schema: aiohttp_admin.Schema = { @@ -123,7 +127,7 @@ async def create_app() -> web.Application: filters={Simple.num: 5})) } for name, permissions in users.items(): - if any(admin["permission_re"].fullmatch(p) is None for p in permissions): + if any(admin[permission_re_key].fullmatch(p) is None for p in permissions): raise ValueError("Not a valid permission.") sess.add(User(username=name, permissions=json.dumps(permissions))) diff --git a/requirements.txt b/requirements.txt index c9211f72..a2730e44 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,6 +1,6 @@ -e . -aiohttp==3.8.6 -aiohttp-security==0.4.0 +aiohttp==3.9.0 +aiohttp-security==0.5.0 aiohttp-session[secure]==2.12.0 aiosqlite==0.19.0 cryptography==41.0.5 diff --git a/setup.py b/setup.py index 279a08d6..be4ba837 100644 --- a/setup.py +++ b/setup.py @@ -40,7 +40,7 @@ def read_version(): download_url="https://github.com/aio-libs/aiohttp-admin", license="Apache 2", packages=find_packages(), - install_requires=("aiohttp>=3.8.2", "aiohttp_security", "aiohttp_session", + install_requires=("aiohttp>=3.9", "aiohttp_security", "aiohttp_session", "cryptography", "pydantic>2,<3", 'typing_extensions>=3.10; python_version<"3.12"'), extras_require={"sa": ["sqlalchemy>=2.0.4,<3"]}, diff --git a/tests/conftest.py b/tests/conftest.py index a4871d3f..b63d22be 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -5,22 +5,38 @@ import pytest from aiohttp import web from aiohttp.test_utils import TestClient -from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine -from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column +from sqlalchemy.ext.asyncio import (AsyncEngine, AsyncSession, + async_sessionmaker, create_async_engine) +from sqlalchemy.orm import DeclarativeBaseNoMeta, Mapped, mapped_column import aiohttp_admin from _auth import check_credentials from aiohttp_admin.backends.sqlalchemy import SAResource -_IdentityCallback = Callable[[str], Awaitable[aiohttp_admin.UserDetails]] +IdentityCallback = Callable[[Optional[str]], Awaitable[aiohttp_admin.UserDetails]] -@pytest.fixture -def base() -> type[DeclarativeBase]: - class Base(DeclarativeBase): - """Base model.""" +class Base(DeclarativeBaseNoMeta): + """Base model.""" + + +class DummyModel(Base): + __tablename__ = "dummy" + + id: Mapped[int] = mapped_column(primary_key=True) + + +class Dummy2Model(Base): + __tablename__ = "dummy2" - return Base + id: Mapped[int] = mapped_column(primary_key=True) + msg: Mapped[Optional[str]] + + +model = web.AppKey[type[DummyModel]]("model") +model2 = web.AppKey[type[Dummy2Model]]("model2") +db = web.AppKey("db", async_sessionmaker[AsyncSession]) +admin = web.AppKey("admin", web.Application) @pytest.fixture @@ -30,28 +46,17 @@ def mock_engine() -> AsyncMock: @pytest.fixture def create_admin_client( - base: DeclarativeBase, aiohttp_client: Callable[[web.Application], Awaitable[TestClient]] -) -> Callable[[Optional[_IdentityCallback]], Awaitable[TestClient]]: - async def admin_client(identity_callback: Optional[_IdentityCallback] = None) -> TestClient: - class DummyModel(base): # type: ignore[misc,valid-type] - __tablename__ = "dummy" - - id: Mapped[int] = mapped_column(primary_key=True) - - class Dummy2Model(base): # type: ignore[misc,valid-type] - __tablename__ = "dummy2" - - id: Mapped[int] = mapped_column(primary_key=True) - msg: Mapped[Optional[str]] - + aiohttp_client: Callable[[web.Application], Awaitable[TestClient]] +) -> Callable[[Optional[IdentityCallback]], Awaitable[TestClient]]: + async def admin_client(identity_callback: Optional[IdentityCallback] = None) -> TestClient: app = web.Application() - app["model"] = DummyModel - app["model2"] = Dummy2Model + app[model] = DummyModel + app[model2] = Dummy2Model engine = create_async_engine("sqlite+aiosqlite:///:memory:") - app["db"] = async_sessionmaker(engine, expire_on_commit=False) + app[db] = async_sessionmaker(engine, expire_on_commit=False) async with engine.begin() as conn: - await conn.run_sync(base.metadata.create_all) - async with app["db"].begin() as sess: + await conn.run_sync(Base.metadata.create_all) + async with app[db].begin() as sess: sess.add(DummyModel()) sess.add(Dummy2Model(msg="Test")) sess.add(Dummy2Model(msg="Test")) @@ -69,7 +74,7 @@ class Dummy2Model(base): # type: ignore[misc,valid-type] } if identity_callback: schema["security"]["identity_callback"] = identity_callback - app["admin"] = aiohttp_admin.setup(app, schema) + app[admin] = aiohttp_admin.setup(app, schema) return await aiohttp_client(app) @@ -85,7 +90,7 @@ async def admin_client(create_admin_client: Callable[[], Awaitable[TestClient]]) def login() -> Callable[[TestClient], Awaitable[dict[str, str]]]: async def do_login(admin_client: TestClient) -> dict[str, str]: assert admin_client.app - url = admin_client.app["admin"].router["token"].url_for() + url = admin_client.app[admin].router["token"].url_for() login = {"username": "admin", "password": "admin123"} async with admin_client.post(url, json=login) as resp: assert resp.status == 200 diff --git a/tests/test_admin.py b/tests/test_admin.py index 082549b6..06b23ab2 100644 --- a/tests/test_admin.py +++ b/tests/test_admin.py @@ -4,7 +4,7 @@ import aiohttp_admin from _auth import check_credentials from _resources import DummyResource -from aiohttp_admin.types import comp, func +from aiohttp_admin.types import comp, func, permission_re_key, state_key def test_path() -> None: @@ -26,7 +26,7 @@ def test_js_module() -> None: "resources": (), "js_module": "/custom_js.js"} admin = aiohttp_admin.setup(app, schema) - assert admin["state"]["js_module"] == "/custom_js.js" + assert admin[state_key]["js_module"] == "/custom_js.js" def test_no_js_module() -> None: @@ -35,7 +35,7 @@ def test_no_js_module() -> None: "resources": ()} admin = aiohttp_admin.setup(app, schema) - assert admin["state"]["js_module"] is None + assert admin[state_key]["js_module"] is None def test_validators() -> None: @@ -51,7 +51,7 @@ def test_validators() -> None: "security": {"check_credentials": check_credentials}, "resources": ({"model": dummy, "validators": {"id": (func("minValue", (3,)),)}},)} admin = aiohttp_admin.setup(app, schema) - validators = admin["state"]["resources"]["dummy"]["inputs"]["id"]["props"]["validate"] + validators = admin[state_key]["resources"]["dummy"]["inputs"]["id"]["props"]["validate"] assert validators == (func("required", ()), func("minValue", (3,))) assert ("minValue", 3) not in dummy.inputs["id"]["props"]["validate"] # type: ignore[operator] @@ -64,7 +64,7 @@ def test_re() -> None: schema: aiohttp_admin.Schema = {"security": {"check_credentials": check_credentials}, "resources": ({"model": test_re},)} admin = aiohttp_admin.setup(app, schema) - r = admin["permission_re"] + r = admin[permission_re_key] assert r.fullmatch("admin.*") assert r.fullmatch("admin.view") @@ -102,7 +102,7 @@ def test_display() -> None: admin = aiohttp_admin.setup(app, schema) - test_state = admin["state"]["resources"]["test"] + test_state = admin[state_key]["resources"]["test"] assert test_state["list_omit"] == ("id",) assert test_state["inputs"]["id"]["props"] == {"validate": (func("required", ()),)} assert test_state["inputs"]["foo"]["props"] == {"alwaysOn": "alwaysOn"} @@ -136,7 +136,7 @@ def test_extra_props() -> None: admin = aiohttp_admin.setup(app, schema) - test_state = admin["state"]["resources"]["test"] + test_state = admin[state_key]["resources"]["test"] assert test_state["fields"]["id"]["props"] == {"textAlign": "left", "placeholder": "foo", "label": "Spam"} assert test_state["inputs"]["id"]["props"] == {"alwaysOn": "alwaysOn", "type": "email", diff --git a/tests/test_backends_abc.py b/tests/test_backends_abc.py index 059ed68d..9f877f4c 100644 --- a/tests/test_backends_abc.py +++ b/tests/test_backends_abc.py @@ -3,13 +3,15 @@ from aiohttp.test_utils import TestClient +from conftest import admin + _Login = Callable[[TestClient], Awaitable[dict[str, str]]] async def test_create_with_null(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_create"].url_for() + url = admin_client.app[admin].router["dummy2_create"].url_for() p = {"data": json.dumps({"msg": None})} async with admin_client.post(url, params=p, headers=h) as resp: assert resp.status == 200, await resp.text() diff --git a/tests/test_backends_sqlalchemy.py b/tests/test_backends_sqlalchemy.py index 44330405..0a260131 100644 --- a/tests/test_backends_sqlalchemy.py +++ b/tests/test_backends_sqlalchemy.py @@ -16,10 +16,19 @@ from _auth import check_credentials from aiohttp_admin.backends.sqlalchemy import FIELD_TYPES, SAResource, permission_for from aiohttp_admin.types import comp, func, regex +from conftest import admin _Login = Callable[[TestClient], Awaitable[dict[str, str]]] +@pytest.fixture +def base() -> type[DeclarativeBase]: + class Base(DeclarativeBase): + """Base model.""" + + return Base + + def test_no_subtypes() -> None: """We don't want any subtypes in the lookup, as this would depend on test ordering.""" assert all({TypeEngine, TypeDecorator} & set(t.__bases__) for t in FIELD_TYPES) @@ -109,13 +118,13 @@ class TestModel(base): # type: ignore[misc,valid-type] }, "resources": ({"model": SAResource(engine, TestModel)},) } - app["admin"] = aiohttp_admin.setup(app, schema) + app[admin] = aiohttp_admin.setup(app, schema) admin_client = await aiohttp_client(app) assert admin_client.app h = await login(admin_client) - url = app["admin"].router["test_get_one"].url_for() + url = app[admin].router["test_get_one"].url_for() async with admin_client.get(url, params={"id": 1}, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": {"id": "1", "binary": "foo"}} @@ -178,13 +187,13 @@ class TestModelParent(base): # type: ignore[misc,valid-type] "resources": ({"model": SAResource(engine, TestModel)}, {"model": SAResource(engine, TestModelParent)}) } - app["admin"] = aiohttp_admin.setup(app, schema) + app[admin] = aiohttp_admin.setup(app, schema) admin_client = await aiohttp_client(app) assert admin_client.app h = await login(admin_client) - url = app["admin"].router["parent_get_one"].url_for() + url = app[admin].router["parent_get_one"].url_for() async with admin_client.get(url, params={"id": 1}, headers=h) as resp: assert resp.status == 200 # child_id must be converted to str ID. @@ -376,13 +385,13 @@ class TestModel(base): # type: ignore[misc,valid-type] }, "resources": ({"model": SAResource(engine, TestModel)},) } - app["admin"] = aiohttp_admin.setup(app, schema) + app[admin] = aiohttp_admin.setup(app, schema) admin_client = await aiohttp_client(app) assert admin_client.app h = await login(admin_client) - url = app["admin"].router["test_get_list"].url_for() + url = app[admin].router["test_get_list"].url_for() p = {"pagination": json.dumps({"page": 1, "perPage": 10}), "sort": json.dumps({"field": "id", "order": "DESC"}), "filter": "{}"} async with admin_client.get(url, params=p, headers=h) as resp: @@ -390,24 +399,24 @@ class TestModel(base): # type: ignore[misc,valid-type] assert await resp.json() == {"data": [{"id": "8", "num": 8, "other": "bar"}, {"id": "5", "num": 5, "other": "foo"}], "total": 2} - url = app["admin"].router["test_get_one"].url_for() + url = app[admin].router["test_get_one"].url_for() async with admin_client.get(url, params={"id": 8}, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": {"id": "8", "num": 8, "other": "bar"}} - url = app["admin"].router["test_get_many"].url_for() + url = app[admin].router["test_get_many"].url_for() async with admin_client.get(url, params={"ids": '["5", "8"]'}, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": [{"id": "5", "num": 5, "other": "foo"}, {"id": "8", "num": 8, "other": "bar"}]} - url = app["admin"].router["test_create"].url_for() + url = app[admin].router["test_create"].url_for() p = {"data": json.dumps({"num": 12, "other": "this"})} async with admin_client.post(url, params=p, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": {"id": "12", "num": 12, "other": "this"}} - url = app["admin"].router["test_update"].url_for() + url = app[admin].router["test_update"].url_for() p1 = {"id": 5, "data": json.dumps({"id": 5, "other": "that"}), "previousData": "{}"} async with admin_client.put(url, params=p1, headers=h) as resp: assert resp.status == 200 @@ -439,19 +448,19 @@ class TestModel(base): # type: ignore[misc,valid-type] }, "resources": ({"model": SAResource(engine, TestModel)},) } - app["admin"] = aiohttp_admin.setup(app, schema) + app[admin] = aiohttp_admin.setup(app, schema) admin_client = await aiohttp_client(app) assert admin_client.app h = await login(admin_client) - url = app["admin"].router["test_get_one"].url_for() + url = app[admin].router["test_get_one"].url_for() async with admin_client.get(url, params={"id": 1}, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": {"id": "1", "date": "2023-04-23", "time": "2023-01-02 03:04:00"}} - url = app["admin"].router["test_create"].url_for() + url = app[admin].router["test_create"].url_for() p = {"data": json.dumps({"date": "2024-05-09", "time": "2020-11-12 03:04:05"})} async with admin_client.post(url, params=p, headers=h) as resp: assert resp.status == 200 @@ -515,13 +524,13 @@ class TestModel(base): # type: ignore[misc,valid-type] }, "resources": ({"model": SAResource(engine, TestModel)},) } - app["admin"] = aiohttp_admin.setup(app, schema) + app[admin] = aiohttp_admin.setup(app, schema) admin_client = await aiohttp_client(app) assert admin_client.app h = await login(admin_client) - url = app["admin"].router["test_create"].url_for() + url = app[admin].router["test_create"].url_for() p = {"data": json.dumps({"foo": True, "bar": 5})} async with admin_client.post(url, params=p, headers=h) as resp: assert resp.status == 200 diff --git a/tests/test_security.py b/tests/test_security.py index 50e4c467..44e64e32 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -4,17 +4,17 @@ from unittest import mock from aiohttp.test_utils import TestClient -from aiohttp_security import AbstractAuthorizationPolicy from aiohttp_admin import Permissions, UserDetails +from conftest import IdentityCallback, admin, db, model2 -_CreateClient = Callable[[AbstractAuthorizationPolicy], Awaitable[TestClient]] +_CreateClient = Callable[[IdentityCallback], Awaitable[TestClient]] _Login = Callable[[TestClient], Awaitable[dict[str, str]]] async def test_no_token(admin_client: TestClient) -> None: assert admin_client.app - url = admin_client.app["admin"].router["dummy_get_list"].url_for() + url = admin_client.app[admin].router["dummy_get_list"].url_for() async with admin_client.get(url) as resp: assert resp.status == 401 assert await resp.text() == "401: Unauthorized" @@ -22,7 +22,7 @@ async def test_no_token(admin_client: TestClient) -> None: async def test_invalid_token(admin_client: TestClient) -> None: assert admin_client.app - url = admin_client.app["admin"].router["dummy_get_one"].url_for() + url = admin_client.app[admin].router["dummy_get_one"].url_for() h = {"Authorization": "invalid"} async with admin_client.get(url, headers=h) as resp: assert resp.status @@ -30,13 +30,13 @@ async def test_invalid_token(admin_client: TestClient) -> None: async def test_valid_login_logout(admin_client: TestClient) -> None: assert admin_client.app - url = admin_client.app["admin"].router["token"].url_for() + url = admin_client.app[admin].router["token"].url_for() login = {"username": "admin", "password": "admin123"} async with admin_client.post(url, json=login) as resp: assert resp.status == 200 token = resp.headers["X-Token"] - get_one_url = admin_client.app["admin"].router["dummy_get_one"].url_for() + get_one_url = admin_client.app[admin].router["dummy_get_one"].url_for() p = {"id": 1} h = {"Authorization": token} async with admin_client.get(get_one_url, params=p, headers=h) as resp: @@ -44,7 +44,7 @@ async def test_valid_login_logout(admin_client: TestClient) -> None: assert await resp.json() == {"data": {"id": "1"}} # Continue to test logout - logout_url = admin_client.app["admin"].router["logout"].url_for() + logout_url = admin_client.app[admin].router["logout"].url_for() async with admin_client.delete(logout_url, headers=h) as resp: assert resp.status == 200 @@ -54,7 +54,7 @@ async def test_valid_login_logout(admin_client: TestClient) -> None: async def test_missing_token(admin_client: TestClient) -> None: assert admin_client.app - url = admin_client.app["admin"].router["token"].url_for() + url = admin_client.app[admin].router["token"].url_for() login = {"username": "admin", "password": "admin123"} async with admin_client.post(url, json=login) as resp: assert resp.status == 200 @@ -63,7 +63,7 @@ async def test_missing_token(admin_client: TestClient) -> None: assert len(cookies) == 1 assert cookies[0]["path"] == "/admin" - url = admin_client.app["admin"].router["dummy_get_one"].url_for() + url = admin_client.app[admin].router["dummy_get_one"].url_for() p = {"id": 1} async with admin_client.get(url, params=p) as resp: assert resp.status == 401 @@ -71,7 +71,7 @@ async def test_missing_token(admin_client: TestClient) -> None: async def test_missing_cookie(admin_client: TestClient) -> None: assert admin_client.app - url = admin_client.app["admin"].router["token"].url_for() + url = admin_client.app[admin].router["token"].url_for() login = {"username": "admin", "password": "admin123"} async with admin_client.post(url, json=login) as resp: assert resp.status == 200 @@ -79,7 +79,7 @@ async def test_missing_cookie(admin_client: TestClient) -> None: admin_client.session.cookie_jar.clear() - url = admin_client.app["admin"].router["dummy_get_one"].url_for() + url = admin_client.app[admin].router["dummy_get_one"].url_for() p = {"id": 1} h = {"Authorization": token} async with admin_client.get(url, params=p, headers=h) as resp: @@ -88,7 +88,7 @@ async def test_missing_cookie(admin_client: TestClient) -> None: async def test_login_invalid_payload(admin_client: TestClient) -> None: assert admin_client.app - url = admin_client.app["admin"].router["token"].url_for() + url = admin_client.app[admin].router["token"].url_for() async with admin_client.post(url, json={"foo": "bar", "password": None}) as resp: assert resp.status == 400 result = await resp.json() @@ -101,7 +101,7 @@ async def test_login_invalid_payload(admin_client: TestClient) -> None: assert result[1]["input"] is None -async def test_list_without_permission(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_list_without_permission(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: assert identity == "admin" @@ -110,7 +110,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy_get_list"].url_for() + url = admin_client.app[admin].router["dummy_get_list"].url_for() p = {"pagination": json.dumps({"page": 1, "perPage": 10}), "sort": json.dumps({"field": "id", "order": "DESC"}), "filter": "{}"} h = await login(admin_client) @@ -121,7 +121,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: # assert await resp.text() == expected -async def test_get_resource_with_permission(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_get_resource_with_permission(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: assert identity == "admin" @@ -130,14 +130,14 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy_get_one"].url_for() + url = admin_client.app[admin].router["dummy_get_one"].url_for() h = await login(admin_client) async with admin_client.get(url, params={"id": 1}, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": {"id": "1"}} -async def test_get_resource_with_wildcard_permission(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_get_resource_with_wildcard_permission(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: assert identity == "admin" @@ -146,14 +146,14 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy_get_one"].url_for() + url = admin_client.app[admin].router["dummy_get_one"].url_for() h = await login(admin_client) async with admin_client.get(url, params={"id": 1}, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": {"id": "1"}} -async def test_get_resource_with_negative_permission(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_get_resource_with_negative_permission(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: assert identity == "admin" @@ -162,7 +162,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy_get_one"].url_for() + url = admin_client.app[admin].router["dummy_get_one"].url_for() h = await login(admin_client) async with admin_client.get(url, params={"id": 1}, headers=h) as resp: assert resp.status == 403 @@ -170,12 +170,12 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: # expected = "403: User does not have 'admin.dummy.view' permission" # assert await resp.text() == expected - url = admin_client.app["admin"].router["dummy2_get_one"].url_for() + url = admin_client.app[admin].router["dummy2_get_one"].url_for() async with admin_client.get(url, params={"id": 1}, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": {"id": "1", "msg": "Test"}} - url = admin_client.app["admin"].router["dummy2_create"].url_for() + url = admin_client.app[admin].router["dummy2_create"].url_for() p = {"data": '{"msg": "Foo"}'} async with admin_client.post(url, params=p, headers=h) as resp: assert resp.status == 403 @@ -183,7 +183,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: # expected = "403: User does not have 'admin.dummy2.create' permission" -async def test_list_resource_finegrained_permission(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_list_resource_finegrained_permission(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: assert identity == "admin" @@ -192,7 +192,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_get_list"].url_for() + url = admin_client.app[admin].router["dummy2_get_list"].url_for() h = await login(admin_client) p = {"pagination": json.dumps({"page": 1, "perPage": 10}), "sort": json.dumps({"field": "id", "order": "DESC"}), "filter": "{}"} @@ -201,7 +201,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert await resp.json() == {"data": [{"id": "3"}, {"id": "2"}, {"id": "1"}], "total": 3} -async def test_get_resource_finegrained_permission(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_get_resource_finegrained_permission(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: assert identity == "admin" @@ -210,14 +210,14 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_get_one"].url_for() + url = admin_client.app[admin].router["dummy2_get_one"].url_for() h = await login(admin_client) async with admin_client.get(url, params={"id": 1}, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": {"id": "1"}} -async def test_get_many_resource_finegrained_permission(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_get_many_resource_finegrained_permission(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: assert identity == "admin" @@ -226,14 +226,14 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_get_many"].url_for() + url = admin_client.app[admin].router["dummy2_get_many"].url_for() h = await login(admin_client) async with admin_client.get(url, params={"ids": '["1"]'}, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": [{"id": "1"}]} -async def test_create_resource_finegrained_permission(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_create_resource_finegrained_permission(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: assert identity == "admin" @@ -242,7 +242,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_create"].url_for() + url = admin_client.app[admin].router["dummy2_create"].url_for() h = await login(admin_client) p = {"data": json.dumps({"msg": "ABC"})} async with admin_client.post(url, params=p, headers=h) as resp: @@ -255,7 +255,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert await resp.json() == {"data": {"id": "4", "msg": None}} -async def test_create_resource_filtered_permission(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_create_resource_filtered_permission(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: assert identity == "admin" @@ -264,7 +264,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_create"].url_for() + url = admin_client.app[admin].router["dummy2_create"].url_for() h = await login(admin_client) p = {"data": json.dumps({"msg": "ABC"})} async with admin_client.post(url, params=p, headers=h) as resp: @@ -277,7 +277,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert await resp.json() == {"data": {"id": "4"}} -async def test_update_resource_finegrained_permission(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_update_resource_finegrained_permission(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: assert identity == "admin" @@ -286,7 +286,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_update"].url_for() + url = admin_client.app[admin].router["dummy2_update"].url_for() h = await login(admin_client) p = {"id": 1, "data": json.dumps({"id": 222, "msg": "ABC"}), "previousData": "{}"} async with admin_client.put(url, params=p, headers=h) as resp: @@ -294,7 +294,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert await resp.json() == {"data": {"id": "222", "msg": "Test"}} -async def test_update_resource_filtered_permission(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_update_resource_filtered_permission(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: assert identity == "admin" @@ -303,21 +303,22 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_update"].url_for() + url = admin_client.app[admin].router["dummy2_update"].url_for() h = await login(admin_client) p = {"id": 1, "data": json.dumps({"id": 222, "msg": "ABC"}), "previousData": "{}"} async with admin_client.put(url, params=p, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": {"id": "222"}} - async with admin_client.app["db"]() as sess: - r = await sess.get(admin_client.app["model2"], 1) + async with admin_client.app[db]() as sess: + r = await sess.get(admin_client.app[model2], 1) assert r is None - r = await sess.get(admin_client.app["model2"], 222) + r = await sess.get(admin_client.app[model2], 222) + assert r is not None assert r.msg == "Test" -async def test_update_many_resource_finegrained_permission( # type: ignore[no-any-unimported] +async def test_update_many_resource_finegrained_permission( create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: assert identity == "admin" @@ -326,7 +327,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_update_many"].url_for() + url = admin_client.app[admin].router["dummy2_update_many"].url_for() h = await login(admin_client) p = {"ids": '["1"]', "data": json.dumps({"msg": "ABC"})} async with admin_client.put(url, params=p, headers=h) as resp: @@ -335,7 +336,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: # expected = "403: User does not have 'admin.dummy2.msg.edit' permission" -async def test_delete_resource_filtered_permission(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_delete_resource_filtered_permission(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: assert identity == "admin" @@ -344,7 +345,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_delete"].url_for() + url = admin_client.app[admin].router["dummy2_delete"].url_for() h = await login(admin_client) p = {"id": 1, "previousData": "{}"} async with admin_client.delete(url, params=p, headers=h) as resp: @@ -352,13 +353,13 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert await resp.json() == {"data": {"id": "1"}} -async def test_permissions_cached(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permissions_cached(create_admin_client: _CreateClient, login: _Login) -> None: identity_callback = mock.AsyncMock(spec_set=(), return_value={"permissions": {"admin.*"}}) admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_get_list"].url_for() + url = admin_client.app[admin].router["dummy2_get_list"].url_for() h = await login(admin_client) identity_callback.assert_called_once() identity_callback.reset_mock() @@ -371,7 +372,7 @@ async def test_permissions_cached(create_admin_client: _CreateClient, # type: i identity_callback.assert_called_once() -async def test_permission_filter_list(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_list(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", 'admin.dummy2.*|msg="Test"|msg="Foo"')} @@ -379,10 +380,10 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - async with admin_client.app["db"].begin() as sess: - sess.add(admin_client.app["model2"](msg="Foo")) + async with admin_client.app[db].begin() as sess: + sess.add(admin_client.app[model2](msg="Foo")) - url = admin_client.app["admin"].router["dummy2_get_list"].url_for() + url = admin_client.app[admin].router["dummy2_get_list"].url_for() p = {"pagination": json.dumps({"page": 1, "perPage": 10}), "sort": json.dumps({"field": "id", "order": "DESC"}), "filter": "{}"} h = await login(admin_client) @@ -394,7 +395,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: "total": 3} -async def test_permission_filter_list2(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_list2(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", 'admin.dummy2.view|msg="Test"')} @@ -402,7 +403,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_get_list"].url_for() + url = admin_client.app[admin].router["dummy2_get_list"].url_for() p = {"pagination": json.dumps({"page": 1, "perPage": 10}), "sort": json.dumps({"field": "id", "order": "DESC"}), "filter": "{}"} h = await login(admin_client) @@ -412,7 +413,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: "data": [{"id": "2", "msg": "Test"}, {"id": "1", "msg": "Test"}], "total": 2} -async def test_permission_filter_get_one(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_get_one(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", 'admin.dummy2.*|msg="Test"')} @@ -420,7 +421,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_get_one"].url_for() + url = admin_client.app[admin].router["dummy2_get_one"].url_for() h = await login(admin_client) async with admin_client.get(url, params={"id": 2}, headers=h) as resp: assert resp.status == 200 @@ -429,7 +430,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert resp.status == 403 -async def test_permission_filter_get_one2(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_get_one2(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", 'admin.dummy2.view|msg="Test"')} @@ -437,7 +438,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_get_one"].url_for() + url = admin_client.app[admin].router["dummy2_get_one"].url_for() h = await login(admin_client) async with admin_client.get(url, params={"id": 2}, headers=h) as resp: assert resp.status == 200 @@ -446,7 +447,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert resp.status == 403 -async def test_permission_filter_get_many(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_get_many(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", 'admin.dummy2.*|msg="Test"')} @@ -454,7 +455,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_get_many"].url_for() + url = admin_client.app[admin].router["dummy2_get_many"].url_for() h = await login(admin_client) async with admin_client.get(url, params={"ids": '["2", "3"]'}, headers=h) as resp: assert resp.status == 200 @@ -464,7 +465,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert await resp.json() == {"data": []} -async def test_permission_filter_get_many2(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_get_many2(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", 'admin.dummy2.view|msg="Test"')} @@ -472,7 +473,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_get_many"].url_for() + url = admin_client.app[admin].router["dummy2_get_many"].url_for() h = await login(admin_client) async with admin_client.get(url, params={"ids": '["2", "3"]'}, headers=h) as resp: assert resp.status == 200 @@ -482,7 +483,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert await resp.json() == {"data": []} -async def test_permission_filter_create(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_create(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", 'admin.dummy2.*|msg="Test"')} @@ -490,7 +491,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_create"].url_for() + url = admin_client.app[admin].router["dummy2_create"].url_for() h = await login(admin_client) p = {"data": json.dumps({"msg": "Test"})} async with admin_client.post(url, params=p, headers=h) as resp: @@ -501,7 +502,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert resp.status == 403 -async def test_permission_filter_create2(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_create2(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", 'admin.dummy2.add|msg="Test"')} @@ -509,7 +510,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_create"].url_for() + url = admin_client.app[admin].router["dummy2_create"].url_for() h = await login(admin_client) p = {"data": json.dumps({"msg": "Test"})} async with admin_client.post(url, params=p, headers=h) as resp: @@ -520,7 +521,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert resp.status == 403 -async def test_permission_filter_update(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_update(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", 'admin.dummy2.*|msg="Test"')} @@ -528,7 +529,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_update"].url_for() + url = admin_client.app[admin].router["dummy2_update"].url_for() h = await login(admin_client) p = {"id": 3, "data": json.dumps({"msg": "Test"}), "previousData": "{}"} async with admin_client.put(url, params=p, headers=h) as resp: @@ -541,7 +542,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert resp.status == 200 -async def test_permission_filter_update2(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_update2(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", 'admin.dummy2.edit|msg="Test"')} @@ -549,7 +550,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_update"].url_for() + url = admin_client.app[admin].router["dummy2_update"].url_for() h = await login(admin_client) p = {"id": 3, "data": json.dumps({"msg": "Test"}), "previousData": "{}"} async with admin_client.put(url, params=p, headers=h) as resp: @@ -562,7 +563,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert resp.status == 200 -async def test_permission_filter_update_many( # type: ignore[no-any-unimported] +async def test_permission_filter_update_many( create_admin_client: _CreateClient, login: _Login ) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: @@ -571,7 +572,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_update_many"].url_for() + url = admin_client.app[admin].router["dummy2_update_many"].url_for() h = await login(admin_client) p = {"ids": '["3"]', "data": json.dumps({"msg": "Test"})} async with admin_client.put(url, params=p, headers=h) as resp: @@ -584,7 +585,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert resp.status == 200 -async def test_permission_filter_update_many2( # type: ignore[no-any-unimported] +async def test_permission_filter_update_many2( create_admin_client: _CreateClient, login: _Login ) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: @@ -593,7 +594,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_update_many"].url_for() + url = admin_client.app[admin].router["dummy2_update_many"].url_for() h = await login(admin_client) p = {"ids": '["3"]', "data": json.dumps({"msg": "Test"})} async with admin_client.put(url, params=p, headers=h) as resp: @@ -606,7 +607,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert resp.status == 200 -async def test_permission_filter_delete(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_delete(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", 'admin.dummy2.*|msg="Test"')} @@ -614,7 +615,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_delete"].url_for() + url = admin_client.app[admin].router["dummy2_delete"].url_for() h = await login(admin_client) p = {"id": 3, "previousData": "{}"} async with admin_client.delete(url, params=p, headers=h) as resp: @@ -624,7 +625,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert resp.status == 200 -async def test_permission_filter_delete2(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_delete2(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", 'admin.dummy2.delete|msg="Test"')} @@ -632,7 +633,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_delete"].url_for() + url = admin_client.app[admin].router["dummy2_delete"].url_for() h = await login(admin_client) p = {"id": 3, "previousData": "{}"} async with admin_client.delete(url, params=p, headers=h) as resp: @@ -642,7 +643,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert resp.status == 200 -async def test_permission_filter_delete_many(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_delete_many(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", 'admin.dummy2.*|msg="Test"')} @@ -650,7 +651,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_delete_many"].url_for() + url = admin_client.app[admin].router["dummy2_delete_many"].url_for() h = await login(admin_client) p = {"ids": '["2", "3"]'} async with admin_client.delete(url, params=p, headers=h) as resp: @@ -664,7 +665,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert await resp.json() == {"data": ["1", "2"]} -async def test_permission_filter_delete_many2(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_delete_many2(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", 'admin.dummy2.delete|msg="Test"')} @@ -672,7 +673,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_delete_many"].url_for() + url = admin_client.app[admin].router["dummy2_delete_many"].url_for() h = await login(admin_client) p = {"ids": '["2", "3"]'} async with admin_client.delete(url, params=p, headers=h) as resp: @@ -686,7 +687,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert await resp.json() == {"data": ["1", "2"]} -async def test_permission_filter_field_list(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_field_list(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", "admin.dummy2.msg.*|id=1|id=2")} @@ -694,7 +695,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_get_list"].url_for() + url = admin_client.app[admin].router["dummy2_get_list"].url_for() p = {"pagination": json.dumps({"page": 1, "perPage": 10}), "sort": json.dumps({"field": "id", "order": "DESC"}), "filter": "{}"} h = await login(admin_client) @@ -705,7 +706,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: "total": 3} -async def test_permission_filter_field_list2(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_field_list2(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", "admin.dummy2.msg.view|id=1|id=3")} @@ -713,7 +714,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_get_list"].url_for() + url = admin_client.app[admin].router["dummy2_get_list"].url_for() p = {"pagination": json.dumps({"page": 1, "perPage": 10}), "sort": json.dumps({"field": "id", "order": "DESC"}), "filter": "{}"} h = await login(admin_client) @@ -724,7 +725,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: "total": 3} -async def test_permission_filter_field_get_one(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_field_get_one(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", "admin.dummy2.msg.*|id=1|id=2")} @@ -732,7 +733,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_get_one"].url_for() + url = admin_client.app[admin].router["dummy2_get_one"].url_for() h = await login(admin_client) async with admin_client.get(url, params={"id": 1}, headers=h) as resp: assert resp.status == 200 @@ -742,7 +743,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert await resp.json() == {"data": {"id": "3"}} -async def test_permission_filter_field_get_one2(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_field_get_one2(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", "admin.dummy2.msg.view|id=1|id=2")} @@ -750,7 +751,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_get_one"].url_for() + url = admin_client.app[admin].router["dummy2_get_one"].url_for() h = await login(admin_client) async with admin_client.get(url, params={"id": 1}, headers=h) as resp: assert resp.status == 200 @@ -760,7 +761,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert await resp.json() == {"data": {"id": "3"}} -async def test_permission_filter_field_get_many(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_field_get_many(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", "admin.dummy2.msg.*|id=1|id=2")} @@ -768,14 +769,14 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_get_many"].url_for() + url = admin_client.app[admin].router["dummy2_get_many"].url_for() h = await login(admin_client) async with admin_client.get(url, params={"ids": '["2", "3"]'}, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": [{"id": "2", "msg": "Test"}, {"id": "3"}]} -async def test_permission_filter_field_get_many2(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_field_get_many2(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", "admin.dummy2.msg.view|id=1|id=2")} @@ -783,14 +784,14 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_get_many"].url_for() + url = admin_client.app[admin].router["dummy2_get_many"].url_for() h = await login(admin_client) async with admin_client.get(url, params={"ids": '["1", "3"]'}, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": [{"id": "1", "msg": "Test"}, {"id": "3"}]} -async def test_permission_filter_field_create(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_field_create(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", "admin.dummy2.msg.*|id=1|id=2")} @@ -798,7 +799,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_create"].url_for() + url = admin_client.app[admin].router["dummy2_create"].url_for() h = await login(admin_client) p = {"data": json.dumps({"msg": "Spam"})} async with admin_client.post(url, params=p, headers=h) as resp: @@ -806,12 +807,13 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: async with admin_client.post(url, params={"data": "{}"}, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": {"id": "4"}} - async with admin_client.app["db"]() as sess: - r = await sess.get(admin_client.app["model2"], 4) + async with admin_client.app[db]() as sess: + r = await sess.get(admin_client.app[model2], 4) + assert r is not None assert r.msg is None -async def test_permission_filter_field_create2(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_field_create2(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", "admin.dummy2.msg.add|id=1|id=2")} @@ -819,7 +821,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_create"].url_for() + url = admin_client.app[admin].router["dummy2_create"].url_for() h = await login(admin_client) p = {"data": json.dumps({"msg": "Spam"})} async with admin_client.post(url, params=p, headers=h) as resp: @@ -827,12 +829,13 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: async with admin_client.post(url, params={"data": "{}"}, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": {"id": "4", "msg": None}} - async with admin_client.app["db"]() as sess: - r = await sess.get(admin_client.app["model2"], 4) + async with admin_client.app[db]() as sess: + r = await sess.get(admin_client.app[model2], 4) + assert r is not None assert r.msg is None -async def test_permission_filter_field_update(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_field_update(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", "admin.dummy2.msg.*|id=1|id=2")} @@ -840,7 +843,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_update"].url_for() + url = admin_client.app[admin].router["dummy2_update"].url_for() h = await login(admin_client) p = {"id": 3, "data": json.dumps({"msg": "Spam"}), "previousData": "{}"} async with admin_client.put(url, params=p, headers=h) as resp: @@ -854,14 +857,15 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: async with admin_client.put(url, params=p, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": {"id": "5"}} - async with admin_client.app["db"]() as sess: - r = await sess.get(admin_client.app["model2"], 2) + async with admin_client.app[db]() as sess: + r = await sess.get(admin_client.app[model2], 2) assert r is None - r = await sess.get(admin_client.app["model2"], 5) + r = await sess.get(admin_client.app[model2], 5) + assert r is not None assert r.msg == "Test" -async def test_permission_filter_field_update2(create_admin_client: _CreateClient, # type: ignore[no-any-unimported] # noqa: B950 +async def test_permission_filter_field_update2(create_admin_client: _CreateClient, login: _Login) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: return {"permissions": ("admin.*", "admin.dummy2.msg.edit|id=1|id=2")} @@ -869,7 +873,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_update"].url_for() + url = admin_client.app[admin].router["dummy2_update"].url_for() h = await login(admin_client) p = {"id": "3", "data": json.dumps({"msg": "Spam"}), "previousData": "{}"} async with admin_client.put(url, params=p, headers=h) as resp: @@ -883,14 +887,15 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: async with admin_client.put(url, params=p, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": {"id": "5", "msg": "Test"}} - async with admin_client.app["db"]() as sess: - r = await sess.get(admin_client.app["model2"], 2) + async with admin_client.app[db]() as sess: + r = await sess.get(admin_client.app[model2], 2) assert r is None - r = await sess.get(admin_client.app["model2"], 5) + r = await sess.get(admin_client.app[model2], 5) + assert r is not None assert r.msg == "Test" -async def test_permission_filter_field_update_many( # type: ignore[no-any-unimported] +async def test_permission_filter_field_update_many( create_admin_client: _CreateClient, login: _Login ) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: @@ -899,7 +904,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_update_many"].url_for() + url = admin_client.app[admin].router["dummy2_update_many"].url_for() h = await login(admin_client) p = {"ids": '["3"]', "data": json.dumps({"msg": "Spam"})} async with admin_client.put(url, params=p, headers=h) as resp: @@ -910,7 +915,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: assert await resp.json() == {"data": ["1", "2"]} -async def test_permission_filter_field_update_many2( # type: ignore[no-any-unimported] +async def test_permission_filter_field_update_many2( create_admin_client: _CreateClient, login: _Login ) -> None: async def identity_callback(identity: Optional[str]) -> UserDetails: @@ -919,7 +924,7 @@ async def identity_callback(identity: Optional[str]) -> UserDetails: admin_client = await create_admin_client(identity_callback) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_update_many"].url_for() + url = admin_client.app[admin].router["dummy2_update_many"].url_for() h = await login(admin_client) p = {"ids": '["3"]', "data": json.dumps({"msg": "Spam"})} async with admin_client.put(url, params=p, headers=h) as resp: diff --git a/tests/test_views.py b/tests/test_views.py index 49f409ac..40724e8c 100644 --- a/tests/test_views.py +++ b/tests/test_views.py @@ -6,13 +6,14 @@ from aiohttp.test_utils import TestClient from aiohttp_admin.types import comp, func +from conftest import admin, db, model, model2 _Login = Callable[[TestClient], Awaitable[dict[str, str]]] async def test_admin_view(admin_client: TestClient) -> None: assert admin_client.app - url = admin_client.app["admin"].router["index"].url_for() + url = admin_client.app[admin].router["index"].url_for() async with admin_client.get(url) as resp: assert resp.status == 200 html = await resp.text() @@ -43,11 +44,11 @@ async def test_admin_view(admin_client: TestClient) -> None: async def test_list_pagination(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - async with admin_client.app["db"].begin() as sess: + async with admin_client.app[db].begin() as sess: for _ in range(25): - sess.add(admin_client.app["model"]()) + sess.add(admin_client.app[model]()) - url = admin_client.app["admin"].router["dummy_get_list"].url_for() + url = admin_client.app[admin].router["dummy_get_list"].url_for() p = {"pagination": '{"page": 1, "perPage": 30}', "sort": '{"field": "id", "order": "ASC"}', "filter": '{}'} async with admin_client.get(url, params=p, headers=h) as resp: @@ -76,11 +77,11 @@ async def test_list_pagination(admin_client: TestClient, login: _Login) -> None: async def test_list_filtering_by_pk(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - async with admin_client.app["db"].begin() as sess: + async with admin_client.app[db].begin() as sess: for _ in range(15): - sess.add(admin_client.app["model"]()) + sess.add(admin_client.app[model]()) - url = admin_client.app["admin"].router["dummy_get_list"].url_for() + url = admin_client.app[admin].router["dummy_get_list"].url_for() p = {"pagination": '{"page": 1, "perPage": 10}', "sort": '{"field": "id", "order": "ASC"}', "filter": '{"id": 3}'} async with admin_client.get(url, params=p, headers=h) as resp: @@ -91,11 +92,11 @@ async def test_list_filtering_by_pk(admin_client: TestClient, login: _Login) -> async def test_list_text_like_filtering(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - async with admin_client.app["db"].begin() as sess: + async with admin_client.app[db].begin() as sess: for _ in range(15): - sess.add(admin_client.app["model"]()) + sess.add(admin_client.app[model]()) - url = admin_client.app["admin"].router["dummy_get_list"].url_for() + url = admin_client.app[admin].router["dummy_get_list"].url_for() p = {"pagination": '{"page": 1, "perPage": 10}', "sort": '{"field": "id", "order": "ASC"}', "filter": '{"id": "3"}'} async with admin_client.get(url, params=p, headers=h) as resp: @@ -106,7 +107,7 @@ async def test_list_text_like_filtering(admin_client: TestClient, login: _Login) async def test_get_one(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - url = admin_client.app["admin"].router["dummy_get_one"].url_for() + url = admin_client.app[admin].router["dummy_get_one"].url_for() async with admin_client.get(url, params={"id": 1}, headers=h) as resp: assert resp.status == 200 @@ -116,7 +117,7 @@ async def test_get_one(admin_client: TestClient, login: _Login) -> None: async def test_get_one_not_exists(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - url = admin_client.app["admin"].router["dummy_get_one"].url_for() + url = admin_client.app[admin].router["dummy_get_one"].url_for() async with admin_client.get(url, params={"id": 5}, headers=h) as resp: assert resp.status == 404 @@ -125,11 +126,11 @@ async def test_get_one_not_exists(admin_client: TestClient, login: _Login) -> No async def test_get_many(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - async with admin_client.app["db"].begin() as sess: + async with admin_client.app[db].begin() as sess: for _ in range(15): - sess.add(admin_client.app["model"]()) + sess.add(admin_client.app[model]()) - url = admin_client.app["admin"].router["dummy_get_many"].url_for() + url = admin_client.app[admin].router["dummy_get_many"].url_for() p = {"ids": '["3", "7", "12"]'} async with admin_client.get(url, params=p, headers=h) as resp: assert resp.status == 200 @@ -139,11 +140,11 @@ async def test_get_many(admin_client: TestClient, login: _Login) -> None: async def test_get_many_not_exists(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - async with admin_client.app["db"].begin() as sess: + async with admin_client.app[db].begin() as sess: for _ in range(5): - sess.add(admin_client.app["model"]()) + sess.add(admin_client.app[model]()) - url = admin_client.app["admin"].router["dummy_get_many"].url_for() + url = admin_client.app[admin].router["dummy_get_many"].url_for() p = {"ids": '["3", "4", "8"]'} async with admin_client.get(url, params=p, headers=h) as resp: assert resp.status == 200 @@ -157,21 +158,22 @@ async def test_get_many_not_exists(admin_client: TestClient, login: _Login) -> N async def test_create(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - url = admin_client.app["admin"].router["dummy_create"].url_for() + url = admin_client.app[admin].router["dummy_create"].url_for() p = {"data": "{}"} async with admin_client.post(url, params=p, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": {"id": "2"}} - async with admin_client.app["db"]() as sess: - r = await sess.get(admin_client.app["model"], 2) + async with admin_client.app[db]() as sess: + r = await sess.get(admin_client.app[model], 2) + assert r is not None assert r.id == 2 async def test_create_duplicate_id(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - url = admin_client.app["admin"].router["dummy_create"].url_for() + url = admin_client.app[admin].router["dummy_create"].url_for() p = {"data": '{"id": 1}'} async with admin_client.post(url, params=p, headers=h) as resp: assert resp.status == 400 @@ -180,24 +182,25 @@ async def test_create_duplicate_id(admin_client: TestClient, login: _Login) -> N async def test_update(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - url = admin_client.app["admin"].router["dummy_update"].url_for() + url = admin_client.app[admin].router["dummy_update"].url_for() p = {"id": 1, "data": '{"id": 4}', "previousData": '{"id": 1}'} async with admin_client.put(url, params=p, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": {"id": "4"}} - async with admin_client.app["db"]() as sess: - r = await sess.get(admin_client.app["model"], 4) + async with admin_client.app[db]() as sess: + r = await sess.get(admin_client.app[model], 4) + assert r is not None assert r.id == 4 - assert await sess.get(admin_client.app["model"], 1) is None - assert await sess.get(admin_client.app["model"], 2) is None + assert await sess.get(admin_client.app[model], 1) is None + assert await sess.get(admin_client.app[model], 2) is None async def test_update_deleted_entity(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - url = admin_client.app["admin"].router["dummy_update"].url_for() + url = admin_client.app[admin].router["dummy_update"].url_for() p = {"id": 2, "data": '{"id": 4}', "previousData": '{"id": 2}'} async with admin_client.put(url, params=p, headers=h) as resp: assert resp.status == 404 @@ -206,7 +209,7 @@ async def test_update_deleted_entity(admin_client: TestClient, login: _Login) -> async def test_update_invalid_attributes(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - url = admin_client.app["admin"].router["dummy_update"].url_for() + url = admin_client.app[admin].router["dummy_update"].url_for() p = {"id": 1, "data": '{"id": 4, "foo": "invalid"}', "previousData": '{"id": 1}'} async with admin_client.put(url, params=p, headers=h) as resp: assert resp.status == 400 @@ -216,23 +219,25 @@ async def test_update_invalid_attributes(admin_client: TestClient, login: _Login async def test_update_many(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - url = admin_client.app["admin"].router["dummy2_update_many"].url_for() + url = admin_client.app[admin].router["dummy2_update_many"].url_for() p = {"ids": '["1", "2"]', "data": json.dumps({"msg": "ABC"})} async with admin_client.put(url, params=p, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": ["1", "2"]} - async with admin_client.app["db"]() as sess: - r = await sess.get(admin_client.app["model2"], 1) + async with admin_client.app[db]() as sess: + r = await sess.get(admin_client.app[model2], 1) + assert r is not None assert r.msg == "ABC" - r = await sess.get(admin_client.app["model2"], 2) + r = await sess.get(admin_client.app[model2], 2) + assert r is not None assert r.msg == "ABC" async def test_update_many_deleted_entity(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - url = admin_client.app["admin"].router["dummy_update_many"].url_for() + url = admin_client.app[admin].router["dummy_update_many"].url_for() p = {"ids": '["2"]', "data": '{"id": 4}'} async with admin_client.put(url, params=p, headers=h) as resp: assert resp.status == 404 @@ -241,7 +246,7 @@ async def test_update_many_deleted_entity(admin_client: TestClient, login: _Logi async def test_update_many_invalid_attributes(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - url = admin_client.app["admin"].router["dummy_update_many"].url_for() + url = admin_client.app[admin].router["dummy_update_many"].url_for() p = {"ids": '["1"]', "data": '{"foo": "invalid"}'} async with admin_client.put(url, params=p, headers=h) as resp: assert resp.status == 400 @@ -251,22 +256,22 @@ async def test_update_many_invalid_attributes(admin_client: TestClient, login: _ async def test_delete(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - url = admin_client.app["admin"].router["dummy_delete"].url_for() + url = admin_client.app[admin].router["dummy_delete"].url_for() p = {"id": 1, "previousData": '{"id": 1}'} async with admin_client.delete(url, params=p, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": {"id": "1"}} - async with admin_client.app["db"]() as sess: - assert await sess.get(admin_client.app["model"], 1) is None - r = await sess.scalars(sa.select(admin_client.app["model"])) + async with admin_client.app[db]() as sess: + assert await sess.get(admin_client.app[model], 1) is None + r = await sess.scalars(sa.select(admin_client.app[model])) assert len(r.all()) == 0 async def test_delete_entity_not_exists(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - url = admin_client.app["admin"].router["dummy_delete"].url_for() + url = admin_client.app[admin].router["dummy_delete"].url_for() p = {"id": 5, "previousData": '{"id": 5}'} async with admin_client.delete(url, params=p, headers=h) as resp: assert resp.status == 404 @@ -275,18 +280,18 @@ async def test_delete_entity_not_exists(admin_client: TestClient, login: _Login) async def test_delete_many(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - async with admin_client.app["db"].begin() as sess: + async with admin_client.app[db].begin() as sess: for _ in range(5): - sess.add(admin_client.app["model"]()) + sess.add(admin_client.app[model]()) - url = admin_client.app["admin"].router["dummy_delete_many"].url_for() + url = admin_client.app[admin].router["dummy_delete_many"].url_for() p = {"ids": '["2", "3", "5"]'} async with admin_client.delete(url, params=p, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": ["2", "3", "5"]} - async with admin_client.app["db"]() as sess: - r = await sess.scalars(sa.select(admin_client.app["model"])) + async with admin_client.app[db]() as sess: + r = await sess.scalars(sa.select(admin_client.app[model])) models = r.all() assert len(models) == 3 assert {m.id for m in models} == {1, 4, 6} @@ -295,27 +300,27 @@ async def test_delete_many(admin_client: TestClient, login: _Login) -> None: async def test_delete_many_not_exists(admin_client: TestClient, login: _Login) -> None: h = await login(admin_client) assert admin_client.app - async with admin_client.app["db"].begin() as sess: + async with admin_client.app[db].begin() as sess: for _ in range(5): - sess.add(admin_client.app["model"]()) + sess.add(admin_client.app[model]()) - url = admin_client.app["admin"].router["dummy_delete_many"].url_for() + url = admin_client.app[admin].router["dummy_delete_many"].url_for() p = {"ids": '["2", "3", "9"]'} async with admin_client.delete(url, params=p, headers=h) as resp: assert resp.status == 200 assert await resp.json() == {"data": ["2", "3"]} - async with admin_client.app["db"]() as sess: - r = await sess.scalars(sa.select(admin_client.app["model"])) + async with admin_client.app[db]() as sess: + r = await sess.scalars(sa.select(admin_client.app[model])) models = r.all() assert len(models) == 4 assert {m.id for m in models} == {1, 4, 5, 6} - url = admin_client.app["admin"].router["dummy_delete_many"].url_for() + url = admin_client.app[admin].router["dummy_delete_many"].url_for() p = {"ids": '["12", "13"]'} async with admin_client.delete(url, params=p, headers=h) as resp: assert resp.status == 404 - async with admin_client.app["db"]() as sess: - r = await sess.scalars(sa.select(admin_client.app["model"])) + async with admin_client.app[db]() as sess: + r = await sess.scalars(sa.select(admin_client.app[model])) assert len(r.all()) == 4