from django.contrib.auth import get_user_model from django.test import TestCase from rest_framework.test import APIClient from services.models import Service from structures.models import Structure User = get_user_model() class ServiceViewSetStructureFilterTestCase(TestCase): """ Test suite for Services API structure_id query parameter filtering. Tests multi-tenant safety, backward compatibility, and edge cases. """ def setUp(self): """Set up test data for all scenarios.""" # Create test user self.user = User.objects.create_user( username="testuser", email="test@example.com", password="testpass123" ) # Create two structures owned by the user self.structure1 = Structure.objects.create( name="Test Structure 1", user=self.user ) self.structure2 = Structure.objects.create( name="Test Structure 2", user=self.user ) # Create active services for structure 1 (3 services) self.service1_1 = Service.objects.create( structure=self.structure1, name="Breakfast", default_price="18.00", vat_rate="10.00", is_active=True ) self.service1_2 = Service.objects.create( structure=self.structure1, name="Parking", default_price="10.00", vat_rate="22.00", is_active=True ) self.service1_3 = Service.objects.create( structure=self.structure1, name="WiFi Premium", default_price="5.00", vat_rate="10.00", is_active=True ) # Create inactive service for structure 1 self.service1_inactive = Service.objects.create( structure=self.structure1, name="Old Service", default_price="20.00", vat_rate="10.00", is_active=False ) # Create active services for structure 2 (2 services) self.service2_1 = Service.objects.create( structure=self.structure2, name="Spa Access", default_price="25.00", vat_rate="10.00", is_active=True ) self.service2_2 = Service.objects.create( structure=self.structure2, name="Airport Shuttle", default_price="30.00", vat_rate="10.00", is_active=True ) # Create inactive service for structure 2 self.service2_inactive = Service.objects.create( structure=self.structure2, name="Deprecated Service", default_price="15.00", vat_rate="10.00", is_active=False ) # API client self.client = APIClient() self.client.force_authenticate(user=self.user) def test_scenario_a_filter_by_structure_id(self): """ Scenario A: structure 1 has 3 active services Request structure_id=1 Only 3 active services returned """ response = self.client.get('/services/', {'structure_id': self.structure1.id}) self.assertEqual(response.status_code, 200) self.assertEqual(len(response.data), 3) # Verify correct services returned service_ids = [s['id'] for s in response.data] self.assertIn(self.service1_1.id, service_ids) self.assertIn(self.service1_2.id, service_ids) self.assertIn(self.service1_3.id, service_ids) # Verify inactive service NOT included self.assertNotIn(self.service1_inactive.id, service_ids) def test_scenario_b_inactive_services_excluded(self): """ Scenario B: inactive services exist Only active services returned when structure_id provided """ response = self.client.get('/services/', {'structure_id': self.structure2.id}) self.assertEqual(response.status_code, 200) self.assertEqual(len(response.data), 2) # Verify only active services service_ids = [s['id'] for s in response.data] self.assertIn(self.service2_1.id, service_ids) self.assertIn(self.service2_2.id, service_ids) self.assertNotIn(self.service2_inactive.id, service_ids) # Verify all returned services are active for service in response.data: self.assertTrue(service['is_active']) def test_scenario_c_invalid_structure_id(self): """ Scenario C: invalid structure_id Returns empty array safely, no 500 error """ # Non-existing structure ID response = self.client.get('/services/', {'structure_id': 999999}) self.assertEqual(response.status_code, 200) self.assertEqual(len(response.data), 0) self.assertEqual(response.data, []) # Invalid string parameter response = self.client.get('/services/', {'structure_id': 'invalid'}) self.assertEqual(response.status_code, 200) self.assertEqual(len(response.data), 0) def test_scenario_d_backward_compatibility(self): """ Scenario D: request without structure_id Existing behavior preserved - returns all user's services """ response = self.client.get('/services/') self.assertEqual(response.status_code, 200) # Should return all active AND inactive services (original behavior) # 3 active + 1 inactive from structure1, 2 active + 1 inactive from structure2 self.assertEqual(len(response.data), 6) # Verify both structures' services included service_ids = [s['id'] for s in response.data] self.assertIn(self.service1_1.id, service_ids) self.assertIn(self.service1_inactive.id, service_ids) self.assertIn(self.service2_1.id, service_ids) self.assertIn(self.service2_inactive.id, service_ids) def test_scenario_e_no_cross_structure_leakage(self): """ Scenario E: verify no cross-structure leakage Request structure_id=1 should NOT return structure 2 services """ response = self.client.get('/services/', {'structure_id': self.structure1.id}) self.assertEqual(response.status_code, 200) # Verify NO structure 2 services leaked service_ids = [s['id'] for s in response.data] self.assertNotIn(self.service2_1.id, service_ids) self.assertNotIn(self.service2_2.id, service_ids) self.assertNotIn(self.service2_inactive.id, service_ids) # All returned services belong to structure 1 for service in response.data: self.assertEqual(service['structure'], self.structure1.id) def test_response_format_unchanged(self): """Verify response schema remains backward compatible.""" response = self.client.get('/services/', {'structure_id': self.structure1.id}) self.assertEqual(response.status_code, 200) self.assertTrue(len(response.data) > 0) # Verify expected fields present service = response.data[0] expected_fields = [ 'id', 'structure', 'name', 'default_price', 'vat_rate', 'is_active', 'created_at', 'updated_at' ] for field in expected_fields: self.assertIn(field, service, f"Field '{field}' missing from response") def test_tenant_isolation(self): """Verify user cannot access services from structures they don't own.""" # Create another user other_user = User.objects.create_user( username="otheruser", email="other@example.com", password="otherpass123" ) # Other user's structure other_structure = Structure.objects.create( name="Other Structure", user=other_user ) # Other user's service Service.objects.create( structure=other_structure, name="Secret Service", default_price="50.00", vat_rate="10.00", is_active=True ) # Current user requests other structure's ID response = self.client.get('/services/', {'structure_id': other_structure.id}) # Should return empty (tenant isolation) self.assertEqual(response.status_code, 200) self.assertEqual(len(response.data), 0) def test_string_structure_id_handling(self): """Verify string structure_id parameter is handled safely.""" # Valid string number response = self.client.get('/services/', {'structure_id': str(self.structure1.id)}) self.assertEqual(response.status_code, 200) self.assertEqual(len(response.data), 3) # Empty string response = self.client.get('/services/', {'structure_id': ''}) self.assertEqual(response.status_code, 200) # Empty string should be falsy, so returns all services self.assertEqual(len(response.data), 6) def test_ordering_preserved(self): """Verify services are ordered by name, id as expected.""" response = self.client.get('/services/', {'structure_id': self.structure1.id}) self.assertEqual(response.status_code, 200) # Verify ordering service_names = [s['name'] for s in response.data] self.assertEqual(service_names, sorted(service_names))