thrillwiki_django_no_react/history_tracking/tests/test_managers.py

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
            )