272 lines
9.5 KiB
Python
Executable File
272 lines
9.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Test script for project merge functionality
|
|
|
|
This script creates two versions of a project, modifies them differently,
|
|
and tests the merge functionality.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
# Add pyPhotoAlbum to path
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
|
|
|
|
from pyPhotoAlbum.project import Project, Page
|
|
from pyPhotoAlbum.models import ImageData, TextBoxData
|
|
from pyPhotoAlbum.project_serializer import save_to_zip, load_from_zip
|
|
from pyPhotoAlbum.merge_manager import MergeManager, MergeStrategy, concatenate_projects
|
|
|
|
|
|
def create_base_project():
|
|
"""Create a base project with some content"""
|
|
project = Project("Base Project")
|
|
|
|
# Add a page with text
|
|
page = Page(page_number=1)
|
|
text = TextBoxData(
|
|
text_content="Original Text",
|
|
x=10, y=10, width=100, height=50
|
|
)
|
|
page.layout.add_element(text)
|
|
project.add_page(page)
|
|
|
|
return project
|
|
|
|
|
|
def test_same_project_merge():
|
|
"""Test merging two versions of the same project"""
|
|
print("=" * 60)
|
|
print("Test 1: Merging Same Project (with conflicts)")
|
|
print("=" * 60)
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
# Create base project
|
|
print("\n1. Creating base project...")
|
|
base_project = create_base_project()
|
|
base_file = os.path.join(temp_dir, "base.ppz")
|
|
success, _ = save_to_zip(base_project, base_file)
|
|
assert success, "Failed to save base project"
|
|
print(f" ✓ Base project saved with project_id: {base_project.project_id}")
|
|
|
|
# Load base project twice to create two versions
|
|
print("\n2. Creating two divergent versions...")
|
|
|
|
# Version A: Modify text content
|
|
project_a = load_from_zip(base_file)
|
|
text_a = project_a.pages[0].layout.elements[0]
|
|
text_a.text_content = "Modified by User A"
|
|
text_a.mark_modified()
|
|
version_a_file = os.path.join(temp_dir, "version_a.ppz")
|
|
save_to_zip(project_a, version_a_file)
|
|
print(f" ✓ Version A: Modified text to '{text_a.text_content}'")
|
|
|
|
# Version B: Modify text position
|
|
project_b = load_from_zip(base_file)
|
|
text_b = project_b.pages[0].layout.elements[0]
|
|
text_b.position = (50, 50)
|
|
text_b.mark_modified()
|
|
version_b_file = os.path.join(temp_dir, "version_b.ppz")
|
|
save_to_zip(project_b, version_b_file)
|
|
print(f" ✓ Version B: Modified position to {text_b.position}")
|
|
|
|
# Detect conflicts
|
|
print("\n3. Detecting conflicts...")
|
|
merge_manager = MergeManager()
|
|
|
|
data_a = project_a.serialize()
|
|
data_b = project_b.serialize()
|
|
|
|
should_merge = merge_manager.should_merge_projects(data_a, data_b)
|
|
assert should_merge, "Projects should be merged (same project_id)"
|
|
print(f" ✓ Projects have same project_id, will merge")
|
|
|
|
conflicts = merge_manager.detect_conflicts(data_a, data_b)
|
|
print(f" ✓ Found {len(conflicts)} conflict(s)")
|
|
|
|
for i, conflict in enumerate(conflicts):
|
|
print(f" - Conflict {i+1}: {conflict.description}")
|
|
|
|
# Auto-resolve using LATEST_WINS strategy
|
|
print("\n4. Auto-resolving with LATEST_WINS strategy...")
|
|
resolutions = merge_manager.auto_resolve_conflicts(MergeStrategy.LATEST_WINS)
|
|
print(f" ✓ Resolutions: {resolutions}")
|
|
|
|
# Apply merge
|
|
merged_data = merge_manager.apply_resolutions(data_a, data_b, resolutions)
|
|
print(f" ✓ Merge applied successfully")
|
|
print(f" ✓ Merged project has {len(merged_data['pages'])} page(s)")
|
|
|
|
print(f"\n{'=' * 60}")
|
|
print("✅ Same project merge test PASSED")
|
|
print(f"{'=' * 60}\n")
|
|
return True
|
|
|
|
|
|
def test_different_project_concatenation():
|
|
"""Test concatenating two different projects"""
|
|
print("=" * 60)
|
|
print("Test 2: Concatenating Different Projects")
|
|
print("=" * 60)
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
# Create two different projects
|
|
print("\n1. Creating two different projects...")
|
|
|
|
project_a = Project("Project A")
|
|
page_a = Page(page_number=1)
|
|
text_a = TextBoxData(text_content="From Project A", x=10, y=10, width=100, height=50)
|
|
page_a.layout.add_element(text_a)
|
|
project_a.add_page(page_a)
|
|
|
|
project_b = Project("Project B")
|
|
page_b = Page(page_number=1)
|
|
text_b = TextBoxData(text_content="From Project B", x=10, y=10, width=100, height=50)
|
|
page_b.layout.add_element(text_b)
|
|
project_b.add_page(page_b)
|
|
|
|
print(f" ✓ Project A: project_id={project_a.project_id}")
|
|
print(f" ✓ Project B: project_id={project_b.project_id}")
|
|
|
|
# Check if should merge
|
|
print("\n2. Checking merge vs concatenate...")
|
|
merge_manager = MergeManager()
|
|
|
|
data_a = project_a.serialize()
|
|
data_b = project_b.serialize()
|
|
|
|
should_merge = merge_manager.should_merge_projects(data_a, data_b)
|
|
assert not should_merge, "Projects should be concatenated (different project_ids)"
|
|
print(f" ✓ Projects have different project_ids, will concatenate")
|
|
|
|
# Concatenate
|
|
print("\n3. Concatenating projects...")
|
|
merged_data = concatenate_projects(data_a, data_b)
|
|
|
|
assert len(merged_data['pages']) == 2, "Should have 2 pages"
|
|
print(f" ✓ Concatenated project has {len(merged_data['pages'])} pages")
|
|
print(f" ✓ Combined name: {merged_data['name']}")
|
|
|
|
print(f"\n{'=' * 60}")
|
|
print("✅ Project concatenation test PASSED")
|
|
print(f"{'=' * 60}\n")
|
|
return True
|
|
|
|
|
|
def test_no_conflicts():
|
|
"""Test merging when there are no conflicts"""
|
|
print("=" * 60)
|
|
print("Test 3: Merging Without Conflicts")
|
|
print("=" * 60)
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
# Create base project with 2 pages
|
|
print("\n1. Creating base project with 2 pages...")
|
|
base_project = Project("Multi-Page Project")
|
|
|
|
page1 = Page(page_number=1)
|
|
text1 = TextBoxData(text_content="Page 1", x=10, y=10, width=100, height=50)
|
|
page1.layout.add_element(text1)
|
|
base_project.add_page(page1)
|
|
|
|
page2 = Page(page_number=2)
|
|
text2 = TextBoxData(text_content="Page 2", x=10, y=10, width=100, height=50)
|
|
page2.layout.add_element(text2)
|
|
base_project.add_page(page2)
|
|
|
|
base_file = os.path.join(temp_dir, "base.ppz")
|
|
save_to_zip(base_project, base_file)
|
|
print(f" ✓ Base project saved with 2 pages")
|
|
|
|
# Version A: Modify page 1
|
|
project_a = load_from_zip(base_file)
|
|
project_a.pages[0].layout.elements[0].text_content = "Page 1 - Modified by A"
|
|
project_a.pages[0].layout.elements[0].mark_modified()
|
|
|
|
# Version B: Modify page 2 (different page, no conflict)
|
|
project_b = load_from_zip(base_file)
|
|
project_b.pages[1].layout.elements[0].text_content = "Page 2 - Modified by B"
|
|
project_b.pages[1].layout.elements[0].mark_modified()
|
|
|
|
print(f" ✓ Version A modified page 1")
|
|
print(f" ✓ Version B modified page 2")
|
|
|
|
# Detect conflicts
|
|
print("\n2. Detecting conflicts...")
|
|
merge_manager = MergeManager()
|
|
|
|
data_a = project_a.serialize()
|
|
data_b = project_b.serialize()
|
|
|
|
conflicts = merge_manager.detect_conflicts(data_a, data_b)
|
|
print(f" ✓ Found {len(conflicts)} conflict(s)")
|
|
|
|
# Should be able to auto-merge
|
|
print("\n3. Auto-merging non-conflicting changes...")
|
|
merged_data = merge_manager.apply_resolutions(data_a, data_b, {})
|
|
|
|
# Verify both changes are present
|
|
merged_project = Project()
|
|
merged_project.deserialize(merged_data)
|
|
|
|
assert len(merged_project.pages) == 2, "Should have 2 pages"
|
|
page1_text = merged_project.pages[0].layout.elements[0].text_content
|
|
page2_text = merged_project.pages[1].layout.elements[0].text_content
|
|
|
|
assert "Modified by A" in page1_text, "Page 1 changes missing"
|
|
assert "Modified by B" in page2_text, "Page 2 changes missing"
|
|
|
|
print(f" ✓ Page 1 text: {page1_text}")
|
|
print(f" ✓ Page 2 text: {page2_text}")
|
|
print(f" ✓ Both changes preserved in merge")
|
|
|
|
print(f"\n{'=' * 60}")
|
|
print("✅ No-conflict merge test PASSED")
|
|
print(f"{'=' * 60}\n")
|
|
return True
|
|
|
|
|
|
def run_all_tests():
|
|
"""Run all merge tests"""
|
|
print("\n" + "=" * 60)
|
|
print("PYPH OTOALBUM MERGE FUNCTIONALITY TESTS")
|
|
print("=" * 60 + "\n")
|
|
|
|
tests = [
|
|
("Same Project Merge", test_same_project_merge),
|
|
("Different Project Concatenation", test_different_project_concatenation),
|
|
("No-Conflict Merge", test_no_conflicts),
|
|
]
|
|
|
|
results = []
|
|
for name, test_func in tests:
|
|
try:
|
|
success = test_func()
|
|
results.append((name, success))
|
|
except Exception as e:
|
|
print(f"\n❌ Test '{name}' FAILED with exception: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
results.append((name, False))
|
|
|
|
# Print summary
|
|
print("\n" + "=" * 60)
|
|
print("TEST SUMMARY")
|
|
print("=" * 60)
|
|
for name, success in results:
|
|
status = "✅ PASS" if success else "❌ FAIL"
|
|
print(f"{status}: {name}")
|
|
|
|
all_passed = all(success for _, success in results)
|
|
print("=" * 60)
|
|
print(f"\nOverall: {'✅ ALL TESTS PASSED' if all_passed else '❌ SOME TESTS FAILED'}\n")
|
|
|
|
return all_passed
|
|
|
|
|
|
if __name__ == "__main__":
|
|
success = run_all_tests()
|
|
sys.exit(0 if success else 1)
|