ronantakizawa/codereview-bench huggingface dataset:
Repository: django/django
File: tests/model_regress/test_state.py (Python)

Code before:
import gc

from django.db.models.base import ModelState, ModelStateFieldsCacheDescriptor
from django.test import SimpleTestCase

from .models import CycleChild, CycleParent


class ModelStateTests(SimpleTestCase):
    def test_fields_cache_descriptor(self):
        self.assertIsInstance(ModelState.fields_cache, ModelStateFieldsCacheDescriptor)

    def test_one_to_one_field_cycle_collection(self):
        """
        Ensure OneToOneField intermediate state does not create reference
        cycles during model initialization.
        """
        self.addCleanup(gc.set_debug, gc.get_debug())
        gc.set_debug(gc.DEBUG_SAVEALL)

        gc.collect()
        del gc.garbage[:]

        p = CycleParent()
        c = CycleChild(parent=p)

        p_id = id(p)

        del p
        del c

        gc.collect()

        leaked = [obj for obj in gc.garbage if id(obj) == p_id]

        self.assertEqual(leaked, [])


Review comment (refactor):
Please use the `garbage_collect()` helper from `django.test.utils`.

Code after:
import gc

from django.db.models.base import ModelState, ModelStateFieldsCacheDescriptor
from django.test import SimpleTestCase
from django.test.utils import garbage_collect

from .models import Worker, WorkerProfile


class ModelStateTests(SimpleTestCase):
    def test_fields_cache_descriptor(self):
        self.assertIsInstance(ModelState.fields_cache, ModelStateFieldsCacheDescriptor)

    def test_one_to_one_field_cycle_collection(self):
        self.addCleanup(gc.set_debug, gc.get_debug())
        gc.set_debug(gc.DEBUG_SAVEALL)

        def clear_garbage():
            del gc.garbage[:]

        self.addCleanup(clear_garbage)

        worker = Worker()
        profile = WorkerProfile(worker=worker)
        worker_id = id(worker)

        del worker
        del profile

        garbage_collect()

        leaked = [obj for obj in gc.garbage if id(obj) == worker_id]
        self.assertEqual(leaked, [])
Repository: django/django
File: tests/model_regress/test_state.py (Python)

Code before:
import gc

from django.db.models.base import ModelState, ModelStateFieldsCacheDescriptor
from django.test import SimpleTestCase

from .models import CycleChild, CycleParent


class ModelStateTests(SimpleTestCase):
    def test_fields_cache_descriptor(self):
        self.assertIsInstance(ModelState.fields_cache, ModelStateFieldsCacheDescriptor)

    def test_one_to_one_field_cycle_collection(self):
        """
        Ensure OneToOneField intermediate state does not create reference
        cycles during model initialization.
        """
        self.addCleanup(gc.set_debug, gc.get_debug())
        gc.set_debug(gc.DEBUG_SAVEALL)

        gc.collect()
        del gc.garbage[:]

        p = CycleParent()
        c = CycleChild(parent=p)

        p_id = id(p)

        del p
        del c

        gc.collect()

        leaked = [obj for obj in gc.garbage if id(obj) == p_id]

        self.assertEqual(leaked, [])


Review comment (suggestion):
```suggestion
```
Thanks for the method name change. Now we can lose the docstring (we avoid "Ensure ..." preambles anyway.)

Code after:
import gc

from django.db.models.base import ModelState, ModelStateFieldsCacheDescriptor
from django.test import SimpleTestCase
from django.test.utils import garbage_collect

from .models import Worker, WorkerProfile


class ModelStateTests(SimpleTestCase):
    def test_fields_cache_descriptor(self):
        self.assertIsInstance(ModelState.fields_cache, ModelStateFieldsCacheDescriptor)

    def test_one_to_one_field_cycle_collection(self):
        self.addCleanup(gc.set_debug, gc.get_debug())
        gc.set_debug(gc.DEBUG_SAVEALL)

        def clear_garbage():
            del gc.garbage[:]

        self.addCleanup(clear_garbage)

        worker = Worker()
        profile = WorkerProfile(worker=worker)
        worker_id = id(worker)

        del worker
        del profile

        garbage_collect()

        leaked = [obj for obj in gc.garbage if id(obj) == worker_id]
        self.assertEqual(leaked, [])
Repository: django/django
File: tests/model_regress/test_state.py (Python)

Code before:

class ModelStateTests(SimpleTestCase):
    def test_fields_cache_descriptor(self):
        self.assertIsInstance(ModelState.fields_cache, ModelStateFieldsCacheDescriptor)

    def test_one_to_one_field_cycle_collection(self):
        """
        Ensure OneToOneField intermediate state does not create reference
        cycles during model initialization.
        """
        self.addCleanup(gc.set_debug, gc.get_debug())
        gc.set_debug(gc.DEBUG_SAVEALL)

        gc.collect()
        del gc.garbage[:]

        p = CycleParent()
        c = CycleChild(parent=p)

        p_id = id(p)

        del p
        del c

        gc.collect()

        leaked = [obj for obj in gc.garbage if id(obj) == p_id]

        self.assertEqual(leaked, [])


Review comment (bug):
After this test I get this warning:
```py
Exception ignored in GC shutdown:
ResourceWarning: gc: 7995 uncollectable objects at shutdown; use gc.set_debug(gc.DEBUG_UNCOLLECTABLE) to list them
```

Code after:


class ModelStateTests(SimpleTestCase):
    def test_fields_cache_descriptor(self):
        self.assertIsInstance(ModelState.fields_cache, ModelStateFieldsCacheDescriptor)

    def test_one_to_one_field_cycle_collection(self):
        self.addCleanup(gc.set_debug, gc.get_debug())
        gc.set_debug(gc.DEBUG_SAVEALL)

        def clear_garbage():
            del gc.garbage[:]

        self.addCleanup(clear_garbage)

        worker = Worker()
        profile = WorkerProfile(worker=worker)
        worker_id = id(worker)

        del worker
        del profile

        garbage_collect()

        leaked = [obj for obj in gc.garbage if id(obj) == worker_id]
        self.assertEqual(leaked, [])
Repository: django/django
File: tests/model_regress/test_state.py (Python)

Code before:
from django.db.models.base import ModelState, ModelStateFieldsCacheDescriptor
from django.test import SimpleTestCase


class ModelStateTests(SimpleTestCase):
    def test_fields_cache_descriptor(self):
        self.assertIsInstance(ModelState.fields_cache, ModelStateFieldsCacheDescriptor)

    def test_model_state_del_clears_cache(self):

        state = ModelState()

        state._fields_cache = {"related_object": "leaked_data"}

        self.assertTrue(hasattr(state, "_fields_cache"))

        state.__del__()

        self.assertFalse(hasattr(state, "_fields_cache"))


Review comment (suggestion):
This test is testing the implementation instead of the underlying issue. It would be better to polish the test you had before.

Code after:
import gc

from django.db.models.base import ModelState, ModelStateFieldsCacheDescriptor
from django.test import SimpleTestCase
from django.test.utils import garbage_collect

from .models import Worker, WorkerProfile


class ModelStateTests(SimpleTestCase):
    def test_fields_cache_descriptor(self):
        self.assertIsInstance(ModelState.fields_cache, ModelStateFieldsCacheDescriptor)

    def test_one_to_one_field_cycle_collection(self):
        self.addCleanup(gc.set_debug, gc.get_debug())
        gc.set_debug(gc.DEBUG_SAVEALL)

        def clear_garbage():
            del gc.garbage[:]

        self.addCleanup(clear_garbage)

        worker = Worker()
        profile = WorkerProfile(worker=worker)
        worker_id = id(worker)

        del worker
        del profile

        garbage_collect()

        leaked = [obj for obj in gc.garbage if id(obj) == worker_id]
        self.assertEqual(leaked, [])
Repository: django/django
File: tests/model_regress/test_state.py (Python)

Code before:
import gc

from django.db.models.base import ModelState, ModelStateFieldsCacheDescriptor
from django.test import SimpleTestCase
from django.test.utils import garbage_collect

from .models import Worker, WorkerProfile


class ModelStateTests(SimpleTestCase):
    def test_fields_cache_descriptor(self):
        self.assertIsInstance(ModelState.fields_cache, ModelStateFieldsCacheDescriptor)

    def test_model_state_del_no_cache(self):
        state = ModelState()
        del state

    def test_one_to_one_field_cycle_collection(self):
        self.addCleanup(gc.set_debug, gc.get_debug())
        gc.set_debug(gc.DEBUG_SAVEALL)

        def clear_garbage():
            del gc.garbage[:]

        self.addCleanup(clear_garbage)

        worker = Worker()
        profile = WorkerProfile(worker=worker)
        worker_id = id(worker)

        del worker
        del profile

        garbage_collect()

        leaked = [obj for obj in gc.garbage if id(obj) == worker_id]
        self.assertEqual(leaked, [])


Review comment (suggestion):
Sorry, I guess this whole test method can go now, now that we're not trying to exercise some separate code path.

Code after:
import gc

from django.db.models.base import ModelState, ModelStateFieldsCacheDescriptor
from django.test import SimpleTestCase
from django.test.utils import garbage_collect

from .models import Worker, WorkerProfile


class ModelStateTests(SimpleTestCase):
    def test_fields_cache_descriptor(self):
        self.assertIsInstance(ModelState.fields_cache, ModelStateFieldsCacheDescriptor)

    def test_one_to_one_field_cycle_collection(self):
        self.addCleanup(gc.set_debug, gc.get_debug())
        gc.set_debug(gc.DEBUG_SAVEALL)

        def clear_garbage():
            del gc.garbage[:]

        self.addCleanup(clear_garbage)

        worker = Worker()
        profile = WorkerProfile(worker=worker)
        worker_id = id(worker)

        del worker
        del profile

        garbage_collect()

        leaked = [obj for obj in gc.garbage if id(obj) == worker_id]
        self.assertEqual(leaked, [])
Repository: django/django
File: tests/migrations/test_autodetector.py (Python)

Code before:
        self.assertNumberMigrations(changes, "app", 1)
        self.assertOperationTypes(changes, "app", 0, ["RemoveField", "AlterField"])
        self.assertOperationAttributes(
            changes,
            "app",
            0,
            0,
            name="pk",
            model_name="foo",
        )
        self.assertOperationAttributes(
            changes,
            "app",
            0,
            1,
            name="id",
            model_name="foo",
            preserve_default=True,
        )

    def test_m2m_target_change_generates_remove_and_add(self):
        # Initial state: Author has M2M to Publisher
        before = [
            self.publisher,
            self.magazine,
            self.author_with_m2m,
        ]
        # After state: Author has M2M to Magazine
        after = [
            self.publisher,
            self.magazine,
            self.author_with_m2m_magazine,
        ]
        changes = self.get_changes(before, after)
        self.assertNumberMigrations(changes, "testapp", 1)
        self.assertOperationTypes(changes, "testapp", 0, ["RemoveField", "AddField"])
        self.assertOperationAttributes(
            changes,
            "testapp",
            0,
            0,
            name="publishers",
            model_name="author",
        )
        self.assertOperationAttributes(
            changes,
            "testapp",
            0,
            1,
            name="publishers",
            model_name="author",

Review comment (security):
```suggestion
        before = [
            self.publisher,
            self.other_publisher,
            self.author_with_m2m,  # m2m to self.publisher.
```

Code after:
            name="pk",
            model_name="foo",
            preserve_default=True,
        )
        self.assertOperationAttributes(
            changes,
            "app",
            0,
            1,
            name="id",
            model_name="foo",
            preserve_default=True,
        )

    def test_remove_composite_pk(self):
        before = [
            ModelState(
                "app",
                "foo",
                [
                    ("pk", models.CompositePrimaryKey("foo_id", "bar_id")),
                    ("id", models.IntegerField()),
                ],
            ),
        ]
        after = [
            ModelState(
                "app",
                "foo",
                [
                    ("id", models.AutoField(primary_key=True)),
                ],
            ),
        ]

        changes = self.get_changes(before, after)
        self.assertNumberMigrations(changes, "app", 1)
        self.assertOperationTypes(changes, "app", 0, ["RemoveField", "AlterField"])
        self.assertOperationAttributes(
            changes,
            "app",
            0,
            0,
            name="pk",
            model_name="foo",
        )
        self.assertOperationAttributes(
            changes,
            "app",
            0,
            1,
Repository: django/django
File: tests/serializers/test_deserialization.py (Python)

Code before:
        deserializer = YamlDeserializer(stream_or_string=stream)

        first_item = next(deserializer)
        second_item = next(deserializer)

        self.assertEqual(first_item.object, self.jane)
        self.assertEqual(second_item.object, self.joe)

    def test_crafted_xml_performance(self):
        """The time to process invalid inputs is not quadratic."""

        def build_crafted_xml(depth, leaf_text_len):
            nested_open = "<nested>" * depth
            nested_close = "</nested>" * depth
            leaf = "x" * leaf_text_len
            field_content = f"{nested_open}{leaf}{nested_close}"
            return f"""
                <django-objects version="1.0">
                   <object model="contenttypes.contenttype" pk="1">
                      <field name="app_label">{field_content}</field>
                      <field name="model">m</field>
                   </object>
                </django-objects>
            """

        crafted_xml = build_crafted_xml(100, 1000)
        with self.assertRaises(SuspiciousOperation):
            list(XMLDeserializer(crafted_xml))


Review comment (refactor):
`build_crafted_xml` seems redundant as a function if it only gets called once. I'd simplify it away.

Code after:
        stream = test_string.encode("utf-8")
        deserializer = YamlDeserializer(stream_or_string=stream)

        first_item = next(deserializer)
        second_item = next(deserializer)

        self.assertEqual(first_item.object, self.jane)
        self.assertEqual(second_item.object, self.joe)

    def test_crafted_xml_rejected(self):
        depth = 100
        leaf_text_len = 1000
        nested_open = "<nested>" * depth
        nested_close = "</nested>" * depth
        leaf = "x" * leaf_text_len
        field_content = f"{nested_open}{leaf}{nested_close}"
        crafted_xml = textwrap.dedent(
            f"""
        <django-objects version="1.0">
            <object model="contenttypes.contenttype" pk="1">
                <field name="app_label">{field_content}</field>
                <field name="model">m</field>
            </object>
        </django-objects>"""
        )

        msg = "Unexpected element: 'nested'"
        with self.assertRaisesMessage(SuspiciousOperation, msg):
            list(XMLDeserializer(crafted_xml))
Repository: django/django
File: tests/backends/sqlite/test_features.py (Python)

Code before:
            "cursor",
            side_effect=OperationalError(msg),
        ):
            with self.assertRaisesMessage(OperationalError, msg):
                connection.features.supports_json_field

    def test_max_query_params_respects_variable_limit(self):
        limit_name = sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER
        current_limit = connection.features.max_query_params
        new_limit = min(42, current_limit)
        try:
            connection.connection.setlimit(limit_name, new_limit)
            self.assertEqual(connection.features.max_query_params, new_limit)
        finally:
            connection.connection.setlimit(limit_name, current_limit)
        self.assertEqual(connection.features.max_query_params, current_limit)

    def test_max_query_params_without_established_connection(self):
        new_connection = connection.copy()
        new_connection.settings_dict = copy.deepcopy(connection.settings_dict)
        self.assertIsNone(new_connection.connection)
        try:
            result = new_connection.features.max_query_params
            self.assertIsInstance(result, int)
        finally:
            new_connection.close()


Review comment (suggestion):
I'm getting the following, when using an in-memory database:

```py
ResourceWarning: unclosed database in <sqlite3.Connection object at 0x10c710f40>
```
It goes away with:
```suggestion
            new_connection._close()
```

Code after:
            "cursor",
            side_effect=OperationalError(msg),
        ):
            with self.assertRaisesMessage(OperationalError, msg):
                connection.features.supports_json_field

    def test_max_query_params_respects_variable_limit(self):
        limit_name = sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER
        current_limit = connection.features.max_query_params
        new_limit = min(42, current_limit)
        try:
            connection.connection.setlimit(limit_name, new_limit)
            self.assertEqual(connection.features.max_query_params, new_limit)
        finally:
            connection.connection.setlimit(limit_name, current_limit)
        self.assertEqual(connection.features.max_query_params, current_limit)

    def test_max_query_params_without_established_connection(self):
        new_connection = connection.copy()
        new_connection.settings_dict = copy.deepcopy(connection.settings_dict)
        self.assertIsNone(new_connection.connection)
        try:
            result = new_connection.features.max_query_params
            self.assertIsInstance(result, int)
        finally:
            new_connection._close()
Repository: django/django
File: tests/get_earliest_or_latest/tests.py (Python)

Code before:
        qs = Comment.objects.values("article").annotate(avg_likes=Avg("likes_count"))
        msg = (
            "Cannot use QuerySet.%s() on an unordered queryset performing aggregation. "
            "Add an ordering with order_by()."
        )
        with self.assertRaisesMessage(TypeError, msg % "first"):
            qs.first()
        with self.assertRaisesMessage(TypeError, msg % "last"):
            qs.last()

    def test_first_last_empty_order_by_no_pk_ordering(self):
        a1 = Article.objects.create(
            headline="Article 1",
            pub_date=datetime(2006, 9, 10),
            expire_date=datetime(2056, 9, 11),
        )
        a2 = Article.objects.create(
            headline="Article 2",
            pub_date=datetime(2007, 9, 10),
            expire_date=datetime(2057, 9, 11),
        )

        # order_by() without arguments followed by first()/last() returns an
        # arbitrary item from the unordered queryset.
        self.assertIn(Article.objects.order_by().first(), [a1, a2])
        self.assertIn(Article.objects.order_by().last(), [a1, a2])


Review comment (bug):
This won't fail if the code change is undone; we'll need something else, e.g. mocking `order_by()` and calling `assert_not_called`.

Code after:

        qs = Comment.objects.values("article").annotate(avg_likes=Avg("likes_count"))
        msg = (
            "Cannot use QuerySet.%s() on an unordered queryset performing aggregation. "
            "Add an ordering with order_by()."
        )
        with self.assertRaisesMessage(TypeError, msg % "first"):
            qs.first()
        with self.assertRaisesMessage(TypeError, msg % "last"):
            qs.last()

    def test_first_last_empty_order_by_has_no_pk_ordering(self):
        Article.objects.create(
            headline="Article 1",
            pub_date=datetime(2006, 9, 10),
            expire_date=datetime(2056, 9, 11),
        )

        qs = Article.objects.order_by()
        with patch.object(type(qs), "order_by") as mock_order_by:
            qs.first()
            mock_order_by.assert_not_called()
            qs.last()
            mock_order_by.assert_not_called()

    def test_first_last_empty_order_by_clears_default_ordering(self):
        OrderedArticle.objects.create(
            headline="Article 1",
            pub_date=datetime(2006, 9, 10),
        )

        qs = OrderedArticle.objects.order_by()
        with patch.object(type(qs), "order_by") as mock_order_by:
            qs.first()
            mock_order_by.assert_not_called()
            qs.last()
            mock_order_by.assert_not_called()
Repository: django/django
File: django/db/backends/postgresql/features.py (Python)

Code before:
import operator

from django.db import DataError, InterfaceError
from django.db.backends.base.features import BaseDatabaseFeatures
from django.db.backends.postgresql.psycopg_any import is_psycopg3
from django.utils.functional import cached_property


class DatabaseFeatures(BaseDatabaseFeatures):
    minimum_database_version = (15,)
    max_query_params = 2**16 - 1
    allows_group_by_selected_pks = True
    can_return_columns_from_insert = True
    can_return_rows_from_bulk_insert = True
    can_return_rows_from_update = True
    has_real_datatype = True
    has_native_uuid_field = True
    has_native_duration_field = True
    has_native_json_field = True
    can_defer_constraint_checks = True
    has_select_for_update = True
    has_select_for_update_nowait = True
    has_select_for_update_of = True
    has_select_for_update_skip_locked = True
    has_select_for_no_key_update = True
    can_release_savepoints = True
    supports_comments = True
    supports_tablespaces = True
    supports_transactions = True
    can_introspect_materialized_views = True
    can_distinct_on_fields = True
    can_rollback_ddl = True
    schema_editor_uses_clientside_param_binding = True
    supports_combined_alters = True
    nulls_order_largest = True
    closed_cursor_error_class = InterfaceError

Review comment (performance):
Since the limit is only present when server-side cursors are used we should make this value a `cached_property`  that introspects `self.connection.settings_dict["OPTIONS"].get("server_side_binding")` and return this value if its enabled and `None` otherwise.

Code after:
import operator

from django.db import DataError, InterfaceError
from django.db.backends.base.features import BaseDatabaseFeatures
from django.db.backends.postgresql.psycopg_any import is_psycopg3
from django.utils.functional import cached_property


class DatabaseFeatures(BaseDatabaseFeatures):
    minimum_database_version = (15,)
    allows_group_by_selected_pks = True
    can_return_columns_from_insert = True
    can_return_rows_from_bulk_insert = True
    can_return_rows_from_update = True
    has_real_datatype = True
    has_native_uuid_field = True
    has_native_duration_field = True
    has_native_json_field = True
    can_defer_constraint_checks = True
    has_select_for_update = True
    has_select_for_update_nowait = True
    has_select_for_update_of = True
    has_select_for_update_skip_locked = True
    has_select_for_no_key_update = True
    can_release_savepoints = True
    supports_comments = True
    supports_tablespaces = True
    supports_transactions = True
    can_introspect_materialized_views = True
    can_distinct_on_fields = True
    can_rollback_ddl = True
    schema_editor_uses_clientside_param_binding = True
    supports_combined_alters = True
    nulls_order_largest = True
    closed_cursor_error_class = InterfaceError
    greatest_least_ignores_nulls = True