ronantakizawa/codereview-bench huggingface dataset:
Repository: ansible/awx
File: awx/main/migrations/0164_remove_inventorysource_update_on_project_update.py (Python)

Code before:
# Generated by Django 3.2.13 on 2022-06-21 21:29

from django.db import migrations
import logging

logger = logging.getLogger("awx")


def forwards(apps, schema_editor):
    InventorySource = apps.get_model('main', 'InventorySource')
    sources = InventorySource.objects.filter(update_on_project_update=True)
    for src in sources:
        if src.update_on_launch == False:
            src.update_on_launch = True
            src.save(update_fields=['update_on_launch'])
        proj = src.source_project
        if proj and proj.update_on_launch is False:
            proj.update_on_launch = True
            proj.save(update_fields=['update_on_launch'])
            logger.warning(f"Setting update_on_launch to True for {proj}")


class Migration(migrations.Migration):

    dependencies = [
        ('main', '0163_convert_job_tags_to_textfield'),
    ]

    operations = [
        migrations.RunPython(forwards, migrations.RunPython.noop),
        migrations.RemoveField(
            model_name='inventorysource',
            name='scm_last_revision',
        ),
        migrations.RemoveField(
            model_name='inventorysource',
            name='update_on_project_update',
        ),
    ]


Review comment (suggestion):
it seems a little in-congruent to not have a log here (like for the project change). Even if it's a lower level, like `logger.info`, that would be good to add.

Code after:
# Generated by Django 3.2.13 on 2022-06-21 21:29

from django.db import migrations
import logging

logger = logging.getLogger("awx")


def forwards(apps, schema_editor):
    InventorySource = apps.get_model('main', 'InventorySource')
    sources = InventorySource.objects.filter(update_on_project_update=True)
    for src in sources:
        if src.update_on_launch == False:
            src.update_on_launch = True
            src.save(update_fields=['update_on_launch'])
            logger.info(f"Setting update_on_launch to True for {src}")
        proj = src.source_project
        if proj and proj.scm_update_on_launch is False:
            proj.scm_update_on_launch = True
            proj.save(update_fields=['scm_update_on_launch'])
            logger.warning(f"Setting scm_update_on_launch to True for {proj}")


class Migration(migrations.Migration):

    dependencies = [
        ('main', '0163_convert_job_tags_to_textfield'),
    ]

    operations = [
        migrations.RunPython(forwards, migrations.RunPython.noop),
        migrations.RemoveField(
            model_name='inventorysource',
            name='scm_last_revision',
        ),
        migrations.RemoveField(
            model_name='inventorysource',
            name='update_on_project_update',
        ),
    ]
Repository: ansible/awx
File: awx/main/tests/unit/utils/test_receptor.py (Python)

Code before:
from awx.main.tasks.receptor import _convert_args_to_cli


def test_file_cleanup_scenario():
    args = _convert_args_to_cli({'exclude_strings': ['awx_423_', 'awx_582_'], 'file_pattern': '/tmp/awx_*_*'})
    assert ' '.join(args) == 'cleanup --exclude-strings=awx_423_ awx_582_ --file-pattern=/tmp/awx_*_*'


def test_image_cleanup_scenario():
    # See input dict in awx.main.tests.unit.models.test_ha
    args = _convert_args_to_cli(
        {
            'file_pattern': '',
            'process_isolation_executable': 'podman',
            'remove_images': ['quay.invalid/foo/bar:latest', 'quay.invalid/foo/bar:devel'],
            'image_prune': True,
        }
    )
    assert (
        ' '.join(args) == 'cleanup --remove-images=quay.invalid/foo/bar:latest quay.invalid/foo/bar:devel --image-prune --process-isolation-executable=podman'
    )


Review comment (suggestion):
As per the backlinked issue, the resultant meaning of this was to exclude any directory that has the `"awx_423_ awx_582_"` literal in its name. That is, in practical terms, excluding no directories.

Code after:
from awx.main.tasks.receptor import _convert_args_to_cli


def test_file_cleanup_scenario():
    args = _convert_args_to_cli({'exclude_strings': ['awx_423_', 'awx_582_'], 'file_pattern': '/tmp/awx_*_*'})
    assert ' '.join(args) == 'cleanup --exclude-strings="awx_423_ awx_582_" --file-pattern=/tmp/awx_*_*'


def test_image_cleanup_scenario():
    # See input dict in awx.main.tests.unit.models.test_ha
    args = _convert_args_to_cli(
        {
            'file_pattern': '',
            'process_isolation_executable': 'podman',
            'remove_images': ['quay.invalid/foo/bar:latest', 'quay.invalid/foo/bar:devel'],
            'image_prune': True,
        }
    )
    assert (
        ' '.join(args) == 'cleanup --remove-images="quay.invalid/foo/bar:latest quay.invalid/foo/bar:devel" --image-prune --process-isolation-executable=podman'
    )
Repository: ansible/awx
File: awx/main/management/commands/check_instance_ready.py (Python)

Code before:
from django.core.management.base import BaseCommand
from awx.main.models.ha import Instance


class Command(BaseCommand):
    help = 'Check if the task manager instance is ready throw error if not ready, can be use as readiness probe for k8s.'

    def handle(self, *args, **options):
        if Instance.objects.me().node_state != Instance.States.READY:
            raise Exception('Instance is not ready')  # so that return code is not 0

        return


Review comment (bug):
The correct way to exit out of Django commands with a failure is to raise `CommandError` (from `django.core.management.base`)

Code after:
from django.core.management.base import BaseCommand, CommandError
from awx.main.models.ha import Instance


class Command(BaseCommand):
    help = 'Check if the task manager instance is ready throw error if not ready, can be use as readiness probe for k8s.'

    def handle(self, *args, **options):
        if Instance.objects.me().node_state != Instance.States.READY:
            raise CommandError('Instance is not ready')  # so that return code is not 0

        return
Repository: ansible/awx
File: awx/main/management/commands/check_instance_ready.py (Python)

Code before:
from django.core.management.base import BaseCommand
from awx.main.models.ha import Instance


class Command(BaseCommand):
    help = 'Check if the task manager instance is ready throw error if not ready, can be use as readiness probe for k8s.'

    def handle(self, *args, **options):
        if Instance.objects.me().node_state != Instance.States.READY:
            raise Exception('Instance is not ready')  # so that return code is not 0

        return


Review comment (suggestion):
```suggestion
from django.core.management.base import CommandError
```

Code after:
from django.core.management.base import BaseCommand, CommandError
from awx.main.models.ha import Instance


class Command(BaseCommand):
    help = 'Check if the task manager instance is ready throw error if not ready, can be use as readiness probe for k8s.'

    def handle(self, *args, **options):
        if Instance.objects.me().node_state != Instance.States.READY:
            raise CommandError('Instance is not ready')  # so that return code is not 0

        return
Repository: ansible/awx
File: awx/main/management/commands/check_instance_ready.py (Python)

Code before:
from django.core.management.base import BaseCommand
from awx.main.models.ha import Instance


class Command(BaseCommand):
    help = 'Check if the task manager instance is ready throw error if not ready, can be use as readiness probe for k8s.'

    def handle(self, *args, **options):
        if Instance.objects.me().node_state != Instance.States.READY:
            raise Exception('Instance is not ready')  # so that return code is not 0

        return


Review comment (suggestion):
```suggestion
            raise CommandError('Instance is not ready')  # so that return code is not 0
```

Code after:
from django.core.management.base import BaseCommand, CommandError
from awx.main.models.ha import Instance


class Command(BaseCommand):
    help = 'Check if the task manager instance is ready throw error if not ready, can be use as readiness probe for k8s.'

    def handle(self, *args, **options):
        if Instance.objects.me().node_state != Instance.States.READY:
            raise CommandError('Instance is not ready')  # so that return code is not 0

        return
Repository: ansible/awx
File: awx/main/tests/functional/dab_rbac/test_managed_roles.py (Python)

Code before:
from ansible_base.rbac.models import RoleDefinition, DABPermission


@pytest.mark.django_db
def test_roles_to_not_create(setup_managed_roles):
    assert RoleDefinition.objects.filter(name='Organization Admin').count() == 1

    SHOULD_NOT_EXIST = ('Organization Organization Admin', 'Organization Team Admin', 'Organization InstanceGroup Admin')

    bad_rds = RoleDefinition.objects.filter(name__in=SHOULD_NOT_EXIST)
    if bad_rds.exists():
        bad_names = list(bad_rds.values_list('name', flat=True))
        raise Exception(f'Found RoleDefinitions that should not exist: {bad_names}')


@pytest.mark.django_db
def test_project_update_role(setup_managed_roles):
    """Role to allow updating a project on the object-level should exist"""
    assert RoleDefinition.objects.filter(name='Project Update').count() == 1


@pytest.mark.django_db
def test_org_child_add_permission(setup_managed_roles):
    for model_name in ('Project', 'NotificationTemplate', 'WorkflowJobTemplate', 'Inventory'):
        rd = RoleDefinition.objects.get(name=f'Organization {model_name} Admin')
        assert 'add_' in str(rd.permissions.values_list('codename', flat=True)), f'The {rd.name} role definition expected to not contain add_ permissions'

    # special case for JobTemplate, anyone can create one with use permission to project/inventory
    assert not DABPermission.objects.filter(codename='add_jobtemplate').exists()


Review comment (security):
```suggestion
        assert 'add_' in str(rd.permissions.values_list('codename', flat=True)), f'The {rd.name} role definition expected to contain add_ permissions'
```

Code after:
from ansible_base.rbac.models import RoleDefinition, DABPermission


@pytest.mark.django_db
def test_roles_to_not_create(setup_managed_roles):
    assert RoleDefinition.objects.filter(name='Organization Admin').count() == 1

    SHOULD_NOT_EXIST = ('Organization Organization Admin', 'Organization Team Admin', 'Organization InstanceGroup Admin')

    bad_rds = RoleDefinition.objects.filter(name__in=SHOULD_NOT_EXIST)
    if bad_rds.exists():
        bad_names = list(bad_rds.values_list('name', flat=True))
        raise Exception(f'Found RoleDefinitions that should not exist: {bad_names}')


@pytest.mark.django_db
def test_project_update_role(setup_managed_roles):
    """Role to allow updating a project on the object-level should exist"""
    assert RoleDefinition.objects.filter(name='Project Update').count() == 1


@pytest.mark.django_db
def test_org_child_add_permission(setup_managed_roles):
    for model_name in ('Project', 'NotificationTemplate', 'WorkflowJobTemplate', 'Inventory'):
        rd = RoleDefinition.objects.get(name=f'Organization {model_name} Admin')
        assert 'add_' in str(rd.permissions.values_list('codename', flat=True)), f'The {rd.name} role definition expected to contain add_ permissions'

    # special case for JobTemplate, anyone can create one with use permission to project/inventory
    assert not DABPermission.objects.filter(codename='add_jobtemplate').exists()
Repository: ansible/awx
File: awx_collection/plugins/lookup/controller_api.py (Python)

Code before:
# (c) 2020 Ansible Project
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function

__metaclass__ = type

DOCUMENTATION = """
name: controller_api
author: John Westcott IV (@john-westcott-iv)
short_description: Search the API for objects
requirements:
  - None
description:
  - Returns GET requests from the AWX API. See the AWX API documentation
    U(https://ansible.readthedocs.io/projects/awx/en/latest/rest_api/index.html) for API usage.
  - For use that is cross-compatible between the awx.awx and ansible.controller collection
    see the controller_meta module
options:
  _terms:
    description:
      - The endpoint to query, i.e. teams, users, tokens, job_templates, etc.
    required: True
  query_params:
    description:
      - The query parameters to search for in the form of key/value pairs.
    type: dict
    required: False
    aliases: [query, data, filter, params]
  expect_objects:
    description:
      - Error if the response does not contain either a detail view or a list view.
    type: boolean
    default: False
    aliases: [expect_object]
  expect_one:
    description:
      - Error if the response contains more than one object.
    type: boolean
    default: False
  return_objects:
    description:

Review comment (suggestion):
This is the same problem as the readme file. The collection is identical with awx and controller, so someone has to make the decision on whether the docs in the collection should be for controller or should be for awx.

My nickel? The collection docs should specify controller where possible. Notice this plugin name is actually controller_api, not awx_api for example. 



Code after:
# (c) 2020 Ansible Project
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function

__metaclass__ = type

DOCUMENTATION = """
name: controller_api
author: John Westcott IV (@john-westcott-iv)
short_description: Search the API for objects
requirements:
  - None
description:
  - Returns GET requests from the Automation Platform Controller API. See
    U(https://docs.ansible.com/automation-controller/latest/html/towerapi/) for API usage.
  - For use that is cross-compatible between the awx.awx and ansible.controller collection
    see the controller_meta module
options:
  _terms:
    description:
      - The endpoint to query, i.e. teams, users, tokens, job_templates, etc.
    required: True
  query_params:
    description:
      - The query parameters to search for in the form of key/value pairs.
    type: dict
    required: False
    aliases: [query, data, filter, params]
  expect_objects:
    description:
      - Error if the response does not contain either a detail view or a list view.
    type: boolean
    default: False
    aliases: [expect_object]
  expect_one:
    description:
      - Error if the response contains more than one object.
    type: boolean
    default: False
  return_objects:
    description:
Repository: ansible/awx
File: awx/__init__.py (Python)

Code before:
    else:
        from setuptools_scm import get_version

        version = get_version(root='..', relative_to=__file__)
        return version


def get_version_from_file():
    vf = version_file()
    if vf:
        with open(vf, 'r') as file:
            return file.read().strip()


def version_file():
    current_dir = os.path.dirname(os.path.abspath(__file__))
    version_file = os.path.join(current_dir, '..', 'VERSION')

    if os.path.exists(version_file):
        return version_file


try:
    __version__ = _get_version('awx')
except PackageNotFoundError:
    __version__ = get_version()

__all__ = ['__version__']


# Check for the presence/absence of "devonly" module to determine if running
# from a source code checkout or release packaage.
try:
    import awx.devonly  # noqa

    MODE = 'development'
except ImportError:  # pragma: no cover
    MODE = 'production'


try:
    import django  # noqa: F401
except ImportError:
    pass
else:
    from django.db import connection


def find_commands(management_dir):
    # Modified version of function from django/core/management/__init__.py.
    command_dir = os.path.join(management_dir, 'commands')

Review comment (suggestion):
LOL, this line errors out with a `NameError` because it never exists prior to the import.

Code after:
    else:
        from setuptools_scm import get_version

        version = get_version(root='..', relative_to=__file__)
        return version


def get_version_from_file():
    vf = version_file()
    if vf:
        with open(vf, 'r') as file:
            return file.read().strip()


def version_file():
    current_dir = os.path.dirname(os.path.abspath(__file__))
    version_file = os.path.join(current_dir, '..', 'VERSION')

    if os.path.exists(version_file):
        return version_file


try:
    __version__ = _get_version('awx')
except PackageNotFoundError:
    __version__ = get_version()

__all__ = ['__version__']


# Check for the presence/absence of "devonly" module to determine if running
# from a source code checkout or release packaage.
try:
    import awx.devonly  # noqa

    MODE = 'development'
except ImportError:  # pragma: no cover
    MODE = 'production'


try:
    import django  # noqa: F401
except ImportError:
    pass
else:
    from django.db import connection


def oauth2_getattribute(self, attr):
    # Custom method to override
    # oauth2_provider.settings.OAuth2ProviderSettings.__getattribute__
Repository: ansible/awx
File: awx/main/management/commands/setup_managed_credential_types.py (Python)

Code before:
# Copyright (c) 2019 Ansible by Red Hat
# All Rights Reserved.

from django.core.management.base import BaseCommand

from awx.main.models import CredentialType


class Command(BaseCommand):
    help = 'Load default managed credential types.'

    def handle(self, *args, **options):
        CredentialType.setup_tower_managed_defaults(wait_for_lock=True)


Review comment (bug):
If someone is explicitly running this management command then they probably want to make sure the code runs to register the CredentialType. Maybe they are debugging an error and want to trigger the code path.

Code after:
# Copyright (c) 2019 Ansible by Red Hat
# All Rights Reserved.

from django.core.management.base import BaseCommand

from awx.main.models import CredentialType


class Command(BaseCommand):
    help = 'Load default managed credential types.'

    def handle(self, *args, **options):
        """
        Note that the call below is almost redundant. The same call as below is called in the Django ready() code path. The ready() code path runs
        before every management command. The one difference in the below call is that the below call is _more_ likely to _actually_ run. The ready() code path
        version _can_ be a NOOP if the lock is not acquired. The below version waits to acquire the lock. This can be useful for recreating bugs or pdb.
        """
        CredentialType.setup_tower_managed_defaults(wait_for_lock=True)
Repository: ansible/awx
File: awx/main/constants.py (Python)

Code before:
# Copyright (c) 2015 Ansible, Inc.
# All Rights Reserved.

import re

from django.utils.translation import gettext_lazy as _

from awx.main.models.inventory import InventorySourceOptions

__all__ = [
    'CLOUD_PROVIDERS',
    'PRIVILEGE_ESCALATION_METHODS',
    'ANSI_SGR_PATTERN',
    'CAN_CANCEL',
    'ACTIVE_STATES',
    'STANDARD_INVENTORY_UPDATE_ENV',
]

plugin_names = tuple(key for key in InventorySourceOptions.injectors.keys())


CLOUD_PROVIDERS = plugin_names
PRIVILEGE_ESCALATION_METHODS = [
    ('sudo', _('Sudo')),
    ('su', _('Su')),
    ('pbrun', _('Pbrun')),
    ('pfexec', _('Pfexec')),
    ('dzdo', _('DZDO')),
    ('pmrun', _('Pmrun')),
    ('runas', _('Runas')),
    ('enable', _('Enable')),
    ('doas', _('Doas')),
    ('ksu', _('Ksu')),
    ('machinectl', _('Machinectl')),
    ('sesu', _('Sesu')),
]
CHOICES_PRIVILEGE_ESCALATION_METHODS = [('', _('None'))] + PRIVILEGE_ESCALATION_METHODS
ANSI_SGR_PATTERN = re.compile(r'\x1b\[[0-9;]*m')
STANDARD_INVENTORY_UPDATE_ENV = {
    # Failure to parse inventory should always be fatal
    'ANSIBLE_INVENTORY_UNPARSED_FAILED': 'True',
    # Always use the --export option for ansible-inventory
    'ANSIBLE_INVENTORY_EXPORT': 'True',
    # Redirecting output to stderr allows JSON parsing to still work with -vvv

Review comment (suggestion):
```suggestion
plugin_names = tuple(InventorySourceOptions.injectors.keys())
```

Code after:
# Copyright (c) 2015 Ansible, Inc.
# All Rights Reserved.

import re

from django.utils.translation import gettext_lazy as _

__all__ = [
    'PRIVILEGE_ESCALATION_METHODS',
    'ANSI_SGR_PATTERN',
    'CAN_CANCEL',
    'ACTIVE_STATES',
    'STANDARD_INVENTORY_UPDATE_ENV',
]

PRIVILEGE_ESCALATION_METHODS = [
    ('sudo', _('Sudo')),
    ('su', _('Su')),
    ('pbrun', _('Pbrun')),
    ('pfexec', _('Pfexec')),
    ('dzdo', _('DZDO')),
    ('pmrun', _('Pmrun')),
    ('runas', _('Runas')),
    ('enable', _('Enable')),
    ('doas', _('Doas')),
    ('ksu', _('Ksu')),
    ('machinectl', _('Machinectl')),
    ('sesu', _('Sesu')),
]
CHOICES_PRIVILEGE_ESCALATION_METHODS = [('', _('None'))] + PRIVILEGE_ESCALATION_METHODS
ANSI_SGR_PATTERN = re.compile(r'\x1b\[[0-9;]*m')
STANDARD_INVENTORY_UPDATE_ENV = {
    # Failure to parse inventory should always be fatal
    'ANSIBLE_INVENTORY_UNPARSED_FAILED': 'True',
    # Always use the --export option for ansible-inventory
    'ANSIBLE_INVENTORY_EXPORT': 'True',
    # Redirecting output to stderr allows JSON parsing to still work with -vvv
    'ANSIBLE_VERBOSE_TO_STDERR': 'True',
    # if ansible-inventory --limit is used for an inventory import, unmatched should be a failure
    'ANSIBLE_HOST_PATTERN_MISMATCH': 'error',
}
CAN_CANCEL = ('new', 'pending', 'waiting', 'running')
ACTIVE_STATES = CAN_CANCEL
ERROR_STATES = ('error',)