leocrm/tests/test_custom_field_definitions.py

"""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