[AAP-89676] Add control plane EE unit/functional tests (#16617)

* [AAP-89071] Add control plane EE unit/functional tests

Add tests for EE resolution, constructed inventory model,
and ansible-sign signature validation tag injection.

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

* [AAP-89071] Fix control plane EE tests

- Test resolve_execution_environment on ProjectUpdate/InventoryUpdate
  (job objects) instead of Project/InventorySource (templates), matching
  the actual code path in BaseTask.run()
- Consolidate inventory EE resolution into single parametrized test
- Remove trivial kind-check test
- Fix OrderedManyToManyField.add() to add one inventory at a time
- Add asymmetric host counts for better aggregation validation

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

* [AAP-89071] Fix InventoryUpdate missing inventory relation

Add inventory=inv to InventoryUpdate.objects.create() so
resolve_execution_environment() can access self.inventory.kind.

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

---------

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

awx/main/tests/functional/models/test_inventory.py

@@ -341,6 +341,33 @@ def test_inventory_update_excessively_long_name(inventory, inventory_source):
     assert iu.name.startswith(inventory.name)
 
 
+@pytest.mark.django_db
+class TestConstructedInventory:
+    def test_constructed_inventory_aggregates_hosts_from_inputs(self, organization):
+        """Create 2 regular inventories with asymmetric host counts, create a constructed
+        inventory with those as inputs, verify both inventories are linked and hosts
+        from each are distinguishable"""
+        inv1 = Inventory.objects.create(name='source-inv-1', organization=organization)
+        inv1.hosts.create(name='host-a')
+        inv1.hosts.create(name='host-b')
+        inv1.hosts.create(name='host-c')
+
+        inv2 = Inventory.objects.create(name='source-inv-2', organization=organization)
+        inv2.hosts.create(name='host-d')
+
+        constructed = Inventory.objects.create(name='my-constructed', kind='constructed', organization=organization)
+        constructed.input_inventories.add(inv1)
+        constructed.input_inventories.add(inv2)
+
+        input_ids = set(constructed.input_inventories.values_list('id', flat=True))
+        assert input_ids == {inv1.id, inv2.id}
+
+        all_host_names = set()
+        for inv in constructed.input_inventories.all():
+            all_host_names.update(inv.hosts.values_list('name', flat=True))
+        assert all_host_names == {'host-a', 'host-b', 'host-c', 'host-d'}
+
+
 @pytest.mark.django_db
 class TestHostManager:
     def test_host_filter_not_smart(self, setup_ec2_gce, organization):

awx/main/tests/functional/utils/test_execution_environments.py

@@ -3,8 +3,9 @@ import pytest
 from django.conf import settings
 from django.test.utils import override_settings
 
+from awx.main.models import Inventory, InventorySource, InventoryUpdate, ProjectUpdate
 from awx.main.models.execution_environments import ExecutionEnvironment
-from awx.main.utils.execution_environments import get_default_execution_environment
+from awx.main.utils.execution_environments import get_default_execution_environment, get_control_plane_execution_environment
 from awx.main.management.commands.register_default_execution_environments import Command
 
 
@@ -44,3 +45,57 @@ def test_user_default(set_up_defaults):
     ee = ExecutionEnvironment.objects.create(name='Steves environment', image='quay.io/ansible/awx-ee')
     with override_settings(DEFAULT_EXECUTION_ENVIRONMENT=ee):
         assert get_default_execution_environment() == ee
+
+
+@pytest.mark.django_db
+def test_project_update_uses_control_plane_ee(set_up_defaults, project):
+    """ProjectUpdate.resolve_execution_environment() should always return the control plane EE"""
+    control_plane_ee = get_control_plane_execution_environment()
+    project_update = ProjectUpdate.objects.create(
+        project=project,
+        scm_type=project.scm_type,
+    )
+    resolved_ee = project_update.resolve_execution_environment()
+    assert resolved_ee == control_plane_ee
+    assert resolved_ee.managed is True
+    assert resolved_ee.organization is None
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    'source,expects_control_plane',
+    [
+        ('constructed', True),
+        ('ec2', False),
+        ('azure_rm', False),
+        ('gce', False),
+        ('vmware', False),
+        ('openstack', False),
+        ('rhv', False),
+        ('satellite6', False),
+        ('controller', False),
+    ],
+)
+def test_inventory_update_ee_resolution(set_up_defaults, organization, source, expects_control_plane):
+    """Constructed inventory updates should resolve to the control plane EE; all others should not"""
+    control_plane_ee = get_control_plane_execution_environment()
+    inv = Inventory.objects.create(
+        name=f'test-inv-{source}',
+        kind='constructed' if source == 'constructed' else '',
+        organization=organization,
+    )
+    inv_source = InventorySource.objects.create(
+        name=f'{source}-source',
+        inventory=inv,
+        source=source,
+    )
+    inv_update = InventoryUpdate.objects.create(
+        inventory=inv,
+        inventory_source=inv_source,
+        source=source,
+    )
+    resolved_ee = inv_update.resolve_execution_environment()
+    if expects_control_plane:
+        assert resolved_ee == control_plane_ee
+    else:
+        assert resolved_ee != control_plane_ee

awx/main/tests/unit/test_tasks.py

@@ -1169,6 +1169,40 @@ class TestProjectUpdateCredentials(TestJobExecution):
         assert env['FOO'] == 'BAR'
 
 
+class TestProjectUpdateSignatureValidation(TestJobExecution):
+    """Tests that ProjectUpdate.save() adds GPG validation job tags when a
+    signature_validation_credential is set on the project."""
+
+    def test_project_update_with_signature_credential_adds_validation_tags(self):
+        gpg_credential_type = CredentialType(pk=1, namespace='gpg_public_key', kind='cryptography')
+        gpg_credential = Credential(pk=1, credential_type=gpg_credential_type)
+        project = Project(pk=1, organization=Organization(pk=1), scm_type='git', signature_validation_credential=gpg_credential)
+        project_update = ProjectUpdate(pk=1, project=project, scm_type='git')
+        project_update.websocket_emit_status = mock.Mock()
+
+        # Simulate the save logic that sets job_tags without hitting the database
+        with mock.patch.object(UnifiedJob, 'save', return_value=None):
+            project_update.save()
+
+        assert 'validation_gpg_public_key' in project_update.job_tags
+        assert 'validation_checksum_manifest' in project_update.job_tags
+        assert 'update_git' in project_update.job_tags
+
+    def test_project_update_without_signature_credential_no_validation_tags(self):
+        project = Project(pk=1, organization=Organization(pk=1), scm_type='git')
+        project_update = ProjectUpdate(pk=1, project=project, scm_type='git')
+        project_update.websocket_emit_status = mock.Mock()
+
+        with mock.patch.object(UnifiedJob, 'save', return_value=None):
+            project_update.save()
+
+        assert 'validation_gpg_public_key' not in project_update.job_tags
+        assert 'validation_checksum_manifest' not in project_update.job_tags
+        assert 'update_git' in project_update.job_tags
+        assert 'install_roles' in project_update.job_tags
+        assert 'install_collections' in project_update.job_tags
+
+
 @pytest.mark.django_db
 class TestProjectUpdateRefspec(TestJobExecution):
     @pytest.fixture