Agent Zero
727d86614e
P0 (7): Auth-bypass removed, migrations fixed, plugin-upload disabled, RLS FORCE+WITH CHECK, plugin double-registration fixed, persistent volume, domain removed P1 (11): User/tenant model, Redis centralized, worker separated, transactional outbox, XSS fixed, DMS chunked streaming, permissions unified, password reset, metrics secured, config/docs fixed, cross-tenant FK P2 (4): Contact model normalized, cross-imports reduced 94%, commands+state machines for contacts/dms/mail/calendar, SPA path-traversal 8 new migrations, 99 unit tests, 13 commands, 8 contracts, 72 files changed
629 lines
22 KiB
Python
629 lines
22 KiB
Python
"""Tests for the Command pattern — Create, Update, Delete, Merge contact commands.
|
|
|
|
Tests cover:
|
|
- Permission checks (viewer denied, admin allowed)
|
|
- Business logic execution
|
|
- Audit log entry creation
|
|
- Outbox event enqueuing
|
|
- State machine validation
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
import uuid as uuid_mod
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.commands.base import CommandResult
|
|
from app.commands.contact_commands import (
|
|
CreateContactCommand,
|
|
UpdateContactCommand,
|
|
DeleteContactCommand,
|
|
MergeContactsCommand,
|
|
)
|
|
from app.core.state_machine import (
|
|
StateMachine,
|
|
StateMachineError,
|
|
contact_state_machine,
|
|
workflow_state_machine,
|
|
)
|
|
from app.models.audit import AuditLog
|
|
from app.models.contact import Contact
|
|
from app.models.outbox import EventOutbox
|
|
|
|
from tests.conftest import ORIGIN_HEADER, login_client, seed_tenant_and_users
|
|
|
|
|
|
# ── State Machine Unit Tests ──
|
|
|
|
|
|
class TestStateMachine:
|
|
"""Unit tests for the StateMachine class."""
|
|
|
|
def test_can_transition_allowed(self):
|
|
"""Allowed transitions return True."""
|
|
assert contact_state_machine.can_transition("lead", "qualified") is True
|
|
assert contact_state_machine.can_transition("qualified", "customer") is True
|
|
assert contact_state_machine.can_transition("customer", "inactive") is True
|
|
assert contact_state_machine.can_transition("inactive", "lead") is True
|
|
|
|
def test_can_transition_disallowed(self):
|
|
"""Disallowed transitions return False."""
|
|
assert contact_state_machine.can_transition("customer", "lead") is False
|
|
assert contact_state_machine.can_transition("inactive", "customer") is False
|
|
|
|
def test_transition_success(self):
|
|
"""Valid transition returns the new state."""
|
|
assert contact_state_machine.transition("lead", "qualified") == "qualified"
|
|
assert workflow_state_machine.transition("active", "paused") == "paused"
|
|
|
|
def test_transition_invalid_raises(self):
|
|
"""Invalid transition raises StateMachineError."""
|
|
with pytest.raises(StateMachineError, match="Invalid state transition"):
|
|
contact_state_machine.transition("customer", "lead")
|
|
|
|
def test_transition_unknown_state_raises(self):
|
|
"""Transition from unknown state raises StateMachineError."""
|
|
with pytest.raises(StateMachineError, match="Invalid state transition"):
|
|
contact_state_machine.transition("nonexistent", "lead")
|
|
|
|
def test_workflow_state_machine_transitions(self):
|
|
"""Workflow state machine has correct transitions."""
|
|
assert workflow_state_machine.can_transition("draft", "active") is True
|
|
assert workflow_state_machine.can_transition("active", "paused") is True
|
|
assert workflow_state_machine.can_transition("paused", "active") is True
|
|
assert workflow_state_machine.can_transition("active", "completed") is True
|
|
assert workflow_state_machine.can_transition("active", "cancelled") is True
|
|
assert workflow_state_machine.can_transition("paused", "cancelled") is True
|
|
assert workflow_state_machine.can_transition("draft", "cancelled") is True
|
|
assert workflow_state_machine.can_transition("completed", "active") is False
|
|
assert workflow_state_machine.can_transition("cancelled", "active") is False
|
|
|
|
def test_custom_state_machine(self):
|
|
"""Custom state machine with own transitions."""
|
|
sm = StateMachine({"a": ["b"], "b": ["c"], "c": []})
|
|
assert sm.can_transition("a", "b") is True
|
|
assert sm.can_transition("b", "c") is True
|
|
assert sm.can_transition("c", "a") is False
|
|
assert sm.transition("a", "b") == "b"
|
|
with pytest.raises(StateMachineError):
|
|
sm.transition("c", "a")
|
|
|
|
|
|
# ── CommandResult Tests ──
|
|
|
|
|
|
class TestCommandResult:
|
|
"""Unit tests for CommandResult."""
|
|
|
|
def test_ok_result(self):
|
|
"""ok() creates a successful result."""
|
|
result = CommandResult.ok(data={"id": "123"})
|
|
assert result.success is True
|
|
assert result.data == {"id": "123"}
|
|
assert result.error is None
|
|
assert result.events == []
|
|
|
|
def test_ok_with_events(self):
|
|
"""ok() with events creates a successful result with events."""
|
|
events = [{"event": "contact.created"}]
|
|
result = CommandResult.ok(data={}, events=events)
|
|
assert result.events == events
|
|
|
|
def test_fail_result(self):
|
|
"""fail() creates a failed result."""
|
|
result = CommandResult.fail("Something went wrong")
|
|
assert result.success is False
|
|
assert result.data is None
|
|
assert result.error == "Something went wrong"
|
|
assert result.events == []
|
|
|
|
|
|
# ── CreateContactCommand Tests ──
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestCreateContactCommand:
|
|
"""Tests for CreateContactCommand."""
|
|
|
|
async def test_create_contact_admin_success(
|
|
self, client, db_session: AsyncSession, redis_client
|
|
):
|
|
"""Admin can create a contact, audit log is created, outbox event enqueued."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
await login_client(client, "admin@tenanta.com")
|
|
|
|
# Use command directly with db_session and redis_client
|
|
current_user = {
|
|
"user_id": str(seed["admin_a"].id),
|
|
"tenant_id": str(seed["tenant_a"].id),
|
|
"role": "admin",
|
|
"is_system_admin": True,
|
|
"permissions": ["*: *"],
|
|
"denied_permissions": [],
|
|
"field_permissions": {},
|
|
}
|
|
|
|
cmd = CreateContactCommand(data={
|
|
"type": "company",
|
|
"name": "Test Company",
|
|
"email_1": "test@example.com",
|
|
})
|
|
result = await cmd.execute(db_session, redis_client, current_user)
|
|
|
|
assert result.success is True
|
|
assert result.data is not None
|
|
assert result.data["name"] == "Test Company"
|
|
assert result.data["displayname"] == "Test Company"
|
|
assert len(result.events) >= 1
|
|
assert result.events[0]["event"] == "contact.created"
|
|
|
|
# Verify audit log was created
|
|
await db_session.flush()
|
|
audit_q = select(AuditLog).where(
|
|
AuditLog.entity_type == "contact",
|
|
AuditLog.action == "create",
|
|
AuditLog.user_id == seed["admin_a"].id,
|
|
)
|
|
audit_result = await db_session.execute(audit_q)
|
|
audit_entries = audit_result.scalars().all()
|
|
assert len(audit_entries) >= 1
|
|
assert audit_entries[0].entity_id == uuid_mod.UUID(result.data["id"])
|
|
|
|
# Verify outbox event was enqueued
|
|
outbox_q = select(EventOutbox).where(
|
|
EventOutbox.event_name == "contact.created"
|
|
)
|
|
outbox_result = await db_session.execute(outbox_q)
|
|
outbox_entries = outbox_result.scalars().all()
|
|
assert len(outbox_entries) >= 1
|
|
|
|
async def test_create_contact_viewer_denied(
|
|
self, db_session: AsyncSession, redis_client
|
|
):
|
|
"""Viewer cannot create a contact (permission denied)."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
|
|
current_user = {
|
|
"user_id": str(seed["viewer_a"].id),
|
|
"tenant_id": str(seed["tenant_a"].id),
|
|
"role": "viewer",
|
|
"is_system_admin": False,
|
|
"permissions": ["contacts:read"],
|
|
"denied_permissions": [],
|
|
"field_permissions": {},
|
|
}
|
|
|
|
cmd = CreateContactCommand(data={
|
|
"type": "company",
|
|
"name": "Denied Company",
|
|
})
|
|
result = await cmd.execute(db_session, redis_client, current_user)
|
|
|
|
assert result.success is False
|
|
assert "Permission denied" in result.error
|
|
|
|
async def test_create_contact_with_invalid_status(
|
|
self, db_session: AsyncSession, redis_client
|
|
):
|
|
"""Creating a contact with invalid status fails."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
|
|
current_user = {
|
|
"user_id": str(seed["admin_a"].id),
|
|
"tenant_id": str(seed["tenant_a"].id),
|
|
"role": "admin",
|
|
"is_system_admin": True,
|
|
"permissions": ["*: *"],
|
|
"denied_permissions": [],
|
|
"field_permissions": {},
|
|
}
|
|
|
|
cmd = CreateContactCommand(data={
|
|
"type": "company",
|
|
"name": "Bad Status Corp",
|
|
"status": "nonexistent",
|
|
})
|
|
result = await cmd.execute(db_session, redis_client, current_user)
|
|
|
|
assert result.success is False
|
|
assert "Invalid contact status" in result.error
|
|
|
|
|
|
# ── UpdateContactCommand Tests ──
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestUpdateContactCommand:
|
|
"""Tests for UpdateContactCommand."""
|
|
|
|
async def test_update_contact_admin_success(
|
|
self, db_session: AsyncSession, redis_client
|
|
):
|
|
"""Admin can update a contact, audit log created, outbox event enqueued."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
|
|
current_user = {
|
|
"user_id": str(seed["admin_a"].id),
|
|
"tenant_id": str(seed["tenant_a"].id),
|
|
"role": "admin",
|
|
"is_system_admin": True,
|
|
"permissions": ["*: *"],
|
|
"denied_permissions": [],
|
|
"field_permissions": {},
|
|
}
|
|
|
|
# First create a contact
|
|
create_cmd = CreateContactCommand(data={
|
|
"type": "company",
|
|
"name": "Original Name",
|
|
})
|
|
create_result = await create_cmd.execute(db_session, redis_client, current_user)
|
|
assert create_result.success is True
|
|
contact_id = create_result.data["id"]
|
|
|
|
# Update the contact
|
|
update_cmd = UpdateContactCommand(
|
|
contact_id=contact_id,
|
|
data={"name": "Updated Name", "email_1": "updated@example.com"},
|
|
)
|
|
update_result = await update_cmd.execute(db_session, redis_client, current_user)
|
|
|
|
assert update_result.success is True
|
|
assert update_result.data["name"] == "Updated Name"
|
|
assert update_result.data["email_1"] == "updated@example.com"
|
|
assert len(update_result.events) >= 1
|
|
assert update_result.events[0]["event"] == "contact.updated"
|
|
|
|
# Verify audit log
|
|
await db_session.flush()
|
|
audit_q = select(AuditLog).where(
|
|
AuditLog.entity_type == "contact",
|
|
AuditLog.action == "update",
|
|
AuditLog.entity_id == uuid_mod.UUID(contact_id),
|
|
)
|
|
audit_result = await db_session.execute(audit_q)
|
|
audit_entries = audit_result.scalars().all()
|
|
assert len(audit_entries) >= 1
|
|
|
|
async def test_update_contact_not_found(
|
|
self, db_session: AsyncSession, redis_client
|
|
):
|
|
"""Updating a non-existent contact fails gracefully."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
|
|
current_user = {
|
|
"user_id": str(seed["admin_a"].id),
|
|
"tenant_id": str(seed["tenant_a"].id),
|
|
"role": "admin",
|
|
"is_system_admin": True,
|
|
"permissions": ["*: *"],
|
|
"denied_permissions": [],
|
|
"field_permissions": {},
|
|
}
|
|
|
|
random_id = str(uuid_mod.uuid4())
|
|
update_cmd = UpdateContactCommand(
|
|
contact_id=random_id,
|
|
data={"name": "New Name"},
|
|
)
|
|
result = await update_cmd.execute(db_session, redis_client, current_user)
|
|
|
|
assert result.success is False
|
|
assert "not found" in result.error.lower()
|
|
|
|
async def test_update_contact_viewer_denied(
|
|
self, db_session: AsyncSession, redis_client
|
|
):
|
|
"""Viewer cannot update a contact."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
|
|
current_user = {
|
|
"user_id": str(seed["viewer_a"].id),
|
|
"tenant_id": str(seed["tenant_a"].id),
|
|
"role": "viewer",
|
|
"is_system_admin": False,
|
|
"permissions": ["contacts:read"],
|
|
"denied_permissions": [],
|
|
"field_permissions": {},
|
|
}
|
|
|
|
cmd = UpdateContactCommand(
|
|
contact_id=str(seed["company_a"].id),
|
|
data={"name": "Hacked"},
|
|
)
|
|
result = await cmd.execute(db_session, redis_client, current_user)
|
|
|
|
assert result.success is False
|
|
assert "Permission denied" in result.error
|
|
|
|
|
|
# ── DeleteContactCommand Tests ──
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestDeleteContactCommand:
|
|
"""Tests for DeleteContactCommand."""
|
|
|
|
async def test_soft_delete_contact_admin_success(
|
|
self, db_session: AsyncSession, redis_client
|
|
):
|
|
"""Admin can soft-delete a contact."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
|
|
current_user = {
|
|
"user_id": str(seed["admin_a"].id),
|
|
"tenant_id": str(seed["tenant_a"].id),
|
|
"role": "admin",
|
|
"is_system_admin": True,
|
|
"permissions": ["*: *"],
|
|
"denied_permissions": [],
|
|
"field_permissions": {},
|
|
}
|
|
|
|
# Create a contact to delete
|
|
create_cmd = CreateContactCommand(data={
|
|
"type": "company",
|
|
"name": "To Delete",
|
|
})
|
|
create_result = await create_cmd.execute(db_session, redis_client, current_user)
|
|
assert create_result.success is True
|
|
contact_id = create_result.data["id"]
|
|
|
|
# Soft-delete it
|
|
delete_cmd = DeleteContactCommand(contact_id=contact_id, hard=False)
|
|
delete_result = await delete_cmd.execute(db_session, redis_client, current_user)
|
|
|
|
assert delete_result.success is True
|
|
assert delete_result.data is None
|
|
assert len(delete_result.events) >= 1
|
|
assert delete_result.events[0]["event"] == "contact.deleted"
|
|
|
|
# Verify audit log
|
|
await db_session.flush()
|
|
audit_q = select(AuditLog).where(
|
|
AuditLog.entity_type == "contact",
|
|
AuditLog.action == "delete",
|
|
AuditLog.entity_id == uuid_mod.UUID(contact_id),
|
|
)
|
|
audit_result = await db_session.execute(audit_q)
|
|
audit_entries = audit_result.scalars().all()
|
|
assert len(audit_entries) >= 1
|
|
|
|
async def test_hard_delete_contact_admin_success(
|
|
self, db_session: AsyncSession, redis_client
|
|
):
|
|
"""Admin can hard-delete a contact."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
|
|
current_user = {
|
|
"user_id": str(seed["admin_a"].id),
|
|
"tenant_id": str(seed["tenant_a"].id),
|
|
"role": "admin",
|
|
"is_system_admin": True,
|
|
"permissions": ["*: *"],
|
|
"denied_permissions": [],
|
|
"field_permissions": {},
|
|
}
|
|
|
|
# Create a contact to hard-delete
|
|
create_cmd = CreateContactCommand(data={
|
|
"type": "company",
|
|
"name": "Hard Delete Me",
|
|
})
|
|
create_result = await create_cmd.execute(db_session, redis_client, current_user)
|
|
assert create_result.success is True
|
|
contact_id = create_result.data["id"]
|
|
|
|
# Hard-delete it
|
|
delete_cmd = DeleteContactCommand(contact_id=contact_id, hard=True)
|
|
delete_result = await delete_cmd.execute(db_session, redis_client, current_user)
|
|
|
|
assert delete_result.success is True
|
|
assert len(delete_result.events) >= 1
|
|
assert delete_result.events[0]["event"] == "contact.hard_deleted"
|
|
|
|
async def test_delete_contact_not_found(
|
|
self, db_session: AsyncSession, redis_client
|
|
):
|
|
"""Deleting a non-existent contact fails gracefully."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
|
|
current_user = {
|
|
"user_id": str(seed["admin_a"].id),
|
|
"tenant_id": str(seed["tenant_a"].id),
|
|
"role": "admin",
|
|
"is_system_admin": True,
|
|
"permissions": ["*: *"],
|
|
"denied_permissions": [],
|
|
"field_permissions": {},
|
|
}
|
|
|
|
random_id = str(uuid_mod.uuid4())
|
|
delete_cmd = DeleteContactCommand(contact_id=random_id)
|
|
result = await delete_cmd.execute(db_session, redis_client, current_user)
|
|
|
|
assert result.success is False
|
|
assert "not found" in result.error.lower()
|
|
|
|
async def test_delete_contact_viewer_denied(
|
|
self, db_session: AsyncSession, redis_client
|
|
):
|
|
"""Viewer cannot delete a contact."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
|
|
current_user = {
|
|
"user_id": str(seed["viewer_a"].id),
|
|
"tenant_id": str(seed["tenant_a"].id),
|
|
"role": "viewer",
|
|
"is_system_admin": False,
|
|
"permissions": ["contacts:read"],
|
|
"denied_permissions": [],
|
|
"field_permissions": {},
|
|
}
|
|
|
|
cmd = DeleteContactCommand(contact_id=str(seed["company_a"].id))
|
|
result = await cmd.execute(db_session, redis_client, current_user)
|
|
|
|
assert result.success is False
|
|
assert "Permission denied" in result.error
|
|
|
|
|
|
# ── MergeContactsCommand Tests ──
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestMergeContactsCommand:
|
|
"""Tests for MergeContactsCommand."""
|
|
|
|
async def test_merge_contacts_admin_success(
|
|
self, db_session: AsyncSession, redis_client
|
|
):
|
|
"""Admin can merge two contacts, audit log created, outbox event enqueued."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
|
|
current_user = {
|
|
"user_id": str(seed["admin_a"].id),
|
|
"tenant_id": str(seed["tenant_a"].id),
|
|
"role": "admin",
|
|
"is_system_admin": True,
|
|
"permissions": ["*: *"],
|
|
"denied_permissions": [],
|
|
"field_permissions": {},
|
|
}
|
|
|
|
# Create two contacts to merge
|
|
create_cmd1 = CreateContactCommand(data={
|
|
"type": "company",
|
|
"name": "Source Company",
|
|
"email_1": "source@example.com",
|
|
})
|
|
result1 = await create_cmd1.execute(db_session, redis_client, current_user)
|
|
assert result1.success is True
|
|
source_id = result1.data["id"]
|
|
|
|
create_cmd2 = CreateContactCommand(data={
|
|
"type": "company",
|
|
"name": "Target Company",
|
|
"email_1": "target@example.com",
|
|
})
|
|
result2 = await create_cmd2.execute(db_session, redis_client, current_user)
|
|
assert result2.success is True
|
|
target_id = result2.data["id"]
|
|
|
|
# Merge source → target
|
|
merge_cmd = MergeContactsCommand(
|
|
source_contact_id=source_id,
|
|
target_contact_id=target_id,
|
|
note="Duplicate detected",
|
|
)
|
|
merge_result = await merge_cmd.execute(db_session, redis_client, current_user)
|
|
|
|
assert merge_result.success is True
|
|
assert merge_result.data is not None
|
|
assert "history" in merge_result.data
|
|
assert merge_result.data["history"]["source_id"] == source_id
|
|
assert merge_result.data["history"]["target_id"] == target_id
|
|
assert len(merge_result.events) >= 1
|
|
assert merge_result.events[0]["event"] == "contact.merged"
|
|
|
|
# Verify audit log
|
|
await db_session.flush()
|
|
audit_q = select(AuditLog).where(
|
|
AuditLog.entity_type == "contact",
|
|
AuditLog.action == "merge",
|
|
AuditLog.entity_id == uuid_mod.UUID(target_id),
|
|
)
|
|
audit_result = await db_session.execute(audit_q)
|
|
audit_entries = audit_result.scalars().all()
|
|
assert len(audit_entries) >= 1
|
|
|
|
async def test_merge_same_contact_fails(
|
|
self, db_session: AsyncSession, redis_client
|
|
):
|
|
"""Merging a contact with itself fails."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
|
|
current_user = {
|
|
"user_id": str(seed["admin_a"].id),
|
|
"tenant_id": str(seed["tenant_a"].id),
|
|
"role": "admin",
|
|
"is_system_admin": True,
|
|
"permissions": ["*: *"],
|
|
"denied_permissions": [],
|
|
"field_permissions": {},
|
|
}
|
|
|
|
# Create a contact
|
|
create_cmd = CreateContactCommand(data={
|
|
"type": "company",
|
|
"name": "Solo Company",
|
|
})
|
|
create_result = await create_cmd.execute(db_session, redis_client, current_user)
|
|
assert create_result.success is True
|
|
contact_id = create_result.data["id"]
|
|
|
|
# Try to merge with itself
|
|
merge_cmd = MergeContactsCommand(
|
|
source_contact_id=contact_id,
|
|
target_contact_id=contact_id,
|
|
)
|
|
result = await merge_cmd.execute(db_session, redis_client, current_user)
|
|
|
|
assert result.success is False
|
|
assert "must be different" in result.error
|
|
|
|
async def test_merge_contacts_viewer_denied(
|
|
self, db_session: AsyncSession, redis_client
|
|
):
|
|
"""Viewer cannot merge contacts."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
|
|
current_user = {
|
|
"user_id": str(seed["viewer_a"].id),
|
|
"tenant_id": str(seed["tenant_a"].id),
|
|
"role": "viewer",
|
|
"is_system_admin": False,
|
|
"permissions": ["contacts:read"],
|
|
"denied_permissions": [],
|
|
"field_permissions": {},
|
|
}
|
|
|
|
merge_cmd = MergeContactsCommand(
|
|
source_contact_id=str(seed["company_a"].id),
|
|
target_contact_id=str(seed["company_a"].id),
|
|
)
|
|
result = await merge_cmd.execute(db_session, redis_client, current_user)
|
|
|
|
assert result.success is False
|
|
assert "Permission denied" in result.error
|
|
|
|
async def test_merge_contact_not_found(
|
|
self, db_session: AsyncSession, redis_client
|
|
):
|
|
"""Merging with a non-existent contact fails."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
|
|
current_user = {
|
|
"user_id": str(seed["admin_a"].id),
|
|
"tenant_id": str(seed["tenant_a"].id),
|
|
"role": "admin",
|
|
"is_system_admin": True,
|
|
"permissions": ["*: *"],
|
|
"denied_permissions": [],
|
|
"field_permissions": {},
|
|
}
|
|
|
|
random_id = str(uuid_mod.uuid4())
|
|
merge_cmd = MergeContactsCommand(
|
|
source_contact_id=random_id,
|
|
target_contact_id=str(seed["company_a"].id),
|
|
)
|
|
result = await merge_cmd.execute(db_session, redis_client, current_user)
|
|
|
|
assert result.success is False
|
|
assert "not found" in result.error.lower()
|