46865882c6
PSA abstraction layer with provider pattern, ConnectWise integration (connection management, ticket linking, note posting, status updates, member mapping), Integrations page UI, Fernet credential encryption, in-memory TTL cache, 6 DB migrations, ConnectWise API reference docs.
60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
"""Tests for PSA connection endpoints — routing and RBAC only.
|
|
|
|
We cannot fully test create/update/test endpoints in CI because they
|
|
call the ConnectWise API. These tests verify routing and authorization.
|
|
"""
|
|
import pytest
|
|
from sqlalchemy import select, update
|
|
from app.models.user import User
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_connection_empty(client, admin_auth_headers):
|
|
"""GET returns null when no connection exists."""
|
|
response = await client.get(
|
|
"/api/v1/integrations/psa/connections",
|
|
headers=admin_auth_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_connection_requires_owner(client, test_user, auth_headers, test_db):
|
|
"""Engineer (non-owner) should get 403 on create."""
|
|
# Downgrade the test user from owner to engineer so require_account_owner rejects
|
|
user_id = test_user["user_data"]["id"]
|
|
await test_db.execute(
|
|
update(User).where(User.id == user_id).values(account_role="engineer")
|
|
)
|
|
await test_db.commit()
|
|
|
|
payload = {
|
|
"provider": "connectwise",
|
|
"display_name": "Test CW",
|
|
"site_url": "https://na.myconnectwise.net",
|
|
"company_id": "testmsp",
|
|
"public_key": "pub123",
|
|
"private_key": "priv456",
|
|
"client_id": "client789",
|
|
}
|
|
response = await client.post(
|
|
"/api/v1/integrations/psa/connections",
|
|
json=payload,
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_nonexistent_returns_404(client, admin_auth_headers):
|
|
"""DELETE with a nonexistent ID returns 404."""
|
|
import uuid
|
|
|
|
fake_id = uuid.uuid4()
|
|
response = await client.delete(
|
|
f"/api/v1/integrations/psa/connections/{fake_id}",
|
|
headers=admin_auth_headers,
|
|
)
|
|
assert response.status_code == 404
|