from django.test import TestCase from django.contrib.contenttypes.models import ContentType from django.core.exceptions import ValidationError from django.db import transaction from django.utils import timezone from history_tracking.models import VersionBranch, ChangeSet from history_tracking.managers import BranchManager, MergeStrategy from parks.models import Park class BranchManagerTests(TestCase): def setUp(self): self.park = Park.objects.create( name='Test Park', slug='test-park', status='OPERATING' ) self.content_type = ContentType.objects.get_for_model(Park) self.manager = BranchManager() self.main_branch = VersionBranch.objects.create( name='main', metadata={'type': 'default_branch'} ) def test_create_branch(self): """Test branch creation with metadata""" branch = self.manager.create_branch( name='feature/test', metadata={'type': 'feature', 'description': 'Test branch'} ) self.assertEqual(branch.name, 'feature/test') self.assertEqual(branch.metadata['type'], 'feature') self.assertTrue(branch.is_active) def test_get_active_branches(self): """Test retrieving only active branches""" # Create some branches feature_branch = self.manager.create_branch( name='feature/active', metadata={'type': 'feature'} ) inactive_branch = self.manager.create_branch( name='feature/inactive', metadata={'type': 'feature'} ) inactive_branch.is_active = False inactive_branch.save() active_branches = self.manager.get_active_branches() self.assertIn(self.main_branch, active_branches) self.assertIn(feature_branch, active_branches) self.assertNotIn(inactive_branch, active_branches) def test_get_branch_changes(self): """Test retrieving changes for a specific branch""" # Create some changes in different branches main_change = ChangeSet.objects.create( branch=self.main_branch, content_type=self.content_type, object_id=self.park.id, data={'name': 'Main Change'}, status='applied' ) feature_branch = self.manager.create_branch(name='feature/test') feature_change = ChangeSet.objects.create( branch=feature_branch, content_type=self.content_type, object_id=self.park.id, data={'name': 'Feature Change'}, status='applied' ) main_changes = self.manager.get_branch_changes(self.main_branch) feature_changes = self.manager.get_branch_changes(feature_branch) self.assertIn(main_change, main_changes) self.assertNotIn(feature_change, main_changes) self.assertIn(feature_change, feature_changes) self.assertNotIn(main_change, feature_changes) def test_merge_branches(self): """Test merging changes between branches""" # Create feature branch with changes feature_branch = self.manager.create_branch(name='feature/test') change = ChangeSet.objects.create( branch=feature_branch, content_type=self.content_type, object_id=self.park.id, data={'name': 'Updated Name'}, status='applied' ) # Merge feature branch into main self.manager.merge_branches( source_branch=feature_branch, target_branch=self.main_branch ) # Verify changes were copied to main branch main_changes = self.manager.get_branch_changes(self.main_branch) self.assertEqual(main_changes.count(), 1) merged_change = main_changes.first() self.assertEqual(merged_change.data, change.data) def test_branch_deletion(self): """Test branch deletion with cleanup""" feature_branch = self.manager.create_branch(name='feature/delete') ChangeSet.objects.create( branch=feature_branch, content_type=self.content_type, object_id=self.park.id, data={'name': 'Test Change'}, status='applied' ) # Delete the branch self.manager.delete_branch(feature_branch) # Verify branch and its changes are gone with self.assertRaises(VersionBranch.DoesNotExist): VersionBranch.objects.get(name='feature/delete') self.assertEqual( ChangeSet.objects.filter(branch=feature_branch).count(), 0 ) class MergeStrategyTests(TestCase): def setUp(self): self.park = Park.objects.create( name='Test Park', slug='test-park', status='OPERATING' ) self.content_type = ContentType.objects.get_for_model(Park) self.main_branch = VersionBranch.objects.create( name='main', metadata={'type': 'default_branch'} ) self.feature_branch = VersionBranch.objects.create( name='feature/test', metadata={'type': 'feature'} ) self.merge_strategy = MergeStrategy() def test_simple_merge(self): """Test merging non-conflicting changes""" # Create changes in feature branch feature_changes = [ ChangeSet.objects.create( branch=self.feature_branch, content_type=self.content_type, object_id=self.park.id, data={'name': 'New Name'}, status='applied', applied_at=timezone.now() ), ChangeSet.objects.create( branch=self.feature_branch, content_type=self.content_type, object_id=self.park.id, data={'description': 'New Description'}, status='applied', applied_at=timezone.now() ) ] # Perform merge with transaction.atomic(): conflicts = self.merge_strategy.merge( source_branch=self.feature_branch, target_branch=self.main_branch ) self.assertEqual(conflicts, []) # No conflicts expected main_changes = ChangeSet.objects.filter(branch=self.main_branch) self.assertEqual(main_changes.count(), 2) def test_conflict_detection(self): """Test detection of conflicting changes""" # Create conflicting changes ChangeSet.objects.create( branch=self.main_branch, content_type=self.content_type, object_id=self.park.id, data={'name': 'Main Name'}, status='applied', applied_at=timezone.now() ) ChangeSet.objects.create( branch=self.feature_branch, content_type=self.content_type, object_id=self.park.id, data={'name': 'Feature Name'}, status='applied', applied_at=timezone.now() ) # Attempt merge with transaction.atomic(): conflicts = self.merge_strategy.merge( source_branch=self.feature_branch, target_branch=self.main_branch ) self.assertTrue(conflicts) # Conflicts should be detected conflict = conflicts[0] self.assertEqual(conflict['field'], 'name') self.assertEqual(conflict['target_value'], 'Main Name') self.assertEqual(conflict['source_value'], 'Feature Name') def test_merge_ordering(self): """Test that changes are merged in the correct order""" # Create sequential changes change1 = ChangeSet.objects.create( branch=self.feature_branch, content_type=self.content_type, object_id=self.park.id, data={'name': 'First Change'}, status='applied', applied_at=timezone.now() ) change2 = ChangeSet.objects.create( branch=self.feature_branch, content_type=self.content_type, object_id=self.park.id, data={'name': 'Second Change'}, status='applied', applied_at=timezone.now() ) # Perform merge with transaction.atomic(): self.merge_strategy.merge( source_branch=self.feature_branch, target_branch=self.main_branch ) # Verify changes were merged in order merged_changes = ChangeSet.objects.filter( branch=self.main_branch ).order_by('applied_at') self.assertEqual( merged_changes[0].data['name'], 'First Change' ) self.assertEqual( merged_changes[1].data['name'], 'Second Change' ) def test_merge_validation(self): """Test validation of merge operations""" # Test merging inactive branch self.feature_branch.is_active = False self.feature_branch.save() with self.assertRaises(ValidationError): self.merge_strategy.merge( source_branch=self.feature_branch, target_branch=self.main_branch ) # Test merging branch into itself with self.assertRaises(ValidationError): self.merge_strategy.merge( source_branch=self.main_branch, target_branch=self.main_branch )