forked from cardosofelipe/fast-next-template
Add comprehensive tests for schemas, validators, and exception handlers
- **Schemas:** Introduce unit tests for `OrganizationBase`, `OrganizationCreate`, and `OrganizationUpdate` schemas. Validate edge cases for slug and name validation. - **Validators:** Add tests for `validate_password_strength`, `validate_phone_number`, `validate_email_format`, and `validate_slug`. Cover edge cases, boundary conditions, and defensive code paths. - **Exception Handlers:** Ensure proper error handling in organization, user, and session CRUD operations. Mock database errors and validate exception responses. - Include test cases to verify robust behavior, normalization, and failure scenarios across schema and validation logic.
This commit is contained in:
@@ -461,3 +461,97 @@ class TestSessionsAdditionalCases:
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["success"] is True
|
||||
|
||||
|
||||
class TestSessionExceptionHandlers:
|
||||
"""
|
||||
Test exception handlers in session routes.
|
||||
Covers lines: 77, 104-106, 181-183, 233-236
|
||||
"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_sessions_with_invalid_token_in_header(self, client, user_token):
|
||||
"""Test list_sessions handles token decode errors gracefully (covers line 77)."""
|
||||
# The token decode happens after successful auth, so we need to mock it
|
||||
from unittest.mock import patch
|
||||
|
||||
# Patch decode_token to raise an exception
|
||||
with patch('app.api.routes.sessions.decode_token', side_effect=Exception("Token decode error")):
|
||||
response = await client.get(
|
||||
"/api/v1/sessions/me",
|
||||
headers={"Authorization": f"Bearer {user_token}"}
|
||||
)
|
||||
|
||||
# Should still succeed (exception is caught and ignored in try/except at line 77)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_sessions_database_error(self, client, user_token):
|
||||
"""Test list_sessions handles database errors (covers lines 104-106)."""
|
||||
from unittest.mock import patch
|
||||
from app.crud import session as session_module
|
||||
|
||||
with patch.object(session_module.session, 'get_user_sessions', side_effect=Exception("Database error")):
|
||||
response = await client.get(
|
||||
"/api/v1/sessions/me",
|
||||
headers={"Authorization": f"Bearer {user_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
|
||||
data = response.json()
|
||||
# The global exception handler wraps it in errors array
|
||||
assert data["errors"][0]["message"] == "Failed to retrieve sessions"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_revoke_session_database_error(self, client, user_token, async_test_db, async_test_user):
|
||||
"""Test revoke_session handles database errors (covers lines 181-183)."""
|
||||
from unittest.mock import patch
|
||||
from uuid import uuid4
|
||||
from app.crud import session as session_module
|
||||
|
||||
# First create a session to revoke
|
||||
from app.crud.session import session as session_crud
|
||||
from app.schemas.sessions import SessionCreate
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
test_engine, AsyncTestingSessionLocal = async_test_db
|
||||
|
||||
async with AsyncTestingSessionLocal() as db:
|
||||
session_in = SessionCreate(
|
||||
user_id=async_test_user.id,
|
||||
refresh_token_jti=str(uuid4()),
|
||||
device_name="Test Device",
|
||||
ip_address="192.168.1.1",
|
||||
user_agent="Mozilla/5.0",
|
||||
last_used_at=datetime.now(timezone.utc),
|
||||
expires_at=datetime.now(timezone.utc) + timedelta(days=60)
|
||||
)
|
||||
user_session = await session_crud.create_session(db, obj_in=session_in)
|
||||
session_id = user_session.id
|
||||
|
||||
# Mock the deactivate method to raise an exception
|
||||
with patch.object(session_module.session, 'deactivate', side_effect=Exception("Database connection lost")):
|
||||
response = await client.delete(
|
||||
f"/api/v1/sessions/{session_id}",
|
||||
headers={"Authorization": f"Bearer {user_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
|
||||
data = response.json()
|
||||
assert data["errors"][0]["message"] == "Failed to revoke session"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cleanup_expired_sessions_database_error(self, client, user_token):
|
||||
"""Test cleanup_expired_sessions handles database errors (covers lines 233-236)."""
|
||||
from unittest.mock import patch
|
||||
from app.crud import session as session_module
|
||||
|
||||
with patch.object(session_module.session, 'cleanup_expired_for_user', side_effect=Exception("Cleanup failed")):
|
||||
response = await client.delete(
|
||||
"/api/v1/sessions/me/expired",
|
||||
headers={"Authorization": f"Bearer {user_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
|
||||
data = response.json()
|
||||
assert data["errors"][0]["message"] == "Failed to cleanup sessions"
|
||||
|
||||
@@ -6,6 +6,7 @@ import pytest
|
||||
from uuid import uuid4
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from unittest.mock import patch, AsyncMock, MagicMock
|
||||
|
||||
from app.crud.organization import organization as organization_crud
|
||||
from app.models.organization import Organization
|
||||
@@ -942,3 +943,193 @@ class TestIsUserOrgAdmin:
|
||||
)
|
||||
|
||||
assert is_admin is False
|
||||
|
||||
|
||||
class TestOrganizationExceptionHandlers:
|
||||
"""
|
||||
Test exception handlers in organization CRUD methods.
|
||||
Uses mocks to trigger database errors and verify proper error handling.
|
||||
Covers lines: 33-35, 57-62, 114-116, 130-132, 207-209, 258-260, 291-294, 326-329, 385-387, 409-411, 466-468, 491-493
|
||||
"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_by_slug_database_error(self, async_test_db):
|
||||
"""Test get_by_slug handles database errors (covers lines 33-35)."""
|
||||
test_engine, AsyncTestingSessionLocal = async_test_db
|
||||
|
||||
async with AsyncTestingSessionLocal() as session:
|
||||
with patch.object(session, 'execute', side_effect=Exception("Database connection lost")):
|
||||
with pytest.raises(Exception, match="Database connection lost"):
|
||||
await organization_crud.get_by_slug(session, slug="test-slug")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_integrity_error_non_slug(self, async_test_db):
|
||||
"""Test create with non-slug IntegrityError (covers lines 56-57)."""
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
test_engine, AsyncTestingSessionLocal = async_test_db
|
||||
|
||||
async with AsyncTestingSessionLocal() as session:
|
||||
async def mock_commit():
|
||||
error = IntegrityError("statement", {}, Exception("foreign key constraint failed"))
|
||||
error.orig = Exception("foreign key constraint failed")
|
||||
raise error
|
||||
|
||||
with patch.object(session, 'commit', side_effect=mock_commit):
|
||||
with patch.object(session, 'rollback', new_callable=AsyncMock):
|
||||
org_in = OrganizationCreate(name="Test", slug="test")
|
||||
with pytest.raises(ValueError, match="Database integrity error"):
|
||||
await organization_crud.create(session, obj_in=org_in)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_unexpected_error(self, async_test_db):
|
||||
"""Test create with unexpected exception (covers lines 58-62)."""
|
||||
test_engine, AsyncTestingSessionLocal = async_test_db
|
||||
|
||||
async with AsyncTestingSessionLocal() as session:
|
||||
with patch.object(session, 'commit', side_effect=RuntimeError("Unexpected error")):
|
||||
with patch.object(session, 'rollback', new_callable=AsyncMock):
|
||||
org_in = OrganizationCreate(name="Test", slug="test")
|
||||
with pytest.raises(RuntimeError, match="Unexpected error"):
|
||||
await organization_crud.create(session, obj_in=org_in)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_multi_with_filters_database_error(self, async_test_db):
|
||||
"""Test get_multi_with_filters handles database errors (covers lines 114-116)."""
|
||||
test_engine, AsyncTestingSessionLocal = async_test_db
|
||||
|
||||
async with AsyncTestingSessionLocal() as session:
|
||||
with patch.object(session, 'execute', side_effect=Exception("Query timeout")):
|
||||
with pytest.raises(Exception, match="Query timeout"):
|
||||
await organization_crud.get_multi_with_filters(session)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_member_count_database_error(self, async_test_db):
|
||||
"""Test get_member_count handles database errors (covers lines 130-132)."""
|
||||
from uuid import uuid4
|
||||
test_engine, AsyncTestingSessionLocal = async_test_db
|
||||
|
||||
async with AsyncTestingSessionLocal() as session:
|
||||
with patch.object(session, 'execute', side_effect=Exception("Count query failed")):
|
||||
with pytest.raises(Exception, match="Count query failed"):
|
||||
await organization_crud.get_member_count(session, organization_id=uuid4())
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_multi_with_member_counts_database_error(self, async_test_db):
|
||||
"""Test get_multi_with_member_counts handles database errors (covers lines 207-209)."""
|
||||
test_engine, AsyncTestingSessionLocal = async_test_db
|
||||
|
||||
async with AsyncTestingSessionLocal() as session:
|
||||
with patch.object(session, 'execute', side_effect=Exception("Complex query failed")):
|
||||
with pytest.raises(Exception, match="Complex query failed"):
|
||||
await organization_crud.get_multi_with_member_counts(session)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_add_user_integrity_error(self, async_test_db, async_test_user):
|
||||
"""Test add_user with IntegrityError (covers lines 258-260)."""
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from unittest.mock import MagicMock
|
||||
test_engine, AsyncTestingSessionLocal = async_test_db
|
||||
|
||||
async with AsyncTestingSessionLocal() as session:
|
||||
# First create org
|
||||
org = Organization(name="Test Org", slug="test-org")
|
||||
session.add(org)
|
||||
await session.commit()
|
||||
org_id = org.id
|
||||
|
||||
async with AsyncTestingSessionLocal() as session:
|
||||
async def mock_commit():
|
||||
raise IntegrityError("statement", {}, Exception("constraint failed"))
|
||||
|
||||
# Mock execute to return None (no existing relationship)
|
||||
async def mock_execute(*args, **kwargs):
|
||||
result = MagicMock()
|
||||
result.scalar_one_or_none = MagicMock(return_value=None)
|
||||
return result
|
||||
|
||||
with patch.object(session, 'execute', side_effect=mock_execute):
|
||||
with patch.object(session, 'commit', side_effect=mock_commit):
|
||||
with patch.object(session, 'rollback', new_callable=AsyncMock):
|
||||
with pytest.raises(ValueError, match="Failed to add user to organization"):
|
||||
await organization_crud.add_user(
|
||||
session,
|
||||
organization_id=org_id,
|
||||
user_id=async_test_user.id
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_remove_user_database_error(self, async_test_db, async_test_user):
|
||||
"""Test remove_user handles database errors (covers lines 291-294)."""
|
||||
from uuid import uuid4
|
||||
test_engine, AsyncTestingSessionLocal = async_test_db
|
||||
|
||||
async with AsyncTestingSessionLocal() as session:
|
||||
with patch.object(session, 'execute', side_effect=Exception("Delete failed")):
|
||||
with pytest.raises(Exception, match="Delete failed"):
|
||||
await organization_crud.remove_user(
|
||||
session,
|
||||
organization_id=uuid4(),
|
||||
user_id=async_test_user.id
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_user_role_database_error(self, async_test_db, async_test_user):
|
||||
"""Test update_user_role handles database errors (covers lines 326-329)."""
|
||||
from uuid import uuid4
|
||||
test_engine, AsyncTestingSessionLocal = async_test_db
|
||||
|
||||
async with AsyncTestingSessionLocal() as session:
|
||||
with patch.object(session, 'execute', side_effect=Exception("Update failed")):
|
||||
with pytest.raises(Exception, match="Update failed"):
|
||||
await organization_crud.update_user_role(
|
||||
session,
|
||||
organization_id=uuid4(),
|
||||
user_id=async_test_user.id,
|
||||
role=OrganizationRole.ADMIN
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_organization_members_database_error(self, async_test_db):
|
||||
"""Test get_organization_members handles database errors (covers lines 385-387)."""
|
||||
from uuid import uuid4
|
||||
test_engine, AsyncTestingSessionLocal = async_test_db
|
||||
|
||||
async with AsyncTestingSessionLocal() as session:
|
||||
with patch.object(session, 'execute', side_effect=Exception("Members query failed")):
|
||||
with pytest.raises(Exception, match="Members query failed"):
|
||||
await organization_crud.get_organization_members(session, organization_id=uuid4())
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_user_organizations_database_error(self, async_test_db, async_test_user):
|
||||
"""Test get_user_organizations handles database errors (covers lines 409-411)."""
|
||||
test_engine, AsyncTestingSessionLocal = async_test_db
|
||||
|
||||
async with AsyncTestingSessionLocal() as session:
|
||||
with patch.object(session, 'execute', side_effect=Exception("User orgs query failed")):
|
||||
with pytest.raises(Exception, match="User orgs query failed"):
|
||||
await organization_crud.get_user_organizations(session, user_id=async_test_user.id)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_user_organizations_with_details_database_error(self, async_test_db, async_test_user):
|
||||
"""Test get_user_organizations_with_details handles database errors (covers lines 466-468)."""
|
||||
test_engine, AsyncTestingSessionLocal = async_test_db
|
||||
|
||||
async with AsyncTestingSessionLocal() as session:
|
||||
with patch.object(session, 'execute', side_effect=Exception("Details query failed")):
|
||||
with pytest.raises(Exception, match="Details query failed"):
|
||||
await organization_crud.get_user_organizations_with_details(session, user_id=async_test_user.id)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_user_role_in_org_database_error(self, async_test_db, async_test_user):
|
||||
"""Test get_user_role_in_org handles database errors (covers lines 491-493)."""
|
||||
from uuid import uuid4
|
||||
test_engine, AsyncTestingSessionLocal = async_test_db
|
||||
|
||||
async with AsyncTestingSessionLocal() as session:
|
||||
with patch.object(session, 'execute', side_effect=Exception("Role query failed")):
|
||||
with pytest.raises(Exception, match="Role query failed"):
|
||||
await organization_crud.get_user_role_in_org(
|
||||
session,
|
||||
user_id=async_test_user.id,
|
||||
organization_id=uuid4()
|
||||
)
|
||||
|
||||
@@ -642,3 +642,54 @@ class TestUtilityMethods:
|
||||
async with AsyncTestingSessionLocal() as session:
|
||||
user = await user_crud.get(session, id=str(async_test_user.id))
|
||||
assert user_crud.is_superuser(user) is False
|
||||
|
||||
|
||||
class TestUserExceptionHandlers:
|
||||
"""
|
||||
Test exception handlers in user CRUD methods.
|
||||
Covers lines: 30-32, 205-208, 257-260
|
||||
"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_by_email_database_error(self, async_test_db):
|
||||
"""Test get_by_email handles database errors (covers lines 30-32)."""
|
||||
from unittest.mock import patch
|
||||
test_engine, AsyncTestingSessionLocal = async_test_db
|
||||
|
||||
async with AsyncTestingSessionLocal() as session:
|
||||
with patch.object(session, 'execute', side_effect=Exception("Database query failed")):
|
||||
with pytest.raises(Exception, match="Database query failed"):
|
||||
await user_crud.get_by_email(session, email="test@example.com")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_update_status_database_error(self, async_test_db, async_test_user):
|
||||
"""Test bulk_update_status handles database errors (covers lines 205-208)."""
|
||||
from unittest.mock import patch, AsyncMock
|
||||
test_engine, AsyncTestingSessionLocal = async_test_db
|
||||
|
||||
async with AsyncTestingSessionLocal() as session:
|
||||
# Mock execute to fail
|
||||
with patch.object(session, 'execute', side_effect=Exception("Bulk update failed")):
|
||||
with patch.object(session, 'rollback', new_callable=AsyncMock):
|
||||
with pytest.raises(Exception, match="Bulk update failed"):
|
||||
await user_crud.bulk_update_status(
|
||||
session,
|
||||
user_ids=[async_test_user.id],
|
||||
is_active=False
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_soft_delete_database_error(self, async_test_db, async_test_user):
|
||||
"""Test bulk_soft_delete handles database errors (covers lines 257-260)."""
|
||||
from unittest.mock import patch, AsyncMock
|
||||
test_engine, AsyncTestingSessionLocal = async_test_db
|
||||
|
||||
async with AsyncTestingSessionLocal() as session:
|
||||
# Mock execute to fail
|
||||
with patch.object(session, 'execute', side_effect=Exception("Bulk delete failed")):
|
||||
with patch.object(session, 'rollback', new_callable=AsyncMock):
|
||||
with pytest.raises(Exception, match="Bulk delete failed"):
|
||||
await user_crud.bulk_soft_delete(
|
||||
session,
|
||||
user_ids=[async_test_user.id]
|
||||
)
|
||||
|
||||
187
backend/tests/schemas/test_organizations.py
Normal file
187
backend/tests/schemas/test_organizations.py
Normal file
@@ -0,0 +1,187 @@
|
||||
"""
|
||||
Tests for organization schemas (app/schemas/organizations.py).
|
||||
|
||||
Covers Pydantic validators for:
|
||||
- Slug validation (lines 26, 28, 30, 32, 62-70)
|
||||
- Name validation (lines 40, 77)
|
||||
"""
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from app.schemas.organizations import (
|
||||
OrganizationBase,
|
||||
OrganizationCreate,
|
||||
OrganizationUpdate,
|
||||
)
|
||||
|
||||
|
||||
class TestOrganizationBaseValidators:
|
||||
"""Test validators in OrganizationBase schema."""
|
||||
|
||||
def test_valid_organization_base(self):
|
||||
"""Test that valid data passes validation."""
|
||||
org = OrganizationBase(
|
||||
name="Test Organization",
|
||||
slug="test-org"
|
||||
)
|
||||
assert org.name == "Test Organization"
|
||||
assert org.slug == "test-org"
|
||||
|
||||
def test_slug_none_returns_none(self):
|
||||
"""Test that None slug is allowed (covers line 26)."""
|
||||
org = OrganizationBase(
|
||||
name="Test Organization",
|
||||
slug=None
|
||||
)
|
||||
assert org.slug is None
|
||||
|
||||
def test_slug_invalid_characters_rejected(self):
|
||||
"""Test slug with invalid characters is rejected (covers line 28)."""
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
OrganizationBase(
|
||||
name="Test Organization",
|
||||
slug="Test_Org!" # Uppercase and special chars
|
||||
)
|
||||
errors = exc_info.value.errors()
|
||||
assert any("lowercase letters, numbers, and hyphens" in str(e['msg']) for e in errors)
|
||||
|
||||
def test_slug_starts_with_hyphen_rejected(self):
|
||||
"""Test slug starting with hyphen is rejected (covers line 30)."""
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
OrganizationBase(
|
||||
name="Test Organization",
|
||||
slug="-test-org"
|
||||
)
|
||||
errors = exc_info.value.errors()
|
||||
assert any("cannot start or end with a hyphen" in str(e['msg']) for e in errors)
|
||||
|
||||
def test_slug_ends_with_hyphen_rejected(self):
|
||||
"""Test slug ending with hyphen is rejected (covers line 30)."""
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
OrganizationBase(
|
||||
name="Test Organization",
|
||||
slug="test-org-"
|
||||
)
|
||||
errors = exc_info.value.errors()
|
||||
assert any("cannot start or end with a hyphen" in str(e['msg']) for e in errors)
|
||||
|
||||
def test_slug_consecutive_hyphens_rejected(self):
|
||||
"""Test slug with consecutive hyphens is rejected (covers line 32)."""
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
OrganizationBase(
|
||||
name="Test Organization",
|
||||
slug="test--org"
|
||||
)
|
||||
errors = exc_info.value.errors()
|
||||
assert any("cannot contain consecutive hyphens" in str(e['msg']) for e in errors)
|
||||
|
||||
def test_name_whitespace_only_rejected(self):
|
||||
"""Test whitespace-only name is rejected (covers line 40)."""
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
OrganizationBase(
|
||||
name=" ",
|
||||
slug="test-org"
|
||||
)
|
||||
errors = exc_info.value.errors()
|
||||
assert any("name cannot be empty" in str(e['msg']) for e in errors)
|
||||
|
||||
def test_name_trimmed(self):
|
||||
"""Test that name is trimmed."""
|
||||
org = OrganizationBase(
|
||||
name=" Test Organization ",
|
||||
slug="test-org"
|
||||
)
|
||||
assert org.name == "Test Organization"
|
||||
|
||||
|
||||
class TestOrganizationCreateValidators:
|
||||
"""Test OrganizationCreate schema inherits validators."""
|
||||
|
||||
def test_valid_organization_create(self):
|
||||
"""Test that valid data passes validation."""
|
||||
org = OrganizationCreate(
|
||||
name="Test Organization",
|
||||
slug="test-org"
|
||||
)
|
||||
assert org.name == "Test Organization"
|
||||
assert org.slug == "test-org"
|
||||
|
||||
def test_slug_validation_inherited(self):
|
||||
"""Test that slug validation is inherited from base."""
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
OrganizationCreate(
|
||||
name="Test",
|
||||
slug="Invalid_Slug!"
|
||||
)
|
||||
errors = exc_info.value.errors()
|
||||
assert any("lowercase letters, numbers, and hyphens" in str(e['msg']) for e in errors)
|
||||
|
||||
|
||||
class TestOrganizationUpdateValidators:
|
||||
"""Test validators in OrganizationUpdate schema."""
|
||||
|
||||
def test_valid_organization_update(self):
|
||||
"""Test that valid update data passes validation."""
|
||||
org = OrganizationUpdate(
|
||||
name="Updated Name",
|
||||
slug="updated-slug"
|
||||
)
|
||||
assert org.name == "Updated Name"
|
||||
assert org.slug == "updated-slug"
|
||||
|
||||
def test_slug_none_returns_none(self):
|
||||
"""Test that None slug is allowed in update (covers line 62)."""
|
||||
org = OrganizationUpdate(slug=None)
|
||||
assert org.slug is None
|
||||
|
||||
def test_update_slug_invalid_characters_rejected(self):
|
||||
"""Test update slug with invalid characters is rejected (covers line 64)."""
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
OrganizationUpdate(slug="Test_Org!")
|
||||
errors = exc_info.value.errors()
|
||||
assert any("lowercase letters, numbers, and hyphens" in str(e['msg']) for e in errors)
|
||||
|
||||
def test_update_slug_starts_with_hyphen_rejected(self):
|
||||
"""Test update slug starting with hyphen is rejected (covers line 66)."""
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
OrganizationUpdate(slug="-test-org")
|
||||
errors = exc_info.value.errors()
|
||||
assert any("cannot start or end with a hyphen" in str(e['msg']) for e in errors)
|
||||
|
||||
def test_update_slug_ends_with_hyphen_rejected(self):
|
||||
"""Test update slug ending with hyphen is rejected (covers line 66)."""
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
OrganizationUpdate(slug="test-org-")
|
||||
errors = exc_info.value.errors()
|
||||
assert any("cannot start or end with a hyphen" in str(e['msg']) for e in errors)
|
||||
|
||||
def test_update_slug_consecutive_hyphens_rejected(self):
|
||||
"""Test update slug with consecutive hyphens is rejected (covers line 68)."""
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
OrganizationUpdate(slug="test--org")
|
||||
errors = exc_info.value.errors()
|
||||
assert any("cannot contain consecutive hyphens" in str(e['msg']) for e in errors)
|
||||
|
||||
def test_update_name_whitespace_only_rejected(self):
|
||||
"""Test whitespace-only name in update is rejected (covers line 77)."""
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
OrganizationUpdate(name=" ")
|
||||
errors = exc_info.value.errors()
|
||||
assert any("name cannot be empty" in str(e['msg']) for e in errors)
|
||||
|
||||
def test_update_name_none_allowed(self):
|
||||
"""Test that None name is allowed in update."""
|
||||
org = OrganizationUpdate(name=None)
|
||||
assert org.name is None
|
||||
|
||||
def test_update_name_trimmed(self):
|
||||
"""Test that update name is trimmed."""
|
||||
org = OrganizationUpdate(name=" Updated Name ")
|
||||
assert org.name == "Updated Name"
|
||||
|
||||
def test_partial_update(self):
|
||||
"""Test that partial updates work (all fields optional)."""
|
||||
org = OrganizationUpdate(name="New Name")
|
||||
assert org.name == "New Name"
|
||||
assert org.slug is None
|
||||
assert org.description is None
|
||||
216
backend/tests/schemas/test_validators.py
Normal file
216
backend/tests/schemas/test_validators.py
Normal file
@@ -0,0 +1,216 @@
|
||||
"""
|
||||
Tests for schema validators (app/schemas/validators.py).
|
||||
|
||||
Covers all edge cases in validation functions:
|
||||
- validate_password_strength
|
||||
- validate_phone_number (lines 115, 119)
|
||||
- validate_email_format (line 148)
|
||||
- validate_slug (lines 170-183)
|
||||
"""
|
||||
import pytest
|
||||
|
||||
from app.schemas.validators import (
|
||||
validate_password_strength,
|
||||
validate_phone_number,
|
||||
validate_email_format,
|
||||
validate_slug,
|
||||
)
|
||||
|
||||
|
||||
class TestPasswordStrengthValidator:
|
||||
"""Test password strength validation."""
|
||||
|
||||
def test_valid_strong_password(self):
|
||||
"""Test that a strong password passes validation."""
|
||||
password = "MySecureP@ss123"
|
||||
result = validate_password_strength(password)
|
||||
assert result == password
|
||||
|
||||
def test_password_too_short(self):
|
||||
"""Test that password shorter than 12 characters is rejected."""
|
||||
with pytest.raises(ValueError, match="at least 12 characters long"):
|
||||
validate_password_strength("Short1!")
|
||||
|
||||
def test_common_password_rejected(self):
|
||||
"""Test that common passwords are rejected."""
|
||||
# "password1234" is in COMMON_PASSWORDS and is 12 chars
|
||||
# Common password check happens before character type checks
|
||||
with pytest.raises(ValueError, match="too common"):
|
||||
validate_password_strength("password1234")
|
||||
|
||||
def test_password_missing_lowercase(self):
|
||||
"""Test that password without lowercase is rejected."""
|
||||
with pytest.raises(ValueError, match="at least one lowercase letter"):
|
||||
validate_password_strength("ALLUPPERCASE123!")
|
||||
|
||||
def test_password_missing_uppercase(self):
|
||||
"""Test that password without uppercase is rejected."""
|
||||
with pytest.raises(ValueError, match="at least one uppercase letter"):
|
||||
validate_password_strength("alllowercase123!")
|
||||
|
||||
def test_password_missing_digit(self):
|
||||
"""Test that password without digit is rejected."""
|
||||
with pytest.raises(ValueError, match="at least one digit"):
|
||||
validate_password_strength("NoDigitsHere!")
|
||||
|
||||
def test_password_missing_special_char(self):
|
||||
"""Test that password without special character is rejected."""
|
||||
with pytest.raises(ValueError, match="at least one special character"):
|
||||
validate_password_strength("NoSpecialChar123")
|
||||
|
||||
|
||||
class TestPhoneNumberValidator:
|
||||
"""Test phone number validation."""
|
||||
|
||||
def test_valid_international_format(self):
|
||||
"""Test valid international phone number."""
|
||||
result = validate_phone_number("+12345678901")
|
||||
assert result == "+12345678901"
|
||||
|
||||
def test_valid_local_format(self):
|
||||
"""Test valid local phone number."""
|
||||
result = validate_phone_number("0123456789")
|
||||
assert result == "0123456789"
|
||||
|
||||
def test_valid_with_formatting(self):
|
||||
"""Test phone number with formatting characters."""
|
||||
result = validate_phone_number("+1 (555) 123-4567")
|
||||
assert result == "+15551234567"
|
||||
|
||||
def test_none_returns_none(self):
|
||||
"""Test that None input returns None."""
|
||||
result = validate_phone_number(None)
|
||||
assert result is None
|
||||
|
||||
def test_empty_string_rejected(self):
|
||||
"""Test that empty string is rejected."""
|
||||
with pytest.raises(ValueError, match="cannot be empty"):
|
||||
validate_phone_number("")
|
||||
|
||||
def test_whitespace_only_rejected(self):
|
||||
"""Test that whitespace-only string is rejected."""
|
||||
with pytest.raises(ValueError, match="cannot be empty"):
|
||||
validate_phone_number(" ")
|
||||
|
||||
def test_invalid_prefix_rejected(self):
|
||||
"""Test that invalid prefix is rejected."""
|
||||
with pytest.raises(ValueError, match="must start with \\+ or 0"):
|
||||
validate_phone_number("12345678901")
|
||||
|
||||
def test_too_short_rejected(self):
|
||||
"""Test that too-short phone number is rejected."""
|
||||
with pytest.raises(ValueError, match="must start with \\+ or 0"):
|
||||
validate_phone_number("+1234567") # Only 7 digits after +
|
||||
|
||||
def test_too_long_rejected(self):
|
||||
"""Test that too-long phone number is rejected."""
|
||||
with pytest.raises(ValueError, match="must start with \\+ or 0"):
|
||||
validate_phone_number("+123456789012345") # 15 digits after +
|
||||
|
||||
def test_multiple_plus_symbols_rejected(self):
|
||||
"""Test phone number with multiple + symbols.
|
||||
|
||||
Note: Line 115 is defensive code - the regex check at line 110 catches this first.
|
||||
The regex ^(?:\+[0-9]{8,14}|0[0-9]{8,14})$ only allows + at the start.
|
||||
"""
|
||||
with pytest.raises(ValueError, match="must start with \\+ or 0 followed by 8-14 digits"):
|
||||
validate_phone_number("+1234+5678901")
|
||||
|
||||
def test_non_digit_after_prefix_rejected(self):
|
||||
"""Test phone number with non-digit characters after prefix.
|
||||
|
||||
Note: Line 119 is defensive code - the regex check at line 110 catches this first.
|
||||
The regex requires all digits after the prefix.
|
||||
"""
|
||||
with pytest.raises(ValueError, match="must start with \\+ or 0"):
|
||||
validate_phone_number("+123abc45678")
|
||||
|
||||
|
||||
class TestEmailFormatValidator:
|
||||
"""Test email format validation."""
|
||||
|
||||
def test_valid_email_lowercase(self):
|
||||
"""Test valid lowercase email."""
|
||||
result = validate_email_format("test@example.com")
|
||||
assert result == "test@example.com"
|
||||
|
||||
def test_email_normalized_to_lowercase(self):
|
||||
"""Test email is normalized to lowercase (covers line 148)."""
|
||||
result = validate_email_format("Test@Example.COM")
|
||||
assert result == "test@example.com"
|
||||
|
||||
def test_email_with_uppercase_domain(self):
|
||||
"""Test email with uppercase domain is normalized."""
|
||||
result = validate_email_format("user@EXAMPLE.COM")
|
||||
assert result == "user@example.com"
|
||||
|
||||
|
||||
class TestSlugValidator:
|
||||
"""Test slug validation."""
|
||||
|
||||
def test_valid_slug_lowercase_letters(self):
|
||||
"""Test valid slug with lowercase letters."""
|
||||
result = validate_slug("test-slug")
|
||||
assert result == "test-slug"
|
||||
|
||||
def test_valid_slug_with_numbers(self):
|
||||
"""Test valid slug with numbers."""
|
||||
result = validate_slug("test-123")
|
||||
assert result == "test-123"
|
||||
|
||||
def test_valid_slug_minimal_length(self):
|
||||
"""Test valid slug with minimal length (2 characters)."""
|
||||
result = validate_slug("ab")
|
||||
assert result == "ab"
|
||||
|
||||
def test_empty_slug_rejected(self):
|
||||
"""Test empty slug is rejected (covers line 170)."""
|
||||
with pytest.raises(ValueError, match="at least 2 characters long"):
|
||||
validate_slug("")
|
||||
|
||||
def test_single_character_slug_rejected(self):
|
||||
"""Test single character slug is rejected (covers line 170)."""
|
||||
with pytest.raises(ValueError, match="at least 2 characters long"):
|
||||
validate_slug("a")
|
||||
|
||||
def test_slug_too_long_rejected(self):
|
||||
"""Test slug longer than 50 characters is rejected (covers line 173)."""
|
||||
long_slug = "a" * 51
|
||||
with pytest.raises(ValueError, match="at most 50 characters long"):
|
||||
validate_slug(long_slug)
|
||||
|
||||
def test_slug_max_length_accepted(self):
|
||||
"""Test slug with exactly 50 characters is accepted."""
|
||||
max_slug = "a" * 50
|
||||
result = validate_slug(max_slug)
|
||||
assert result == max_slug
|
||||
|
||||
def test_slug_starts_with_hyphen_rejected(self):
|
||||
"""Test slug starting with hyphen is rejected (covers line 177)."""
|
||||
with pytest.raises(ValueError, match="cannot start or end with a hyphen"):
|
||||
validate_slug("-test")
|
||||
|
||||
def test_slug_ends_with_hyphen_rejected(self):
|
||||
"""Test slug ending with hyphen is rejected (covers line 177)."""
|
||||
with pytest.raises(ValueError, match="cannot start or end with a hyphen"):
|
||||
validate_slug("test-")
|
||||
|
||||
def test_slug_consecutive_hyphens_rejected(self):
|
||||
"""Test slug with consecutive hyphens is rejected (covers line 177)."""
|
||||
with pytest.raises(ValueError, match="cannot contain consecutive hyphens"):
|
||||
validate_slug("test--slug")
|
||||
|
||||
def test_slug_uppercase_letters_rejected(self):
|
||||
"""Test slug with uppercase letters is rejected (covers line 177)."""
|
||||
with pytest.raises(ValueError, match="only contain lowercase letters"):
|
||||
validate_slug("Test-Slug")
|
||||
|
||||
def test_slug_special_characters_rejected(self):
|
||||
"""Test slug with special characters is rejected (covers line 177)."""
|
||||
with pytest.raises(ValueError, match="only contain lowercase letters"):
|
||||
validate_slug("test_slug")
|
||||
|
||||
def test_slug_spaces_rejected(self):
|
||||
"""Test slug with spaces is rejected (covers line 177)."""
|
||||
with pytest.raises(ValueError, match="only contain lowercase letters"):
|
||||
validate_slug("test slug")
|
||||
Reference in New Issue
Block a user