Files
dataset-uploader/features/environment.py

87 lines
3.1 KiB
Python

"""Behave environment configuration with setup/teardown hooks.
This module handles:
- Temporary directory creation and cleanup
- Shared test fixtures
- Logging configuration for tests
"""
from __future__ import annotations
import logging
import os
import shutil
import sys
import tempfile
from pathlib import Path
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from behave.model import Feature, Scenario
from behave.runner import Context
# Add scripts directory to Python path for imports
scripts_dir = Path(__file__).parent.parent / "scripts"
if str(scripts_dir) not in sys.path:
sys.path.insert(0, str(scripts_dir))
# Suppress verbose logging from libraries during tests
logging.getLogger("rdflib").setLevel(logging.ERROR)
logging.getLogger("datasets").setLevel(logging.ERROR)
logging.getLogger("urllib3").setLevel(logging.ERROR)
logging.getLogger("filelock").setLevel(logging.ERROR)
def before_all(context: Context) -> None:
"""Run once before all tests - create base temp directory."""
context.base_temp_dir = Path(tempfile.mkdtemp(prefix="behave_rdf_tests_"))
context.cleanup_dirs = [] # list[Path]
def after_all(context: Context) -> None:
"""Run once after all tests - cleanup base temp directory."""
if hasattr(context, "base_temp_dir") and context.base_temp_dir.exists():
shutil.rmtree(context.base_temp_dir, ignore_errors=True)
# Cleanup any remaining registered directories
if hasattr(context, "cleanup_dirs"):
for cleanup_dir in context.cleanup_dirs:
if cleanup_dir.exists():
shutil.rmtree(cleanup_dir, ignore_errors=True)
def before_feature(context: Context, feature: Feature) -> None:
"""Run before each feature file."""
# Create feature-specific temp directory
feature_name = feature.name.replace(" ", "_").lower()[:30]
context.feature_temp_dir = context.base_temp_dir / f"feature_{feature_name}"
context.feature_temp_dir.mkdir(parents=True, exist_ok=True)
def after_feature(context: Context, feature: Feature) -> None:
"""Run after each feature file - cleanup feature temp directory."""
if hasattr(context, "feature_temp_dir") and context.feature_temp_dir.exists():
shutil.rmtree(context.feature_temp_dir, ignore_errors=True)
def before_scenario(context: Context, scenario: Scenario) -> None:
"""Run before each scenario - create scenario-specific temp directory."""
scenario_name = scenario.name.replace(" ", "_").lower()[:30]
context.scenario_temp_dir = context.feature_temp_dir / f"scenario_{scenario_name}"
context.scenario_temp_dir.mkdir(parents=True, exist_ok=True)
# Initialize scenario-specific attributes
context.input_path = None
context.output_path = None
context.result = None
context.rdf_format = None
context.expected_triple_count = None
def after_scenario(context: Context, scenario: Scenario) -> None:
"""Run after each scenario - cleanup scenario temp directory."""
if hasattr(context, "scenario_temp_dir") and context.scenario_temp_dir.exists():
shutil.rmtree(context.scenario_temp_dir, ignore_errors=True)