feat(#357): custom_field_definitions generisch — W4b-Muster (422/403-Entity-Checks, {items,total}-Shape, ACL-Fix), zentrale Helper, Plural-Ableitungs-Fix

This commit is contained in:
Agent Zero
2026-08-29 01:27:24 +02:00
parent 36dd7c5101
commit b5036a1fc0
8 changed files with 442 additions and 71 deletions
+1 -1
View File
@@ -9,7 +9,7 @@
1. **Saved-Filters 422-Test** — ✅ erledigt (d7b3c7c... der Fix wurde 2026-08-28 in `0d052ab` deployed)
2. **AppShell ×4 + Router ×2 Mock-Fixes** — ✅ erledigt (2026-08-28): AppShell war bereits grün (useCurrentUser-Mock existierte in Zeile 9; Plan-Eintrag veraltet, bewiesen: Lauf 23:00 'Tests 2 failed (6)' = nur Router). Router.test.tsx gefixt: QueryClientProvider + Mocks useCurrentUser/useUserPermissions → **2/2 passed (2.03s)**
3. **ContactEditModal File-Level** — ✅ erledigt (2026-08-28): Geister-Test nach §10 gelöscht. Ursache bewiesen: ContactEditModal.tsx (349 Z.) wurde in db4701b (BUG-080/082) gelöscht, Test blieb → Vitest 'Failed to resolve import' (1 failed, no tests). Ersatz ContactEditForm.tsx lebt und wird von ContactsList/ContactDetailPage genutzt
4. **custom_field_definitions generisch machen**ENTITY_PLUGIN_OWNERS-Muster wie W4b anwenden (app/routes/custom_field_definitions.py + app/services/entity_permission_service.py)
4. **custom_field_definitions generisch machen**✅ erledigt (2026-08-29): W4b-Muster komplett angewendet. Route: 422-Entity-Validierung + 403-Owner-Modul-Check (contacts:read etc.) auf create/list/update/delete, ACL-Loch gefixt (delete übergab user_id nie → 500 für alle Nicht-Admins), PermissionError→403-Handler. Shape-Fix: Backend lieferte Array, alle 6 Frontend-Consumer lesen `data.items` → CustomFields-UI zeigte seit jeher leer; jetzt `{items,total}` (CustomFieldDefinitionListResponse). entity_permission_service: Plural-Ableitungs-Fix ('workflow'→workflows:read statt Phantom contacts:read; 'address'→addresses:read via +es), unregister_entity_model räumt ENTITY_PLUGIN_OWNERS mit auf (Lifecycle-Leak), zentrale Helper validate_entity_type/check_entity_read_permission — saved_filters+saved_views Duplikate entfernt (Aliase, Call-Sites unverändert). Neue Suite tests/test_custom_field_definitions.py **13 Tests**: Rot bewiesen 10 failed/4 passed → Grün **25 passed** (13 cfd + 12 saved_filters-Regression), Permission-Suiten 22/22, custom_fields+lifecycle 13/13, ruff exit=0, create_app OK. 3er-Kombi-Failures (custom_fields+entity_registry+lifecycle) per Stash als identischer Vorbestand bewiesen (clean HEAD: gleiche 7 Failures — Suite-Isolation, kein Zusammenhang mit Änderung). api-documentation.md ergänzt (4 Endpoints)
5. **Sidebar /contacts statische Route entfernen** — routes/index.tsx Zeile 250; erst PluginRouteRenderer beweisen (Kritikpunkt 21: Production-Build + Reload-Test), dann entfernen
6. **Kontakt-Model ins ContactsPlugin** — groß (~30 Import-Stellen, Alembic-Kette), bewusst zurückgestellt
7. **Phase L: Dokumente-Generator** — PLATFORM_ROADMAP.md 'Phase L' (L1-L5, ~9-15 Tage), user-abgestimmt, Basis: report_generator-Plugin
+61 -10
View File
@@ -1,4 +1,10 @@
"""API routes for CustomFieldDefinition CRUD."""
"""API routes for CustomFieldDefinition CRUD — generic across entities.
Paket 4 (#357): applies the W4b pattern — entity validation (422) and
owner-module read check (403) via ENTITY_PLUGIN_OWNERS, central helpers
from entity_permission_service, and the {items, total} list shape all
frontend consumers read.
"""
from __future__ import annotations
@@ -11,17 +17,22 @@ from app.core.db import get_db
from app.deps import get_current_user, require_permission
from app.schemas.custom_field_definition import (
CustomFieldDefinitionCreate,
CustomFieldDefinitionListResponse,
CustomFieldDefinitionResponse,
CustomFieldDefinitionUpdate,
)
from app.services import custom_field_service
from app.services.entity_permission_service import (
check_entity_read_permission,
validate_entity_type,
)
router = APIRouter(prefix="/api/v1/custom-fields", tags=["custom-fields-definitions"])
@router.get(
"/definitions",
response_model=list[CustomFieldDefinitionResponse],
response_model=CustomFieldDefinitionListResponse,
dependencies=[Depends(require_permission("custom_fields:read"))],
)
async def list_definitions(
@@ -29,10 +40,28 @@ async def list_definitions(
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""List all active custom field definitions for the current tenant."""
"""List all active custom field definitions for the current tenant.
Returns ``{items, total}`` the shape the frontend API client and
all six consumers (CustomFields page, ContactsList, SortPanel,
GroupPanel, FilterPanel) read.
"""
tenant_id = uuid.UUID(current_user["tenant_id"])
definitions = await custom_field_service.list_definitions(db, tenant_id, entity=entity)
return definitions
if entity:
validate_entity_type(entity)
check_entity_read_permission(current_user, entity)
definitions = await custom_field_service.list_definitions(
db,
tenant_id,
entity=entity,
user_id=uuid.UUID(current_user["user_id"]),
is_system_admin=current_user.get("is_system_admin", False),
)
return CustomFieldDefinitionListResponse(
items=[CustomFieldDefinitionResponse.model_validate(d) for d in definitions],
total=len(definitions),
)
@router.post(
@@ -46,7 +75,16 @@ async def create_definition(
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""Create a new custom field definition."""
"""Create a new custom field definition.
W4b: the entity must be a registered entity type (422) and the user
needs the owning module's read permission (e.g. contacts:read) on top
of custom_fields:write writing definitions for modules you cannot
even read would bypass module isolation.
"""
validate_entity_type(body.entity)
check_entity_read_permission(current_user, body.entity)
tenant_id = uuid.UUID(current_user["tenant_id"])
user_id = uuid.UUID(current_user["user_id"])
definition = await custom_field_service.create_definition(
@@ -69,6 +107,7 @@ async def update_definition(
"""Update an existing custom field definition."""
tenant_id = uuid.UUID(current_user["tenant_id"])
user_id = uuid.UUID(current_user["user_id"])
is_admin = current_user.get("is_system_admin", False)
try:
def_id = uuid.UUID(definition_id)
except (ValueError, TypeError):
@@ -79,11 +118,16 @@ async def update_definition(
if not update_data:
raise HTTPException(400, detail={"detail": "No fields to update", "code": "no_updates"})
definition = await custom_field_service.update_definition(
db, tenant_id, def_id, update_data, user_id=user_id
)
try:
definition = await custom_field_service.update_definition(
db, tenant_id, def_id, update_data, user_id=user_id, is_system_admin=is_admin
)
except PermissionError as e:
raise HTTPException(status_code=403, detail={"detail": str(e), "code": "forbidden"}) from e
if definition is None:
raise HTTPException(404, detail={"detail": "Definition not found", "code": "not_found"})
# W4b: changing a definition still requires read access to its entity module
check_entity_read_permission(current_user, definition.entity)
return definition
@@ -99,12 +143,19 @@ async def delete_definition(
):
"""Delete a custom field definition."""
tenant_id = uuid.UUID(current_user["tenant_id"])
user_id = uuid.UUID(current_user["user_id"])
is_admin = current_user.get("is_system_admin", False)
try:
def_id = uuid.UUID(definition_id)
except (ValueError, TypeError):
raise HTTPException(400, detail={"detail": "Invalid definition_id", "code": "invalid_id"}) from None
deleted = await custom_field_service.delete_definition(db, tenant_id, def_id)
try:
deleted = await custom_field_service.delete_definition(
db, tenant_id, def_id, user_id=user_id, is_system_admin=is_admin
)
except PermissionError as e:
raise HTTPException(status_code=403, detail={"detail": str(e), "code": "forbidden"}) from e
if not deleted:
raise HTTPException(404, detail={"detail": "Definition not found", "code": "not_found"})
return None
+6 -28
View File
@@ -18,37 +18,15 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.core.db import get_db
from app.deps import get_current_user
from app.models.saved_filter import SavedFilter
from app.services.entity_permission_service import (
check_entity_read_permission as _check_entity_read,
)
from app.services.entity_permission_service import (
validate_entity_type as _validate_entity_type,
)
router = APIRouter(prefix="/api/v1/saved-filters", tags=["saved-filters"])
VALID_ENTITY_TYPES = None # Dynamic — validated against ENTITY_MODELS at runtime
def _validate_entity_type(entity_type: str) -> None:
"""Validate entity_type against ENTITY_MODELS. Raises HTTPException if invalid."""
from app.services.entity_permission_service import ENTITY_MODELS
if entity_type not in ENTITY_MODELS:
from fastapi import HTTPException
valid = sorted(ENTITY_MODELS.keys())
raise HTTPException(422, detail={
"detail": f"Invalid entity_type: {entity_type}",
"code": "invalid_entity_type",
"valid_types": valid,
})
def _check_entity_read(current_user: dict, entity_type: str) -> None:
"""Check that the user has read permission for the entity type."""
from app.core.permissions import check_permission
from app.services.entity_permission_service import get_entity_read_permission
perm = get_entity_read_permission(entity_type)
if not check_permission(current_user, perm):
raise HTTPException(403, detail={
"detail": f"Permission '{perm}' required",
"code": "forbidden",
})
class SavedFilterCreate(BaseModel):
"""Schema for creating a saved filter."""
+6 -28
View File
@@ -18,37 +18,15 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.core.db import get_db
from app.deps import get_current_user
from app.models.saved_view import SavedView
from app.services.entity_permission_service import (
check_entity_read_permission as _check_entity_read,
)
from app.services.entity_permission_service import (
validate_entity_type as _validate_entity_type,
)
router = APIRouter(prefix="/api/v1/saved-views", tags=["saved-views"])
VALID_ENTITY_TYPES = None # Dynamic — validated against ENTITY_MODELS at runtime
def _validate_entity_type(entity_type: str) -> None:
"""Validate entity_type against ENTITY_MODELS. Raises HTTPException if invalid."""
from app.services.entity_permission_service import ENTITY_MODELS
if entity_type not in ENTITY_MODELS:
from fastapi import HTTPException
valid = sorted(ENTITY_MODELS.keys())
raise HTTPException(422, detail={
"detail": f"Invalid entity_type: {entity_type}",
"code": "invalid_entity_type",
"valid_types": valid,
})
def _check_entity_read(current_user: dict, entity_type: str) -> None:
"""Check that the user has read permission for the entity type."""
from app.core.permissions import check_permission
from app.services.entity_permission_service import get_entity_read_permission
perm = get_entity_read_permission(entity_type)
if not check_permission(current_user, perm):
raise HTTPException(403, detail={
"detail": f"Permission '{perm}' required",
"code": "forbidden",
})
class SavedViewCreate(BaseModel):
"""Schema for creating a saved view."""
+7
View File
@@ -62,3 +62,10 @@ class CustomFieldDefinitionResponse(BaseModel):
updated_by: uuid.UUID | None = None
created_at: datetime
updated_at: datetime
class CustomFieldDefinitionListResponse(BaseModel):
"""List response — {items, total} shape consumed by the frontend."""
items: list[CustomFieldDefinitionResponse]
total: int
+45 -4
View File
@@ -84,10 +84,15 @@ def get_entity_read_permission(entity_type: str) -> str:
owner = ENTITY_PLUGIN_OWNERS.get(entity_type)
if owner:
return f"{owner}:read"
# Core entities: derive from module name (e.g. workflows → workflows:read)
module = entity_type.rstrip("s")
candidates = [k for k in _core_module_keys(module, "read")]
return candidates[0] if candidates else "contacts:read"
# Core entities: derive from the module name in CORE_PERMISSIONS. Modules
# are mostly plural ("workflows", "addresses") while entity types are
# mostly singular ("workflow", "address") — try exact, singular and
# plural forms before the contacts:read fallback.
for module in (entity_type, entity_type.rstrip("s"), f"{entity_type}s", f"{entity_type}es"):
candidates = _core_module_keys(module, "read")
if candidates:
return candidates[0]
return "contacts:read"
def _core_module_keys(module: str, action: str) -> list[str]:
@@ -100,6 +105,41 @@ def _core_module_keys(module: str, action: str) -> list[str]:
if p.get("module") == module and p["key"].endswith(f":{action}")
]
def validate_entity_type(entity_type: str) -> None:
"""Validate entity_type against ENTITY_MODELS (W4b pattern).
Central helper for entity-typed CRUD (saved filters/views,
custom field definitions). Raises fastapi HTTPException 422
with the valid types so clients can self-correct.
"""
if entity_type not in ENTITY_MODELS:
from fastapi import HTTPException
raise HTTPException(422, detail={
"detail": f"Invalid entity_type: {entity_type}",
"code": "invalid_entity_type",
"valid_types": sorted(ENTITY_MODELS.keys()),
})
def check_entity_read_permission(current_user: dict, entity_type: str) -> None:
"""Check that the user may read the entity type's owning module (W4b).
Raises fastapi HTTPException 403 when the derived module read
permission (e.g. contacts:read, workflows:read) is missing.
"""
from fastapi import HTTPException
from app.core.permissions import check_permission
perm = get_entity_read_permission(entity_type)
if not check_permission(current_user, perm):
raise HTTPException(403, detail={
"detail": f"Permission '{perm}' required",
"code": "forbidden",
})
# Core models with OwnedMixin (Phase 2 additions)
try:
from app.models.entity_attachment import EntityAttachment
@@ -130,6 +170,7 @@ def register_entity_model(
def unregister_entity_model(entity_type: str) -> None:
"""Unregister an entity model (called during plugin deactivation)."""
ENTITY_MODELS.pop(entity_type, None)
ENTITY_PLUGIN_OWNERS.pop(entity_type, None)
def _get_entity_model(entity_type: str) -> type:
+11
View File
@@ -228,6 +228,17 @@ List endpoints use `page` (1-based) and `page_size` (1-100) query parameters. Re
| PATCH | `/api/v1/sequences/{sequence_id}` | Update a sequence. |
| DELETE | `/api/v1/sequences/{sequence_id}` | Delete a sequence. |
### custom-fields-definitions (4 endpoints)
Generic CRUD for tenant-wide custom field definitions per entity type (W4b pattern: entity must be a registered entity type — 422; user needs the owning module's read permission, e.g. `contacts:read`, on top of `custom_fields:read/write` — 403).
| Method | Path | Description |
|--------|------|-------------|
| GET | `/api/v1/custom-fields/definitions` | List definitions, optional `?entity=` filter. **Response**: `{items: CustomFieldDefinition[], total: int}` |
| POST | `/api/v1/custom-fields/definitions` | Create a definition. **Request**: `CustomFieldDefinitionCreate` |
| PATCH | `/api/v1/custom-fields/definitions/{definition_id}` | Update a definition. **Request**: `CustomFieldDefinitionUpdate` |
| DELETE | `/api/v1/custom-fields/definitions/{definition_id}` | Delete a definition (204). |
### system-settings (6 endpoints)
| Method | Path | Description |
+305
View File
@@ -0,0 +1,305 @@
"""Custom field definitions tests — generic entity CRUD (W4b pattern).
Paket 4 (#357): custom_field_definitions generisch machen —
entity validation (422), owner-module read check (403),
{items,total} response shape, PermissionError 403 (not 500).
"""
from __future__ import annotations
import pytest
from httpx import AsyncClient
from tests.conftest import ORIGIN_HEADER, login_client, seed_tenant_and_users
DEF_BASE = "/api/v1/custom-fields/definitions"
async def _create_cf_only_user(db, seed) -> None:
"""User with custom_fields:read/write but NO contacts:read (owner check)."""
from app.core.auth import hash_password
from app.models.role import Role
from app.models.user import User, UserTenant
role = Role(
tenant_id=seed["tenant_a"].id,
name="cf_only",
permissions={"custom_fields": {"read": True, "write": True}},
denied_permissions=[],
field_permissions={},
)
db.add(role)
await db.flush()
user = User(
email="cfonly@tenanta.com",
name="CF Only",
password_hash=hash_password("TestPass123!"),
is_active=True,
preferences={},
)
db.add(user)
await db.flush()
ut = UserTenant(
user_id=user.id,
tenant_id=seed["tenant_a"].id,
is_default=True,
role="cf_only",
role_id=role.id,
)
db.add(ut)
await db.flush()
await db.commit()
async def _create_definition(client: AsyncClient) -> str:
"""Create a contact definition as admin, return its id."""
resp = await client.post(
DEF_BASE,
json={
"entity": "contact",
"name": "test_field",
"label": "Test Field",
"field_type": "text",
},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 201, f"Setup create failed: {resp.status_code} {resp.text}"
return resp.json()["id"]
@pytest.mark.asyncio
class TestCustomFieldDefinitionCreate:
"""POST /api/v1/custom-fields/definitions"""
async def test_create_definition_returns_201(self, client: AsyncClient, db_session):
"""Admin can create a definition for a registered entity."""
await seed_tenant_and_users(db_session)
await login_client(client, "admin@tenanta.com")
resp = await client.post(
DEF_BASE,
json={
"entity": "contact",
"name": "lead_score",
"label": "Lead Score",
"field_type": "number",
},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 201
data = resp.json()
assert data["entity"] == "contact"
assert data["name"] == "lead_score"
async def test_create_definition_invalid_entity_returns_422(
self, client: AsyncClient, db_session
):
"""Create with unregistered entity type returns 422 (W4b pattern)."""
await seed_tenant_and_users(db_session)
await login_client(client, "admin@tenanta.com")
resp = await client.post(
DEF_BASE,
json={
"entity": "invalid_entity_xyz",
"name": "bad_field",
"label": "Bad",
"field_type": "text",
},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 422
async def test_create_definition_without_entity_read_returns_403(
self, client: AsyncClient, db_session
):
"""cf-only user (custom_fields:write, no contacts:read) gets 403."""
seed = await seed_tenant_and_users(db_session)
await _create_cf_only_user(db_session, seed)
await login_client(client, "cfonly@tenanta.com")
resp = await client.post(
DEF_BASE,
json={
"entity": "contact",
"name": "no_perm_field",
"label": "No Perm",
"field_type": "text",
},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 403
@pytest.mark.asyncio
class TestCustomFieldDefinitionList:
"""GET /api/v1/custom-fields/definitions"""
async def test_list_definitions_returns_items_shape(
self, client: AsyncClient, db_session
):
"""List returns {items, total} — the shape all 6 frontend consumers read."""
await seed_tenant_and_users(db_session)
await login_client(client, "admin@tenanta.com")
await _create_definition(client)
resp = await client.get(DEF_BASE, headers=ORIGIN_HEADER)
assert resp.status_code == 200
data = resp.json()
assert isinstance(data, dict)
assert "items" in data
assert "total" in data
assert data["total"] == 1
assert len(data["items"]) == 1
assert data["items"][0]["entity"] == "contact"
async def test_list_definitions_with_entity_filter(
self, client: AsyncClient, db_session
):
"""List ?entity=contact returns only contact definitions."""
await seed_tenant_and_users(db_session)
await login_client(client, "admin@tenanta.com")
await _create_definition(client)
resp = await client.get(
f"{DEF_BASE}?entity=contact", headers=ORIGIN_HEADER
)
assert resp.status_code == 200
data = resp.json()
assert data["total"] == 1
assert all(item["entity"] == "contact" for item in data["items"])
async def test_list_definitions_invalid_entity_returns_422(
self, client: AsyncClient, db_session
):
"""List with unregistered entity filter returns 422 (consistent with create)."""
await seed_tenant_and_users(db_session)
await login_client(client, "admin@tenanta.com")
resp = await client.get(
f"{DEF_BASE}?entity=invalid_entity_xyz", headers=ORIGIN_HEADER
)
assert resp.status_code == 422
async def test_list_definitions_without_entity_read_returns_403(
self, client: AsyncClient, db_session
):
"""cf-only user listing contact definitions gets 403 (owner read check)."""
seed = await seed_tenant_and_users(db_session)
await _create_cf_only_user(db_session, seed)
await login_client(client, "cfonly@tenanta.com")
resp = await client.get(
f"{DEF_BASE}?entity=contact", headers=ORIGIN_HEADER
)
assert resp.status_code == 403
@pytest.mark.asyncio
class TestCustomFieldDefinitionUpdate:
"""PATCH /api/v1/custom-fields/definitions/{id}"""
async def test_update_definition_returns_200(
self, client: AsyncClient, db_session
):
"""Owner can update their own definition."""
await seed_tenant_and_users(db_session)
await login_client(client, "admin@tenanta.com")
def_id = await _create_definition(client)
resp = await client.patch(
f"{DEF_BASE}/{def_id}",
json={"label": "Updated Label"},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 200
assert resp.json()["label"] == "Updated Label"
async def test_update_definition_other_owner_returns_403_not_500(
self, client: AsyncClient, db_session
):
"""Viewer updating admin's definition gets 403, never a 500."""
await seed_tenant_and_users(db_session)
await login_client(client, "admin@tenanta.com")
def_id = await _create_definition(client)
await login_client(client, "viewer@tenanta.com")
resp = await client.patch(
f"{DEF_BASE}/{def_id}",
json={"label": "Hacked"},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 403
@pytest.mark.asyncio
class TestCustomFieldDefinitionDelete:
"""DELETE /api/v1/custom-fields/definitions/{id}"""
async def test_delete_definition_returns_204(
self, client: AsyncClient, db_session
):
"""Owner can delete their own definition."""
await seed_tenant_and_users(db_session)
await login_client(client, "admin@tenanta.com")
def_id = await _create_definition(client)
resp = await client.delete(f"{DEF_BASE}/{def_id}", headers=ORIGIN_HEADER)
assert resp.status_code == 204
async def test_delete_definition_other_owner_returns_403_not_500(
self, client: AsyncClient, db_session
):
"""Viewer deleting admin's definition gets 403, never a 500.
Today delete_definition is called WITHOUT user_id ACL check runs
with user_id=None PermissionError unhandled 500.
"""
await seed_tenant_and_users(db_session)
await login_client(client, "admin@tenanta.com")
def_id = await _create_definition(client)
await login_client(client, "viewer@tenanta.com")
resp = await client.delete(f"{DEF_BASE}/{def_id}", headers=ORIGIN_HEADER)
assert resp.status_code == 403
class TestEntityPermissionDerivation:
"""Unit tests: get_entity_read_permission + owner-registry cleanup."""
def test_get_entity_read_permission_plural_core_module(self):
"""'workflow' derives workflows:read — not the contacts:read fallback."""
from app.services.entity_permission_service import (
ENTITY_PLUGIN_OWNERS,
get_entity_read_permission,
)
saved_owner = ENTITY_PLUGIN_OWNERS.pop("workflow", None)
try:
assert get_entity_read_permission("workflow") == "workflows:read"
finally:
if saved_owner is not None:
ENTITY_PLUGIN_OWNERS["workflow"] = saved_owner
def test_get_entity_read_permission_singular_core_module(self):
"""'address' derives addresses:read (module is plural in registry)."""
from app.services.entity_permission_service import (
ENTITY_PLUGIN_OWNERS,
get_entity_read_permission,
)
saved_owner = ENTITY_PLUGIN_OWNERS.pop("address", None)
try:
assert get_entity_read_permission("address") == "addresses:read"
finally:
if saved_owner is not None:
ENTITY_PLUGIN_OWNERS["address"] = saved_owner
def test_unregister_entity_model_clears_owner_mapping(self):
"""unregister_entity_model must clean ENTITY_PLUGIN_OWNERS too."""
from app.models.workflow import Workflow
from app.services.entity_permission_service import (
ENTITY_MODELS,
ENTITY_PLUGIN_OWNERS,
get_entity_read_permission,
register_entity_model,
unregister_entity_model,
)
register_entity_model("zzz_unit_test", Workflow, plugin_name="zzz_plugin")
assert ENTITY_PLUGIN_OWNERS["zzz_unit_test"] == "zzz_plugin"
assert get_entity_read_permission("zzz_unit_test") == "zzz_plugin:read"
unregister_entity_model("zzz_unit_test")
assert "zzz_unit_test" not in ENTITY_PLUGIN_OWNERS
assert "zzz_unit_test" not in ENTITY_MODELS