128 lines
4.9 KiB
Python
128 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Simple test script to verify the log parsing and monitoring functionality
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import tempfile
|
|
import time
|
|
from main import LogParser, Config, LogMonitor
|
|
|
|
def test_log_parsing():
|
|
"""Test log parsing functionality"""
|
|
print("Testing log parsing...")
|
|
|
|
# Test access log
|
|
access_line = '45.153.34.68 - - [02/Mar/2026:21:22:38 +0000] "GET / HTTP/1.1" 444 0 "-" "Mozilla/5.0"'
|
|
parsed = LogParser.parse_log_line(access_line)
|
|
assert parsed is not None, "Failed to parse access log"
|
|
assert parsed['type'] == 'access', "Wrong log type"
|
|
assert parsed['ip'] == '45.153.34.68', "Wrong IP"
|
|
print("✓ Access log parsing works")
|
|
|
|
# Test error log
|
|
error_line = '2026/03/02 04:11:13 [error] 1083281#1083281: *3381930 connect() failed (113: No route to host) while connecting to upstream, client: 185.71.113.95, server: jellyfin.nussnougate.net, request: "POST /Sessions/Playing/Progress HTTP/1.1", upstream: "http://192.168.100.101:8096/Sessions/Playing/Progress", host: "jellyfin.nussnougate.net"'
|
|
parsed = LogParser.parse_log_line(error_line)
|
|
assert parsed is not None, "Failed to parse error log"
|
|
assert parsed['type'] == 'error', "Wrong log type"
|
|
assert parsed['ip'] == '185.71.113.95', "Wrong IP"
|
|
print("✓ Error log parsing works")
|
|
|
|
# Test NAXSI log
|
|
naxsi_line = '{"ip":"192.253.248.11","server":"jellyfin.nussnougate.net","uri":"/","config":"block","rid":"a3da57ce31e7a5489d0625abe261f6e0","cscore0":"$UWA","score0":8,"zone0":"HEADERS","id0":10000034,"var_name0":"user-agent"}, client: 192.253.248.11, server: jellyfin.nussnougate.net, request: "GET / HTTP/1.1", host: "jellyfin.nussnougate.net", referrer: "http://jellyfin.nussnougate.net//.git/HEAD"'
|
|
parsed = LogParser.parse_log_line(naxsi_line)
|
|
assert parsed is not None, "Failed to parse NAXSI log"
|
|
assert parsed['type'] == 'error', "Wrong log type"
|
|
assert parsed['subtype'] == 'naxsi', "Wrong subtype"
|
|
assert parsed['ip'] == '192.253.248.11', "Wrong IP"
|
|
print("✓ NAXSI log parsing works")
|
|
|
|
def test_config_loading():
|
|
"""Test configuration loading"""
|
|
print("\nTesting configuration loading...")
|
|
|
|
config = Config()
|
|
logbull_config = config.get_logbull_config()
|
|
assert 'host' in logbull_config, "Missing LogBull host in config"
|
|
assert 'project_id' in logbull_config, "Missing LogBull project_id in config"
|
|
|
|
log_files = config.get_log_files()
|
|
assert len(log_files) > 0, "No log files configured"
|
|
|
|
service_config = config.get_service_config()
|
|
assert 'poll_interval' in service_config, "Missing poll_interval in config"
|
|
|
|
print("✓ Configuration loading works")
|
|
|
|
def test_file_monitoring():
|
|
"""Test file monitoring functionality"""
|
|
print("\nTesting file monitoring...")
|
|
|
|
# Create temporary log file
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.log') as f:
|
|
temp_log_file = f.name
|
|
f.write('45.153.34.68 - - [02/Mar/2026:21:22:38 +0000] "GET / HTTP/1.1" 444 0 "-" "Mozilla/5.0"\n')
|
|
f.write('2026/03/02 04:11:13 [error] 1083281#1083281: *3381930 connect() failed (113: No route to host) while connecting to upstream, client: 185.71.113.95, server: test.example.com, request: "POST /test HTTP/1.1", host: "test.example.com"\n')
|
|
|
|
try:
|
|
# Create a test config with our temp file
|
|
test_config_content = """
|
|
logbull:
|
|
host: "http://localhost:4005"
|
|
project_id: "778e67d7-5ec6-4c48-b199-cfbded605557"
|
|
flush_interval: 1
|
|
|
|
log_files:
|
|
- "{}"
|
|
|
|
service:
|
|
poll_interval: 0.1
|
|
max_lines_per_batch: 10
|
|
log_level: "DEBUG"
|
|
""".format(temp_log_file)
|
|
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.yaml') as f:
|
|
temp_config_file = f.name
|
|
f.write(test_config_content)
|
|
|
|
# Test monitor initialization
|
|
config = Config(temp_config_file)
|
|
monitor = LogMonitor(config)
|
|
|
|
# Process logs
|
|
monitor.process_logs()
|
|
|
|
print("✓ File monitoring works")
|
|
|
|
finally:
|
|
# Clean up
|
|
os.unlink(temp_log_file)
|
|
os.unlink(temp_config_file)
|
|
|
|
def main():
|
|
"""Run all tests"""
|
|
print("Running tests for proxy-to-logbull...\n")
|
|
|
|
try:
|
|
test_log_parsing()
|
|
test_config_loading()
|
|
test_file_monitoring()
|
|
|
|
print("\n🎉 All tests passed! The service is ready to use.")
|
|
print("\nTo run the service:")
|
|
print(" python3 main.py")
|
|
print("\nTo run as a systemd service:")
|
|
print(" sudo cp proxy-to-logbull.service /etc/systemd/system/")
|
|
print(" sudo systemctl daemon-reload")
|
|
print(" sudo systemctl start proxy-to-logbull")
|
|
print(" sudo systemctl enable proxy-to-logbull")
|
|
|
|
except Exception as e:
|
|
print(f"\n❌ Test failed: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main() |