leocrm-bot
434 lines
14 KiB
Python
434 lines
14 KiB
Python
"""Tests for Entity Links plugin — link, unlink, reverse links, multi-links, event cleanup."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
|
|
|
|
from app.core.db import close_engine, reset_engine_for_testing
|
|
from app.core.event_bus import get_event_bus
|
|
from app.core.service_container import get_container
|
|
from app.main import create_app
|
|
from app.plugins.builtins.entity_links import EntityLinksPlugin
|
|
from app.plugins.registry import reset_registry_for_testing
|
|
from app.services.plugin_service import reset_plugin_service_for_testing
|
|
from tests.conftest import ORIGIN_HEADER, login_client, seed_tenant_and_users
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def plugin_app(engine: AsyncEngine, redis_client):
|
|
"""FastAPI app with entity_links plugin registered."""
|
|
reset_engine_for_testing(engine)
|
|
app = create_app()
|
|
|
|
registry = reset_registry_for_testing()
|
|
registry.initialize(engine, app)
|
|
|
|
container = get_container()
|
|
await container.initialize()
|
|
|
|
registry.register_plugin(EntityLinksPlugin())
|
|
reset_plugin_service_for_testing(registry)
|
|
|
|
yield app
|
|
await close_engine()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def plugin_client(plugin_app) -> AsyncClient:
|
|
transport = ASGITransport(app=plugin_app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as c:
|
|
yield c
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def authed_client(plugin_client: AsyncClient, db_session: AsyncSession) -> AsyncClient:
|
|
"""Authenticated admin client with seeded data."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
await login_client(plugin_client, "admin@tenanta.com")
|
|
resp = await plugin_client.post("/api/v1/plugins/entity_links/install", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
resp = await plugin_client.post("/api/v1/plugins/entity_links/activate", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
return plugin_client, seed
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_link_file_to_company(authed_client: AsyncClient):
|
|
"""AC2: POST /api/v1/dms/files/{id}/link → 200, file linked to entity."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
company_id = str(seed["company_a"].id)
|
|
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/link",
|
|
json={"entity_type": "company", "entity_id": company_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["file_id"] == file_id
|
|
assert data["entity_type"] == "company"
|
|
assert data["entity_id"] == company_id
|
|
assert data["already_linked"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_link_file_to_contact(authed_client: AsyncClient):
|
|
"""POST /api/v1/dms/files/{id}/link → 200, file linked to contact."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
# Use a random UUID for contact (no contact seeded, but link is N:M metadata)
|
|
contact_id = str(uuid.uuid4())
|
|
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/link",
|
|
json={"entity_type": "contact", "entity_id": contact_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["entity_type"] == "contact"
|
|
assert data["entity_id"] == contact_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unlink_file_from_entity(authed_client: AsyncClient):
|
|
"""AC3: DELETE /api/v1/dms/files/{id}/link → 204, link removed."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
company_id = str(seed["company_a"].id)
|
|
|
|
# Link first
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/link",
|
|
json={"entity_type": "company", "entity_id": company_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
# Unlink
|
|
resp = await client.request(
|
|
"DELETE",
|
|
f"/api/v1/dms/files/{file_id}/link",
|
|
json={"entity_type": "company", "entity_id": company_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
# Verify links list is empty
|
|
resp = await client.get(f"/api/v1/dms/files/{file_id}/links", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
assert resp.json() == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_file_links(authed_client: AsyncClient):
|
|
"""GET /api/v1/dms/files/{id}/links → 200, list all linked entities for file."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
company_id = str(seed["company_a"].id)
|
|
contact_id = str(uuid.uuid4())
|
|
|
|
# Link to company
|
|
await client.post(
|
|
f"/api/v1/dms/files/{file_id}/link",
|
|
json={"entity_type": "company", "entity_id": company_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
# Link to contact
|
|
await client.post(
|
|
f"/api/v1/dms/files/{file_id}/link",
|
|
json={"entity_type": "contact", "entity_id": contact_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
|
|
resp = await client.get(f"/api/v1/dms/files/{file_id}/links", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multi_links_one_file_many_entities(authed_client: AsyncClient):
|
|
"""Multi-links: one file → many entities."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
|
|
# Link to 3 different companies
|
|
for _ in range(3):
|
|
entity_id = str(uuid.uuid4())
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/link",
|
|
json={"entity_type": "company", "entity_id": entity_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
resp = await client.get(f"/api/v1/dms/files/{file_id}/links", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()) == 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reverse_link_company_files(authed_client: AsyncClient):
|
|
"""GET /api/v1/companies/{id}/files → 200, list linked files for company."""
|
|
client, seed = authed_client
|
|
company_id = str(seed["company_a"].id)
|
|
|
|
# Link 2 files to the company
|
|
for _ in range(2):
|
|
file_id = str(uuid.uuid4())
|
|
await client.post(
|
|
f"/api/v1/dms/files/{file_id}/link",
|
|
json={"entity_type": "company", "entity_id": company_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
|
|
resp = await client.get(f"/api/v1/companies/{company_id}/files", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data) == 2
|
|
assert all(link["entity_type"] == "company" for link in data)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reverse_link_contact_files(authed_client: AsyncClient):
|
|
"""GET /api/v1/contacts/{id}/files → 200, list linked files for contact."""
|
|
client, seed = authed_client
|
|
contact_id = str(uuid.uuid4())
|
|
|
|
# Link 1 file to the contact
|
|
file_id = str(uuid.uuid4())
|
|
await client.post(
|
|
f"/api/v1/dms/files/{file_id}/link",
|
|
json={"entity_type": "contact", "entity_id": contact_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
|
|
resp = await client.get(f"/api/v1/contacts/{contact_id}/files", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data) == 1
|
|
assert data[0]["entity_type"] == "contact"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_event_cleanup_on_company_deleted(authed_client: AsyncClient):
|
|
"""AC13: DMS plugin listens to company.deleted event → linked files cleanup."""
|
|
client, seed = authed_client
|
|
company_id = seed["company_a"].id
|
|
tenant_id = seed["tenant_a"].id
|
|
|
|
# Link a file to the company
|
|
file_id = str(uuid.uuid4())
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/link",
|
|
json={"entity_type": "company", "entity_id": str(company_id)},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
# Verify link exists
|
|
resp = await client.get(f"/api/v1/companies/{company_id}/files", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()) == 1
|
|
|
|
# Publish company.deleted event
|
|
event_bus = get_event_bus()
|
|
await event_bus.publish(
|
|
"company.deleted",
|
|
{
|
|
"entity_id": str(company_id),
|
|
"company_id": str(company_id),
|
|
"tenant_id": str(tenant_id),
|
|
},
|
|
)
|
|
|
|
# Allow async event handler to complete
|
|
import asyncio
|
|
|
|
await asyncio.sleep(0.1)
|
|
|
|
# Verify link is cleaned up
|
|
resp = await client.get(f"/api/v1/companies/{company_id}/files", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
assert resp.json() == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_event_cleanup_on_contact_deleted(authed_client: AsyncClient):
|
|
"""Event cleanup on contact.deleted → linked files removed."""
|
|
client, seed = authed_client
|
|
contact_id = uuid.uuid4()
|
|
tenant_id = seed["tenant_a"].id
|
|
|
|
# Link a file to the contact
|
|
file_id = str(uuid.uuid4())
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/link",
|
|
json={"entity_type": "contact", "entity_id": str(contact_id)},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
# Verify link exists
|
|
resp = await client.get(f"/api/v1/contacts/{contact_id}/files", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()) == 1
|
|
|
|
# Publish contact.deleted event
|
|
event_bus = get_event_bus()
|
|
await event_bus.publish(
|
|
"contact.deleted",
|
|
{
|
|
"entity_id": str(contact_id),
|
|
"contact_id": str(contact_id),
|
|
"tenant_id": str(tenant_id),
|
|
},
|
|
)
|
|
|
|
# Allow async event handler to complete
|
|
import asyncio
|
|
|
|
await asyncio.sleep(0.1)
|
|
|
|
# Verify link is cleaned up
|
|
resp = await client.get(f"/api/v1/contacts/{contact_id}/files", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
assert resp.json() == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_link_invalid_file_id(authed_client: AsyncClient):
|
|
"""POST /api/v1/dms/files/{invalid}/link → 400."""
|
|
client, seed = authed_client
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/bad-uuid/link",
|
|
json={"entity_type": "company", "entity_id": str(uuid.uuid4())},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_link_invalid_entity_id(authed_client: AsyncClient):
|
|
"""POST /api/v1/dms/files/{id}/link with invalid entity_id → 400."""
|
|
client, seed = authed_client
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{uuid.uuid4()}/link",
|
|
json={"entity_type": "company", "entity_id": "bad-uuid"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_link_invalid_entity_type(authed_client: AsyncClient):
|
|
"""POST /api/v1/dms/files/{id}/link with invalid entity_type → 400."""
|
|
client, seed = authed_client
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{uuid.uuid4()}/link",
|
|
json={"entity_type": "invalid", "entity_id": str(uuid.uuid4())},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_link_already_linked(authed_client: AsyncClient):
|
|
"""POST /api/v1/dms/files/{id}/link twice → already_linked=True."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
company_id = str(seed["company_a"].id)
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/link",
|
|
json={"entity_type": "company", "entity_id": company_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["already_linked"] is False
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/link",
|
|
json={"entity_type": "company", "entity_id": company_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["already_linked"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unlink_not_found(authed_client: AsyncClient):
|
|
"""DELETE /api/v1/dms/files/{id}/link with nonexistent link → 404."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
company_id = str(seed["company_a"].id)
|
|
resp = await client.request(
|
|
"DELETE",
|
|
f"/api/v1/dms/files/{file_id}/link",
|
|
json={"entity_type": "company", "entity_id": company_id},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unlink_invalid_file_id(authed_client: AsyncClient):
|
|
"""DELETE /api/v1/dms/files/{invalid}/link → 400."""
|
|
client, seed = authed_client
|
|
resp = await client.request(
|
|
"DELETE",
|
|
"/api/v1/dms/files/bad-uuid/link",
|
|
json={"entity_type": "company", "entity_id": str(uuid.uuid4())},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_file_links_invalid_id(authed_client: AsyncClient):
|
|
"""GET /api/v1/dms/files/{invalid}/links → 400."""
|
|
client, seed = authed_client
|
|
resp = await client.get("/api/v1/dms/files/bad-uuid/links", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_company_files_invalid_id(authed_client: AsyncClient):
|
|
"""GET /api/v1/companies/{invalid}/files → 400."""
|
|
client, seed = authed_client
|
|
resp = await client.get("/api/v1/companies/bad-uuid/files", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_contact_files_invalid_id(authed_client: AsyncClient):
|
|
"""GET /api/v1/contacts/{invalid}/files → 400."""
|
|
client, seed = authed_client
|
|
resp = await client.get("/api/v1/contacts/bad-uuid/files", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_company_files_empty(authed_client: AsyncClient):
|
|
"""GET /api/v1/companies/{id}/files with no links → 200 + empty list."""
|
|
client, seed = authed_client
|
|
company_id = str(seed["company_a"].id)
|
|
resp = await client.get(f"/api/v1/companies/{company_id}/files", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
assert resp.json() == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_contact_files_empty(authed_client: AsyncClient):
|
|
"""GET /api/v1/contacts/{id}/files with no links → 200 + empty list."""
|
|
client, seed = authed_client
|
|
contact_id = str(uuid.uuid4())
|
|
resp = await client.get(f"/api/v1/contacts/{contact_id}/files", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
assert resp.json() == []
|