Repository: ansible/awx
File: awx/main/tests/functional/test_inventory_source_injectors.py (Python)

Code before:
            raise AssertionError(
                "File {} is not referenced. References and files:\n{}\n{}".format(abs_file_path, json.dumps(env, indent=4), json.dumps(dir_contents, indent=4))
            )
        file_content = private_key_regex.sub('{{private_key}}', file_content)
        content[file_aliases[abs_file_path]] = file_content

    return (env, content)


def create_reference_data(source_dir, env, content):
    if not os.path.exists(source_dir):
        os.mkdir(source_dir)
    if content:
        files_dir = os.path.join(source_dir, 'files')
        if not os.path.exists(files_dir):
            os.mkdir(files_dir)
        for env_name, content in content.items():
            with open(os.path.join(files_dir, env_name), 'w') as f:
                f.write(content)
    if env:
        with open(os.path.join(source_dir, 'env.json'), 'w') as f:
            json.dump(env, f, indent=4, sort_keys=True)


@pytest.mark.django_db
@pytest.mark.parametrize('this_kind', discover_available_cloud_provider_plugin_names())
def test_inventory_update_injected_content(this_kind, inventory, fake_credential_factory, mock_me):
    ExecutionEnvironment.objects.create(name='Control Plane EE', managed=True)
    ExecutionEnvironment.objects.create(name='Default Job EE', managed=False)

    injector = InventorySource.injectors[this_kind]
    if injector.plugin_name is None:
        pytest.skip('Use of inventory plugin is not enabled for this source')

    src_vars = dict(base_source_var='value_of_var')
    src_vars['plugin'] = injector.get_proper_name()
    inventory_source = InventorySource.objects.create(
        inventory=inventory,
        source=this_kind,
        source_vars=src_vars,
    )
    inventory_source.credentials.add(fake_credential_factory(this_kind))
    inventory_update = inventory_source.create_unified_job()
    task = RunInventoryUpdate()

    def substitute_run(awx_receptor_job):
        """This method will replace run_pexpect
        instead of running, it will read the private data directory contents
        It will make assertions that the contents are correct
        If MAKE_INVENTORY_REFERENCE_FILES is set, it will produce reference files
        """

Review comment (bug):
Looking into the test failures, it seems to me that my suspicion was likely right and this might need to be moved away from import-time invocation. And I'd try converting this to use `pytest-subtests`.

Code after:
            raise AssertionError(
                "File {} is not referenced. References and files:\n{}\n{}".format(abs_file_path, json.dumps(env, indent=4), json.dumps(dir_contents, indent=4))
            )
        file_content = private_key_regex.sub('{{private_key}}', file_content)
        content[file_aliases[abs_file_path]] = file_content

    return (env, content)


def create_reference_data(source_dir, env, content):
    if not os.path.exists(source_dir):
        os.mkdir(source_dir)
    if content:
        files_dir = os.path.join(source_dir, 'files')
        if not os.path.exists(files_dir):
            os.mkdir(files_dir)
        for env_name, content in content.items():
            with open(os.path.join(files_dir, env_name), 'w') as f:
                f.write(content)
    if env:
        with open(os.path.join(source_dir, 'env.json'), 'w') as f:
            json.dump(env, f, indent=4, sort_keys=True)


@pytest.mark.django_db
@pytest.mark.parametrize('this_kind', discover_available_cloud_provider_plugin_names())
def test_inventory_update_injected_content(this_kind, inventory, fake_credential_factory, mock_me):
    if this_kind.endswith('_supported'):
        this_kind = this_kind[:-10]

    ExecutionEnvironment.objects.create(name='Control Plane EE', managed=True)
    ExecutionEnvironment.objects.create(name='Default Job EE', managed=False)

    injector = InventorySource.injectors[this_kind]

    src_vars = dict(base_source_var='value_of_var')
    src_vars['plugin'] = injector.get_proper_name()
    inventory_source = InventorySource.objects.create(
        inventory=inventory,
        source=this_kind,
        source_vars=src_vars,
    )
    inventory_source.credentials.add(fake_credential_factory(this_kind))
    inventory_update = inventory_source.create_unified_job()
    task = RunInventoryUpdate()

    def substitute_run(awx_receptor_job):
        """This method will replace run_pexpect
        instead of running, it will read the private data directory contents
        It will make assertions that the contents are correct
        If MAKE_INVENTORY_REFERENCE_FILES is set, it will produce reference files