forked from cleveragents/cleveragents-core
103 lines
3.1 KiB
Python
103 lines
3.1 KiB
Python
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
import traceback
|
|
from pathlib import Path
|
|
|
|
from cleveragents.core.config import ConfigurationManager
|
|
|
|
# Add the src directory to the path so we can import the package
|
|
sys.path.insert(
|
|
0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../src"))
|
|
)
|
|
|
|
|
|
def before_all(context):
|
|
# Set up any global context here
|
|
|
|
# Add custom exception hook for better error reporting
|
|
original_excepthook = sys.excepthook
|
|
|
|
def custom_excepthook(exc_type, exc_value, exc_traceback):
|
|
print("\n\n===== UNCAUGHT EXCEPTION =====")
|
|
print(f"Type: {exc_type.__name__}")
|
|
print(f"Value: {exc_value}")
|
|
print("Traceback:")
|
|
traceback.print_tb(exc_traceback)
|
|
print("==============================\n")
|
|
# Call the original excepthook
|
|
original_excepthook(exc_type, exc_value, exc_traceback)
|
|
|
|
sys.excepthook = custom_excepthook
|
|
|
|
|
|
def after_all(context):
|
|
# Clean up after all tests
|
|
pass
|
|
|
|
|
|
def before_feature(context, feature):
|
|
# Set up context for each feature
|
|
pass
|
|
|
|
|
|
def after_feature(context, feature):
|
|
# Clean up after each feature
|
|
pass
|
|
|
|
|
|
def before_scenario(context, scenario):
|
|
# Set up context for each scenario
|
|
context.temp_dir = Path(tempfile.mkdtemp())
|
|
context.config_manager = ConfigurationManager()
|
|
context.error = None
|
|
context.cleanups = []
|
|
context.cleanup_dirs = []
|
|
|
|
|
|
def after_scenario(context, scenario):
|
|
# Clean up after each scenario
|
|
# Capture any errors that occurred during the scenario
|
|
if scenario.status == "failed":
|
|
print(f"\n\n===== SCENARIO FAILED: {scenario.name} =====")
|
|
if hasattr(context, "error") and context.error:
|
|
print(f"Exception Type: {type(context.error).__name__}")
|
|
print(f"Exception Message: {str(context.error)}")
|
|
if hasattr(context.error, "__traceback__"):
|
|
print("--- Traceback ---")
|
|
tb_lines = traceback.format_exception(
|
|
type(context.error), context.error, context.error.__traceback__
|
|
)
|
|
print("".join(tb_lines), end="")
|
|
print("-------------------")
|
|
print("===============================\n")
|
|
|
|
if hasattr(context, "env_patchers"):
|
|
for patcher in reversed(context.env_patchers):
|
|
patcher.stop()
|
|
delattr(context, "env_patchers")
|
|
|
|
if hasattr(context, "cleanups"):
|
|
for cleanup in reversed(context.cleanups):
|
|
try:
|
|
cleanup()
|
|
except Exception as e:
|
|
print(f"Error during cleanup: {e}")
|
|
delattr(context, "cleanups")
|
|
|
|
def _cleanup_dir(temp_dir_obj):
|
|
if isinstance(temp_dir_obj, tempfile.TemporaryDirectory):
|
|
temp_dir_obj.cleanup()
|
|
elif temp_dir_obj and os.path.exists(temp_dir_obj):
|
|
shutil.rmtree(temp_dir_obj)
|
|
|
|
if hasattr(context, "cleanup_dirs"):
|
|
for temp_dir in context.cleanup_dirs:
|
|
_cleanup_dir(temp_dir)
|
|
del context.cleanup_dirs
|
|
|
|
if hasattr(context, "temp_dir"):
|
|
_cleanup_dir(context.temp_dir)
|
|
delattr(context, "temp_dir")
|