119 lines
4.1 KiB
Python
119 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script for Frigate MQTT Counter Service
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import sqlite3
|
|
from datetime import datetime
|
|
import threading
|
|
import time
|
|
|
|
# Add the current directory to Python path to import our module
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
def test_database_creation():
|
|
"""Test that database is created properly"""
|
|
print("Testing database creation...")
|
|
|
|
# Create a temporary directory to test database creation
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
# Change to temp directory to avoid conflicts
|
|
original_dir = os.getcwd()
|
|
os.chdir(temp_dir)
|
|
|
|
try:
|
|
# Import after changing directory to avoid conflicts
|
|
from frigate_counter import FrigateCounter
|
|
|
|
# Create a temporary counter instance to test database
|
|
counter = FrigateCounter()
|
|
|
|
# Check if database file exists
|
|
if os.path.exists(counter.db_path):
|
|
print("✓ Database file created successfully")
|
|
|
|
# Check table structure
|
|
conn = sqlite3.connect(counter.db_path)
|
|
cursor = conn.cursor()
|
|
cursor.execute("PRAGMA table_info(karung_counts)")
|
|
columns = cursor.fetchall()
|
|
conn.close()
|
|
|
|
expected_columns = ['id', 'camera_name', 'date', 'counter_value', 'timestamp']
|
|
actual_columns = [col[1] for col in columns]
|
|
|
|
if set(expected_columns) <= set(actual_columns):
|
|
print("✓ Database table structure is correct")
|
|
else:
|
|
print("✗ Database table structure is incorrect")
|
|
print(f"Expected: {expected_columns}")
|
|
print(f"Actual: {actual_columns}")
|
|
else:
|
|
print("✗ Database file was not created")
|
|
finally:
|
|
os.chdir(original_dir)
|
|
|
|
def test_counter_logic():
|
|
"""Test counter logic"""
|
|
print("\nTesting counter logic...")
|
|
|
|
# For now, just verify the class can be instantiated
|
|
# without actually connecting to MQTT brokers
|
|
try:
|
|
# Mock the MQTT connection to avoid errors during testing
|
|
import frigate_counter
|
|
|
|
# Create a simple test that doesn't initialize MQTT connections
|
|
class TestCounter(frigate_counter.FrigateCounter):
|
|
def __init__(self):
|
|
# Initialize with minimal setup
|
|
self.db_path = 'test_karung_counts.db'
|
|
self.init_database()
|
|
self.pintu_kiri_buka_detected = False
|
|
self.pintu_kanan_buka_detected = False
|
|
self.timer_active = False
|
|
self.timer_start_time = None
|
|
self.counter = 0
|
|
self.counter_lock = threading.Lock()
|
|
self.topic = 'test'
|
|
self.site_name = 'test'
|
|
|
|
test_counter = TestCounter()
|
|
print("✓ FrigateCounter class can be instantiated successfully")
|
|
print(f"✓ Default topic: {test_counter.topic}")
|
|
print(f"✓ Default site name: {test_counter.site_name}")
|
|
print(f"✓ Database path: {test_counter.db_path}")
|
|
|
|
except Exception as e:
|
|
print(f"✗ Error in counter logic test: {e}")
|
|
|
|
def test_environment_variables():
|
|
"""Test environment variable handling"""
|
|
print("\nTesting environment variables...")
|
|
|
|
# Set some test values
|
|
os.environ['FRIGATE_MQTT_HOST'] = 'test-frigate-host'
|
|
os.environ['REPORT_MQTT_HOST'] = 'test-report-host'
|
|
os.environ['TOPIC'] = 'test-topic'
|
|
os.environ['SITE_NAME'] = 'test-site'
|
|
|
|
# Import after setting environment variables
|
|
from frigate_counter import FrigateCounter
|
|
counter = FrigateCounter()
|
|
|
|
print(f"✓ FRIGATE_MQTT_HOST: {counter.frigate_mqtt_host}")
|
|
print(f"✓ REPORT_MQTT_HOST: {counter.report_mqtt_host}")
|
|
print(f"✓ TOPIC: {counter.topic}")
|
|
print(f"✓ SITE_NAME: {counter.site_name}")
|
|
|
|
if __name__ == "__main__":
|
|
print("Running Frigate Counter Service Tests\n")
|
|
|
|
test_database_creation()
|
|
test_counter_logic()
|
|
test_environment_variables()
|
|
|
|
print("\nTest completed.") |