""" Django management command to test schema separation functionality. This command verifies that: 1. Each tenant has its own schema 2. Data is properly isolated between tenants 3. CRUD operations work correctly in tenant schemas """ from django.core.management.base import BaseCommand from django.db import connection from tenants.models import Tenant from tenants.schema_utils import ( get_tenant_schema_name, set_current_tenant_schema, get_current_tenant_schema ) from customers.models import Customer from inventory.models import Product, Category import logging logger = logging.getLogger(__name__) class Command(BaseCommand): help = 'Test schema separation functionality and data isolation' def add_arguments(self, parser): parser.add_argument( '--create-test-data', action='store_true', help='Create test data in each tenant schema', ) def handle(self, *args, **options): create_test_data = options.get('create_test_data', False) self.stdout.write( self.style.SUCCESS('=== Testing Schema Separation ===') ) # Get all tenants tenants = Tenant.objects.all() if not tenants.exists(): self.stdout.write( self.style.ERROR('No tenants found. Create a tenant first.') ) return self.stdout.write(f'Testing with {tenants.count()} tenant(s)') for tenant in tenants: self.test_tenant_schema(tenant, create_test_data) self.stdout.write( self.style.SUCCESS('Schema separation test completed!') ) def test_tenant_schema(self, tenant, create_test_data=False): """Test schema functionality for a single tenant""" schema_name = get_tenant_schema_name(tenant.tenant_code) self.stdout.write(f'\n=== Testing tenant: {tenant.name} (schema: {schema_name}) ===') # Test schema switching try: set_current_tenant_schema(tenant.tenant_code) current_schema = get_current_tenant_schema() if current_schema == schema_name: self.stdout.write(f'✓ Schema switching works: {current_schema}') else: self.stdout.write( self.style.ERROR(f'✗ Schema mismatch: expected {schema_name}, got {current_schema}') ) return except Exception as e: self.stdout.write( self.style.ERROR(f'✗ Schema switching failed: {e}') ) return # Test database queries in tenant schema self.test_tenant_queries(tenant, create_test_data) # Test data isolation self.test_data_isolation(tenant) def test_tenant_queries(self, tenant, create_test_data=False): """Test CRUD operations in tenant schema""" self.stdout.write('Testing CRUD operations...') try: # Test Customer model initial_customer_count = Customer.objects.count() self.stdout.write(f' - Initial customers: {initial_customer_count}') if create_test_data: # Create test customer test_customer = Customer.objects.create( first_name=f'Test Customer {tenant.tenant_code}', last_name='Schema Test', phone=f'+254700{tenant.id:06d}', # Unique phone per tenant email=f'test.{tenant.tenant_code}@example.com' ) self.stdout.write(f' ✓ Created test customer: {test_customer}') # Verify customer exists customer_exists = Customer.objects.filter(id=test_customer.id).exists() if customer_exists: self.stdout.write(' ✓ Customer query successful') else: self.stdout.write(' ✗ Customer query failed') # Test Category model initial_category_count = Category.objects.count() self.stdout.write(f' - Initial categories: {initial_category_count}') if create_test_data: # Create test category test_category = Category.objects.create( name=f'Test Category {tenant.tenant_code}', description=f'Test category for tenant {tenant.name}' ) self.stdout.write(f' ✓ Created test category: {test_category}') # Test Product model initial_product_count = Product.objects.count() self.stdout.write(f' - Initial products: {initial_product_count}') except Exception as e: self.stdout.write( self.style.ERROR(f' ✗ CRUD operations failed: {e}') ) def test_data_isolation(self, tenant): """Verify that this tenant cannot see other tenant's data""" self.stdout.write('Testing data isolation...') try: # Get current tenant's data counts current_customers = Customer.objects.count() current_categories = Category.objects.count() current_products = Product.objects.count() self.stdout.write( f' - Current tenant data: {current_customers} customers, ' f'{current_categories} categories, {current_products} products' ) # Test raw SQL to verify schema isolation with connection.cursor() as cursor: cursor.execute("SELECT current_schema()") current_schema = cursor.fetchone()[0] self.stdout.write(f' - Current PostgreSQL schema: {current_schema}') # Verify schema isolation by checking if we can see other schemas cursor.execute(""" SELECT schema_name FROM information_schema.schemata WHERE schema_name LIKE 'tenant_%' ORDER BY schema_name """) all_tenant_schemas = [row[0] for row in cursor.fetchall()] self.stdout.write(f' - Available tenant schemas: {", ".join(all_tenant_schemas)}') # Verify we're in the correct schema expected_schema = get_tenant_schema_name(tenant.tenant_code) if current_schema == expected_schema: self.stdout.write(' ✓ Data isolation verified - in correct schema') else: self.stdout.write( self.style.WARNING( f' ⚠ Schema mismatch: expected {expected_schema}, got {current_schema}' ) ) except Exception as e: self.stdout.write( self.style.ERROR(f' ✗ Data isolation test failed: {e}') ) def test_cross_schema_isolation(self): """Test that tenants cannot access each other's data""" self.stdout.write('\n=== Testing Cross-Schema Isolation ===') tenants = list(Tenant.objects.all()) if len(tenants) < 2: self.stdout.write('Need at least 2 tenants to test cross-schema isolation') return tenant1, tenant2 = tenants[0], tenants[1] # Switch to tenant 1 and get data count set_current_tenant_schema(tenant1.code) tenant1_customers = Customer.objects.count() # Switch to tenant 2 and get data count set_current_tenant_schema(tenant2.code) tenant2_customers = Customer.objects.count() self.stdout.write( f'Tenant {tenant1.code}: {tenant1_customers} customers' ) self.stdout.write( f'Tenant {tenant2.code}: {tenant2_customers} customers' ) if tenant1_customers != tenant2_customers: self.stdout.write('✓ Cross-schema isolation verified - different data counts') else: self.stdout.write('⚠ Same data counts - may indicate shared data')