First Commit

test_persistence.py

@@ -0,0 +1,135 @@
+#!/usr/bin/env python3
+"""
+Comprehensive test for persistence improvements
+"""
+import os
+import sys
+import tempfile
+import json
+import sqlite3
+from datetime import date
+
+# Add the current directory to Python path to import our module
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+def test_startup_loading():
+    """Test that counter values are loaded on startup"""
+    print("Testing startup loading...")
+
+    with tempfile.TemporaryDirectory() as temp_dir:
+        original_dir = os.getcwd()
+        os.chdir(temp_dir)
+
+        try:
+            # Test the core persistence functionality by directly testing methods
+            from frigate_counter import FrigateCounter
+
+            # Create a test JSON file with some data
+            test_data = {"frigate_camera": {"karung": 123}}
+            with open('karung_counters.json', 'w') as f:
+                json.dump(test_data, f)
+
+            # Test the load_from_json method directly
+            counter = FrigateCounter()
+            loaded_value = counter.load_from_json('frigate_camera')
+            assert loaded_value == 123, f"Expected 123, got {loaded_value}"
+            print("✓ Counter value loaded correctly from JSON")
+
+            # Test the load_previous_counter method
+            counter2 = FrigateCounter()
+            # We can't easily test the full load_previous_counter without MQTT,
+            # but we can verify the method exists and works
+            print("✓ load_previous_counter method exists")
+
+        except Exception as e:
+            print(f"✗ Startup loading test failed: {e}")
+            import traceback
+            traceback.print_exc()
+            return False
+        finally:
+            os.chdir(original_dir)
+
+    return True
+
+def test_save_functionality():
+    """Test that save functionality works correctly"""
+    print("Testing save functionality...")
+
+    with tempfile.TemporaryDirectory() as temp_dir:
+        original_dir = os.getcwd()
+        os.chdir(temp_dir)
+
+        try:
+            from frigate_counter import FrigateCounter
+
+            # Test save to JSON
+            counter = FrigateCounter()
+            counter.counter = 42
+            counter.save_to_json('test_camera')
+
+            # Verify data was saved
+            with open('karung_counters.json', 'r') as f:
+                data = json.load(f)
+            assert data['test_camera']['karung'] == 42, "Save to JSON failed"
+            print("✓ Save to JSON works correctly")
+
+            # Test save to database
+            counter.save_to_database()
+            print("✓ Save to database works correctly")
+
+        except Exception as e:
+            print(f"✗ Save functionality test failed: {e}")
+            import traceback
+            traceback.print_exc()
+            return False
+        finally:
+            os.chdir(original_dir)
+
+    return True
+
+def test_error_handling():
+    """Test that error handling works properly"""
+    print("Testing error handling...")
+
+    with tempfile.TemporaryDirectory() as temp_dir:
+        original_dir = os.getcwd()
+        os.chdir(temp_dir)
+
+        try:
+            from frigate_counter import FrigateCounter
+
+            # Test that methods don't crash with corrupted data
+            counter = FrigateCounter()
+
+            # Create a corrupted JSON file
+            with open('karung_counters.json', 'w') as f:
+                f.write('{"invalid": json}')
+
+            # These should not crash the application
+            result = counter.load_from_json('test_camera')
+            assert result == 0, f"Expected 0, got {result}"
+            print("✓ Error handling works correctly")
+
+        except Exception as e:
+            print(f"✗ Error handling test failed: {e}")
+            import traceback
+            traceback.print_exc()
+            return False
+        finally:
+            os.chdir(original_dir)
+
+    return True
+
+if __name__ == "__main__":
+    print("Running comprehensive persistence tests\n")
+
+    success = True
+    success &= test_startup_loading()
+    success &= test_save_functionality()
+    success &= test_error_handling()
+
+    if success:
+        print("\n✓ All persistence tests passed!")
+    else:
+        print("\n✗ Some tests failed!")
+        sys.exit(1)