BE: test fixes

Signed-off-by: jokob-sk <jokob.sk@gmail.com>

test/integration/integration_test.py

@@ -1,448 +1,237 @@
-#!/usr/bin/env python3
-"""
-NetAlertX SQL Injection Fix - Integration Testing
-Validates the complete implementation as requested by maintainer jokob-sk
-"""
-
-import sys
 import os
 import sqlite3
-import json
-import unittest
-from unittest.mock import Mock, patch, MagicMock
 import tempfile
-import subprocess
+import pytest
+from unittest.mock import Mock, patch
 
 # Add server paths
+import sys
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'server'))
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'server', 'db'))
 
-# Import our modules
 from db.sql_safe_builder import SafeConditionBuilder, create_safe_condition_builder
 from messaging.reporting import get_notifications
 
-class NetAlertXIntegrationTest(unittest.TestCase):
+# -----------------------------
-    """
+# Fixtures
-    Comprehensive integration tests to validate:
+# -----------------------------
-    1. Fresh install compatibility
+@pytest.fixture
-    2. Existing DB/config compatibility  
+def test_db_path():
-    3. Notification system integration
+    path = tempfile.mktemp(suffix=".db")
-    4. Settings persistence
+    yield path
-    5. Device operations
+    if os.path.exists(path):
-    6. Plugin functionality
+        os.remove(path)
-    7. Error handling
+
-    """
+@pytest.fixture
-    
+def builder():
-    def setUp(self):
+    return create_safe_condition_builder()
-        """Set up test environment"""
+
-        self.test_db_path = tempfile.mktemp(suffix='.db')
+@pytest.fixture
-        self.builder = SafeConditionBuilder()
+def test_db(test_db_path):
-        self.create_test_database()
+    conn = sqlite3.connect(test_db_path)
-    
+    cur = conn.cursor()
-    def tearDown(self):
+
-        """Clean up test environment"""
+    # Minimal schema for integration testing
-        if os.path.exists(self.test_db_path):
+    cur.execute('''
-            os.remove(self.test_db_path)
+        CREATE TABLE IF NOT EXISTS Events_Devices (
-    
+            eve_MAC TEXT,
-    def create_test_database(self):
+            eve_DateTime TEXT,
-        """Create test database with NetAlertX schema"""
+            devLastIP TEXT,
-        conn = sqlite3.connect(self.test_db_path)
+            eve_EventType TEXT,
-        cursor = conn.cursor()
+            devName TEXT,
-        
+            devComments TEXT,
-        # Create minimal schema for testing
+            eve_PendingAlertEmail INTEGER
-        cursor.execute('''
+        )
-            CREATE TABLE IF NOT EXISTS Events_Devices (
+    ''')
-                eve_MAC TEXT,
+
-                eve_DateTime TEXT,
+    cur.execute('''
-                devLastIP TEXT,
+        CREATE TABLE IF NOT EXISTS Devices (
-                eve_EventType TEXT,
+            devMac TEXT PRIMARY KEY,
-                devName TEXT,
+            devName TEXT,
-                devComments TEXT,
+            devComments TEXT,
-                eve_PendingAlertEmail INTEGER
+            devAlertEvents INTEGER DEFAULT 1,
-            )
+            devAlertDown INTEGER DEFAULT 1
-        ''')
+        )
-        
+    ''')
-        cursor.execute('''
+
-            CREATE TABLE IF NOT EXISTS Devices (
+    cur.execute('''
-                devMac TEXT PRIMARY KEY,
+        CREATE TABLE IF NOT EXISTS Events (
-                devName TEXT,
+            eve_MAC TEXT,
-                devComments TEXT,
+            eve_DateTime TEXT,
-                devAlertEvents INTEGER DEFAULT 1,
+            eve_EventType TEXT,
-                devAlertDown INTEGER DEFAULT 1
+            eve_PendingAlertEmail INTEGER
-            )
+        )
-        ''')
+    ''')
-        
+
-        cursor.execute('''
+    cur.execute('''
-            CREATE TABLE IF NOT EXISTS Events (
+        CREATE TABLE IF NOT EXISTS Plugins_Events (
-                eve_MAC TEXT,
+            Plugin TEXT,
-                eve_DateTime TEXT,
+            Object_PrimaryId TEXT,
-                eve_EventType TEXT,
+            Object_SecondaryId TEXT,
-                eve_PendingAlertEmail INTEGER
+            DateTimeChanged TEXT,
-            )
+            Watched_Value1 TEXT,
-        ''')
+            Watched_Value2 TEXT,
-        
+            Watched_Value3 TEXT,
-        cursor.execute('''
+            Watched_Value4 TEXT,
-            CREATE TABLE IF NOT EXISTS Plugins_Events (
+            Status TEXT
-                Plugin TEXT,
+        )
-                Object_PrimaryId TEXT,
+    ''')
-                Object_SecondaryId TEXT,
+
-                DateTimeChanged TEXT,
+    # Insert test data
-                Watched_Value1 TEXT,
+    test_data = [
-                Watched_Value2 TEXT,
+        ('aa:bb:cc:dd:ee:ff', '2024-01-01 12:00:00', '192.168.1.100', 'New Device', 'Test Device', 'Test Comment', 1),
-                Watched_Value3 TEXT,
+        ('11:22:33:44:55:66', '2024-01-01 12:01:00', '192.168.1.101', 'Connected', 'Test Device 2', 'Another Comment', 1),
-                Watched_Value4 TEXT,
+        ('77:88:99:aa:bb:cc', '2024-01-01 12:02:00', '192.168.1.102', 'Disconnected', 'Test Device 3', 'Third Comment', 1),
-                Status TEXT
+    ]
-            )
+    cur.executemany('''
-        ''')
+        INSERT INTO Events_Devices (eve_MAC, eve_DateTime, devLastIP, eve_EventType, devName, devComments, eve_PendingAlertEmail)
-        
+        VALUES (?, ?, ?, ?, ?, ?, ?)
-        # Insert test data
+    ''', test_data)
-        test_data = [
+
-            ('aa:bb:cc:dd:ee:ff', '2024-01-01 12:00:00', '192.168.1.100', 'New Device', 'Test Device', 'Test Comment', 1),
+    conn.commit()
-            ('11:22:33:44:55:66', '2024-01-01 12:01:00', '192.168.1.101', 'Connected', 'Test Device 2', 'Another Comment', 1),
+    conn.close()
-            ('77:88:99:aa:bb:cc', '2024-01-01 12:02:00', '192.168.1.102', 'Disconnected', 'Test Device 3', 'Third Comment', 1),
+    return test_db_path
-        ]
+
-        
+# -----------------------------
-        cursor.executemany('''
+# Tests
-            INSERT INTO Events_Devices (eve_MAC, eve_DateTime, devLastIP, eve_EventType, devName, devComments, eve_PendingAlertEmail)
+# -----------------------------
-            VALUES (?, ?, ?, ?, ?, ?, ?)
+
-        ''', test_data)
+def test_fresh_install_compatibility(builder):
-        
+    condition, params = builder.get_safe_condition_legacy("")
-        conn.commit()
+    assert condition == ""
-        conn.close()
+    assert params == {}
-    
+
-    def test_1_fresh_install_compatibility(self):
+    condition, params = builder.get_safe_condition_legacy("AND devName = 'TestDevice'")
-        """Test 1: Fresh install (no DB/config)"""
+    assert "devName = :" in condition
-        print("\n=== TEST 1: Fresh Install Compatibility ===")
+    assert 'TestDevice' in params.values()
-        
+
-        # Test SafeConditionBuilder initialization
+def test_existing_db_compatibility():
-        builder = create_safe_condition_builder()
+    mock_db = Mock()
-        self.assertIsInstance(builder, SafeConditionBuilder)
+    mock_result = Mock()
-        
+    mock_result.columnNames = ['MAC', 'Datetime', 'IP', 'Event Type', 'Device name', 'Comments']
-        # Test empty condition handling
+    mock_result.json = {'data': []}
-        condition, params = builder.get_safe_condition_legacy("")
+    mock_db.get_table_as_json.return_value = mock_result
-        self.assertEqual(condition, "")
+
-        self.assertEqual(params, {})
+    with patch('messaging.reporting.get_setting_value') as s:
-        
+        s.side_effect = lambda k: {
-        # Test basic valid condition
+            'NTFPRCS_INCLUDED_SECTIONS': ['new_devices', 'events'],
-        condition, params = builder.get_safe_condition_legacy("AND devName = 'TestDevice'")
+            'NTFPRCS_new_dev_condition': "AND devName = 'TestDevice'",
-        self.assertIn("devName = :", condition)
+            'NTFPRCS_event_condition': "AND devComments LIKE '%test%'",
-        self.assertIn('TestDevice', list(params.values()))
+            'NTFPRCS_alert_down_time': '60'
-        
+        }.get(k, '')
-        print("✅ Fresh install compatibility: PASSED")
+
-    
+        with patch('messaging.reporting.get_timezone_offset', return_value='+00:00'):
-    def test_2_existing_db_compatibility(self):
-        """Test 2: Existing DB/config compatibility"""
-        print("\n=== TEST 2: Existing DB/Config Compatibility ===")
-        
-        # Mock database connection
-        mock_db = Mock()
-        mock_sql = Mock()
-        mock_db.sql = mock_sql
-        mock_db.get_table_as_json = Mock()
-        
-        # Mock return value for get_table_as_json
-        mock_result = Mock()
-        mock_result.columnNames = ['MAC', 'Datetime', 'IP', 'Event Type', 'Device name', 'Comments']
-        mock_result.json = {'data': []}
-        mock_db.get_table_as_json.return_value = mock_result
-        
-        # Mock settings
-        with patch('messaging.reporting.get_setting_value') as mock_settings:
-            mock_settings.side_effect = lambda key: {
-                'NTFPRCS_INCLUDED_SECTIONS': ['new_devices', 'events'],
-                'NTFPRCS_new_dev_condition': "AND devName = 'TestDevice'",
-                'NTFPRCS_event_condition': "AND devComments LIKE '%test%'",
-                'NTFPRCS_alert_down_time': '60'
-            }.get(key, '')
-            
-            with patch('messaging.reporting.get_timezone_offset', return_value='+00:00'):
-                # Test get_notifications function
-                result = get_notifications(mock_db)
-                
-                # Verify structure
-                self.assertIn('new_devices', result)
-                self.assertIn('events', result)
-                self.assertIn('new_devices_meta', result)
-                self.assertIn('events_meta', result)
-                
-                # Verify parameterized queries were called
-                self.assertTrue(mock_db.get_table_as_json.called)
-                
-                # Check that calls used parameters (not direct concatenation)
-                calls = mock_db.get_table_as_json.call_args_list
-                for call in calls:
-                    args, kwargs = call
-                    if len(args) > 1:  # Has parameters
-                        self.assertIsInstance(args[1], dict)  # Parameters should be dict
-        
-        print("✅ Existing DB/config compatibility: PASSED")
-    
-    def test_3_notification_system_integration(self):
-        """Test 3: Notification testing integration"""
-        print("\n=== TEST 3: Notification System Integration ===")
-        
-        # Test that SafeConditionBuilder integrates with notification queries
-        builder = create_safe_condition_builder()
-        
-        # Test email notification conditions
-        email_condition = "AND devName = 'EmailTestDevice'"
-        condition, params = builder.get_safe_condition_legacy(email_condition)
-        self.assertIn("devName = :", condition)
-        self.assertIn('EmailTestDevice', list(params.values()))
-        
-        # Test Apprise notification conditions
-        apprise_condition = "AND eve_EventType = 'Connected'"
-        condition, params = builder.get_safe_condition_legacy(apprise_condition)
-        self.assertIn("eve_EventType = :", condition)
-        self.assertIn('Connected', list(params.values()))
-        
-        # Test webhook notification conditions
-        webhook_condition = "AND devComments LIKE '%webhook%'"
-        condition, params = builder.get_safe_condition_legacy(webhook_condition)
-        self.assertIn("devComments LIKE :", condition)
-        self.assertIn('%webhook%', list(params.values()))
-        
-        # Test MQTT notification conditions
-        mqtt_condition = "AND eve_MAC = 'aa:bb:cc:dd:ee:ff'"
-        condition, params = builder.get_safe_condition_legacy(mqtt_condition)
-        self.assertIn("eve_MAC = :", condition)
-        self.assertIn('aa:bb:cc:dd:ee:ff', list(params.values()))
-        
-        print("✅ Notification system integration: PASSED")
-    
-    def test_4_settings_persistence(self):
-        """Test 4: Settings persistence"""
-        print("\n=== TEST 4: Settings Persistence ===")
-        
-        # Test various setting formats that should be supported
-        test_settings = [
-            "AND devName = 'Persistent Device'",
-            "AND devComments = {s-quote}Legacy Quote{s-quote}",
-            "AND eve_EventType IN ('Connected', 'Disconnected')",
-            "AND devLastIP = '192.168.1.1'",
-            ""  # Empty setting should work
-        ]
-        
-        builder = create_safe_condition_builder()
-        
-        for setting in test_settings:
-            try:
-                condition, params = builder.get_safe_condition_legacy(setting)
-                # Should not raise exception
-                self.assertIsInstance(condition, str)
-                self.assertIsInstance(params, dict)
-            except Exception as e:
-                if setting != "":  # Empty is allowed to "fail" gracefully
-                    self.fail(f"Setting '{setting}' failed: {e}")
-        
-        print("✅ Settings persistence: PASSED")
-    
-    def test_5_device_operations(self):
-        """Test 5: Device operations"""
-        print("\n=== TEST 5: Device Operations ===")
-        
-        # Test device-related conditions
-        builder = create_safe_condition_builder()
-        
-        device_conditions = [
-            "AND devName = 'Updated Device'",
-            "AND devMac = 'aa:bb:cc:dd:ee:ff'",
-            "AND devComments = 'Device updated successfully'",
-            "AND devLastIP = '192.168.1.200'"
-        ]
-        
-        for condition in device_conditions:
-            safe_condition, params = builder.get_safe_condition_legacy(condition)
-            self.assertTrue(len(params) > 0 or safe_condition == "")
-            # Ensure no direct string concatenation in output
-            self.assertNotIn("'", safe_condition)  # No literal quotes in SQL
-        
-        print("✅ Device operations: PASSED")
-    
-    def test_6_plugin_functionality(self):
-        """Test 6: Plugin functionality"""
-        print("\n=== TEST 6: Plugin Functionality ===")
-        
-        # Test plugin-related conditions that might be used
-        builder = create_safe_condition_builder()
-        
-        plugin_conditions = [
-            "AND Plugin = 'TestPlugin'",
-            "AND Object_PrimaryId = 'primary123'",
-            "AND Status = 'Active'"
-        ]
-        
-        for condition in plugin_conditions:
-            safe_condition, params = builder.get_safe_condition_legacy(condition)
-            if safe_condition:  # If condition was accepted
-                self.assertIn(":", safe_condition)  # Should have parameter placeholder
-                self.assertTrue(len(params) > 0)  # Should have parameters
-        
-        # Test that plugin data structure is preserved
-        mock_db = Mock()
-        mock_db.sql = Mock()
-        mock_result = Mock()
-        mock_result.columnNames = ['Plugin', 'Object_PrimaryId', 'Status']
-        mock_result.json = {'data': []}
-        mock_db.get_table_as_json.return_value = mock_result
-        
-        with patch('messaging.reporting.get_setting_value') as mock_settings:
-            mock_settings.side_effect = lambda key: {
-                'NTFPRCS_INCLUDED_SECTIONS': ['plugins']
-            }.get(key, '')
-            
             result = get_notifications(mock_db)
-            self.assertIn('plugins', result)
-            self.assertIn('plugins_meta', result)
-        
-        print("✅ Plugin functionality: PASSED")
-    
-    def test_7_sql_injection_prevention(self):
-        """Test 7: SQL injection prevention (critical security test)"""
-        print("\n=== TEST 7: SQL Injection Prevention ===")
-        
-        # Test malicious inputs are properly blocked
-        malicious_inputs = [
-            "'; DROP TABLE Events_Devices; --",
-            "' OR '1'='1",
-            "1' UNION SELECT * FROM Devices --",
-            "'; INSERT INTO Events VALUES ('hacked'); --",
-            "' AND (SELECT COUNT(*) FROM sqlite_master) > 0 --"
-        ]
-        
-        builder = create_safe_condition_builder()
-        
-        for malicious_input in malicious_inputs:
-            condition, params = builder.get_safe_condition_legacy(malicious_input)
-            # All malicious inputs should result in empty/safe condition
-            self.assertEqual(condition, "", f"Malicious input not blocked: {malicious_input}")
-            self.assertEqual(params, {}, f"Parameters returned for malicious input: {malicious_input}")
-        
-        print("✅ SQL injection prevention: PASSED")
-    
-    def test_8_error_log_inspection(self):
-        """Test 8: Error handling and logging"""
-        print("\n=== TEST 8: Error Handling and Logging ===")
-        
-        # Test that invalid inputs are logged properly
-        builder = create_safe_condition_builder()
-        
-        # This should log an error but not crash
-        invalid_condition = "INVALID SQL SYNTAX HERE"
-        condition, params = builder.get_safe_condition_legacy(invalid_condition)
-        
-        # Should return empty/safe values
-        self.assertEqual(condition, "")
-        self.assertEqual(params, {})
-        
-        # Test edge cases
-        edge_cases = [
-            None,  # This would cause TypeError in unpatched version
-            "",
-            "   ",
-            "\n\t",
-            "AND column_not_in_whitelist = 'value'"
-        ]
-        
-        for case in edge_cases:
-            try:
-                if case is not None:
-                    condition, params = builder.get_safe_condition_legacy(case)
-                    self.assertIsInstance(condition, str)
-                    self.assertIsInstance(params, dict)
-            except Exception as e:
-                # Should not crash on any input
-                self.fail(f"Unexpected exception for input {case}: {e}")
-        
-        print("✅ Error handling and logging: PASSED")
-    
-    def test_9_backward_compatibility(self):
-        """Test 9: Backward compatibility with legacy settings"""
-        print("\n=== TEST 9: Backward Compatibility ===")
-        
-        # Test legacy {s-quote} placeholder support
-        builder = create_safe_condition_builder()
-        
-        legacy_conditions = [
-            "AND devName = {s-quote}Legacy Device{s-quote}",
-            "AND devComments = {s-quote}Old Style Quote{s-quote}",
-            "AND devName = 'Normal Quote'"  # Modern style should still work
-        ]
-        
-        for legacy_condition in legacy_conditions:
-            condition, params = builder.get_safe_condition_legacy(legacy_condition)
-            if condition:  # If accepted as valid
-                # Should not contain the {s-quote} placeholder in output
-                self.assertNotIn("{s-quote}", condition)
-                # Should have proper parameter binding
-                self.assertIn(":", condition)
-                self.assertTrue(len(params) > 0)
-        
-        print("✅ Backward compatibility: PASSED")
-    
-    def test_10_performance_impact(self):
-        """Test 10: Performance impact measurement"""
-        print("\n=== TEST 10: Performance Impact ===")
-        
-        import time
-        
-        builder = create_safe_condition_builder()
-        
-        # Test performance of condition building
-        test_condition = "AND devName = 'Performance Test Device'"
-        
-        start_time = time.time()
-        for _ in range(1000):  # Run 1000 times
-            condition, params = builder.get_safe_condition_legacy(test_condition)
-        end_time = time.time()
-        
-        total_time = end_time - start_time
-        avg_time_ms = (total_time / 1000) * 1000
-        
-        print(f"Average condition building time: {avg_time_ms:.3f}ms")
-        
-        # Should be under 1ms per condition
-        self.assertLess(avg_time_ms, 1.0, "Performance regression detected")
-        
-        print("✅ Performance impact: PASSED")
 
-def run_integration_tests():
+    assert 'new_devices' in result
-    """Run all integration tests and generate report"""
+    assert 'events' in result
-    print("=" * 70)
+    assert 'new_devices_meta' in result
-    print("NetAlertX SQL Injection Fix - Integration Test Suite")
+    assert 'events_meta' in result
-    print("Validating PR #1182 as requested by maintainer jokob-sk")
+    assert mock_db.get_table_as_json.called
-    print("=" * 70)
-    
-    # Run tests
-    suite = unittest.TestLoader().loadTestsFromTestCase(NetAlertXIntegrationTest)
-    runner = unittest.TextTestRunner(verbosity=2)
-    result = runner.run(suite)
-    
-    # Generate summary
-    print("\n" + "=" * 70)
-    print("INTEGRATION TEST SUMMARY")
-    print("=" * 70)
-    
-    total_tests = result.testsRun
-    failures = len(result.failures)
-    errors = len(result.errors)
-    passed = total_tests - failures - errors
-    
-    print(f"Total Tests: {total_tests}")
-    print(f"Passed: {passed}")
-    print(f"Failed: {failures}")
-    print(f"Errors: {errors}")
-    print(f"Success Rate: {(passed/total_tests)*100:.1f}%")
-    
-    if failures == 0 and errors == 0:
-        print("\n🎉 ALL INTEGRATION TESTS PASSED!")
-        print("✅ Ready for maintainer approval")
-        return True
-    else:
-        print("\n❌ INTEGRATION TESTS FAILED")
-        print("🚫 Requires fixes before approval")
-        return False
 
-if __name__ == "__main__":
+def test_notification_system_integration(builder):
-    success = run_integration_tests()
+    email_condition = "AND devName = 'EmailTestDevice'"
-    sys.exit(0 if success else 1)
+    condition, params = builder.get_safe_condition_legacy(email_condition)
+    assert "devName = :" in condition
+    assert 'EmailTestDevice' in params.values()
+
+    apprise_condition = "AND eve_EventType = 'Connected'"
+    condition, params = builder.get_safe_condition_legacy(apprise_condition)
+    assert "eve_EventType = :" in condition
+    assert 'Connected' in params.values()
+
+    webhook_condition = "AND devComments LIKE '%webhook%'"
+    condition, params = builder.get_safe_condition_legacy(webhook_condition)
+    assert "devComments LIKE :" in condition
+    assert '%webhook%' in params.values()
+
+    mqtt_condition = "AND eve_MAC = 'aa:bb:cc:dd:ee:ff'"
+    condition, params = builder.get_safe_condition_legacy(mqtt_condition)
+    assert "eve_MAC = :" in condition
+    assert 'aa:bb:cc:dd:ee:ff' in params.values()
+
+def test_settings_persistence(builder):
+    test_settings = [
+        "AND devName = 'Persistent Device'",
+        "AND devComments = {s-quote}Legacy Quote{s-quote}",
+        "AND eve_EventType IN ('Connected', 'Disconnected')",
+        "AND devLastIP = '192.168.1.1'",
+        ""
+    ]
+    for setting in test_settings:
+        condition, params = builder.get_safe_condition_legacy(setting)
+        assert isinstance(condition, str)
+        assert isinstance(params, dict)
+
+def test_device_operations(builder):
+    device_conditions = [
+        "AND devName = 'Updated Device'",
+        "AND devMac = 'aa:bb:cc:dd:ee:ff'",
+        "AND devComments = 'Device updated successfully'",
+        "AND devLastIP = '192.168.1.200'"
+    ]
+    for cond in device_conditions:
+        safe_condition, params = builder.get_safe_condition_legacy(cond)
+        assert len(params) > 0 or safe_condition == ""
+        assert "'" not in safe_condition
+
+def test_plugin_functionality(builder):
+    plugin_conditions = [
+        "AND Plugin = 'TestPlugin'",
+        "AND Object_PrimaryId = 'primary123'",
+        "AND Status = 'Active'"
+    ]
+    for cond in plugin_conditions:
+        safe_condition, params = builder.get_safe_condition_legacy(cond)
+        if safe_condition:
+            assert ":" in safe_condition
+            assert len(params) > 0
+
+def test_sql_injection_prevention(builder):
+    malicious_inputs = [
+        "'; DROP TABLE Events_Devices; --",
+        "' OR '1'='1",
+        "1' UNION SELECT * FROM Devices --",
+        "'; INSERT INTO Events VALUES ('hacked'); --",
+        "' AND (SELECT COUNT(*) FROM sqlite_master) > 0 --"
+    ]
+    for payload in malicious_inputs:
+        condition, params = builder.get_safe_condition_legacy(payload)
+        assert condition == ""
+        assert params == {}
+
+def test_error_handling(builder):
+    invalid_condition = "INVALID SQL SYNTAX HERE"
+    condition, params = builder.get_safe_condition_legacy(invalid_condition)
+    assert condition == ""
+    assert params == {}
+
+    edge_cases = [None, "", "   ", "\n\t", "AND column_not_in_whitelist = 'value'"]
+    for case in edge_cases:
+        if case is not None:
+            condition, params = builder.get_safe_condition_legacy(case)
+            assert isinstance(condition, str)
+            assert isinstance(params, dict)
+
+def test_backward_compatibility(builder):
+    legacy_conditions = [
+        "AND devName = {s-quote}Legacy Device{s-quote}",
+        "AND devComments = {s-quote}Old Style Quote{s-quote}",
+        "AND devName = 'Normal Quote'"
+    ]
+    for cond in legacy_conditions:
+        condition, params = builder.get_safe_condition_legacy(cond)
+        if condition:
+            assert "{s-quote}" not in condition
+            assert ":" in condition
+            assert len(params) > 0
+
+def test_performance_impact(builder):
+    import time
+    test_condition = "AND devName = 'Performance Test Device'"
+    start = time.time()
+    for _ in range(1000):
+        condition, params = builder.get_safe_condition_legacy(test_condition)
+    end = time.time()
+    avg_ms = (end - start) / 1000 * 1000
+    assert avg_ms < 1.0

test/integration/test_sql_injection_fix.py

@@ -1,139 +0,0 @@
-#!/usr/bin/env python3
-"""
-Test script to validate SQL injection fixes for issue #1179
-"""
-import re
-import sys
-
-def test_datetime_injection_fix():
-    """Test that datetime injection vulnerability is fixed"""
-    
-    # Read the reporting.py file
-    with open('server/messaging/reporting.py', 'r') as f:
-        content = f.read()
-    
-    # Check for vulnerable f-string patterns with datetime and user input
-    vulnerable_patterns = [
-        r"datetime\('now',\s*f['\"].*{get_setting_value\('NTFPRCS_alert_down_time'\)}",
-        r"datetime\('now',\s*f['\"].*{get_timezone_offset\(\)}"
-    ]
-    
-    vulnerabilities_found = []
-    for pattern in vulnerable_patterns:
-        matches = re.findall(pattern, content)
-        if matches:
-            vulnerabilities_found.extend(matches)
-    
-    if vulnerabilities_found:
-        print("❌ SECURITY TEST FAILED: Vulnerable datetime patterns found:")
-        for vuln in vulnerabilities_found:
-            print(f"  - {vuln}")
-        return False
-    
-    # Check for the secure patterns
-    secure_patterns = [
-        r"minutes = int\(get_setting_value\('NTFPRCS_alert_down_time'\) or 0\)",
-        r"tz_offset = get_timezone_offset\(\)"
-    ]
-    
-    secure_found = 0
-    for pattern in secure_patterns:
-        if re.search(pattern, content):
-            secure_found += 1
-    
-    if secure_found >= 2:
-        print("✅ SECURITY TEST PASSED: Secure datetime handling implemented")
-        return True
-    else:
-        print("⚠️  SECURITY TEST WARNING: Expected secure patterns not fully found")
-        return False
-
-def test_notification_instance_fix():
-    """Test that the clearPendingEmailFlag function is secure"""
-    
-    with open('server/models/notification_instance.py', 'r') as f:
-        content = f.read()
-    
-    # Check for vulnerable f-string patterns in clearPendingEmailFlag
-    clearflag_section = ""
-    in_function = False
-    lines = content.split('\n')
-    
-    for line in lines:
-        if 'def clearPendingEmailFlag' in line:
-            in_function = True
-        elif in_function and line.strip() and not line.startswith('    ') and not line.startswith('\t'):
-            break
-        
-        if in_function:
-            clearflag_section += line + '\n'
-    
-    # Check for vulnerable patterns
-    vulnerable_patterns = [
-        r"f['\"].*{get_setting_value\('NTFPRCS_alert_down_time'\)}",
-        r"f['\"].*{get_timezone_offset\(\)}"
-    ]
-    
-    vulnerabilities_found = []
-    for pattern in vulnerable_patterns:
-        matches = re.findall(pattern, clearflag_section)
-        if matches:
-            vulnerabilities_found.extend(matches)
-    
-    if vulnerabilities_found:
-        print("❌ SECURITY TEST FAILED: clearPendingEmailFlag still vulnerable:")
-        for vuln in vulnerabilities_found:
-            print(f"  - {vuln}")
-        return False
-    
-    print("✅ SECURITY TEST PASSED: clearPendingEmailFlag appears secure")
-    return True
-
-def test_code_quality():
-    """Test basic code quality and imports"""
-    
-    # Check if the modified files can be imported (basic syntax check)
-    try:
-        import subprocess
-        result = subprocess.run([
-            'python3', '-c', 
-            'import sys; sys.path.append("server"); from messaging import reporting'
-        ], capture_output=True, text=True, cwd='.')
-        
-        if result.returncode == 0:
-            print("✅ CODE QUALITY TEST PASSED: reporting.py imports successfully")
-            return True
-        else:
-            print(f"❌ CODE QUALITY TEST FAILED: Import error: {result.stderr}")
-            return False
-    except Exception as e:
-        print(f"⚠️  CODE QUALITY TEST WARNING: Could not test imports: {e}")
-        return True  # Don't fail for environment issues
-
-if __name__ == "__main__":
-    print("🔒 Running SQL Injection Security Tests for Issue #1179\n")
-    
-    tests = [
-        ("Datetime Injection Fix", test_datetime_injection_fix),
-        ("Notification Instance Security", test_notification_instance_fix),
-        ("Code Quality", test_code_quality)
-    ]
-    
-    results = []
-    for test_name, test_func in tests:
-        print(f"Running: {test_name}")
-        result = test_func()
-        results.append(result)
-        print()
-    
-    passed = sum(results)
-    total = len(results)
-    
-    print(f"🔒 Security Test Summary: {passed}/{total} tests passed")
-    
-    if passed == total:
-        print("✅ All security tests passed! The SQL injection fixes are working correctly.")
-        sys.exit(0)
-    else:
-        print("❌ Some security tests failed. Please review the fixes.")
-        sys.exit(1)