AAP-53946 [devel-only] Forever ensure that ansible_base.lib.* is importable (ansible#844)

Makes static content in ansible_base.lib importable in absence of Django
Fixes ansible#443

Additionally adds a check so that we keep this importable.

Author: AlanCoding

.github/workflows/linting.yml

@@ -22,6 +22,8 @@ jobs:
             command: check_black
           - name: api-isort
             command: check_isort
+          - name: pure-python-imports
+            command: check_pure_python_imports
     steps:
       - name: Install make
         run: sudo apt install make
@@ -38,5 +40,18 @@ jobs:
       - name: Install requirments
         run: pip3.11 install -r requirements/requirements_dev.txt
 
+      - name: Install test dependencies for pure-python-imports
+        if: matrix.tests.name == 'pure-python-imports'
+        run: |
+          pip3.11 install -r requirements/requirements.in
+          pip3.11 install -r requirements/requirements_testing.txt
+          pip3.11 install -r requirements/requirements_channels.in
+          pip3.11 install -r requirements/requirements_redis_client.in
+
+      - name: Run pure-python-imports check
+        if: matrix.tests.name == 'pure-python-imports'
+        run: python tools/scripts/check_pure_python_imports.py
+
       - name: Run check ${{ matrix.tests.name }}
+        if: matrix.tests.name != 'pure-python-imports'
         run: make ${{ matrix.tests.command }}

Makefile

@@ -49,6 +49,7 @@ check_flake8:
 check_isort:
 	tox -e isort -- --check $(CHECK_SYNTAX_FILES)
 
+
 ## Starts a postgres container in the background if one is not running
 # Options:
 #  -d, --detatch: run the container in background

ansible_base/jwt_consumer/common/util.py

@@ -7,7 +7,6 @@
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import padding
 
-from ansible_base.jwt_consumer.common.cert import JWTCert, JWTCertException
 from ansible_base.lib.utils.settings import get_setting
 
 logger = logging.getLogger('ansible_base.jwt_consumer.common.util')
@@ -32,6 +31,8 @@ def generate_x_trusted_proxy_header(key: str) -> str:
 
 
 def validate_x_trusted_proxy_header(header_value: str, ignore_cache=False) -> bool:
+    from ansible_base.jwt_consumer.common.cert import JWTCert, JWTCertException
+
     try:
         cert = JWTCert()
         cert.get_decryption_key(ignore_cache=ignore_cache)

ansible_base/lib/utils/create_system_user.py

@@ -1,20 +1,22 @@
-import logging
-from typing import Optional, Tuple, Type
+from __future__ import annotations
 
-from django.core.exceptions import ImproperlyConfigured
-from django.db import models
-from django.utils.translation import gettext as _
+import logging
+from typing import TYPE_CHECKING, Optional, Tuple, Type
 
 from ansible_base.lib.utils.settings import get_setting
 
+if TYPE_CHECKING:
+    from django.db import models
+
 logger = logging.getLogger('ansible_base.lib.utils.create_system_user')
 
 """
 These functions are in its own file because it is loaded during migrations so it has no access to models.
 """
 
 
-def create_system_user(user_model: Type[models.Model]) -> models.Model:  # Note: We can't load models here so we can typecast to anything better than Model
+def create_system_user(user_model: Type[models.Model]) -> models.Model:
+    # Note: We can't load models here so we can typecast to anything better than Model
     from ansible_base.lib.abstract_models.user import AbstractDABUser
 
     #
@@ -63,4 +65,8 @@ def get_system_username() -> Tuple[Optional[str], str]:
         return str(value), setting_name
 
     logger.error(f"Expected get_setting to return a string for {setting_name}, got {type(value)}")
+
+    from django.core.exceptions import ImproperlyConfigured
+    from django.utils.translation import gettext as _
+
     raise ImproperlyConfigured(_("Setting %(setting_name)s needs to be a string not a %(type)s") % {'setting_name': setting_name, 'type': type(value)})

ansible_base/lib/utils/requests.py

@@ -1,12 +1,13 @@
-import logging
-from typing import Optional
+from __future__ import annotations
 
-from crum import get_current_request
-from django.http import HttpRequest
+import logging
+from typing import TYPE_CHECKING, Optional
 
-from ansible_base.jwt_consumer.common.util import validate_x_trusted_proxy_header
 from ansible_base.lib.utils.settings import get_setting
 
+if TYPE_CHECKING:
+    from django.http import HttpRequest
+
 logger = logging.getLogger('ansible_base.lib.uitls.requests')
 
 
@@ -43,6 +44,8 @@ def get_remote_hosts(request: HttpRequest, get_first_only: bool = False) -> list
     # If we are connected to from a trusted proxy then we can add some additional headers
     try:
         if 'HTTP_X_TRUSTED_PROXY' in request.META:
+            from ansible_base.jwt_consumer.common.util import validate_x_trusted_proxy_header
+
             if validate_x_trusted_proxy_header(request.META['HTTP_X_TRUSTED_PROXY']):
                 # The last entry in x-forwarded-for from envoy can be trusted implicitly
                 # https://www.envoyproxy.io/docs/envoy/latest/configuration/http/http_conn_man/headers#x-forwarded-for
@@ -68,10 +71,14 @@ def get_remote_hosts(request: HttpRequest, get_first_only: bool = False) -> list
 def is_proxied_request(request: Optional[HttpRequest] = None) -> bool:
     "Return true if request claims to be from a proxy and the header validates as such."
     if request is None:
+        from crum import get_current_request
+
         request = get_current_request()
     if request is None:
         # e.g. being called by CLI or something
         return False
     if x_trusted_proxy := request.META.get("HTTP_X_TRUSTED_PROXY"):
+        from ansible_base.jwt_consumer.common.util import validate_x_trusted_proxy_header
+
         return validate_x_trusted_proxy_header(x_trusted_proxy)
     return False

test_app/tests/jwt_consumer/common/test_util.py

@@ -13,7 +13,7 @@ def test_validate_trusted_proxy_header_bad_cached_key_but_correct_setting(self,
             {"key": rsa_keypair.public, "cached": False},
         ]
         with override_settings(ANSIBLE_BASE_JWT_KEY=rsa_keypair.public):
-            with mock.patch("ansible_base.jwt_consumer.common.util.JWTCert.get_decryption_key", create_mock_method(field_dicts)):
+            with mock.patch("ansible_base.jwt_consumer.common.cert.JWTCert.get_decryption_key", create_mock_method(field_dicts)):
                 assert validate_x_trusted_proxy_header(generate_x_trusted_proxy_header(rsa_keypair.private))
 
     def test_validate_trusted_proxy_header_no_key(self, caplog):

test_app/tests/lib/utils/test_requests.py

@@ -103,7 +103,7 @@ def test_get_remote_hosts_alternate_headers_behind_trusted_proxy(rsa_keypair):
             assert remote_hosts == ['5.6.7.8', '9.10.11.12', '5.6.7.8', 'a.b.c.d']
 
 
-@mock.patch("ansible_base.lib.utils.requests.validate_x_trusted_proxy_header", side_effect=Exception)
+@mock.patch("ansible_base.jwt_consumer.common.util.validate_x_trusted_proxy_header", side_effect=Exception)
 def test_get_remote_hosts_validate_trusted_proxy_header_exception(mock_validate: mock.MagicMock, rsa_keypair, caplog):
     with override_settings(ANSIBLE_BASE_JWT_KEY=rsa_keypair.public):
         headers = {

tools/scripts/check_pure_python_imports.py

@@ -0,0 +1,147 @@
+#!/usr/bin/env python3
+# Generated by Claude Code (Sonnet 4)
+
+"""
+Script to test all modules in ansible_base/lib for pure Python compatibility.
+This script ensures that modules in the lib folder can be imported without
+requiring Django apps to be initialized.
+"""
+
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+
+def find_python_modules(lib_path):
+    """Find all Python modules in the lib directory."""
+    modules = []
+    lib_path = Path(lib_path)
+
+    for py_file in lib_path.rglob("*.py"):
+        if py_file.name == "__init__.py":
+            # Convert path to module name (including ansible_base prefix)
+            relative_path = py_file.parent.relative_to(lib_path.parent.parent)
+            module_name = str(relative_path).replace(os.sep, ".")
+            modules.append(module_name)
+        else:
+            # Convert path to module name (including ansible_base prefix)
+            relative_path = py_file.relative_to(lib_path.parent.parent)
+            module_name = str(relative_path)[:-3].replace(os.sep, ".")  # Remove .py extension
+            modules.append(module_name)
+
+    return sorted(modules)
+
+
+def test_pure_python_import(module_name, base_path):
+    """Test if a module can be imported with pure Python (no Django setup)."""
+    test_script = f'''
+import sys
+sys.path.insert(0, "{base_path}")
+
+try:
+    import {module_name}
+    print("SUCCESS")
+except Exception as e:
+    print(f"ERROR: {{e}}")
+'''
+
+    # Run in a clean environment without Django setup
+    env = os.environ.copy()
+    if 'DJANGO_SETTINGS_MODULE' in env:
+        del env['DJANGO_SETTINGS_MODULE']
+
+    try:
+        result = subprocess.run(['python', '-c', test_script], capture_output=True, text=True, env=env, timeout=30)
+
+        if result.returncode == 0 and "SUCCESS" in result.stdout:
+            return True, None
+        else:
+            error_msg = result.stderr.strip() or result.stdout.strip()
+            return False, error_msg
+    except subprocess.TimeoutExpired:
+        return False, "Import test timed out"
+    except Exception as e:
+        return False, f"Failed to run test: {e}"
+
+
+def main():
+    """Main function to test all modules in ansible_base/lib."""
+    script_dir = Path(__file__).parent.parent.parent  # Go up two levels since we're in tools/scripts/
+    lib_path = script_dir / "ansible_base" / "lib"
+
+    if not lib_path.exists():
+        print(f"ERROR: {lib_path} does not exist")
+        sys.exit(1)
+
+    print("Testing pure Python import compatibility for ansible_base/lib modules...")
+    print("=" * 70)
+
+    modules = find_python_modules(lib_path)
+
+    # Modules that are allowed to fail because they inherently require Django
+    allowed_failures = {
+        # Abstract models inherently require Django to be initialized
+        'ansible_base.lib.abstract_models',
+        'ansible_base.lib.abstract_models.common',
+        'ansible_base.lib.abstract_models.immutable',
+        'ansible_base.lib.abstract_models.organization',
+        'ansible_base.lib.abstract_models.team',
+        'ansible_base.lib.abstract_models.user',
+        # View classes require Django REST framework
+        'ansible_base.lib.utils.views.ansible_base',
+        'ansible_base.lib.utils.views.django_app_api',
+        'ansible_base.lib.utils.views.permissions',
+        'ansible_base.lib.utils.views.urls',
+        # Router classes require Django REST framework
+        'ansible_base.lib.routers',
+        'ansible_base.lib.routers.association_resource_router',
+        # These modules require Django settings to be configured at import time
+        'ansible_base.lib.backends.prefixed_user_auth',
+        'ansible_base.lib.dynamic_config.dynamic_urls',
+        'ansible_base.lib.serializers.common',
+        'ansible_base.lib.testing.fixtures',
+        'ansible_base.lib.testing.util',
+        'ansible_base.lib.utils.auth',
+    }
+
+    failed_modules = []
+    passed_modules = []
+    expected_failures = []
+
+    for module_name in modules:
+        print(f"Testing {module_name}... ", end="", flush=True)
+        success, error = test_pure_python_import(module_name, str(script_dir))
+
+        if success:
+            print("✓ PASS")
+            passed_modules.append(module_name)
+        else:
+            if module_name in allowed_failures:
+                print("✗ EXPECTED FAIL")
+                expected_failures.append((module_name, error))
+            else:
+                print("✗ UNEXPECTED FAIL")
+                print(f"  Error: {error}")
+                failed_modules.append((module_name, error))
+
+    print("\n" + "=" * 70)
+    print(f"Results: {len(passed_modules)} passed, {len(failed_modules)} unexpected failures, {len(expected_failures)} expected failures")
+
+    if failed_modules:
+        print("\nUnexpected failures:")
+        for module_name, error in failed_modules:
+            print(f"  - {module_name}: {error}")
+
+        print("\nThese modules should be importable with pure Python.")
+        print("Consider moving Django-specific imports inline within functions.")
+        sys.exit(1)
+    else:
+        print("\nAll importable modules passed! ✓")
+        if expected_failures:
+            print(f"({len(expected_failures)} modules have expected failures due to Django dependencies)")
+        sys.exit(0)
+
+
+if __name__ == "__main__":
+    main()