Write logic to combing workflow labels, IGs with nodes

Additionally, move the inventory-specific hacks of yesteryear
  into the prompts_dict method of the WorkflowJob model
  try to make it clear exactly what this is hacking and why

Correctly summarize label prompts, and add missing EE

Expand unit tests to apply more fields

adding missing fields to preserve during copy to workflow.py

Fix bug where empty workflow job vars blanked node vars (#12904)

* Fix bug where empty workflow job vars blanked node vars

* Fix bug where workflow job has no extra_vars, add test

* Add empty workflow job extra vars to assure fix

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

@@ -1,7 +1,7 @@
 import pytest
 
 # AWX
-from awx.main.models.jobs import JobTemplate, JobLaunchConfig, LaunchTimeConfigBase
+from awx.main.models.jobs import JobTemplate, LaunchTimeConfigBase
 from awx.main.models.execution_environments import ExecutionEnvironment
 
 
@@ -12,18 +12,6 @@ def full_jt(inventory, project, machine_credential):
     return jt
 
 
-@pytest.fixture
-def config_factory(full_jt):
-    def return_config(data):
-        job = full_jt.create_unified_job(**data)
-        try:
-            return job.launch_config
-        except JobLaunchConfig.DoesNotExist:
-            return None
-
-    return return_config
-
-
 @pytest.mark.django_db
 class TestConfigCreation:
     """
@@ -41,39 +29,51 @@ class TestConfigCreation:
         assert config.limit == 'foobar'
         assert config.char_prompts == {'limit': 'foobar'}
 
-    def test_added_credential(self, full_jt, credential):
-        job = full_jt.create_unified_job(credentials=[credential])
+    def test_added_related(self, full_jt, credential, default_instance_group, label):
+        job = full_jt.create_unified_job(credentials=[credential], instance_groups=[default_instance_group], labels=[label])
         config = job.launch_config
         assert set(config.credentials.all()) == set([credential])
+        assert set(config.labels.all()) == set([label])
+        assert set(config.instance_groups.all()) == set([default_instance_group])
 
     def test_survey_passwords_ignored(self, inventory_source):
         iu = inventory_source.create_unified_job(survey_passwords={'foo': '$encrypted$'})
         assert iu.launch_config.prompts_dict() == {}
 
 
+@pytest.fixture
+def full_prompts_dict(inventory, credential, label, default_instance_group):
+    ee = ExecutionEnvironment.objects.create(name='test-ee', image='quay.io/foo/bar')
+    r = {
+        'limit': 'foobar',
+        'inventory': inventory,
+        'credentials': [credential],
+        'execution_environment': ee,
+        'labels': [label],
+        'instance_groups': [default_instance_group],
+        'verbosity': 3,
+        'scm_branch': 'non_dev',
+        'diff_mode': True,
+        'skip_tags': 'foobar',
+        'job_tags': 'untagged',
+        'forks': 26,
+        'job_slice_count': 2,
+        'timeout': 200,
+        'extra_vars': {'prompted_key': 'prompted_val'},
+        'job_type': 'check',
+    }
+    assert set(JobTemplate.get_ask_mapping().keys()) - set(r.keys()) == set()  # make fixture comprehensive
+    return r
+
+
 @pytest.mark.django_db
-class TestConfigReversibility:
+def test_config_reversibility(full_jt, full_prompts_dict):
     """
     Checks that a blob of saved prompts will be re-created in the
     prompts_dict for launching new jobs
     """
-
-    def test_char_field_only(self, config_factory):
-        config = config_factory({'limit': 'foobar'})
-        assert config.prompts_dict() == {'limit': 'foobar'}
-
-    def test_related_objects(self, config_factory, inventory, credential, label, default_instance_group):
-        ee = ExecutionEnvironment.objects.create(name='test-ee', image='quay.io/foo/bar')
-        prompts = {
-            'limit': 'foobar',
-            'inventory': inventory,
-            'credentials': [credential],
-            'execution_environment': ee,
-            'labels': [label],
-            'instance_groups': [default_instance_group],
-        }
-        config = config_factory(prompts)
-        assert config.prompts_dict() == prompts
+    config = full_jt.create_unified_job(**full_prompts_dict).launch_config
+    assert config.prompts_dict() == full_prompts_dict
 
 
 @pytest.mark.django_db

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

@@ -12,6 +12,9 @@ from awx.main.models.workflow import (
 )
 from awx.main.models.jobs import JobTemplate, Job
 from awx.main.models.projects import ProjectUpdate
+from awx.main.models.credential import Credential, CredentialType
+from awx.main.models.label import Label
+from awx.main.models.ha import InstanceGroup
 from awx.main.scheduler.dag_workflow import WorkflowDAG
 from awx.api.versioning import reverse
 from awx.api.views import WorkflowJobTemplateNodeSuccessNodesList
@@ -229,6 +232,65 @@ class TestWorkflowJob:
         assert queued_node.get_job_kwargs()['extra_vars'] == {'a': 42, 'b': 43}
         assert queued_node.ancestor_artifacts == {'a': 42, 'b': 43}
 
+    def test_combine_prompts_WFJT_to_node(self, project, inventory, organization):
+        """
+        Test that complex prompts like variables, credentials, labels, etc
+        are properly combined from the workflow-level with the node-level
+        """
+        jt = JobTemplate.objects.create(
+            project=project,
+            inventory=inventory,
+            ask_variables_on_launch=True,
+            ask_credential_on_launch=True,
+            ask_instance_groups_on_launch=True,
+            ask_labels_on_launch=True,
+            ask_limit_on_launch=True,
+        )
+        wj = WorkflowJob.objects.create(name='test-wf-job', extra_vars='{}')
+
+        common_ig = InstanceGroup.objects.create(name='common')
+        common_ct = CredentialType.objects.create(name='common')
+
+        node = WorkflowJobNode.objects.create(workflow_job=wj, unified_job_template=jt, extra_vars={'node_key': 'node_val'})
+        node.limit = 'node_limit'
+        node.save()
+        node_cred_unique = Credential.objects.create(credential_type=CredentialType.objects.create(name='node'))
+        node_cred_conflicting = Credential.objects.create(credential_type=common_ct)
+        node.credentials.add(node_cred_unique, node_cred_conflicting)
+        node_labels = [Label.objects.create(name='node1', organization=organization), Label.objects.create(name='node2', organization=organization)]
+        node.labels.add(*node_labels)
+        node_igs = [common_ig, InstanceGroup.objects.create(name='node')]
+        for ig in node_igs:
+            node.instance_groups.add(ig)
+
+        # assertions for where node has prompts but workflow job does not
+        data = node.get_job_kwargs()
+        assert data['extra_vars'] == {'node_key': 'node_val'}
+        assert set(data['credentials']) == set([node_cred_conflicting, node_cred_unique])
+        assert data['instance_groups'] == node_igs
+        assert set(data['labels']) == set(node_labels)
+        assert data['limit'] == 'node_limit'
+
+        # add prompts to the WorkflowJob
+        wj.limit = 'wj_limit'
+        wj.extra_vars = {'wj_key': 'wj_val'}
+        wj.save()
+        wj_cred_unique = Credential.objects.create(credential_type=CredentialType.objects.create(name='wj'))
+        wj_cred_conflicting = Credential.objects.create(credential_type=common_ct)
+        wj.credentials.add(wj_cred_unique, wj_cred_conflicting)
+        wj.labels.add(Label.objects.create(name='wj1', organization=organization), Label.objects.create(name='wj2', organization=organization))
+        wj_igs = [InstanceGroup.objects.create(name='wj'), common_ig]
+        for ig in wj_igs:
+            wj.instance_groups.add(ig)
+
+        # assertions for behavior where node and workflow jobs have prompts
+        data = node.get_job_kwargs()
+        assert data['extra_vars'] == {'node_key': 'node_val', 'wj_key': 'wj_val'}
+        assert set(data['credentials']) == set([wj_cred_unique, wj_cred_conflicting, node_cred_unique])
+        assert data['instance_groups'] == wj_igs
+        assert set(data['labels']) == set(node_labels)  # as exception, WFJT labels not applied
+        assert data['limit'] == 'wj_limit'
+
 
 @pytest.mark.django_db
 class TestWorkflowJobTemplate: