Pushing code to migrate git

This commit is contained in:
2025-08-15 00:57:39 -08:00
parent 53f3e8a44d
commit 455bdac187
50 changed files with 14761 additions and 225 deletions
+139
View File
@@ -0,0 +1,139 @@
"""
PunimTag Test Configuration
Shared fixtures and configuration for all tests.
"""
import pytest
import sqlite3
import tempfile
import os
import sys
from pathlib import Path
# Add src to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from backend.app import app
@pytest.fixture
def test_db():
"""Create a temporary test database."""
db_fd, db_path = tempfile.mkstemp()
# Create test database schema
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Create tables
cursor.execute('''
CREATE TABLE images (
id INTEGER PRIMARY KEY,
filename TEXT NOT NULL,
path TEXT NOT NULL,
date_taken TEXT
)
''')
cursor.execute('''
CREATE TABLE faces (
id INTEGER PRIMARY KEY,
image_id INTEGER,
person_id INTEGER,
encoding BLOB,
left INTEGER,
top INTEGER,
right INTEGER,
bottom INTEGER
)
''')
cursor.execute('''
CREATE TABLE people (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
cursor.execute('''
CREATE TABLE tags (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL UNIQUE
)
''')
cursor.execute('''
CREATE TABLE image_tags (
image_id INTEGER,
tag_id INTEGER,
PRIMARY KEY (image_id, tag_id)
)
''')
conn.commit()
conn.close()
yield db_path
# Cleanup
os.close(db_fd)
os.unlink(db_path)
@pytest.fixture
def client(test_db):
"""Create a test client with test database."""
app.config['TESTING'] = True
app.config['DATABASE_PATH'] = test_db
with app.test_client() as client:
yield client
@pytest.fixture
def sample_photos(test_db):
"""Add sample photos to the test database."""
conn = sqlite3.connect(test_db)
cursor = conn.cursor()
photos = [
('photo1.jpg', '/test/path/photo1.jpg', '2023-01-01'),
('photo2.jpg', '/test/path/photo2.jpg', '2023-01-02'),
('photo3.jpg', '/test/path/photo3.jpg', '2023-01-03')
]
cursor.executemany(
'INSERT INTO images (filename, path, date_taken) VALUES (?, ?, ?)',
photos
)
conn.commit()
conn.close()
return photos
@pytest.fixture
def sample_faces(test_db):
"""Add sample faces to the test database."""
conn = sqlite3.connect(test_db)
cursor = conn.cursor()
# Add a person first
cursor.execute('INSERT INTO people (name) VALUES (?)', ('Test Person',))
person_id = cursor.lastrowid
# Add faces
faces = [
(1, person_id, b'fake_encoding_1', 100, 100, 200, 200),
(2, person_id, b'fake_encoding_2', 150, 150, 250, 250),
(3, None, b'fake_encoding_3', 200, 200, 300, 300), # Unidentified face
]
cursor.executemany(
'INSERT INTO faces (image_id, person_id, encoding, left, top, right, bottom) VALUES (?, ?, ?, ?, ?, ?, ?)',
faces
)
conn.commit()
conn.close()
return faces
+448
View File
@@ -0,0 +1,448 @@
#!/usr/bin/env python3
"""
Comprehensive Backend Test Suite for PunimTag
Tests all backend functionality including face clustering, enhanced recognition, and complex queries
"""
import os
import tempfile
import shutil
import unittest
import uuid
import pickle
from datetime import datetime, timedelta
import numpy as np
from punimtag import PunimTag
from config import PunimTagConfig, create_default_config
from typing import List
class TestBackendFunctionality(unittest.TestCase):
"""Test all backend features thoroughly"""
def setUp(self):
"""Set up test environment with temporary database and config"""
self.test_dir = tempfile.mkdtemp()
self.db_path = os.path.join(self.test_dir, 'test.db')
self.photos_dir = os.path.join(self.test_dir, 'photos')
self.config_path = os.path.join(self.test_dir, 'test_config.json')
os.makedirs(self.photos_dir, exist_ok=True)
# Create test configuration
self.config = PunimTagConfig(self.config_path)
self.config.face_recognition.confidence_threshold = 0.5
self.config.auto_tagging.enabled = True
self.config.processing.batch_size = 50
self.config.save()
# Initialize PunimTag with test database
self.tagger = PunimTag(db_path=self.db_path, photos_dir=self.photos_dir)
def tearDown(self):
"""Clean up test environment"""
self.tagger.close()
shutil.rmtree(self.test_dir)
def test_configuration_system(self):
"""Test configuration loading and saving"""
# Test default values
self.assertEqual(self.config.face_recognition.confidence_threshold, 0.5)
self.assertTrue(self.config.auto_tagging.enabled)
# Test updating settings
success = self.config.update_setting('face_recognition', 'confidence_threshold', 0.7)
self.assertTrue(success)
self.assertEqual(self.config.face_recognition.confidence_threshold, 0.7)
# Test getting settings
value = self.config.get_setting('processing', 'batch_size')
self.assertEqual(value, 50)
# Test tag suggestions
event_tags = self.config.get_tag_suggestions('event')
self.assertIn('wedding', event_tags)
self.assertIn('bar_mitzvah', event_tags)
def test_jewish_org_tags(self):
"""Test Jewish organization specific tag functionality"""
# Test adding Jewish event tags
for tag_name in ['shabbat', 'chanukah', 'passover']:
tag_id = self.tagger.add_tag(tag_name, 'event')
self.assertIsNotNone(tag_id)
# Test location tags
for tag_name in ['synagogue', 'sanctuary', 'sukkah']:
tag_id = self.tagger.add_tag(tag_name, 'location')
self.assertIsNotNone(tag_id)
# Verify tags exist in database
c = self.tagger.conn.cursor()
c.execute("SELECT COUNT(*) FROM tags WHERE category = 'event'")
event_count = c.fetchone()[0]
self.assertGreaterEqual(event_count, 3)
def test_face_clustering(self):
"""Test face clustering functionality"""
# Create mock face data
face_ids = self._create_mock_faces(10)
# Test clustering
clusters = self.tagger.cluster_unknown_faces()
self.assertIsInstance(clusters, dict)
# Test getting cluster data
cluster_data = self.tagger.get_face_clusters()
self.assertIsInstance(cluster_data, list)
# Each cluster should have required fields
for cluster in cluster_data:
self.assertIn('cluster_id', cluster)
self.assertIn('face_count', cluster)
self.assertIn('face_ids', cluster)
self.assertIn('representative_face', cluster)
def test_cluster_assignment(self):
"""Test assigning clusters to people"""
# Create mock faces and cluster them
face_ids = self._create_mock_faces(5)
clusters = self.tagger.cluster_unknown_faces()
if clusters:
cluster_id = list(clusters.keys())[0]
success = self.tagger.assign_cluster_to_person(cluster_id, "Test Person")
self.assertTrue(success)
# Verify assignment
c = self.tagger.conn.cursor()
c.execute("SELECT COUNT(*) FROM faces WHERE person_id IS NOT NULL")
assigned_count = c.fetchone()[0]
self.assertGreater(assigned_count, 0)
def test_most_common_faces(self):
"""Test getting most frequently photographed people"""
# Add some people and faces
person1_id = self.tagger.add_person("John Doe")
person2_id = self.tagger.add_person("Jane Smith")
# Create mock faces assigned to people
face_ids = self._create_mock_faces(10)
# Assign faces to people
for i, face_id in enumerate(face_ids[:5]):
self.tagger.assign_face_to_person(face_id, person1_id, True)
for face_id in face_ids[5:7]:
self.tagger.assign_face_to_person(face_id, person2_id, True)
# Test getting most common faces
common_faces = self.tagger.get_most_common_faces(limit=10)
self.assertIsInstance(common_faces, list)
if common_faces:
# Should be sorted by face count (John Doe should be first)
self.assertEqual(common_faces[0]['name'], "John Doe")
self.assertEqual(common_faces[0]['face_count'], 5)
def test_face_verification(self):
"""Test face verification functionality"""
person_id = self.tagger.add_person("Test Person")
face_ids = self._create_mock_faces(3)
# Assign faces to person
for face_id in face_ids:
self.tagger.assign_face_to_person(face_id, person_id, True)
# Test verification
faces = self.tagger.verify_person_faces(person_id)
self.assertEqual(len(faces), 3)
# Test removing incorrect assignment
self.tagger.remove_incorrect_face_assignment(face_ids[0])
# Verify removal
faces_after = self.tagger.verify_person_faces(person_id)
self.assertEqual(len(faces_after), 2)
def test_batch_processing(self):
"""Test batch image processing"""
# Create mock image paths
image_paths = [
os.path.join(self.photos_dir, f'test_{i}.jpg')
for i in range(5)
]
# Create empty test files
for path in image_paths:
with open(path, 'w') as f:
f.write('') # Empty file for testing
# Test batch processing (will fail on actual processing but test the logic)
try:
results = self.tagger.batch_process_images(image_paths, batch_size=2)
self.assertIn('processed', results)
self.assertIn('errors', results)
self.assertIn('skipped', results)
except Exception:
# Expected to fail with empty files, but structure should be correct
pass
def test_advanced_search(self):
"""Test advanced search functionality"""
# Setup test data
person_id = self.tagger.add_person("Search Test Person")
tag_id = self.tagger.add_tag("test_event", "event")
# Create mock image
image_id = self._create_mock_image()
# Add mock face and tag
face_id = self._create_mock_face(image_id)
self.tagger.assign_face_to_person(face_id, person_id, True)
self.tagger.tag_image(image_id, tag_id)
# Test various search scenarios
# Search by person
results = self.tagger.advanced_search(people=["Search Test Person"])
self.assertIsInstance(results, list)
# Search by tag
results = self.tagger.advanced_search(tags=["test_event"])
self.assertIsInstance(results, list)
# Search by person and tag
results = self.tagger.advanced_search(
people=["Search Test Person"],
tags=["test_event"]
)
self.assertIsInstance(results, list)
# Search with date range
today = datetime.now()
yesterday = today - timedelta(days=1)
tomorrow = today + timedelta(days=1)
results = self.tagger.advanced_search(
date_from=yesterday,
date_to=tomorrow
)
self.assertIsInstance(results, list)
# Search with location bounds
results = self.tagger.advanced_search(
latitude_min=40.0,
latitude_max=41.0,
longitude_min=-74.0,
longitude_max=-73.0
)
self.assertIsInstance(results, list)
# Search with minimum people requirement
results = self.tagger.advanced_search(min_people=1)
self.assertIsInstance(results, list)
# Search with limit
results = self.tagger.advanced_search(limit=5)
self.assertIsInstance(results, list)
self.assertLessEqual(len(results), 5)
def test_face_quality_calculation(self):
"""Test face quality scoring"""
# Test with different face sizes and encodings
small_face = (10, 30, 30, 10) # 20x20 face
large_face = (10, 110, 110, 10) # 100x100 face
encoding = np.random.rand(128)
small_quality = self.tagger.calculate_face_quality(encoding, small_face)
large_quality = self.tagger.calculate_face_quality(encoding, large_face)
# Larger faces should have higher quality scores
self.assertGreater(large_quality, small_quality)
# Quality should be between 0 and 1
self.assertGreaterEqual(small_quality, 0)
self.assertLessEqual(small_quality, 1)
self.assertGreaterEqual(large_quality, 0)
self.assertLessEqual(large_quality, 1)
def test_database_integrity(self):
"""Test database integrity and relationships"""
# Test foreign key relationships
person_id = self.tagger.add_person("Integrity Test")
image_id = self._create_mock_image()
face_id = self._create_mock_face(image_id)
tag_id = self.tagger.add_tag("integrity_test")
# Test assignments
self.tagger.assign_face_to_person(face_id, person_id, True)
self.tagger.tag_image(image_id, tag_id)
# Verify relationships exist
c = self.tagger.conn.cursor()
# Check face-person relationship
c.execute("SELECT person_id FROM faces WHERE id = ?", (face_id,))
result = c.fetchone()
self.assertEqual(result[0], person_id)
# Check image-tag relationship
c.execute("SELECT tag_id FROM image_tags WHERE image_id = ?", (image_id,))
result = c.fetchone()
self.assertEqual(result[0], tag_id)
def test_search_edge_cases(self):
"""Test search functionality with edge cases"""
# Search with empty parameters
results = self.tagger.advanced_search()
self.assertIsInstance(results, list)
# Search with non-existent person
results = self.tagger.advanced_search(people=["Non Existent Person"])
self.assertEqual(len(results), 0)
# Search with non-existent tag
results = self.tagger.advanced_search(tags=["non_existent_tag"])
self.assertEqual(len(results), 0)
# Search with invalid date range
future_date = datetime.now() + timedelta(days=365)
past_date = datetime.now() - timedelta(days=365)
results = self.tagger.advanced_search(
date_from=future_date,
date_to=past_date
)
self.assertEqual(len(results), 0)
# Helper methods
def _create_mock_image(self) -> int:
"""Create a mock image entry in database"""
import uuid
unique_path = f'test_path_{uuid.uuid4().hex[:8]}.jpg'
c = self.tagger.conn.cursor()
c.execute('''INSERT INTO images
(path, filename, date_taken, width, height, file_size)
VALUES (?, ?, ?, ?, ?, ?)''',
(unique_path, unique_path, datetime.now(),
800, 600, 12345))
self.tagger.conn.commit()
return c.lastrowid
def _create_mock_face(self, image_id: int) -> int:
"""Create a mock face entry in database"""
import pickle
encoding = np.random.rand(128)
encoding_blob = pickle.dumps(encoding)
c = self.tagger.conn.cursor()
c.execute('''INSERT INTO faces
(image_id, top, right, bottom, left, encoding)
VALUES (?, ?, ?, ?, ?, ?)''',
(image_id, 10, 110, 110, 10, encoding_blob))
self.tagger.conn.commit()
return c.lastrowid
def _create_mock_faces(self, count: int) -> List[int]:
"""Create multiple mock faces"""
face_ids = []
for i in range(count):
image_id = self._create_mock_image()
face_id = self._create_mock_face(image_id)
face_ids.append(face_id)
return face_ids
def run_performance_tests():
"""Run performance tests with larger datasets"""
print("\nRunning Performance Tests")
print("=" * 50)
with tempfile.TemporaryDirectory() as temp_dir:
db_path = os.path.join(temp_dir, 'perf_test.db')
tagger = PunimTag(db_path=db_path)
try:
# Test with larger numbers of faces
print("Creating 1000 mock faces...")
start_time = datetime.now()
face_ids = []
for i in range(1000):
# Create image
c = tagger.conn.cursor()
c.execute('''INSERT INTO images
(path, filename, date_taken, width, height, file_size)
VALUES (?, ?, ?, ?, ?, ?)''',
(f'perf_test_{i}_{uuid.uuid4().hex[:8]}.jpg', f'perf_test_{i}.jpg',
datetime.now(), 800, 600, 12345))
image_id = c.lastrowid
# Create face
encoding = np.random.rand(128)
encoding_blob = pickle.dumps(encoding)
c.execute('''INSERT INTO faces
(image_id, top, right, bottom, left, encoding)
VALUES (?, ?, ?, ?, ?, ?)''',
(image_id, 10, 110, 110, 10, encoding_blob))
face_ids.append(c.lastrowid)
if i % 100 == 0:
print(f"Created {i} faces...")
tagger.conn.commit()
creation_time = (datetime.now() - start_time).total_seconds()
print(f"Created 1000 faces in {creation_time:.2f} seconds")
# Test clustering performance
print("Testing clustering performance...")
start_time = datetime.now()
clusters = tagger.cluster_unknown_faces()
clustering_time = (datetime.now() - start_time).total_seconds()
print(f"Clustered faces in {clustering_time:.2f} seconds")
print(f"Found {len(clusters)} clusters")
# Test search performance
print("Testing search performance...")
start_time = datetime.now()
results = tagger.advanced_search(limit=100)
search_time = (datetime.now() - start_time).total_seconds()
print(f"Search completed in {search_time:.2f} seconds")
print(f"Found {len(results)} results")
finally:
tagger.close()
def main():
"""Main test runner"""
print("PunimTag Backend Test Suite")
print("=" * 50)
# Run unit tests
print("Running unit tests...")
loader = unittest.TestLoader()
suite = loader.loadTestsFromTestCase(TestBackendFunctionality)
runner = unittest.TextTestRunner(verbosity=2)
result = runner.run(suite)
if result.wasSuccessful():
print("\n✅ All unit tests passed!")
# Run performance tests
run_performance_tests()
print("\n🎉 Backend testing completed successfully!")
print("\nBackend is ready for UI development.")
else:
print("\n❌ Some tests failed. Please fix issues before proceeding.")
return False
return True
if __name__ == "__main__":
success = main()
exit(0 if success else 1)
+200
View File
@@ -0,0 +1,200 @@
"""
Main test suite for PunimTag
Consolidated tests covering core functionality.
"""
import json
import os
import sys
from pathlib import Path
# Add src to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
def test_imports():
"""Test that all modules can be imported."""
try:
from backend import app
print("✅ Flask app imported successfully")
# Test if we can access the app instance
if hasattr(app, 'app'):
print("✅ Flask app instance found")
else:
print("⚠️ Flask app instance not found, but module imported")
return True
except ImportError as e:
print(f"❌ Import error: {e}")
return False
def test_database_connection():
"""Test database connection and basic operations."""
try:
# Test if we can connect to the database
import sqlite3
db_path = Path(__file__).parent.parent / "data" / "punimtag_simple.db"
if db_path.exists():
conn = sqlite3.connect(str(db_path))
cursor = conn.cursor()
cursor.execute("SELECT 1")
result = cursor.fetchone()
conn.close()
if result and result[0] == 1:
print("✅ Database connection successful")
return True
else:
print("❌ Database query failed")
return False
else:
print("⚠️ Database file not found, but this is normal for fresh installs")
return True
except Exception as e:
print(f"❌ Database error: {e}")
return False
def test_face_recognition_import():
"""Test face recognition module import."""
try:
from backend import visual_identifier
print("✅ Face recognition module imported successfully")
return True
except ImportError as e:
print(f"❌ Face recognition import error: {e}")
return False
def test_config_loading():
"""Test configuration loading."""
try:
# Test if config directory exists and has files
config_dir = Path(__file__).parent.parent / "config"
if config_dir.exists():
config_files = list(config_dir.glob("*.py"))
if config_files:
print(f"✅ Configuration directory found with {len(config_files)} files")
return True
else:
print("⚠️ Configuration directory exists but no Python files found")
return True
else:
print("❌ Configuration directory not found")
return False
except Exception as e:
print(f"❌ Configuration error: {e}")
return False
def test_directory_structure():
"""Test that all required directories exist."""
required_dirs = [
"src/backend",
"src/frontend",
"src/utils",
"tests",
"data",
"config",
"docs",
"photos",
"scripts",
"assets"
]
missing_dirs = []
for dir_path in required_dirs:
if not os.path.exists(dir_path):
missing_dirs.append(dir_path)
if missing_dirs:
print(f"❌ Missing directories: {missing_dirs}")
return False
else:
print("✅ All required directories exist")
return True
def test_steering_documents():
"""Test that steering documents exist."""
required_docs = [
"docs/product.md",
"docs/structure.md",
"docs/tech.md",
"docs/api-standards.md",
"docs/testing-standards.md",
"docs/code-conventions.md"
]
missing_docs = []
for doc_path in required_docs:
if not os.path.exists(doc_path):
missing_docs.append(doc_path)
if missing_docs:
print(f"❌ Missing steering documents: {missing_docs}")
return False
else:
print("✅ All steering documents exist")
return True
def test_main_app_file():
"""Test that the main application file exists and is accessible."""
try:
main_app_path = Path(__file__).parent.parent / "src" / "backend" / "app.py"
if main_app_path.exists():
print(f"✅ Main app file found: {main_app_path}")
# Test if we can read the file
with open(main_app_path, 'r') as f:
content = f.read()
if 'Flask' in content and 'app' in content:
print("✅ Main app file contains Flask app")
return True
else:
print("⚠️ Main app file exists but doesn't contain expected Flask content")
return True
else:
print("❌ Main app file not found")
return False
except Exception as e:
print(f"❌ Main app file error: {e}")
return False
def run_all_tests():
"""Run all tests and report results."""
print("🧪 Running PunimTag Test Suite")
print("=" * 50)
tests = [
test_imports,
test_database_connection,
test_face_recognition_import,
test_config_loading,
test_directory_structure,
test_steering_documents,
test_main_app_file
]
passed = 0
total = len(tests)
for test in tests:
try:
if test():
passed += 1
except Exception as e:
print(f"❌ Test {test.__name__} failed with exception: {e}")
print("=" * 50)
print(f"📊 Test Results: {passed}/{total} tests passed")
if passed == total:
print("🎉 All tests passed!")
return True
else:
print("⚠️ Some tests failed")
return False
if __name__ == "__main__":
success = run_all_tests()
sys.exit(0 if success else 1)
+250
View File
@@ -0,0 +1,250 @@
#!/usr/bin/env python3
"""
Test script for PunimTag
Tests core functionality including face detection, recognition, tagging, and search
"""
import os
import shutil
import tempfile
import unittest
from datetime import datetime
from punimtag import PunimTag
import numpy as np
class TestPunimTag(unittest.TestCase):
def setUp(self):
"""Set up test environment"""
# Create temporary directory for test database
self.test_dir = tempfile.mkdtemp()
self.db_path = os.path.join(self.test_dir, 'test.db')
self.photos_dir = os.path.join(self.test_dir, 'photos')
os.makedirs(self.photos_dir, exist_ok=True)
# Initialize PunimTag with test database
self.tagger = PunimTag(db_path=self.db_path, photos_dir=self.photos_dir)
def tearDown(self):
"""Clean up test environment"""
self.tagger.close()
shutil.rmtree(self.test_dir)
def test_database_creation(self):
"""Test that database tables are created correctly"""
c = self.tagger.conn.cursor()
# Check tables exist
c.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = {row[0] for row in c.fetchall()}
expected_tables = {'images', 'people', 'faces', 'tags', 'image_tags'}
self.assertEqual(tables & expected_tables, expected_tables)
def test_add_person(self):
"""Test adding people to database"""
# Add person
person_id = self.tagger.add_person("John Doe")
self.assertIsNotNone(person_id)
# Verify person exists
c = self.tagger.conn.cursor()
c.execute("SELECT name FROM people WHERE id = ?", (person_id,))
result = c.fetchone()
self.assertEqual(result[0], "John Doe")
# Test duplicate handling
person_id2 = self.tagger.add_person("John Doe")
self.assertEqual(person_id, person_id2)
def test_add_tag(self):
"""Test tag creation"""
# Add tag without category
tag_id1 = self.tagger.add_tag("vacation")
self.assertIsNotNone(tag_id1)
# Add tag with category
tag_id2 = self.tagger.add_tag("beach", "location")
self.assertIsNotNone(tag_id2)
# Verify tags exist
c = self.tagger.conn.cursor()
c.execute("SELECT name, category FROM tags WHERE id = ?", (tag_id2,))
result = c.fetchone()
self.assertEqual(result[0], "beach")
self.assertEqual(result[1], "location")
def test_metadata_extraction(self):
"""Test metadata extraction from images"""
# Test with a non-existent file - should handle gracefully
try:
metadata = self.tagger.extract_metadata("nonexistent.jpg")
# If it doesn't raise an exception, check default values
self.assertIsNone(metadata['date_taken'])
self.assertIsNone(metadata['latitude'])
self.assertIsNone(metadata['longitude'])
except FileNotFoundError:
# This is also acceptable behavior
pass
def test_face_identification(self):
"""Test face identification logic"""
# Test with no known faces
result = self.tagger.identify_face(np.random.rand(128))
self.assertEqual(result, (None, None))
# Would need actual face encodings for more thorough testing
def test_search_functionality(self):
"""Test search capabilities"""
# Search with no data should return empty
results = self.tagger.search_images()
self.assertEqual(len(results), 0)
# Test with filters
results = self.tagger.search_images(
people=["John Doe"],
tags=["vacation"],
date_from=datetime(2023, 1, 1),
date_to=datetime(2023, 12, 31)
)
self.assertEqual(len(results), 0)
def test_unidentified_faces(self):
"""Test getting unidentified faces"""
faces = self.tagger.get_unidentified_faces()
self.assertEqual(len(faces), 0) # Should be empty initially
class TestImageProcessing(unittest.TestCase):
"""Test image processing with actual images"""
@classmethod
def setUpClass(cls):
"""Create test images"""
cls.test_dir = tempfile.mkdtemp()
cls.photos_dir = os.path.join(cls.test_dir, 'photos')
os.makedirs(cls.photos_dir, exist_ok=True)
# Create test images (simple colored squares)
try:
from PIL import Image
# Create a few test images
for i, color in enumerate(['red', 'green', 'blue']):
img = Image.new('RGB', (100, 100), color)
img.save(os.path.join(cls.photos_dir, f'test_{color}.jpg'))
except ImportError:
print("PIL not available, skipping image creation")
@classmethod
def tearDownClass(cls):
"""Clean up test images"""
shutil.rmtree(cls.test_dir)
def setUp(self):
"""Set up for each test"""
self.db_path = os.path.join(self.test_dir, 'test.db')
self.tagger = PunimTag(db_path=self.db_path, photos_dir=self.photos_dir)
def tearDown(self):
"""Clean up after each test"""
self.tagger.close()
if os.path.exists(self.db_path):
os.remove(self.db_path)
def test_process_directory(self):
"""Test processing a directory of images"""
# Process all images
processed = self.tagger.process_directory()
# Should process the test images (if created)
self.assertGreaterEqual(processed, 0)
# Check images were added to database
c = self.tagger.conn.cursor()
c.execute("SELECT COUNT(*) FROM images")
count = c.fetchone()[0]
self.assertEqual(count, processed)
def test_with_sample_images(image_paths):
"""
Test PunimTag with actual image files
Args:
image_paths: List of paths to test images
"""
print("Testing PunimTag with sample images")
print("=" * 50)
# Create temporary database
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as tmp:
db_path = tmp.name
try:
# Initialize PunimTag
tagger = PunimTag(db_path=db_path)
# Process each image
print(f"\nProcessing {len(image_paths)} images...")
for path in image_paths:
if os.path.exists(path):
print(f"Processing: {path}")
try:
image_id = tagger.process_image(path)
print(f" ✓ Added to database with ID: {image_id}")
except Exception as e:
print(f" ✗ Error: {e}")
else:
print(f" ✗ File not found: {path}")
# Show statistics
c = tagger.conn.cursor()
c.execute("SELECT COUNT(*) FROM images")
image_count = c.fetchone()[0]
print(f"\nTotal images: {image_count}")
c.execute("SELECT COUNT(*) FROM faces")
face_count = c.fetchone()[0]
print(f"Total faces detected: {face_count}")
# Get unidentified faces
unidentified = tagger.get_unidentified_faces()
print(f"Unidentified faces: {len(unidentified)}")
# Close connection
tagger.close()
print("\n✓ Test completed successfully!")
finally:
# Clean up
if os.path.exists(db_path):
os.remove(db_path)
def main():
"""Main test runner"""
print("PunimTag Test Suite")
print("=" * 50)
# Run unit tests
print("\nRunning unit tests...")
unittest.main(argv=[''], exit=False, verbosity=2)
# Optional: Test with actual images
print("\n" + "=" * 50)
print("To test with actual images, call:")
print("python test_punimtag.py image1.jpg image2.jpg ...")
# Check if images were provided as arguments
import sys
if len(sys.argv) > 1:
image_paths = sys.argv[1:]
test_with_sample_images(image_paths)
if __name__ == "__main__":
main()
+286
View File
@@ -0,0 +1,286 @@
#!/usr/bin/env python3
"""
API Test Suite for PunimTag Web GUI
Tests all web endpoints to identify issues with pre-load check
"""
import requests
import json
import time
import sys
from urllib.parse import urljoin
class WebAPITester:
def __init__(self, base_url="http://127.0.0.1:5000"):
self.base_url = base_url
self.session = requests.Session()
self.session.headers.update({
'Content-Type': 'application/json',
'Accept': 'application/json'
})
def test_endpoint(self, endpoint, method='GET', data=None, timeout=10, expected_status=200):
"""Test a single endpoint with timeout and error handling"""
url = urljoin(self.base_url, endpoint)
print(f"Testing {method} {endpoint}...")
start_time = time.time()
try:
if method == 'GET':
response = self.session.get(url, timeout=timeout)
elif method == 'POST':
response = self.session.post(url, json=data, timeout=timeout)
else:
raise ValueError(f"Unsupported method: {method}")
elapsed = time.time() - start_time
if response.status_code == expected_status:
print(f" ✅ SUCCESS ({elapsed:.2f}s) - Status: {response.status_code}")
try:
return response.json()
except:
return response.text
else:
print(f" ❌ FAILED ({elapsed:.2f}s) - Status: {response.status_code}")
print(f" Response: {response.text[:200]}")
return None
except requests.exceptions.Timeout:
elapsed = time.time() - start_time
print(f" ⏰ TIMEOUT ({elapsed:.2f}s) - Endpoint took too long")
return None
except requests.exceptions.ConnectionError:
print(f" 🔌 CONNECTION ERROR - Cannot connect to {self.base_url}")
return None
except Exception as e:
elapsed = time.time() - start_time
print(f" 💥 ERROR ({elapsed:.2f}s) - {str(e)}")
return None
def test_preload_endpoints(self):
"""Test all endpoints used in pre-load check"""
print("\n🔍 Testing Pre-load Check Endpoints")
print("=" * 50)
# Test database connection
db_result = self.test_endpoint('/check_database')
if not db_result:
print(" ❌ Database check failed - this will cause pre-load issues")
return False
# Test system status
status_result = self.test_endpoint('/system_status')
if not status_result:
print(" ❌ System status failed - this will cause pre-load issues")
return False
# Test debug endpoint
debug_result = self.test_endpoint('/debug/preload_test')
if not debug_result:
print(" ❌ Debug endpoint failed - this will cause pre-load issues")
return False
print(" ✅ All pre-load endpoints working correctly")
return True
def test_main_endpoints(self):
"""Test main application endpoints"""
print("\n📱 Testing Main Application Endpoints")
print("=" * 50)
# Test main page
main_result = self.test_endpoint('/', expected_status=200)
if not main_result:
print(" ❌ Main page failed")
return False
# Test photos endpoint
photos_result = self.test_endpoint('/get_photos?tab=all_photos&page=1&per_page=1')
if not photos_result:
print(" ❌ Photos endpoint failed")
return False
# Test faces endpoint
faces_result = self.test_endpoint('/get_faces?tab=unidentified&page=1&per_page=1')
if not faces_result:
print(" ❌ Faces endpoint failed")
return False
print(" ✅ All main endpoints working correctly")
return True
def test_thumbnail_endpoints(self):
"""Test thumbnail generation endpoints"""
print("\n🖼️ Testing Thumbnail Endpoints")
print("=" * 50)
# First get a face ID to test with
faces_result = self.test_endpoint('/get_faces?tab=unidentified&page=1&per_page=1')
if not faces_result or not isinstance(faces_result, dict) or not faces_result.get('faces'):
print(" ⚠️ No faces available for thumbnail testing")
return True
faces = faces_result.get('faces', [])
if not faces:
print(" ⚠️ No faces available for thumbnail testing")
return True
face_id = faces[0].get('face_id')
if not face_id:
print(" ⚠️ No valid face ID found for thumbnail testing")
return True
# Test face thumbnail
thumbnail_result = self.test_endpoint(f'/get_thumbnail/{face_id}')
if not thumbnail_result:
print(" ❌ Face thumbnail endpoint failed")
return False
# Test photo thumbnail
photos_result = self.test_endpoint('/get_photos?tab=all_photos&page=1&per_page=1')
if photos_result and isinstance(photos_result, dict) and photos_result.get('photos'):
photos = photos_result.get('photos', [])
if photos:
photo_id = photos[0].get('image_id')
if photo_id:
photo_thumbnail_result = self.test_endpoint(f'/get_photo_thumbnail/{photo_id}')
if not photo_thumbnail_result:
print(" ❌ Photo thumbnail endpoint failed")
return False
print(" ✅ All thumbnail endpoints working correctly")
return True
def test_performance(self):
"""Test endpoint performance"""
print("\n⚡ Performance Testing")
print("=" * 50)
endpoints = [
'/check_database',
'/system_status',
'/debug/preload_test',
'/get_photos?tab=all_photos&page=1&per_page=1',
'/get_faces?tab=unidentified&page=1&per_page=1'
]
performance_results = {}
for endpoint in endpoints:
times = []
for i in range(3): # Test each endpoint 3 times
start_time = time.time()
result = self.test_endpoint(endpoint, timeout=30)
elapsed = time.time() - start_time
times.append(elapsed)
time.sleep(0.5) # Small delay between tests
avg_time = sum(times) / len(times)
performance_results[endpoint] = {
'avg_time': avg_time,
'min_time': min(times),
'max_time': max(times),
'success': all(t < 30 for t in times) # All under 30s timeout
}
status = "" if performance_results[endpoint]['success'] else ""
print(f" {status} {endpoint}: {avg_time:.2f}s avg ({min(times):.2f}s-{max(times):.2f}s)")
return performance_results
def test_browser_simulation(self):
"""Simulate browser behavior for pre-load check"""
print("\n🌐 Browser Simulation Test")
print("=" * 50)
# Simulate the exact pre-load check sequence
checks = [
{ 'name': 'Database Connection', 'endpoint': '/check_database' },
{ 'name': 'System Status', 'endpoint': '/system_status' },
{ 'name': 'Debug Test', 'endpoint': '/debug/preload_test' }
]
all_passed = True
for check in checks:
print(f"Testing {check['name']}...")
result = self.test_endpoint(check['endpoint'], timeout=10)
if result:
print(f"{check['name']} passed")
else:
print(f"{check['name']} failed")
all_passed = False
if all_passed:
print(" 🎉 All browser simulation tests passed!")
else:
print(" 💥 Some browser simulation tests failed!")
return all_passed
def run_all_tests(self):
"""Run all tests"""
print("🚀 Starting PunimTag Web API Test Suite")
print("=" * 60)
# Test server connectivity first
print("\n🔌 Testing Server Connectivity")
print("-" * 30)
try:
response = self.session.get(self.base_url, timeout=5)
print(f"✅ Server is running at {self.base_url}")
except Exception as e:
print(f"❌ Cannot connect to server: {e}")
print("Make sure the server is running with: python simple_web_gui.py")
return False
# Run all test suites
results = {
'preload': self.test_preload_endpoints(),
'main': self.test_main_endpoints(),
'thumbnails': self.test_thumbnail_endpoints(),
'performance': self.test_performance(),
'browser_sim': self.test_browser_simulation()
}
# Summary
print("\n📊 Test Summary")
print("=" * 60)
passed = sum(1 for result in results.values() if result)
total = len(results)
for test_name, result in results.items():
status = "✅ PASS" if result else "❌ FAIL"
print(f" {status} {test_name.replace('_', ' ').title()}")
print(f"\nOverall: {passed}/{total} test suites passed")
if passed == total:
print("🎉 All tests passed! The web API is working correctly.")
print("If the browser is still stuck, the issue might be:")
print(" - Browser cache (try Ctrl+F5)")
print(" - CORS issues (check browser console)")
print(" - JavaScript errors (check browser console)")
else:
print("💥 Some tests failed. Check the output above for details.")
return passed == total
def main():
"""Main test runner"""
if len(sys.argv) > 1:
base_url = sys.argv[1]
else:
base_url = "http://127.0.0.1:5000"
tester = WebAPITester(base_url)
success = tester.run_all_tests()
sys.exit(0 if success else 1)
if __name__ == "__main__":
main()