learningcircuit--local-deep-research/tests/settings/env_vars/test_env_var_usage.py

#!/usr/bin/env python3
"""
Test that environment variables are accessed correctly throughout the codebase.
This validates that all env var access goes through SettingsManager where appropriate.
"""

import ast
import sys
from pathlib import Path
from typing import List, Tuple

# Make pytest optional for standalone execution
try:
    import pytest
except ImportError:
    pytest = None

# Add src to path to import settings
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "src"))

from local_deep_research.settings.manager import SettingsManager
from local_deep_research.settings.env_registry import registry as env_registry


# Get all registered environment variables
ALL_ENV_VARS = set(env_registry.get_all_env_vars().keys())
BOOTSTRAP_VARS = set(SettingsManager.get_bootstrap_env_vars().keys())


# Files/patterns where direct os.environ access is allowed
ALLOWED_PATTERNS = {
    # Configuration and settings modules
    "settings/",
    "config/",
    "setup.py",
    # Test files
    "test_",
    "_test.py",
    "tests/",
    # Scripts and utilities
    "scripts/",
    "utils/",
    "cli.py",
    ".pre-commit-hooks/",
    # Example scripts
    "examples/",
    # Specific modules that need direct access
    "log_utils.py",
    "paths.py",
    "queue_config.py",
    "server_config.py",
    "rate_limiter.py",  # Rate limiter config (decorator needs env at module load)
    # Database and migrations
    "alembic/",
    "migrations/",
    "encrypted_db.py",
    "sqlcipher_utils.py",
}

# System environment variables that are always allowed
SYSTEM_VARS = {
    "PATH",
    "HOME",
    "USER",
    "PYTHONPATH",
    "TMPDIR",
    "TEMP",
    "PWD",
    "SHELL",
    "TERM",
    "LANG",
    "LC_ALL",
    "TZ",
    "CI",
    "GITHUB_ACTIONS",
    "TESTING",  # External testing flag
    "PYTEST_CURRENT_TEST",  # Pytest sets this when running tests
    "DOCKER_CONTAINER",
    "WERKZEUG_RUN_MAIN",  # Flask/Werkzeug debug reloader detection
}


class EnvVarChecker(ast.NodeVisitor):
    def __init__(self, filename: str):
        self.filename = filename
        self.errors = []
        self.has_settings_manager_import = False

    def visit_ImportFrom(self, node):
        if node.module and "settings" in node.module.lower():
            for alias in node.names:
                if "SettingsManager" in (alias.name or ""):
                    self.has_settings_manager_import = True
        self.generic_visit(node)

    def visit_Call(self, node):
        # Check for os.environ.get() or os.getenv()
        is_environ_get = False
        env_var_name = None

        # Pattern 1: os.environ.get("VAR_NAME")
        if (
            isinstance(node.func, ast.Attribute)
            and node.func.attr == "get"
            and isinstance(node.func.value, ast.Attribute)
            and node.func.value.attr == "environ"
            and isinstance(node.func.value.value, ast.Name)
            and node.func.value.value.id == "os"
        ):
            is_environ_get = True
            if node.args and isinstance(node.args[0], ast.Constant):
                env_var_name = node.args[0].value

        # Pattern 2: os.getenv("VAR_NAME")
        elif (
            isinstance(node.func, ast.Attribute)
            and node.func.attr == "getenv"
            and isinstance(node.func.value, ast.Name)
            and node.func.value.id == "os"
        ):
            is_environ_get = True
            if node.args and isinstance(node.args[0], ast.Constant):
                env_var_name = node.args[0].value

        if is_environ_get and env_var_name:
            self._check_env_var_usage(node.lineno, env_var_name)

        self.generic_visit(node)

    def visit_Subscript(self, node):
        # Check for os.environ["VAR_NAME"] pattern
        if (
            isinstance(node.value, ast.Attribute)
            and node.value.attr == "environ"
            and isinstance(node.value.value, ast.Name)
            and node.value.value.id == "os"
            and isinstance(node.slice, ast.Constant)
        ):
            env_var_name = node.slice.value
            self._check_env_var_usage(node.lineno, env_var_name)

        self.generic_visit(node)

    def _check_env_var_usage(self, line_num: int, env_var_name: str):
        """Check if this environment variable usage is allowed."""
        # Always allow system vars
        if env_var_name in SYSTEM_VARS:
            return

        # Always allow registered vars in allowed files
        if env_var_name in ALL_ENV_VARS or env_var_name in BOOTSTRAP_VARS:
            if self._is_file_allowed():
                return

        # Check if file is allowed to use os.environ
        if not self._is_file_allowed():
            # For registered LDR vars, provide specific guidance
            if env_var_name in ALL_ENV_VARS or env_var_name in BOOTSTRAP_VARS:
                # Find the setting key for this env var
                setting_key = self._find_setting_key(env_var_name)
                if setting_key:
                    self.errors.append(
                        (
                            line_num,
                            f"Environment variable '{env_var_name}' should be accessed via "
                            f"SettingsManager.get_setting('{setting_key}')",
                        )
                    )
                else:
                    self.errors.append(
                        (
                            line_num,
                            f"Registered environment variable '{env_var_name}' should be accessed through SettingsManager",
                        )
                    )
            # For unregistered LDR vars, flag as potential issue
            elif env_var_name.startswith("LDR_"):
                self.errors.append(
                    (
                        line_num,
                        f"Unregistered LDR environment variable '{env_var_name}' - should this be added to settings?",
                    )
                )
            # For other vars, suggest evaluation
            else:
                self.errors.append(
                    (
                        line_num,
                        f"Direct access to environment variable '{env_var_name}' - consider if this should use SettingsManager",
                    )
                )

    def _is_file_allowed(self) -> bool:
        """Check if this file is allowed to use os.environ directly."""
        for pattern in ALLOWED_PATTERNS:
            if pattern in self.filename:
                return True
        return False

    def _find_setting_key(self, env_var_name: str) -> str:
        """Find the setting key for a given environment variable."""
        try:
            # Check all registered settings
            for setting_key in env_registry.list_all_settings():
                if env_registry.get_env_var(setting_key) == env_var_name:
                    return setting_key
        except Exception:
            pass
        return ""


def check_file(filepath: Path) -> List[Tuple[int, str]]:
    """Check a single Python file for environment variable issues."""
    try:
        content = filepath.read_text(encoding="utf-8")
    except Exception as e:
        print(f"Error reading {filepath}: {e}")
        return []

    try:
        tree = ast.parse(content, str(filepath))
        checker = EnvVarChecker(str(filepath))
        checker.visit(tree)
        return checker.errors
    except SyntaxError:
        # Skip files with syntax errors
        return []
    except Exception as e:
        print(f"Error parsing {filepath}: {e}")
        return []


class TestEnvironmentVariableUsage:
    """Test that environment variables are accessed correctly in the codebase."""

    def all_python_files(self):
        """Get all Python files in the repository."""
        repo_root = Path(__file__).parent.parent.parent.parent
        python_files = []

        for py_file in repo_root.rglob("*.py"):
            # Skip virtual environments and build directories
            if any(
                part in py_file.parts
                for part in [
                    ".venv",
                    "venv",
                    "build",
                    "dist",
                    "__pycache__",
                    ".git",
                ]
            ):
                continue
            python_files.append(py_file)

        return python_files

    def test_env_vars_loaded(self):
        """Test that we successfully loaded env var definitions."""
        assert len(ALL_ENV_VARS) > 0, (
            "No environment variables loaded from registry"
        )
        assert len(BOOTSTRAP_VARS) > 0, "No bootstrap variables loaded"
        print(f"✓ Loaded {len(ALL_ENV_VARS)} registered environment variables")
        print(f"✓ Loaded {len(BOOTSTRAP_VARS)} bootstrap environment variables")

    def test_no_invalid_env_var_access(self, all_python_files=None):
        """Test that no files have invalid direct environment variable access."""
        # If not provided (when running without pytest), get the files
        if all_python_files is None:
            all_python_files = self.all_python_files()

        all_errors = []
        files_checked = 0

        for py_file in all_python_files:
            files_checked += 1
            errors = check_file(py_file)
            if errors:
                all_errors.append((py_file, errors))

        print(f"\n📊 Checked {files_checked} Python files")

        if all_errors:
            # Format error message
            repo_root = Path(__file__).parent.parent.parent.parent
            error_msg = [
                f"\n❌ Found environment variable issues in {len(all_errors)} files:\n"
            ]

            for filepath, errors in all_errors:
                rel_path = filepath.relative_to(repo_root)
                error_msg.append(f"\n{rel_path}:")
                for line_num, error in errors:
                    error_msg.append(f"  Line {line_num}: {error}")

            error_msg.append("\n📚 Guidelines:")
            error_msg.append(
                "1. Use SettingsManager.get_setting() for application settings"
            )
            error_msg.append("2. Direct os.environ access is only allowed in:")
            error_msg.append("   - Configuration/settings modules")
            error_msg.append("   - Test files")
            error_msg.append("   - Database initialization modules")
            error_msg.append(
                "3. All LDR_ environment variables should be registered in settings"
            )

            error_message = "\n".join(error_msg)
            # Use pytest.fail if available, otherwise raise AssertionError
            try:
                import pytest

                pytest.fail(error_message)
            except ImportError:
                raise AssertionError(error_message)
        else:
            print("✅ All environment variable usage is compliant!")


if __name__ == "__main__":
    # Allow running as a standalone script (for CI without full pytest deps)
    import sys

    if pytest:
        # Run with pytest if available
        sys.exit(pytest.main([__file__, "-v"]))
    else:
        # Run without pytest
        print("Running validation without pytest...")
        test = TestEnvironmentVariableUsage()

        # Load env vars
        test.test_env_vars_loaded()

        # Run validation
        try:
            test.test_no_invalid_env_var_access()
            sys.exit(0)
        except AssertionError as e:
            print(str(e))
            sys.exit(1)