forked from cleveragents/cleveragents-core
532f128a49
Replaced all 28 uses of the deprecated tempfile.mktemp() function with the atomic tempfile.mkstemp() + os.close(fd) pattern across 16 test infrastructure and benchmark files. Added import os where it was not already present. tempfile.mktemp() has been deprecated since Python 2.3 due to a TOCTOU race condition: between name generation and file creation, another process could create a file at the predicted path. Under 16-worker parallel test execution, there was also a non-zero risk of path collisions. Files modified: - features/environment.py (3 call sites) - robot/helper_resource_registry_migration.py (3 call sites) - robot/helper_plan_phase_migration.py (1 call site) - robot/helper_plan_persistence_e2e.py (1 call site) - robot/helper_persistence_lifecycle.py (1 call site) - robot/helper_db_lifecycle_models.py (1 call site) - robot/helper_automation_profiles.py (1 call site) - features/steps/skill_discovery_steps.py (2 call sites) - features/steps/resource_registry_tables_steps.py (2 call sites) - features/steps/plan_persistence_steps.py (1 call site) - features/steps/persistence_robot_alignment_steps.py (1 call site) - features/steps/garbage_collection_cli_steps.py (3 call sites) - features/steps/decision_recording_steps.py (1 call site) - features/steps/coverage_boost_steps.py (1 call site) - features/steps/cli_streaming_steps.py (1 call site) - features/steps/automation_profiles_guards_steps.py (1 call site) - benchmarks/persistence_robot_bench.py (4 call sites) All 10,640 BDD scenarios pass. No performance regression observed (within normal run-to-run variance). Zero remaining uses of tempfile.mktemp() in the codebase. ISSUES CLOSED: #730
149 lines
4.8 KiB
Python
149 lines
4.8 KiB
Python
"""Step definitions for coverage boost tests."""
|
|
|
|
import os
|
|
|
|
from behave import then, when
|
|
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.platform import ensure_cli_importable
|
|
|
|
|
|
@when("I create a Settings instance")
|
|
def step_create_settings(context):
|
|
"""Create a Settings instance."""
|
|
# Clear any existing singleton
|
|
if hasattr(Settings, "_instance"):
|
|
Settings._instance = None
|
|
# Remove database URL env vars so we test the actual pydantic defaults,
|
|
# not the per-scenario temp paths injected by environment.py.
|
|
for key in ("CLEVERAGENTS_DATABASE_URL", "CLEVERAGENTS_TEST_DATABASE_URL"):
|
|
os.environ.pop(key, None)
|
|
context.settings = Settings()
|
|
|
|
|
|
@then("it should have default values")
|
|
def step_check_default_values(context):
|
|
"""Check Settings has default values."""
|
|
assert context.settings.server_host == "0.0.0.0"
|
|
assert context.settings.server_port == 8080
|
|
assert context.settings.database_url == "sqlite:///cleveragents.db"
|
|
assert context.settings.debug_log_level == "INFO"
|
|
assert context.settings.storage_base_path.name == "data"
|
|
|
|
|
|
@when("I check if Settings is in production mode")
|
|
def step_check_production_mode(context):
|
|
"""Check production mode."""
|
|
settings = Settings()
|
|
context.is_production = settings.is_production
|
|
|
|
|
|
@then("it should return the correct production status")
|
|
def step_verify_production_status(context):
|
|
"""Verify production status."""
|
|
# Default should be production (debug_enabled=False, server_reload=False)
|
|
assert context.is_production
|
|
|
|
|
|
@when("I get the database URL from Settings")
|
|
def step_get_database_url(context):
|
|
"""Get database URL."""
|
|
# Clear singleton and database URL env vars so we test the actual
|
|
# pydantic defaults, not per-scenario temp paths from environment.py.
|
|
if hasattr(Settings, "_instance"):
|
|
Settings._instance = None
|
|
for key in ("CLEVERAGENTS_DATABASE_URL", "CLEVERAGENTS_TEST_DATABASE_URL"):
|
|
os.environ.pop(key, None)
|
|
settings = Settings()
|
|
context.db_url = settings.get_database_url()
|
|
context.test_db_url = settings.get_database_url(test=True)
|
|
|
|
|
|
@then("it should return the configured database URL")
|
|
def step_check_database_url(context):
|
|
"""Check database URL."""
|
|
assert context.db_url == "sqlite:///cleveragents.db"
|
|
assert context.test_db_url == "sqlite:///cleveragents_test.db"
|
|
|
|
|
|
@when("I check if any provider is configured in Settings")
|
|
def step_check_provider_configured(context):
|
|
"""Check if provider is configured."""
|
|
import shutil
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
# Clear any existing settings
|
|
from cleveragents.config import settings as settings_module
|
|
from cleveragents.config.settings import Settings
|
|
|
|
settings_module._settings = None
|
|
|
|
# Store current env vars
|
|
stored_vars = {}
|
|
for key in [
|
|
"OPENAI_API_KEY",
|
|
"ANTHROPIC_API_KEY",
|
|
"GOOGLE_API_KEY",
|
|
"AZURE_API_KEY",
|
|
"OPENROUTER_API_KEY",
|
|
"GEMINI_API_KEY",
|
|
"HF_TOKEN",
|
|
]:
|
|
stored_vars[key] = os.environ.get(key)
|
|
os.environ.pop(key, None)
|
|
|
|
# Move .env file temporarily if it exists
|
|
env_file = Path(".env")
|
|
temp_env = None
|
|
if env_file.exists():
|
|
fd, _tmp_env = tempfile.mkstemp(suffix=".env")
|
|
os.close(fd)
|
|
temp_env = Path(_tmp_env)
|
|
shutil.move(str(env_file), str(temp_env))
|
|
|
|
try:
|
|
# Test with no providers
|
|
settings1 = Settings()
|
|
context.no_provider = settings1.has_provider_configured()
|
|
|
|
# Test with a provider
|
|
os.environ["OPENAI_API_KEY"] = "test-key"
|
|
# Force new settings instance
|
|
settings_module._settings = None
|
|
settings2 = Settings()
|
|
context.with_provider = settings2.has_provider_configured()
|
|
finally:
|
|
# Restore .env file
|
|
if temp_env and temp_env.exists():
|
|
shutil.move(str(temp_env), str(env_file))
|
|
|
|
# Clean up and restore env vars
|
|
os.environ.pop("OPENAI_API_KEY", None)
|
|
for key, val in stored_vars.items():
|
|
if val:
|
|
os.environ[key] = val
|
|
settings_module._settings = None
|
|
|
|
|
|
@then("it should return the provider status")
|
|
def step_verify_provider_status(context):
|
|
"""Verify provider status."""
|
|
assert not context.no_provider
|
|
assert context.with_provider
|
|
|
|
|
|
@when("I import and use the platform module")
|
|
def step_use_platform_module(context):
|
|
"""Use platform module."""
|
|
context.cli_module = ensure_cli_importable()
|
|
|
|
|
|
@then("ensure_cli_importable should work")
|
|
def step_check_cli_importable(context):
|
|
"""Check ensure_cli_importable worked."""
|
|
assert context.cli_module is not None
|
|
assert context.cli_module.__name__ == "cleveragents.cli"
|
|
assert hasattr(context.cli_module, "main")
|
|
assert hasattr(context.cli_module, "app")
|