66 lines
2.2 KiB
Python
66 lines
2.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Quick test harness for AutomationLogger diagnostic verification.
|
|
This script loads the AutomationLogger module by path and runs a few methods to
|
|
create a test run and write logs. It prints the activity log and diagnostic file.
|
|
"""
|
|
import importlib.util
|
|
import sys
|
|
import os
|
|
from pathlib import Path
|
|
|
|
MODULE_PATH = '/data/app/igny8/backend/igny8_core/business/automation/services/automation_logger.py'
|
|
|
|
spec = importlib.util.spec_from_file_location('automation_logger', MODULE_PATH)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
AutomationLogger = mod.AutomationLogger
|
|
|
|
BASE_LOG_DIR = '/data/app/logs/automation'
|
|
SHARED_DIR = '/data/app/logs/automation/all_runs_test'
|
|
|
|
logger = AutomationLogger(base_log_dir=BASE_LOG_DIR, shared_log_dir=SHARED_DIR)
|
|
|
|
print('Using base_log_dir =', logger.base_log_dir)
|
|
print('Using shared_log_dir =', logger.shared_log_dir)
|
|
|
|
# Run a test flow
|
|
run_id = logger.start_run(999, 999, 'test')
|
|
print('Created run_id:', run_id)
|
|
|
|
logger.log_stage_progress(run_id, 999, 999, 0, 'Diagnostic: stage progress test')
|
|
logger.log_stage_error(run_id, 999, 999, 0, 'Diagnostic: simulated error')
|
|
logger.log_stage_complete(run_id, 999, 999, 0, 3, '0m 1s', 0)
|
|
|
|
# Print activity log via get_activity_log
|
|
activity = logger.get_activity_log(999, 999, run_id, last_n=50)
|
|
print('\nActivity log (last lines):')
|
|
for line in activity:
|
|
print(line)
|
|
|
|
# Print diagnostic file tail
|
|
diag_file = os.path.join(BASE_LOG_DIR, 'automation_diagnostic.log')
|
|
print('\nDiagnostic file path:', diag_file)
|
|
if os.path.exists(diag_file):
|
|
print('\nDiagnostic log tail:')
|
|
with open(diag_file, 'r') as f:
|
|
lines = f.readlines()
|
|
for line in lines[-50:]:
|
|
print(line.rstrip())
|
|
else:
|
|
print('Diagnostic file not found')
|
|
|
|
# List created directories for quick verification
|
|
print('\nListing created run dirs under base:')
|
|
for p in sorted(Path(BASE_LOG_DIR).rglob(run_id)):
|
|
print(p)
|
|
|
|
print('\nShared run dir listing:')
|
|
shared_run = os.path.join(SHARED_DIR, run_id)
|
|
if os.path.exists(shared_run):
|
|
for root, dirs, files in os.walk(shared_run):
|
|
for f in files:
|
|
print(os.path.join(root, f))
|
|
else:
|
|
print('Shared run dir not found')
|