ronantakizawa/codereview-bench huggingface dataset:
Repository: ansible/awx
File: awxkit/test/api/pages/test_base.py (Python)

Code before:
@pytest.mark.parametrize(
    "kwargs,url,token",
    [
        ({"client_id": "foo", "client_secret": "bar"}, "/o/token/", "my_token"),
        ({"client_id": "foo"}, "/o/token/", "my_token"),
        ({}, "/api/gateway/v1/tokens/", "my_personal_token"),
    ],
)
def test_get_oauth2_token_from_gateway(mocker, response, kwargs, url, token):
    post = mocker.patch("requests.Session.post", return_value=response)
    base = Base()
    ret = base.get_oauth2_token(**kwargs)
    assert post.call_count == 1
    assert post.call_args.args[0] == url
    assert ret == token


@pytest.mark.parametrize(
    "kwargs,url,token",
    [
        ({"client_id": "foo", "client_secret": "bar"}, "/api/o/token/", "my_token"),
        ({"client_id": "foo"}, "/api/o/token/", "my_token"),
        ({}, "/api/v2/users/foo/personal_tokens/", "my_personal_token"),
    ],
)
def test_get_oauth2_token_from_controller(mocker, response, kwargs, url, token):
    type(response).ok = PropertyMock(side_effect=[False, True])
    post = mocker.patch("requests.Session.post", return_value=response)
    base = Base()
    ret = base.get_oauth2_token(**kwargs)
    assert post.call_count == 2
    assert post.call_args.args[0] == url
    assert ret == token


Review comment (security):
Mind annotating this with types?
```suggestion
def test_get_oauth2_token_from_controller(mocker: MockerFixture, response: Response, kwargs: dict[str, str], url: str, token: str) -> None:
```
This needs a `from requests import Response` and `from pytest_mock import MockerFixture` at the top.

Code after:
@pytest.mark.parametrize(
    ("auth_creds", "url", "token"),
    [
        ({"client_id": "foo", "client_secret": "bar"}, "/o/token/", "my_token"),
        ({"client_id": "foo"}, "/o/token/", "my_token"),
        ({}, "/api/gateway/v1/tokens/", "my_personal_token"),
    ],
)
def test_get_oauth2_token_from_gateway(mocker: MockerFixture, response: Response, auth_creds, url, token):
    post = mocker.patch("requests.Session.post", return_value=response)
    base = Base()
    ret = base.get_oauth2_token(**auth_creds)
    assert post.call_count == 1
    assert post.call_args.args[0] == url
    assert ret == token


@pytest.mark.parametrize(
    ("auth_creds", "url", "token"),
    [
        ({"client_id": "foo", "client_secret": "bar"}, "/api/o/token/", "my_token"),
        ({"client_id": "foo"}, "/api/o/token/", "my_token"),
        ({}, "/api/v2/users/foo/personal_tokens/", "my_personal_token"),
    ],
)
def test_get_oauth2_token_from_controller(mocker: MockerFixture, response: Response, auth_creds, url, token):
    type(response).ok = mocker.PropertyMock(side_effect=[False, True])
    post = mocker.patch("requests.Session.post", return_value=response)
    base = Base()
    ret = base.get_oauth2_token(**auth_creds)
    assert post.call_count == 2
    assert post.call_args.args[0] == url
    assert ret == token
Repository: ansible/awx
File: awxkit/test/api/pages/test_base.py (Python)

Code before:
    r.json.return_value = {
        "token": "my_personal_token",
        "access_token": "my_token",
    }
    return r


@pytest.mark.parametrize(
    "kwargs,url,token",
    [
        ({"client_id": "foo", "client_secret": "bar"}, "/o/token/", "my_token"),
        ({"client_id": "foo"}, "/o/token/", "my_token"),
        ({}, "/api/gateway/v1/tokens/", "my_personal_token"),
    ],
)
def test_get_oauth2_token_from_gateway(mocker, response, kwargs, url, token):
    post = mocker.patch("requests.Session.post", return_value=response)
    base = Base()
    ret = base.get_oauth2_token(**kwargs)
    assert post.call_count == 1
    assert post.call_args.args[0] == url
    assert ret == token


@pytest.mark.parametrize(
    "kwargs,url,token",
    [
        ({"client_id": "foo", "client_secret": "bar"}, "/api/o/token/", "my_token"),
        ({"client_id": "foo"}, "/api/o/token/", "my_token"),
        ({}, "/api/v2/users/foo/personal_tokens/", "my_personal_token"),
    ],
)
def test_get_oauth2_token_from_controller(mocker, response, kwargs, url, token):
    type(response).ok = PropertyMock(side_effect=[False, True])
    post = mocker.patch("requests.Session.post", return_value=response)
    base = Base()
    ret = base.get_oauth2_token(**kwargs)
    assert post.call_count == 2
    assert post.call_args.args[0] == url
    assert ret == token


Review comment (suggestion):
> [!tip]
> *Pro tip*: It is possible to use an iterable of strings for param names.
> ```suggestion
>     ("kwargs", "url", "token"),
> ```

Code after:
    r.json.return_value = {
        "token": "my_personal_token",
        "access_token": "my_token",
    }
    return r


@pytest.mark.parametrize(
    ("auth_creds", "url", "token"),
    [
        ({"client_id": "foo", "client_secret": "bar"}, "/o/token/", "my_token"),
        ({"client_id": "foo"}, "/o/token/", "my_token"),
        ({}, "/api/gateway/v1/tokens/", "my_personal_token"),
    ],
)
def test_get_oauth2_token_from_gateway(mocker: MockerFixture, response: Response, auth_creds, url, token):
    post = mocker.patch("requests.Session.post", return_value=response)
    base = Base()
    ret = base.get_oauth2_token(**auth_creds)
    assert post.call_count == 1
    assert post.call_args.args[0] == url
    assert ret == token


@pytest.mark.parametrize(
    ("auth_creds", "url", "token"),
    [
        ({"client_id": "foo", "client_secret": "bar"}, "/api/o/token/", "my_token"),
        ({"client_id": "foo"}, "/api/o/token/", "my_token"),
        ({}, "/api/v2/users/foo/personal_tokens/", "my_personal_token"),
    ],
)
def test_get_oauth2_token_from_controller(mocker: MockerFixture, response: Response, auth_creds, url, token):
    type(response).ok = mocker.PropertyMock(side_effect=[False, True])
    post = mocker.patch("requests.Session.post", return_value=response)
    base = Base()
    ret = base.get_oauth2_token(**auth_creds)
    assert post.call_count == 2
    assert post.call_args.args[0] == url
    assert ret == token
Repository: ansible/awx
File: awxkit/test/api/pages/test_base.py (Python)

Code before:
    "kwargs,url,token",
    [
        ({"client_id": "foo", "client_secret": "bar"}, "/o/token/", "my_token"),
        ({"client_id": "foo"}, "/o/token/", "my_token"),
        ({}, "/api/gateway/v1/tokens/", "my_personal_token"),
    ],
)
def test_get_oauth2_token_from_gateway(mocker, response, kwargs, url, token):
    post = mocker.patch("requests.Session.post", return_value=response)
    base = Base()
    ret = base.get_oauth2_token(**kwargs)
    assert post.call_count == 1
    assert post.call_args.args[0] == url
    assert ret == token


@pytest.mark.parametrize(
    "kwargs,url,token",
    [
        ({"client_id": "foo", "client_secret": "bar"}, "/api/o/token/", "my_token"),
        ({"client_id": "foo"}, "/api/o/token/", "my_token"),
        ({}, "/api/v2/users/foo/personal_tokens/", "my_personal_token"),
    ],
)
def test_get_oauth2_token_from_controller(mocker, response, kwargs, url, token):
    type(response).ok = PropertyMock(side_effect=[False, True])
    post = mocker.patch("requests.Session.post", return_value=response)
    base = Base()
    ret = base.get_oauth2_token(**kwargs)
    assert post.call_count == 2
    assert post.call_args.args[0] == url
    assert ret == token


Review comment (suggestion):
There's no need to import this as it's already provided by the fixture of choice:
```suggestion
    type(response).ok = mocker.PropertyMock(side_effect=[False, True])
```

Code after:
    ("auth_creds", "url", "token"),
    [
        ({"client_id": "foo", "client_secret": "bar"}, "/o/token/", "my_token"),
        ({"client_id": "foo"}, "/o/token/", "my_token"),
        ({}, "/api/gateway/v1/tokens/", "my_personal_token"),
    ],
)
def test_get_oauth2_token_from_gateway(mocker: MockerFixture, response: Response, auth_creds, url, token):
    post = mocker.patch("requests.Session.post", return_value=response)
    base = Base()
    ret = base.get_oauth2_token(**auth_creds)
    assert post.call_count == 1
    assert post.call_args.args[0] == url
    assert ret == token


@pytest.mark.parametrize(
    ("auth_creds", "url", "token"),
    [
        ({"client_id": "foo", "client_secret": "bar"}, "/api/o/token/", "my_token"),
        ({"client_id": "foo"}, "/api/o/token/", "my_token"),
        ({}, "/api/v2/users/foo/personal_tokens/", "my_personal_token"),
    ],
)
def test_get_oauth2_token_from_controller(mocker: MockerFixture, response: Response, auth_creds, url, token):
    type(response).ok = mocker.PropertyMock(side_effect=[False, True])
    post = mocker.patch("requests.Session.post", return_value=response)
    base = Base()
    ret = base.get_oauth2_token(**auth_creds)
    assert post.call_count == 2
    assert post.call_args.args[0] == url
    assert ret == token
Repository: ansible/awx
File: awxkit/test/api/pages/test_base.py (Python)

Code before:
from unittest.mock import Mock, PropertyMock

from http.client import NOT_FOUND
import pytest

from awxkit.api.pages import Base
from awxkit.config import config


@pytest.fixture(autouse=True)
def setup_config():
    config.credentials = {"default": {"username": "foo", "password": "bar"}}
    config.base_url = ""


@pytest.fixture
def response():
    r = Mock()
    r.status_code = NOT_FOUND
    r.json.return_value = {
        "token": "my_personal_token",
        "access_token": "my_token",
    }
    return r


@pytest.mark.parametrize(
    "kwargs,url,token",
    [
        ({"client_id": "foo", "client_secret": "bar"}, "/o/token/", "my_token"),
        ({"client_id": "foo"}, "/o/token/", "my_token"),
        ({}, "/api/gateway/v1/tokens/", "my_personal_token"),
    ],
)
def test_get_oauth2_token_from_gateway(mocker, response, kwargs, url, token):
    post = mocker.patch("requests.Session.post", return_value=response)
    base = Base()
    ret = base.get_oauth2_token(**kwargs)
    assert post.call_count == 1
    assert post.call_args.args[0] == url
    assert ret == token



Review comment (suggestion):
It's possible to drop the `unittest` import with the following:
```suggestion
def response(mocker: MockerFixture):
    r = mocker.Mock()
```

Code after:
from http.client import NOT_FOUND
import pytest
from pytest_mock import MockerFixture
from requests import Response

from awxkit.api.pages import Base
from awxkit.config import config


@pytest.fixture(autouse=True)
def setup_config(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(config, "credentials", {"default": {"username": "foo", "password": "bar"}}, raising=False)
    monkeypatch.setattr(config, "base_url", "", raising=False)


@pytest.fixture
def response(mocker):
    r = mocker.Mock()
    r.status_code = NOT_FOUND
    r.json.return_value = {
        "token": "my_personal_token",
        "access_token": "my_token",
    }
    return r


@pytest.mark.parametrize(
    ("auth_creds", "url", "token"),
    [
        ({"client_id": "foo", "client_secret": "bar"}, "/o/token/", "my_token"),
        ({"client_id": "foo"}, "/o/token/", "my_token"),
        ({}, "/api/gateway/v1/tokens/", "my_personal_token"),
    ],
)
def test_get_oauth2_token_from_gateway(mocker: MockerFixture, response: Response, auth_creds, url, token):
    post = mocker.patch("requests.Session.post", return_value=response)
    base = Base()
    ret = base.get_oauth2_token(**auth_creds)
    assert post.call_count == 1
    assert post.call_args.args[0] == url
    assert ret == token
Repository: ansible/awx
File: awxkit/test/api/pages/test_base.py (Python)

Code before:
from unittest.mock import Mock, PropertyMock

from http.client import NOT_FOUND
import pytest

from awxkit.api.pages import Base
from awxkit.config import config


@pytest.fixture(autouse=True)
def setup_config():
    config.credentials = {"default": {"username": "foo", "password": "bar"}}
    config.base_url = ""


@pytest.fixture
def response():
    r = Mock()
    r.status_code = NOT_FOUND
    r.json.return_value = {
        "token": "my_personal_token",
        "access_token": "my_token",
    }
    return r


@pytest.mark.parametrize(
    "kwargs,url,token",
    [
        ({"client_id": "foo", "client_secret": "bar"}, "/o/token/", "my_token"),
        ({"client_id": "foo"}, "/o/token/", "my_token"),
        ({}, "/api/gateway/v1/tokens/", "my_personal_token"),
    ],
)
def test_get_oauth2_token_from_gateway(mocker, response, kwargs, url, token):
    post = mocker.patch("requests.Session.post", return_value=response)
    base = Base()
    ret = base.get_oauth2_token(**kwargs)

Review comment (security):
Make sure that the patching is undone once testing this module is complete. We should be careful not to create side effects for any other tests:
```suggestion
def setup_config(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(
        "credentials",
        {"default": {"username": "foo", "password": "bar"}},
    )
    monkeypatch.setattr("base_url", "")
```

Code after:
from http.client import NOT_FOUND
import pytest
from pytest_mock import MockerFixture
from requests import Response

from awxkit.api.pages import Base
from awxkit.config import config


@pytest.fixture(autouse=True)
def setup_config(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(config, "credentials", {"default": {"username": "foo", "password": "bar"}}, raising=False)
    monkeypatch.setattr(config, "base_url", "", raising=False)


@pytest.fixture
def response(mocker):
    r = mocker.Mock()
    r.status_code = NOT_FOUND
    r.json.return_value = {
        "token": "my_personal_token",
        "access_token": "my_token",
    }
    return r


@pytest.mark.parametrize(
    ("auth_creds", "url", "token"),
    [
        ({"client_id": "foo", "client_secret": "bar"}, "/o/token/", "my_token"),
        ({"client_id": "foo"}, "/o/token/", "my_token"),
        ({}, "/api/gateway/v1/tokens/", "my_personal_token"),
    ],
)
def test_get_oauth2_token_from_gateway(mocker: MockerFixture, response: Response, auth_creds, url, token):
    post = mocker.patch("requests.Session.post", return_value=response)
    base = Base()
    ret = base.get_oauth2_token(**auth_creds)
Repository: ansible/awx
File: awx/main/utils/plugins.py (Python)

Code before:
    """
    from awx.main.models.inventory import InventorySourceOptions

    plugin_description_list = [(plugin_name, plugin.plugin_description) for plugin_name, plugin in InventorySourceOptions.injectors.items()]

    plugin_description = dict(plugin_description_list)

    return plugin_description


@cache
def load_combined_inventory_source_options() -> dict[str, str]:
    """
    Return a dictionary of cloud provider plugin names and 'file'.

    The 'file' entry is included separately since it needs to be consumed directly by the serializer.

    :returns: A dictionary of cloud provider plugin names (as both keys and values) plus the 'file' entry.
    :rtype: dict[str, str]
    """

    plugins = compute_cloud_inventory_sources()

    plugin_description = discover_available_cloud_provider_descriptions()

    plugin_description['scm'] = 'Sourced from a Project'

    plugin_description['file'] = 'File-based inventory source'

    result = {plugin: plugin_description.get(plugin, plugin) for plugin in plugins}

    return result


Review comment (suggestion):
Since this is a plugin we don't want to assume that the `ssh` plugin will be discovered. Change this to something like:

```
if 'scm' in plugin_description:
    plugin_description['scm'] = 'Sourced from a Project'
```

Code after:
    """
    from awx.main.models.inventory import InventorySourceOptions

    plugin_description_list = [(plugin_name, plugin.plugin_description) for plugin_name, plugin in InventorySourceOptions.injectors.items()]

    plugin_description = dict(plugin_description_list)

    return plugin_description


@cache
def load_combined_inventory_source_options() -> dict[str, str]:
    """
    Return a dictionary of cloud provider plugin names and 'file'.

    The 'file' entry is included separately since it needs to be consumed directly by the serializer.

    :returns: A dictionary of cloud provider plugin names (as both keys and values) plus the 'file' entry.
    :rtype: dict[str, str]
    """

    plugins = compute_cloud_inventory_sources()

    plugin_description = discover_available_cloud_provider_descriptions()

    if 'scm' in plugins:
        plugin_description['scm'] = 'Sourced from a Project'

    if 'file' in plugins:
        plugin_description['file'] = 'File-based inventory source'

    result = {plugin: plugin_description.get(plugin, plugin) for plugin in plugins}

    return result
Repository: ansible/awx
File: awx/main/models/inventory.py (Python)

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

# Python
import datetime
import time
import logging
import re
import copy
import os.path
from urllib.parse import urljoin
from importlib.metadata import entry_points

# Django
from django.conf import settings
from django.db import models, connection
from django.utils.translation import gettext_lazy as _
from django.db import transaction
from django.core.exceptions import ValidationError
from django.urls import resolve
from django.utils.timezone import now
from django.db.models import Q

# REST Framework
from rest_framework.exceptions import ParseError

from ansible_base.lib.utils.models import prevent_search
from awx_plugins.interfaces._temporary_private_licensing_api import detect_server_product_name

# AWX
from awx.api.versioning import reverse
from awx.main.utils.plugins import discover_available_cloud_provider_plugin_names, compute_cloud_inventory_sources
from awx.main.consumers import emit_channel_notification
from awx.main.fields import (
    ImplicitRoleField,
    SmartFilterField,
    OrderedManyToManyField,

Review comment (suggestion):
```suggestion
from importlib.metadata import entry_points, EntryPoint
```

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

# Python
import datetime
import time
import logging
import re
import copy
import os.path
from urllib.parse import urljoin

# Django
from django.conf import settings
from django.db import models, connection
from django.utils.translation import gettext_lazy as _
from django.db import transaction
from django.core.exceptions import ValidationError
from django.urls import resolve
from django.utils.timezone import now
from django.db.models import Q

# REST Framework
from rest_framework.exceptions import ParseError

from ansible_base.lib.utils.models import prevent_search

# AWX
from awx.api.versioning import reverse
from awx.main.utils.plugins import discover_available_cloud_provider_plugin_names, compute_cloud_inventory_sources
from awx.main.consumers import emit_channel_notification
from awx.main.fields import (
    ImplicitRoleField,
    SmartFilterField,
    OrderedManyToManyField,
)
from awx.main.managers import HostManager, HostMetricActiveManager
Repository: ansible/awx
File: awx/main/utils/common.py (Python)

Code before:
                raise
            sqlstate_str = psycopg.errors.lookup(sqlstate)
            logger.error('SQL Error state: {} - {}'.format(sqlstate, sqlstate_str))
        raise


def cleanup_new_process(func):
    """
    Cleanup django connection, cache connection, before executing new thread or processes entry point, func.
    """

    @functools.wraps(func)
    def wrapper_cleanup_new_process(*args, **kwargs):
        from awx.conf.settings import SettingsWrapper  # noqa

        django_connection.close()
        django_cache.close()
        SettingsWrapper.initialize()
        return func(*args, **kwargs)

    return wrapper_cleanup_new_process


def log_excess_runtime(func_logger, cutoff=5.0, debug_cutoff=5.0, msg=None, add_log_data=False):
    def log_excess_runtime_decorator(func):
        @functools.wraps(func)
        def _new_func(*args, **kwargs):
            start_time = time.time()
            log_data = {'name': repr(func.__name__)}

            if add_log_data:
                return_value = func(*args, log_data=log_data, **kwargs)
            else:
                return_value = func(*args, **kwargs)

            log_data['delta'] = time.time() - start_time
            if isinstance(return_value, dict):
                log_data.update(return_value)

            if msg is None:
                record_msg = 'Running {name} took {delta:.2f}s'
            else:
                record_msg = msg
            if log_data['delta'] > cutoff:
                func_logger.info(record_msg.format(**log_data))
            elif log_data['delta'] > debug_cutoff:
                func_logger.debug(record_msg.format(**log_data))
            return return_value

        return _new_func


Review comment (suggestion):
Can you explain what the `/` is doing here? Maybe it's a new python thing I'm unaware of, but maybe it's unintended.

Code after:
            logger.error('SQL Error state: {} - {}'.format(sqlstate, sqlstate_str))
        raise


def cleanup_new_process(func):
    """
    Cleanup django connection, cache connection, before executing new thread or processes entry point, func.
    """

    @functools.wraps(func)
    def wrapper_cleanup_new_process(*args, **kwargs):
        from awx.conf.settings import SettingsWrapper  # noqa

        django_connection.close()
        django_cache.close()
        SettingsWrapper.initialize()
        return func(*args, **kwargs)

    return wrapper_cleanup_new_process


def unified_job_class_to_event_table_name(job_class):
    return f'main_{job_class().event_class.__name__.lower()}'


def load_all_entry_points_for(entry_point_subsections: list[str], /) -> dict[str, EntryPoint]:
    return {ep.name: ep for entry_point_category in entry_point_subsections for ep in entry_points(group=f'awx_plugins.{entry_point_category}')}