Forward port external query files from stable-2.6 (#16312)

* Revert "AAP-58452 Add version fallback for external query files (#16309)"

This reverts commit 0f2692b504.

* AAP-58441: Add runtime integration for external query collection (#7208)

Extend build_private_data_files() to copy vendor collections from
/var/lib/awx/vendor_collections/ to the job's private_data_dir,
making external query files available to the indirect node counting
callback plugin in execution environments.

Changes:
- Copy vendor_collections to private_data_dir during job preparation
- Add vendor_collections path to ANSIBLE_COLLECTIONS_PATH in build_env()
- Gracefully handle missing source directory with warning log
- Feature gated by FEATURE_INDIRECT_NODE_COUNTING_ENABLED flag

This enables external query file discovery for indirect node counting
across all deployment types (RPM, Podman, OpenShift, Kubernetes) using
the existing private_data_dir mechanism.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

* [stable-2.6] AAP-58451: Add callback plugin discovery for external query files (#7223)

* AAP-58451: Add callback plugin discovery for external query files

Extend the indirect_instance_count callback plugin to discover and load
external query files from the bundled redhat.indirect_accounting collection
when embedded queries are not present in the target collection.

Changes:
- Add external query discovery with precedence (embedded queries first)
- External query path: redhat.indirect_accounting/extensions/audit/
  external_queries/{namespace}.{name}.{version}.yml
- Use self._display.v() for external query messages (visible with -v)
- Use self._display.vv() for embedded query messages (visible with -vv)
- Fix: Change .exists() to .is_file() per Traversable ABC
- Handle missing external query collection gracefully (ModuleNotFoundError)

Note: This implements exact version match only. Version fallback logic
is covered in AAP-58452.

* fix CI error when using Traversable.is_file

* Add minimal implementation for AAP-58451

* Fix formatting

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

* AAP-58452 Add version fallback for external query files (#7254)

* AAP-58456 unit test suite for external query handling (#7283)

* Add unit tests for external query handling

* Refactor unit tests for external query handling

* Refactor indirect node counting callback code to improve testing code

* Refactor unit tests for external query handling for improved callback code

* Fix test for majore version boundary check

* Fix weaknesses in some unit tests

* Make callback plugin module self contained, independent from awx

* AAP-58470 integration tests (core) for external queries (#7278)

* Add collection for testing external queries

* Add query files for testing external query file runtime integration

* Add live tests for external query file runtime integration

* Remove redundant wait for events and refactor test data folders

* Fix unit tests: mock flag_enabled to avoid DB access

The AAP-58441 cherry-pick added a flag_enabled() call in
BaseTask.build_private_data_files(), which is called by all task types.
Tests for RunInventoryUpdate and RunJob credentials now hit this code
path and need the flag mocked to avoid database access in unit tests.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: attempt exact query file match before Version parsing (#7345)

The exact-version filename check does not require PEP440 parsing, but
Version() was called first, causing early return on non-PEP440 version
strings even when an exact file exists on disk. Move the exact file
check before Version parsing so fallback logic only parses when needed.

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

* Do no longer mutate global sys.modules (#7337)

* [stable-2.6] AAP-58452 fix: Add queries_dir guard (#7338)

* Add queries_dir guard

* fix: update unit tests to mock _get_query_file_dir instead of files

The TestVersionFallback tests mocked `files()` with chainable path
mocks, but `find_external_query_with_fallback` now uses
`_get_query_file_dir()` which returns the queries directory directly.
Mock the helper instead for simpler, correct tests.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

* fix: remove unused EXTERNAL_QUERY_PATH constant (#7336)

The constant was defined but never referenced — the path is constructed
inline via Traversable's `/` operator which requires individual segments,
not a slash-separated string.

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

* fix: restore original feature flag state in test fixture (#7347)

The enable_indirect_host_counting fixture unconditionally disabled the
FEATURE_INDIRECT_NODE_COUNTING_ENABLED flag on teardown, even when it
was already enabled before the test (as is the case in development via
development_defaults.py). This caused test_indirect_host_counting to
fail when run after the external query tests, because the callback
plugin was no longer enabled.

Save and restore the original flag state instead.

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Dirk Julich <djulich@redhat.com>
Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

awx/main/tasks/jobs.py

@@ -419,6 +419,19 @@ class BaseTask(object):
                     private_data_files['credentials'][credential] = self.write_private_data_file(private_data_dir, None, data, sub_dir='env')
             for credential, data in private_data.get('certificates', {}).items():
                 self.write_private_data_file(private_data_dir, 'ssh_key_data-cert.pub', data, sub_dir=os.path.join('artifacts', str(self.instance.id)))
+
+        # Copy vendor collections to private_data_dir for indirect node counting
+        # This makes external query files available to the callback plugin in EEs
+        if flag_enabled("FEATURE_INDIRECT_NODE_COUNTING_ENABLED"):
+            vendor_src = '/var/lib/awx/vendor_collections'
+            vendor_dest = os.path.join(private_data_dir, 'vendor_collections')
+            if os.path.exists(vendor_src):
+                try:
+                    shutil.copytree(vendor_src, vendor_dest)
+                    logger.debug(f"Copied vendor collections from {vendor_src} to {vendor_dest}")
+                except Exception as e:
+                    logger.warning(f"Failed to copy vendor collections: {e}")
+
         return private_data_files, ssh_key_data
 
     def build_passwords(self, instance, runtime_passwords):
@@ -1143,6 +1156,11 @@ class RunJob(SourceControlMixin, BaseTask):
             if 'callbacks_enabled' in config_values:
                 env['ANSIBLE_CALLBACKS_ENABLED'] += ':' + config_values['callbacks_enabled']
 
+            # Add vendor collections path for external query file discovery
+            vendor_collections_path = os.path.join(CONTAINER_ROOT, 'vendor_collections')
+            env['ANSIBLE_COLLECTIONS_PATH'] = f"{vendor_collections_path}:{env['ANSIBLE_COLLECTIONS_PATH']}"
+            logger.debug(f"ANSIBLE_COLLECTIONS_PATH updated for vendor collections: {env['ANSIBLE_COLLECTIONS_PATH']}")
+
         return env
 
     def build_args(self, job, private_data_dir, passwords):

awx/main/tests/data/collections/host_query_external_v1_0_0/galaxy.yml

@@ -0,0 +1,19 @@
+---
+authors:
+  - AWX Project Contributors <awx-project@googlegroups.com>
+dependencies: {}
+description: External query testing collection. No embedded query file. Not for use in production.
+documentation: https://github.com/ansible/awx
+homepage: https://github.com/ansible/awx
+issues: https://github.com/ansible/awx
+license:
+  - GPL-3.0-or-later
+name: external
+namespace: demo
+readme: README.md
+repository: https://github.com/ansible/awx
+tags:
+  - demo
+  - testing
+  - external_query
+version: 1.0.0

awx/main/tests/data/collections/host_query_external_v1_0_0/plugins/modules/example.py

@@ -0,0 +1,78 @@
+#!/usr/bin/python
+
+# Same licensing as AWX
+from __future__ import absolute_import, division, print_function
+
+__metaclass__ = type
+
+DOCUMENTATION = r'''
+---
+module: example
+
+short_description: Module for specific live tests
+
+version_added: "2.0.0"
+
+description: This module is part of a test collection in local source. Used for external query testing.
+
+options:
+    host_name:
+        description: Name to return as the host name.
+        required: false
+        type: str
+
+author:
+    - AWX Live Tests
+'''
+
+EXAMPLES = r'''
+- name: Test with defaults
+  demo.external.example:
+
+- name: Test with custom host name
+  demo.external.example:
+    host_name: foo_host
+'''
+
+RETURN = r'''
+direct_host_name:
+    description: The name of the host, this will be collected with the feature.
+    type: str
+    returned: always
+    sample: 'foo_host'
+'''
+
+from ansible.module_utils.basic import AnsibleModule
+
+
+def run_module():
+    module_args = dict(
+        host_name=dict(type='str', required=False, default='foo_host_default'),
+    )
+
+    result = dict(
+        changed=False,
+        other_data='sample_string',
+    )
+
+    module = AnsibleModule(argument_spec=module_args, supports_check_mode=True)
+
+    if module.check_mode:
+        module.exit_json(**result)
+
+    result['direct_host_name'] = module.params['host_name']
+    result['nested_host_name'] = {'host_name': module.params['host_name']}
+    result['name'] = 'vm-foo'
+
+    # non-cononical facts
+    result['device_type'] = 'Fake Host'
+
+    module.exit_json(**result)
+
+
+def main():
+    run_module()
+
+
+if __name__ == '__main__':
+    main()

awx/main/tests/data/collections/host_query_external_v1_5_0/galaxy.yml

@@ -0,0 +1,19 @@
+---
+authors:
+  - AWX Project Contributors <awx-project@googlegroups.com>
+dependencies: {}
+description: External query testing collection v1.5.0. No embedded query file. Not for use in production.
+documentation: https://github.com/ansible/awx
+homepage: https://github.com/ansible/awx
+issues: https://github.com/ansible/awx
+license:
+  - GPL-3.0-or-later
+name: external
+namespace: demo
+readme: README.md
+repository: https://github.com/ansible/awx
+tags:
+  - demo
+  - testing
+  - external_query
+version: 1.5.0

awx/main/tests/data/collections/host_query_external_v1_5_0/plugins/modules/example.py

@@ -0,0 +1,78 @@
+#!/usr/bin/python
+
+# Same licensing as AWX
+from __future__ import absolute_import, division, print_function
+
+__metaclass__ = type
+
+DOCUMENTATION = r'''
+---
+module: example
+
+short_description: Module for specific live tests
+
+version_added: "2.0.0"
+
+description: This module is part of a test collection in local source. Used for external query testing.
+
+options:
+    host_name:
+        description: Name to return as the host name.
+        required: false
+        type: str
+
+author:
+    - AWX Live Tests
+'''
+
+EXAMPLES = r'''
+- name: Test with defaults
+  demo.external.example:
+
+- name: Test with custom host name
+  demo.external.example:
+    host_name: foo_host
+'''
+
+RETURN = r'''
+direct_host_name:
+    description: The name of the host, this will be collected with the feature.
+    type: str
+    returned: always
+    sample: 'foo_host'
+'''
+
+from ansible.module_utils.basic import AnsibleModule
+
+
+def run_module():
+    module_args = dict(
+        host_name=dict(type='str', required=False, default='foo_host_default'),
+    )
+
+    result = dict(
+        changed=False,
+        other_data='sample_string',
+    )
+
+    module = AnsibleModule(argument_spec=module_args, supports_check_mode=True)
+
+    if module.check_mode:
+        module.exit_json(**result)
+
+    result['direct_host_name'] = module.params['host_name']
+    result['nested_host_name'] = {'host_name': module.params['host_name']}
+    result['name'] = 'vm-foo'
+
+    # non-cononical facts
+    result['device_type'] = 'Fake Host'
+
+    module.exit_json(**result)
+
+
+def main():
+    run_module()
+
+
+if __name__ == '__main__':
+    main()

awx/main/tests/data/collections/host_query_external_v3_0_0/galaxy.yml

@@ -0,0 +1,19 @@
+---
+authors:
+  - AWX Project Contributors <awx-project@googlegroups.com>
+dependencies: {}
+description: External query testing collection v3.0.0. No embedded query file. Not for use in production.
+documentation: https://github.com/ansible/awx
+homepage: https://github.com/ansible/awx
+issues: https://github.com/ansible/awx
+license:
+  - GPL-3.0-or-later
+name: external
+namespace: demo
+readme: README.md
+repository: https://github.com/ansible/awx
+tags:
+  - demo
+  - testing
+  - external_query
+version: 3.0.0

awx/main/tests/data/collections/host_query_external_v3_0_0/plugins/modules/example.py

@@ -0,0 +1,78 @@
+#!/usr/bin/python
+
+# Same licensing as AWX
+from __future__ import absolute_import, division, print_function
+
+__metaclass__ = type
+
+DOCUMENTATION = r'''
+---
+module: example
+
+short_description: Module for specific live tests
+
+version_added: "2.0.0"
+
+description: This module is part of a test collection in local source. Used for external query testing.
+
+options:
+    host_name:
+        description: Name to return as the host name.
+        required: false
+        type: str
+
+author:
+    - AWX Live Tests
+'''
+
+EXAMPLES = r'''
+- name: Test with defaults
+  demo.external.example:
+
+- name: Test with custom host name
+  demo.external.example:
+    host_name: foo_host
+'''
+
+RETURN = r'''
+direct_host_name:
+    description: The name of the host, this will be collected with the feature.
+    type: str
+    returned: always
+    sample: 'foo_host'
+'''
+
+from ansible.module_utils.basic import AnsibleModule
+
+
+def run_module():
+    module_args = dict(
+        host_name=dict(type='str', required=False, default='foo_host_default'),
+    )
+
+    result = dict(
+        changed=False,
+        other_data='sample_string',
+    )
+
+    module = AnsibleModule(argument_spec=module_args, supports_check_mode=True)
+
+    if module.check_mode:
+        module.exit_json(**result)
+
+    result['direct_host_name'] = module.params['host_name']
+    result['nested_host_name'] = {'host_name': module.params['host_name']}
+    result['name'] = 'vm-foo'
+
+    # non-cononical facts
+    result['device_type'] = 'Fake Host'
+
+    module.exit_json(**result)
+
+
+def main():
+    run_module()
+
+
+if __name__ == '__main__':
+    main()

awx/main/tests/data/projects/test_host_query_external_v1_0_0/collections/requirements.yml

@@ -0,0 +1,5 @@
+---
+collections:
+  - name: 'file:///tmp/live_tests/host_query_external_v1_0_0'
+    type: git
+    version: devel

awx/main/tests/data/projects/test_host_query_external_v1_0_0/run_task.yml

@@ -0,0 +1,8 @@
+---
+- hosts: all
+  gather_facts: false
+  connection: local
+  tasks:
+    - demo.external.example:
+      register: result
+    - debug: var=result

awx/main/tests/data/projects/test_host_query_external_v1_5_0/collections/requirements.yml

@@ -0,0 +1,5 @@
+---
+collections:
+  - name: 'file:///tmp/live_tests/host_query_external_v1_5_0'
+    type: git
+    version: devel

awx/main/tests/data/projects/test_host_query_external_v1_5_0/run_task.yml

@@ -0,0 +1,8 @@
+---
+- hosts: all
+  gather_facts: false
+  connection: local
+  tasks:
+    - demo.external.example:
+      register: result
+    - debug: var=result

awx/main/tests/data/projects/test_host_query_external_v3_0_0/collections/requirements.yml

@@ -0,0 +1,5 @@
+---
+collections:
+  - name: 'file:///tmp/live_tests/host_query_external_v3_0_0'
+    type: git
+    version: devel

awx/main/tests/data/projects/test_host_query_external_v3_0_0/run_task.yml

@@ -0,0 +1,8 @@
+---
+- hosts: all
+  gather_facts: false
+  connection: local
+  tasks:
+    - demo.external.example:
+      register: result
+    - debug: var=result

awx/main/tests/live/tests/conftest.py

@@ -25,6 +25,7 @@ logger = logging.getLogger(__name__)
 
 
 PROJ_DATA = os.path.join(os.path.dirname(data.__file__), 'projects')
+COLL_DATA = os.path.join(os.path.dirname(data.__file__), 'collections')
 
 
 def _copy_folders(source_path, dest_path, clear=False):
@@ -56,6 +57,7 @@ def live_tmp_folder():
         shutil.rmtree(path)
     os.mkdir(path)
     _copy_folders(PROJ_DATA, path)
+    _copy_folders(COLL_DATA, path)
     for dirname in os.listdir(path):
         source_dir = os.path.join(path, dirname)
         subprocess.run(GIT_COMMANDS, cwd=source_dir, shell=True)

awx/main/tests/live/tests/test_indirect_counting_external_queries.py

@@ -0,0 +1,347 @@
+"""
+Integration tests for external query file functionality (AAP-58470).
+
+Tests verify the end-to-end external query file workflow for indirect node
+counting using real AWX job execution. A fixture-created vendor collection
+at /var/lib/awx/vendor_collections/ provides external query files, simulating
+what the build-time (AAP-58426) and deployment (AAP-58557) integrations will
+provide once available.
+
+Test data:
+- Collection 'demo.external' at various versions (no embedded query)
+- External query files in mock redhat.indirect_accounting collection
+"""
+
+import os
+import shutil
+import time
+import yaml
+
+import pytest
+from flags.state import enable_flag, disable_flag, flag_enabled
+
+from awx.main.tests.live.tests.conftest import wait_for_events, unified_job_stdout
+from awx.main.tasks.host_indirect import save_indirect_host_entries
+from awx.main.models.indirect_managed_node_audit import IndirectManagedNodeAudit
+from awx.main.models.event_query import EventQuery
+from awx.main.models import Job
+
+# --- Constants ---
+
+EXTERNAL_QUERY_JQ = '{name: .name, canonical_facts: {host_name: .direct_host_name}, facts: {device_type: .device_type}}'
+
+EXTERNAL_QUERY_CONTENT = yaml.dump(
+    {'demo.external.example': {'query': EXTERNAL_QUERY_JQ}},
+    default_flow_style=False,
+)
+
+# For precedence test: different jq (no device_type in facts) so we can detect which query was used
+EXTERNAL_QUERY_FOR_DEMO_QUERY_JQ = '{name: .name, canonical_facts: {host_name: .direct_host_name}, facts: {}}'
+
+EXTERNAL_QUERY_FOR_DEMO_QUERY_CONTENT = yaml.dump(
+    {'demo.query.example': {'query': EXTERNAL_QUERY_FOR_DEMO_QUERY_JQ}},
+    default_flow_style=False,
+)
+
+VENDOR_COLLECTIONS_BASE = '/var/lib/awx/vendor_collections'
+
+
+# --- Fixtures ---
+
+
+@pytest.fixture
+def enable_indirect_host_counting():
+    """Enable FEATURE_INDIRECT_NODE_COUNTING_ENABLED flag for the test.
+
+    Only creates a FlagState DB record if the flag isn't already enabled
+    (e.g. via development_defaults.py), to avoid UniqueViolation errors
+    and to avoid leaking state to other tests.
+    """
+    flag_name = "FEATURE_INDIRECT_NODE_COUNTING_ENABLED"
+    was_enabled = flag_enabled(flag_name)
+    if not was_enabled:
+        enable_flag(flag_name)
+    yield
+    if not was_enabled:
+        disable_flag(flag_name)
+
+
+@pytest.fixture
+def vendor_collections_dir():
+    """Set up mock redhat.indirect_accounting collection at /var/lib/awx/vendor_collections/.
+
+    Creates the collection structure with external query files:
+    - demo.external.1.0.0.yml  (exact match for v1.0.0)
+    - demo.external.1.1.0.yml  (fallback target for v1.5.0)
+    - demo.query.0.0.1.yml     (for precedence test with embedded-query collection)
+    """
+    base = os.path.join(VENDOR_COLLECTIONS_BASE, 'ansible_collections', 'redhat', 'indirect_accounting')
+    queries_path = os.path.join(base, 'extensions', 'audit', 'external_queries')
+    meta_path = os.path.join(base, 'meta')
+
+    os.makedirs(queries_path, exist_ok=True)
+    os.makedirs(meta_path, exist_ok=True)
+
+    # galaxy.yml for valid collection structure
+    with open(os.path.join(base, 'galaxy.yml'), 'w') as f:
+        yaml.dump(
+            {
+                'namespace': 'redhat',
+                'name': 'indirect_accounting',
+                'version': '1.0.0',
+                'description': 'Test fixture for external query integration tests',
+                'authors': ['AWX Tests'],
+                'dependencies': {},
+            },
+            f,
+        )
+
+    # meta/runtime.yml
+    with open(os.path.join(meta_path, 'runtime.yml'), 'w') as f:
+        yaml.dump({'requires_ansible': '>=2.15.0'}, f)
+
+    # External query files for demo.external collection
+    for version in ('1.0.0', '1.1.0'):
+        with open(os.path.join(queries_path, f'demo.external.{version}.yml'), 'w') as f:
+            f.write(EXTERNAL_QUERY_CONTENT)
+
+    # External query file for demo.query collection (precedence test)
+    with open(os.path.join(queries_path, 'demo.query.0.0.1.yml'), 'w') as f:
+        f.write(EXTERNAL_QUERY_FOR_DEMO_QUERY_CONTENT)
+
+    yield base
+
+    # Cleanup
+    shutil.rmtree(VENDOR_COLLECTIONS_BASE, ignore_errors=True)
+
+
+@pytest.fixture(autouse=True)
+def cleanup_test_data():
+    """Clean up EventQuery and IndirectManagedNodeAudit records after each test."""
+    yield
+    EventQuery.objects.filter(fqcn='demo.external').delete()
+    EventQuery.objects.filter(fqcn='demo.query').delete()
+    IndirectManagedNodeAudit.objects.filter(job__name__icontains='external_query').delete()
+
+
+# --- Helpers ---
+
+
+def run_external_query_job(run_job_from_playbook, live_tmp_folder, test_name, project_dir, jt_params=None):
+    """Run a job and return the Job object after waiting for indirect host processing."""
+    scm_url = f'file://{live_tmp_folder}/{project_dir}'
+    run_job_from_playbook(test_name, 'run_task.yml', scm_url=scm_url, jt_params=jt_params)
+
+    job = Job.objects.filter(name__icontains=test_name).order_by('-created').first()
+    assert job is not None, f'Job not found for test {test_name}'
+    wait_for_events(job)
+
+    return job
+
+
+def wait_for_indirect_processing(job, expect_records=True, timeout=5):
+    """Wait for indirect host processing to complete.
+
+    Follows the same pattern as test_indirect_host_counting.py:53-72.
+    """
+    # Ensure indirect host processing runs (wait_for_events already called by caller)
+    job.refresh_from_db()
+    if job.event_queries_processed is False:
+        save_indirect_host_entries.delay(job.id, wait_for_events=False)
+
+    if expect_records:
+        # Poll for audit records to appear
+        for _ in range(20):
+            if IndirectManagedNodeAudit.objects.filter(job=job).exists():
+                break
+            time.sleep(0.25)
+        else:
+            raise RuntimeError(f'No IndirectManagedNodeAudit records populated for job_id={job.id}')
+    else:
+        # For negative tests, wait a reasonable time to confirm no records appear
+        time.sleep(timeout)
+        job.refresh_from_db()
+
+
+# --- AC8.1: External query populates IndirectManagedNodeAudit correctly ---
+
+
+def test_external_query_populates_audit_table(live_tmp_folder, run_job_from_playbook, enable_indirect_host_counting, vendor_collections_dir):
+    """AC8.1: Job using demo.external.example with external query file populates
+    IndirectManagedNodeAudit table correctly.
+
+    Uses demo.external v1.0.0 with exact-match external query file demo.external.1.0.0.yml.
+    """
+    job = run_external_query_job(
+        run_job_from_playbook,
+        live_tmp_folder,
+        'external_query_ac8_1',
+        'test_host_query_external_v1_0_0',
+    )
+    wait_for_indirect_processing(job, expect_records=True)
+
+    # Verify installed_collections captured demo.external
+    assert 'demo.external' in job.installed_collections
+    assert 'host_query' in job.installed_collections['demo.external']
+
+    # Verify IndirectManagedNodeAudit records
+    assert IndirectManagedNodeAudit.objects.filter(job=job).count() == 1
+    host_audit = IndirectManagedNodeAudit.objects.filter(job=job).first()
+
+    assert host_audit.canonical_facts == {'host_name': 'foo_host_default'}
+    assert host_audit.facts == {'device_type': 'Fake Host'}
+    assert host_audit.name == 'vm-foo'
+    assert host_audit.organization == job.organization
+    assert 'demo.external.example' in host_audit.events
+
+
+# --- AC8.2: Precedence - embedded query takes precedence over external ---
+
+
+def test_embedded_query_takes_precedence(live_tmp_folder, run_job_from_playbook, enable_indirect_host_counting, vendor_collections_dir):
+    """AC8.2: When collection has both embedded and external query files,
+    the embedded query takes precedence.
+
+    Uses demo.query v0.0.1 which HAS an embedded query (extensions/audit/event_query.yml).
+    An external query (demo.query.0.0.1.yml) also exists but uses a different jq expression
+    (no device_type in facts). By checking the audit record's facts, we verify which query was used.
+    """
+    # Run with demo.query collection (has embedded query)
+    job = run_external_query_job(
+        run_job_from_playbook,
+        live_tmp_folder,
+        'external_query_ac8_2',
+        'test_host_query',
+    )
+    wait_for_indirect_processing(job, expect_records=True)
+
+    # Verify the embedded query was used (includes device_type in facts)
+    host_audit = IndirectManagedNodeAudit.objects.filter(job=job).first()
+    assert host_audit.facts == {'device_type': 'Fake Host'}, (
+        'Expected embedded query output (with device_type). ' 'If facts is {}, the external query was incorrectly used instead.'
+    )
+
+
+# --- AC8.3: Version fallback to compatible version ---
+
+
+def test_fallback_to_compatible_version(live_tmp_folder, run_job_from_playbook, enable_indirect_host_counting, vendor_collections_dir):
+    """AC8.3: Job using collection version with no exact query file falls back
+    correctly to compatible version.
+
+    Uses demo.external v1.5.0. No demo.external.1.5.0.yml exists, but
+    demo.external.1.1.0.yml is available (same major version, highest <= 1.5.0).
+    The fallback should find and use the 1.1.0 query.
+    """
+    job = run_external_query_job(
+        run_job_from_playbook,
+        live_tmp_folder,
+        'external_query_ac8_3',
+        'test_host_query_external_v1_5_0',
+    )
+    wait_for_indirect_processing(job, expect_records=True)
+
+    # Verify installed_collections captured demo.external at v1.5.0
+    assert 'demo.external' in job.installed_collections
+    assert job.installed_collections['demo.external']['version'] == '1.5.0'
+
+    # Verify IndirectManagedNodeAudit records were created via fallback
+    assert IndirectManagedNodeAudit.objects.filter(job=job).count() == 1
+    host_audit = IndirectManagedNodeAudit.objects.filter(job=job).first()
+
+    assert host_audit.canonical_facts == {'host_name': 'foo_host_default'}
+    assert host_audit.facts == {'device_type': 'Fake Host'}
+    assert host_audit.name == 'vm-foo'
+
+
+# --- AC8.4: Fallback queries don't overcount ---
+
+
+def test_fallback_does_not_overcount(live_tmp_folder, run_job_from_playbook, enable_indirect_host_counting, vendor_collections_dir):
+    """AC8.4: Fallback queries don't count MORE nodes than exact-version queries.
+
+    Runs two jobs:
+    1. Exact match scenario (demo.external v1.0.0 -> demo.external.1.0.0.yml)
+    2. Fallback scenario (demo.external v1.5.0 -> falls back to demo.external.1.1.0.yml)
+
+    Verifies that fallback record count <= exact record count.
+    """
+    # Run exact-match job
+    exact_job = run_external_query_job(
+        run_job_from_playbook,
+        live_tmp_folder,
+        'external_query_ac8_4_exact',
+        'test_host_query_external_v1_0_0',
+    )
+    wait_for_indirect_processing(exact_job, expect_records=True)
+    exact_count = IndirectManagedNodeAudit.objects.filter(job=exact_job).count()
+
+    # Run fallback job
+    fallback_job = run_external_query_job(
+        run_job_from_playbook,
+        live_tmp_folder,
+        'external_query_ac8_4_fallback',
+        'test_host_query_external_v1_5_0',
+    )
+    wait_for_indirect_processing(fallback_job, expect_records=True)
+    fallback_count = IndirectManagedNodeAudit.objects.filter(job=fallback_job).count()
+
+    # Critical safety check: fallback must never count MORE than exact
+    assert fallback_count <= exact_count, (
+        f'Overcounting detected! Fallback produced {fallback_count} records ' f'but exact match produced only {exact_count} records.'
+    )
+
+    # Both use the same jq expression and same module, so counts should be equal
+    assert exact_count == fallback_count
+
+
+# --- AC8.5: Warning logs contain correct version information ---
+
+
+def test_fallback_log_contains_version_info(live_tmp_folder, run_job_from_playbook, enable_indirect_host_counting, vendor_collections_dir):
+    """AC8.5: Warning logs contain correct version information when fallback is used.
+
+    Runs a job with verbosity=1 so callback plugin verbose output is captured.
+    Verifies the log contains the installed version (1.5.0), fallback version (1.1.0),
+    and collection FQCN (demo.external).
+    """
+    job = run_external_query_job(
+        run_job_from_playbook,
+        live_tmp_folder,
+        'external_query_ac8_5',
+        'test_host_query_external_v1_5_0',
+        jt_params={'verbosity': 1},
+    )
+    wait_for_indirect_processing(job, expect_records=True)
+
+    # Get job stdout to check for fallback log message
+    stdout = unified_job_stdout(job)
+
+    # The callback plugin emits: "Using external query {version_used} for {fqcn} v{ver}."
+    assert '1.1.0' in stdout, f'Fallback version 1.1.0 not found in job stdout. stdout:\n{stdout}'
+    assert 'demo.external' in stdout, f'Collection FQCN demo.external not found in job stdout. stdout:\n{stdout}'
+    assert '1.5.0' in stdout, f'Installed version 1.5.0 not found in job stdout. stdout:\n{stdout}'
+
+
+# --- AC8.6: No counting when no compatible fallback exists ---
+
+
+def test_no_counting_without_compatible_fallback(live_tmp_folder, run_job_from_playbook, enable_indirect_host_counting, vendor_collections_dir):
+    """AC8.6: No counting occurs when no compatible fallback exists.
+
+    Uses demo.external v3.0.0 with only v1.x external query files available.
+    Since major versions differ (3 vs 1), no fallback should occur and no
+    IndirectManagedNodeAudit records should be created.
+    """
+    job = run_external_query_job(
+        run_job_from_playbook,
+        live_tmp_folder,
+        'external_query_ac8_6',
+        'test_host_query_external_v3_0_0',
+    )
+    wait_for_indirect_processing(job, expect_records=False)
+
+    # No audit records should exist for this job
+    assert IndirectManagedNodeAudit.objects.filter(job=job).count() == 0, (
+        'IndirectManagedNodeAudit records were created despite no compatible ' 'fallback existing for demo.external v3.0.0 (only v1.x queries available).'
+    )

awx/main/tests/unit/test_indirect_query_discovery.py

@@ -0,0 +1,431 @@
+"""
+Unit tests for external query discovery and version fallback logic.
+Tests for AAP-58456: Unit Test Suite for External Query Handling
+"""
+
+import sys
+from io import StringIO
+from unittest import mock
+
+import pytest
+from packaging.version import Version
+
+
+# Helper for mocking importlib.resources.files() path traversal
+def create_chainable_path_mock(final_mock, depth=3):
+    """Mock that supports chained / operations: mock / 'a' / 'b' / 'c' -> final_mock"""
+
+    class ChainableMock:
+        def __init__(self, d=0):
+            self.d = d
+
+        def __truediv__(self, other):
+            return final_mock if self.d >= depth - 1 else ChainableMock(self.d + 1)
+
+    return ChainableMock()
+
+
+def create_queries_dir_mock(file_lookup_func):
+    """Mock for queries_dir: mock / 'filename' -> file_lookup_func('filename')"""
+
+    class QueriesDirMock:
+        def __truediv__(self, filename):
+            return file_lookup_func(filename)
+
+    return QueriesDirMock()
+
+
+# Ansible mocking required for importing the module (it imports from ansible.plugins.callback.CallbackBase)
+class MockCallbackBase:
+    def __init__(self):
+        self._display = mock.MagicMock()
+
+    def v2_playbook_on_stats(self, stats):
+        pass
+
+
+_mock_callback_module = mock.MagicMock()
+_mock_callback_module.CallbackBase = MockCallbackBase
+
+
+@pytest.fixture(autouse=True)
+def _mock_ansible_modules():
+    """Temporarily inject fake ansible modules so the callback plugin can be imported."""
+    with mock.patch.dict(
+        sys.modules,
+        {
+            'ansible': mock.MagicMock(),
+            'ansible.plugins': mock.MagicMock(),
+            'ansible.plugins.callback': _mock_callback_module,
+            'ansible.cli': mock.MagicMock(),
+            'ansible.cli.galaxy': mock.MagicMock(),
+            'ansible.release': mock.MagicMock(__version__='2.16.0'),
+            'ansible.galaxy': mock.MagicMock(),
+            'ansible.galaxy.collection': mock.MagicMock(),
+            'ansible.utils': mock.MagicMock(),
+            'ansible.utils.collection_loader': mock.MagicMock(),
+            'ansible.constants': mock.MagicMock(),
+        },
+    ):
+        yield
+
+
+class TestListExternalQueries:
+    """Tests for list_external_queries function."""
+
+    @mock.patch('awx.playbooks.library.indirect_instance_count.files')
+    def test_returns_empty_when_collection_not_installed(self, mock_files):
+        from awx.playbooks.library.indirect_instance_count import list_external_queries
+
+        mock_files.side_effect = ModuleNotFoundError("No module named 'ansible_collections.redhat'")
+
+        result = list_external_queries('demo', 'external')
+
+        assert result == []
+
+    @mock.patch('awx.playbooks.library.indirect_instance_count.files')
+    def test_parses_version_from_filenames(self, mock_files):
+        from awx.playbooks.library.indirect_instance_count import list_external_queries
+
+        mock_file_1 = mock.Mock()
+        mock_file_1.name = 'demo.external.1.0.0.yml'
+        mock_file_2 = mock.Mock()
+        mock_file_2.name = 'demo.external.2.1.0.yml'
+        mock_file_other = mock.Mock()
+        mock_file_other.name = 'other.collection.1.0.0.yml'
+
+        mock_queries_dir = mock.Mock()
+        mock_queries_dir.iterdir.return_value = [mock_file_1, mock_file_2, mock_file_other]
+        mock_files.return_value = create_chainable_path_mock(mock_queries_dir)
+
+        result = list_external_queries('demo', 'external')
+
+        assert len(result) == 2
+        assert Version('1.0.0') in result
+        assert Version('2.1.0') in result
+
+    @mock.patch('awx.playbooks.library.indirect_instance_count.files')
+    def test_skips_invalid_versions(self, mock_files):
+        from awx.playbooks.library.indirect_instance_count import list_external_queries
+
+        mock_file_valid = mock.Mock()
+        mock_file_valid.name = 'demo.external.1.0.0.yml'
+        mock_file_invalid = mock.Mock()
+        mock_file_invalid.name = 'demo.external.invalid.yml'
+
+        mock_queries_dir = mock.Mock()
+        mock_queries_dir.iterdir.return_value = [mock_file_valid, mock_file_invalid]
+        mock_files.return_value = create_chainable_path_mock(mock_queries_dir)
+
+        result = list_external_queries('demo', 'external')
+
+        assert len(result) == 1
+        assert Version('1.0.0') in result
+
+
+class TestVersionFallback:
+    """Tests for version fallback logic (AC7.4-AC7.9)."""
+
+    @mock.patch('awx.playbooks.library.indirect_instance_count._get_query_file_dir')
+    def test_exact_match_preferred(self, mock_get_dir):
+        """AC7.4: Exact version match is preferred over fallback version."""
+        from awx.playbooks.library.indirect_instance_count import find_external_query_with_fallback
+
+        mock_exact_file = mock.Mock()
+        mock_exact_file.exists.return_value = True
+        mock_exact_file.open.return_value.__enter__ = mock.Mock(return_value=StringIO('exact_version_query'))
+        mock_exact_file.open.return_value.__exit__ = mock.Mock(return_value=False)
+
+        mock_get_dir.return_value = create_queries_dir_mock(lambda f: mock_exact_file)
+
+        content, fallback_used, version = find_external_query_with_fallback('demo', 'external', '2.5.0')
+
+        assert content == 'exact_version_query'
+        assert fallback_used is False
+        assert version == '2.5.0'
+
+    @mock.patch('awx.playbooks.library.indirect_instance_count.list_external_queries')
+    @mock.patch('awx.playbooks.library.indirect_instance_count._get_query_file_dir')
+    def test_fallback_nearest_lower_same_major(self, mock_get_dir, mock_list):
+        """AC7.5: Fallback selects nearest lower version within same major version.
+
+        When installed is 4.5.0 and 4.0.0/4.1.0 are available, selects 4.1.0.
+        """
+        from awx.playbooks.library.indirect_instance_count import find_external_query_with_fallback
+
+        mock_list.return_value = [Version('4.0.0'), Version('4.1.0')]
+
+        mock_exact_file = mock.Mock(exists=mock.Mock(return_value=False))
+        mock_fallback_file = mock.Mock()
+        mock_fallback_file.exists.return_value = True
+        mock_fallback_file.open.return_value.__enter__ = mock.Mock(return_value=StringIO('fallback_query'))
+        mock_fallback_file.open.return_value.__exit__ = mock.Mock(return_value=False)
+
+        def file_lookup(filename):
+            return mock_fallback_file if '4.1.0' in filename else mock_exact_file
+
+        mock_get_dir.return_value = create_queries_dir_mock(file_lookup)
+
+        content, fallback_used, version = find_external_query_with_fallback('community', 'vmware', '4.5.0')
+
+        assert content == 'fallback_query'
+        assert fallback_used is True
+        assert version == '4.1.0'
+
+    @mock.patch('awx.playbooks.library.indirect_instance_count.list_external_queries')
+    @mock.patch('awx.playbooks.library.indirect_instance_count._get_query_file_dir')
+    def test_fallback_respects_major_version_boundary(self, mock_get_dir, mock_list):
+        """Test that fallback does NOT cross major version boundaries.
+
+        When installed version is 6.0.0 and only 5.0.0 query exists,
+        no fallback should occur because major versions differ.
+        """
+        from awx.playbooks.library.indirect_instance_count import find_external_query_with_fallback
+
+        mock_list.return_value = [Version('5.0.0')]
+
+        # Mock exact file (6.0.0) to not exist
+        mock_exact_file = mock.Mock(exists=mock.Mock(return_value=False))
+        # Mock fallback file (5.0.0) to exist - if major version check is broken,
+        # this file would be incorrectly selected
+        mock_fallback_file = mock.Mock()
+        mock_fallback_file.exists.return_value = True
+        mock_fallback_file.open.return_value.__enter__ = mock.Mock(return_value=StringIO('wrong_major_version_query'))
+        mock_fallback_file.open.return_value.__exit__ = mock.Mock(return_value=False)
+
+        def file_lookup(filename):
+            return mock_fallback_file if '5.0.0' in filename else mock_exact_file
+
+        mock_get_dir.return_value = create_queries_dir_mock(file_lookup)
+
+        content, fallback_used, version = find_external_query_with_fallback('community', 'vmware', '6.0.0')
+
+        # Should NOT fall back to 5.0.0 because major version differs (5 vs 6)
+        assert content is None
+        assert fallback_used is False
+
+    @mock.patch('awx.playbooks.library.indirect_instance_count.list_external_queries')
+    @mock.patch('awx.playbooks.library.indirect_instance_count._get_query_file_dir')
+    def test_no_fallback_when_incompatible(self, mock_get_dir, mock_list):
+        """AC7.7: No fallback when all available versions are higher than installed.
+
+        When installed version is 3.8.0 and only 4.0.0 and 5.0.0 exist,
+        no fallback should occur because both are higher than installed.
+        """
+        from awx.playbooks.library.indirect_instance_count import find_external_query_with_fallback
+
+        mock_list.return_value = [Version('4.0.0'), Version('5.0.0')]
+
+        # Mock exact file (3.8.0) to not exist
+        mock_exact_file = mock.Mock(exists=mock.Mock(return_value=False))
+        # Mock available files to exist - if version filtering is broken,
+        # one of these would be incorrectly selected
+        mock_available_file = mock.Mock()
+        mock_available_file.exists.return_value = True
+        mock_available_file.open.return_value.__enter__ = mock.Mock(return_value=StringIO('higher_version_query'))
+        mock_available_file.open.return_value.__exit__ = mock.Mock(return_value=False)
+
+        def file_lookup(filename):
+            if '4.0.0' in filename or '5.0.0' in filename:
+                return mock_available_file
+            return mock_exact_file
+
+        mock_get_dir.return_value = create_queries_dir_mock(file_lookup)
+
+        content, fallback_used, version = find_external_query_with_fallback('community', 'vmware', '3.8.0')
+
+        # Should NOT fall back to 4.0.0 or 5.0.0 because both are higher than 3.8.0
+        assert content is None
+        assert fallback_used is False
+
+    @mock.patch('awx.playbooks.library.indirect_instance_count.list_external_queries')
+    @mock.patch('awx.playbooks.library.indirect_instance_count._get_query_file_dir')
+    def test_fallback_selection_logic(self, mock_get_dir, mock_list):
+        """AC7.9: Complex fallback scenario with multiple candidates.
+
+        When installed is 4.5.0 and 4.0.0, 4.1.0, 5.0.0 are available,
+        selects 4.1.0 (highest compatible within same major, <= installed).
+        """
+        from awx.playbooks.library.indirect_instance_count import find_external_query_with_fallback
+
+        mock_list.return_value = [Version('4.0.0'), Version('4.1.0'), Version('5.0.0')]
+
+        mock_exact_file = mock.Mock(exists=mock.Mock(return_value=False))
+        mock_fallback_file = mock.Mock()
+        mock_fallback_file.exists.return_value = True
+        mock_fallback_file.open.return_value.__enter__ = mock.Mock(return_value=StringIO('query_4.1.0'))
+        mock_fallback_file.open.return_value.__exit__ = mock.Mock(return_value=False)
+
+        def file_lookup(filename):
+            return mock_fallback_file if '4.1.0' in filename else mock_exact_file
+
+        mock_get_dir.return_value = create_queries_dir_mock(file_lookup)
+
+        content, fallback_used, version = find_external_query_with_fallback('community', 'vmware', '4.5.0')
+
+        assert version == '4.1.0'
+        assert fallback_used is True
+        assert content == 'query_4.1.0'
+
+
+class TestExternalQueryDiscovery:
+    """Tests for callback plugin query discovery (AC7.1-AC7.3)."""
+
+    @mock.patch('awx.playbooks.library.indirect_instance_count.list_collections')
+    @mock.patch('awx.playbooks.library.indirect_instance_count.files')
+    @mock.patch('awx.playbooks.library.indirect_instance_count.find_external_query_with_fallback')
+    @mock.patch.dict('os.environ', {'AWX_ISOLATED_DATA_DIR': '/tmp/artifacts'})
+    def test_precedence_embedded_over_external(self, mock_fallback, mock_files, mock_list_collections):
+        """AC7.1: Embedded query takes precedence when both embedded and external exist."""
+        from awx.playbooks.library.indirect_instance_count import CallbackModule
+
+        mock_list_collections.return_value = [mock.Mock(namespace='demo', name='query', ver='1.0.0', fqcn='demo.query')]
+
+        mock_embedded_file = mock.Mock()
+        mock_embedded_file.exists.return_value = True
+        mock_embedded_file.open.return_value.__enter__ = mock.Mock(return_value=StringIO('embedded_query'))
+        mock_embedded_file.open.return_value.__exit__ = mock.Mock(return_value=False)
+        mock_files.return_value = create_chainable_path_mock(mock_embedded_file)
+
+        callback = CallbackModule()
+        callback._display = mock.Mock()
+
+        with mock.patch('builtins.open', mock.mock_open()):
+            with mock.patch('json.dumps', return_value='{}'):
+                callback.v2_playbook_on_stats(mock.Mock())
+
+        mock_fallback.assert_not_called()
+        callback._display.vv.assert_called()
+
+    @mock.patch('awx.playbooks.library.indirect_instance_count.list_collections')
+    @mock.patch('awx.playbooks.library.indirect_instance_count.files')
+    @mock.patch('awx.playbooks.library.indirect_instance_count.find_external_query_with_fallback')
+    @mock.patch.dict('os.environ', {'AWX_ISOLATED_DATA_DIR': '/tmp/artifacts'})
+    def test_external_query_when_embedded_missing(self, mock_fallback, mock_files, mock_list_collections):
+        """AC7.2: External query is discovered when embedded query is missing."""
+        from awx.playbooks.library.indirect_instance_count import CallbackModule
+
+        mock_candidate = mock.Mock()
+        mock_candidate.namespace = 'demo'
+        mock_candidate.name = 'external'
+        mock_candidate.ver = '2.5.0'
+        mock_candidate.fqcn = 'demo.external'
+        mock_list_collections.return_value = [mock_candidate]
+
+        mock_embedded_file = mock.Mock(exists=mock.Mock(return_value=False))
+        mock_files.return_value = create_chainable_path_mock(mock_embedded_file)
+        mock_fallback.return_value = ('external_query_content', False, '2.5.0')
+
+        callback = CallbackModule()
+        callback._display = mock.Mock()
+
+        with mock.patch('builtins.open', mock.mock_open()):
+            with mock.patch('json.dumps', return_value='{}'):
+                callback.v2_playbook_on_stats(mock.Mock())
+
+        mock_fallback.assert_called_once_with('demo', 'external', '2.5.0')
+        callback._display.v.assert_called()
+
+    @mock.patch('awx.playbooks.library.indirect_instance_count.list_collections')
+    @mock.patch('awx.playbooks.library.indirect_instance_count.files')
+    @mock.patch('awx.playbooks.library.indirect_instance_count.find_external_query_with_fallback')
+    @mock.patch.dict('os.environ', {'AWX_ISOLATED_DATA_DIR': '/tmp/artifacts'})
+    def test_no_query_when_both_missing(self, mock_fallback, mock_files, mock_list_collections):
+        """AC7.3: No query is used when both embedded and external queries are missing."""
+        from awx.playbooks.library.indirect_instance_count import CallbackModule
+
+        mock_list_collections.return_value = [mock.Mock(namespace='unknown', name='collection', ver='1.0.0', fqcn='unknown.collection')]
+
+        mock_embedded_file = mock.Mock(exists=mock.Mock(return_value=False))
+        mock_files.return_value = create_chainable_path_mock(mock_embedded_file)
+        mock_fallback.return_value = (None, False, None)
+
+        callback = CallbackModule()
+        callback._display = mock.Mock()
+
+        with mock.patch('builtins.open', mock.mock_open()):
+            with mock.patch('json.dumps', return_value='{}'):
+                callback.v2_playbook_on_stats(mock.Mock())
+
+        mock_fallback.assert_called_once()
+
+    @mock.patch('awx.playbooks.library.indirect_instance_count.list_collections')
+    @mock.patch('awx.playbooks.library.indirect_instance_count.files')
+    @mock.patch('awx.playbooks.library.indirect_instance_count.find_external_query_with_fallback')
+    @mock.patch.dict('os.environ', {'AWX_ISOLATED_DATA_DIR': '/tmp/artifacts'})
+    def test_info_log_on_fallback(self, mock_fallback, mock_files, mock_list_collections):
+        """AC7.8: Log message is emitted when fallback version is used.
+
+        Verifies that when a fallback version is used, a log message is emitted
+        containing both the fallback version and the collection FQCN.
+
+        Note: AC7.8 specifies 'warning logs' but implementation uses verbose/info
+        level (_display.v) as this is informational rather than a warning condition.
+        """
+        from awx.playbooks.library.indirect_instance_count import CallbackModule
+
+        mock_list_collections.return_value = [mock.Mock(namespace='community', name='vmware', ver='4.5.0', fqcn='community.vmware')]
+
+        mock_embedded_file = mock.Mock(exists=mock.Mock(return_value=False))
+        mock_files.return_value = create_chainable_path_mock(mock_embedded_file)
+        mock_fallback.return_value = ('fallback_query_content', True, '4.1.0')
+
+        callback = CallbackModule()
+        callback._display = mock.Mock()
+
+        with mock.patch('builtins.open', mock.mock_open()):
+            with mock.patch('json.dumps', return_value='{}'):
+                callback.v2_playbook_on_stats(mock.Mock())
+
+        callback._display.v.assert_called()
+        call_args = callback._display.v.call_args[0][0]
+        assert '4.1.0' in call_args
+        assert 'community.vmware' in call_args
+
+
+class TestPrivateDataDirIntegration:
+    """Tests for vendor collection copying (AC7.10-AC7.11)."""
+
+    @mock.patch('awx.main.tasks.jobs.flag_enabled')
+    @mock.patch('awx.main.tasks.jobs.shutil.copytree')
+    @mock.patch('awx.main.tasks.jobs.os.path.exists')
+    def test_vendor_collections_copied(self, mock_exists, mock_copytree, mock_flag):
+        """AC7.10: build_private_data_files() copies vendor collections to private_data_dir."""
+        from awx.main.tasks.jobs import BaseTask
+
+        mock_flag.return_value = True
+        mock_exists.return_value = True
+
+        task = BaseTask()
+        task.instance = mock.Mock()
+        task.cleanup_paths = []
+        task.build_private_data = mock.Mock(return_value=None)
+
+        private_data_dir = '/tmp/awx_123_abc'
+        task.build_private_data_files(task.instance, private_data_dir)
+
+        mock_copytree.assert_called_once_with('/var/lib/awx/vendor_collections', f'{private_data_dir}/vendor_collections')
+
+    @mock.patch('awx.main.tasks.jobs.flag_enabled')
+    @mock.patch('awx.main.tasks.jobs.logger')
+    @mock.patch('awx.main.tasks.jobs.shutil.copytree')
+    @mock.patch('awx.main.tasks.jobs.os.path.exists')
+    def test_missing_source_handled_gracefully(self, mock_exists, mock_copytree, mock_logger, mock_flag):
+        """AC7.11: Collection copy handles missing source directory gracefully."""
+        from awx.main.tasks.jobs import BaseTask
+
+        mock_flag.return_value = True
+        mock_exists.return_value = False
+
+        task = BaseTask()
+        task.instance = mock.Mock()
+        task.cleanup_paths = []
+        task.build_private_data = mock.Mock(return_value=None)
+
+        private_data_dir = '/tmp/awx_123_abc'
+        result = task.build_private_data_files(task.instance, private_data_dir)
+
+        # copytree should not be called when source doesn't exist
+        mock_copytree.assert_not_called()
+        # Function should complete without raising an exception
+        assert result is not None

awx/main/tests/unit/test_tasks.py

@@ -625,6 +625,11 @@ class TestAdhocRun(TestJobExecution):
 
 
 class TestJobCredentials(TestJobExecution):
+    @pytest.fixture(autouse=True)
+    def mock_flag_enabled(self):
+        with mock.patch('awx.main.tasks.jobs.flag_enabled', return_value=False):
+            yield
+
     @pytest.fixture
     def job(self, execution_environment):
         job = Job(pk=1, inventory=Inventory(pk=1), project=Project(pk=1))
@@ -1158,6 +1163,11 @@ class TestProjectUpdateRefspec(TestJobExecution):
 
 
 class TestInventoryUpdateCredentials(TestJobExecution):
+    @pytest.fixture(autouse=True)
+    def mock_flag_enabled(self):
+        with mock.patch('awx.main.tasks.jobs.flag_enabled', return_value=False):
+            yield
+
     @pytest.fixture
     def inventory_update(self, execution_environment):
         return InventoryUpdate(pk=1, execution_environment=execution_environment, inventory_source=InventorySource(pk=1, inventory=Inventory(pk=1)))

awx/playbooks/library/indirect_instance_count.py

@@ -44,33 +44,35 @@ from ansible.galaxy.collection import find_existing_collections
 from ansible.utils.collection_loader import AnsibleCollectionConfig
 import ansible.constants as C
 
-
-@with_collection_artifacts_manager
-def list_collections(artifacts_manager=None):
-    artifacts_manager.require_build_metadata = False
-
-    default_collections_path = set(C.COLLECTIONS_PATHS)
-    collections_search_paths = default_collections_path | set(AnsibleCollectionConfig.collection_paths)
-    collections = list(find_existing_collections(list(collections_search_paths), artifacts_manager, dedupe=False))
-    return collections
-
-
 # External query path constants
 EXTERNAL_QUERY_COLLECTION = 'ansible_collections.redhat.indirect_accounting'
-EXTERNAL_QUERY_PATH = 'extensions/audit/external_queries'
+
+
+def _get_query_file_dir():
+    """Return the query file directory or None."""
+    try:
+        queries_dir = files(EXTERNAL_QUERY_COLLECTION) / 'extensions' / 'audit' / 'external_queries'
+    except ModuleNotFoundError:
+        return None
+    if not queries_dir.is_dir():
+        return None
+    return queries_dir
 
 
 def list_external_queries(namespace, name):
     """List all available external query versions for a collection.
 
-    Returns a list of Version objects for all available query files
-    matching the namespace.name pattern.
+    Args:
+        namespace: Collection namespace (e.g., 'community')
+        name: Collection name (e.g., 'vmware')
+
+    Returns:
+        List of Version objects for all available query files
+        matching the namespace.name pattern.
     """
     versions = []
 
-    try:
-        queries_dir = files(EXTERNAL_QUERY_COLLECTION) / 'extensions' / 'audit' / 'external_queries'
-    except ModuleNotFoundError:
+    if not (queries_dir := _get_query_file_dir()):
         return versions
 
     # Pattern: namespace.name.X.Y.Z.yml where X.Y.Z is the version
@@ -89,14 +91,13 @@ def list_external_queries(namespace, name):
     return versions
 
 
-def find_external_query_with_fallback(namespace, name, installed_version, display=None):
+def find_external_query_with_fallback(namespace, name, installed_version):
     """Find external query file with semantic version fallback.
 
     Args:
         namespace: Collection namespace (e.g., 'community')
         name: Collection name (e.g., 'vmware')
         installed_version: Version string of installed collection (e.g., '4.5.0')
-        display: Ansible display object for logging
 
     Returns:
         Tuple of (query_content, fallback_used, fallback_version) or (None, False, None)
@@ -104,32 +105,31 @@ def find_external_query_with_fallback(namespace, name, installed_version, displa
         - fallback_used: True if a fallback version was used instead of exact match
         - fallback_version: The version string used (for logging)
     """
-    try:
-        installed_version_object = Version(installed_version)
-    except InvalidVersion:
-        # Invalid version string - can't do version comparison
-        return None, False, None
-    try:
-        queries_dir = files(EXTERNAL_QUERY_COLLECTION) / 'extensions' / 'audit' / 'external_queries'
-    except ModuleNotFoundError:
+    if not (queries_dir := _get_query_file_dir()):
         return None, False, None
 
-    # 1. Try exact version match first (AC5.2)
+    # 1. Try exact version match first
     exact_file = queries_dir / f'{namespace}.{name}.{installed_version}.yml'
     if exact_file.exists():
         with exact_file.open('r') as f:
             return f.read(), False, installed_version
 
     # 2. Find compatible fallback (same major version, nearest lower version)
+    try:
+        installed_version_object = Version(installed_version)
+    except InvalidVersion:
+        # Can't do version comparison for fallback
+        return None, False, None
     available_versions = list_external_queries(namespace, name)
     if not available_versions:
         return None, False, None
-    # Filter to same major version and versions <= installed version (AC5.3, AC5.5)
+
+    # Filter to same major version and versions <= installed version
     compatible_versions = [v for v in available_versions if v.major == installed_version_object.major and v <= installed_version_object]
     if not compatible_versions:
-        # No compatible fallback exists (AC5.7)
         return None, False, None
-    # Select nearest lower version - highest compatible version (AC5.4)
+
+    # Select nearest lower version - highest compatible version
     fallback_version_object = max(compatible_versions)
     fallback_version_str = str(fallback_version_object)
     fallback_file = queries_dir / f'{namespace}.{name}.{fallback_version_str}.yml'
@@ -140,6 +140,16 @@ def find_external_query_with_fallback(namespace, name, installed_version, displa
     return None, False, None
 
 
+@with_collection_artifacts_manager
+def list_collections(artifacts_manager=None):
+    artifacts_manager.require_build_metadata = False
+
+    default_collections_path = set(C.COLLECTIONS_PATHS)
+    collections_search_paths = default_collections_path | set(AnsibleCollectionConfig.collection_paths)
+    collections = list(find_existing_collections(list(collections_search_paths), artifacts_manager, dedupe=False))
+    return collections
+
+
 class CallbackModule(CallbackBase):
     """
     logs playbook results, per host, in /var/log/ansible/hosts
@@ -165,9 +175,10 @@ class CallbackModule(CallbackBase):
                 'version': candidate.ver,
             }
 
-            query_file = files(f'ansible_collections.{candidate.namespace}.{candidate.name}') / 'extensions' / 'audit' / 'event_query.yml'
-            if query_file.exists():
-                with query_file.open('r') as f:
+            # 1. Check for embedded query file (takes precedence)
+            embedded_query_file = files(f'ansible_collections.{candidate.namespace}.{candidate.name}') / 'extensions' / 'audit' / 'event_query.yml'
+            if embedded_query_file.exists():
+                with embedded_query_file.open('r') as f:
                     collection_print['host_query'] = f.read()
                 self._display.vv(f"Using embedded query for {candidate.fqcn} v{candidate.ver}")
             else: