ronantakizawa/codereview-bench huggingface dataset:
Repository: django/django
File: tests/foreign_object/models/customers.py (Python)

Code before:
from django.db import models
from django.db.models import Q


class Address(models.Model):
    company = models.CharField(max_length=1)
    customer_id = models.IntegerField()

    class Meta:
        unique_together = [
            ("company", "customer_id"),
        ]


class Customer(models.Model):
    company = models.CharField(max_length=1)
    customer_id = models.IntegerField()
    address = models.ForeignObject(
        Address,
        models.CASCADE,
        null=True,
        # order mismatches the Contact ForeignObject.
        from_fields=["company", "customer_id"],
        to_fields=["company", "customer_id"],
    )

    class Meta:

Review comment (suggestion):
The peer files seem to just use `models.Q` instead of a discrete import.

Code after:
from django.db import models


class Address(models.Model):
    company = models.CharField(max_length=1)
    customer_id = models.IntegerField()

    class Meta:
        unique_together = [
            ("company", "customer_id"),
        ]


class Customer(models.Model):
    company = models.CharField(max_length=1)
    customer_id = models.IntegerField()
    address = models.ForeignObject(
        Address,
        models.CASCADE,
        null=True,
        # order mismatches the Contact ForeignObject.
        from_fields=["company", "customer_id"],
        to_fields=["company", "customer_id"],
    )

    class Meta:
        unique_together = [
Repository: django/django
File: tests/foreign_object/models/customers.py (Python)

Code before:
    )

    class Meta:
        unique_together = [
            ("company", "customer_id"),
        ]


class Contact(models.Model):
    company_code = models.CharField(max_length=1)
    customer_code = models.IntegerField()
    customer = models.ForeignObject(
        Customer,
        models.CASCADE,
        related_name="contacts",
        to_fields=["customer_id", "company"],
        from_fields=["customer_code", "company_code"],
    )


class Patron(models.Model):
    pass


class PatronTab(models.Model):
    patron_id = models.IntegerField(null=True)
    patron = models.ForeignObject(
        Patron,
        from_fields=["patron_id"],
        to_fields=["id"],
        on_delete=models.CASCADE,
        null=True,
    )

    class Meta:
        required_db_features = {"supports_table_check_constraints"}
        constraints = [
            models.CheckConstraint(
                condition=Q(patron__isnull=False),
                name="patron_not_null",
            ),
        ]


Review comment (suggestion):
The on_delete arg is ignored by `ForeignObject`; let's just remove.
```suggestion
```

Code after:

    class Meta:
        unique_together = [
            ("company", "customer_id"),
        ]


class Contact(models.Model):
    company_code = models.CharField(max_length=1)
    customer_code = models.IntegerField()
    customer = models.ForeignObject(
        Customer,
        models.CASCADE,
        related_name="contacts",
        to_fields=["customer_id", "company"],
        from_fields=["customer_code", "company_code"],
    )


class CustomerTab(models.Model):
    customer_id = models.IntegerField()
    customer = models.ForeignObject(
        Customer,
        from_fields=["customer_id"],
        to_fields=["id"],
        on_delete=models.CASCADE,
    )

    class Meta:
        required_db_features = {"supports_table_check_constraints"}
        constraints = [
            models.CheckConstraint(
                condition=models.Q(customer__lt=1000),
                name="customer_id_limit",
            ),
        ]
Repository: django/django
File: tests/foreign_object/tests.py (Python)

Code before:
        self.assertIn("reverse_path_infos", foreign_object_restored.__dict__)


class ForeignObjectModelValidationTests(TestCase):
    @skipUnlessDBFeature("supports_table_check_constraints")
    def test_validate_constraints_with_foreign_object(self):
        customer_tab = CustomerTab(customer_id=1500)
        with self.assertRaisesMessage(ValidationError, "customer_id_limit"):
            customer_tab.validate_constraints()

    @skipUnlessDBFeature("supports_table_check_constraints")
    def test_validate_constraints_success_case_single_query(self):
        customer_tab = CustomerTab(customer_id=500)
        with CaptureQueriesContext(connection) as ctx:
            customer_tab.validate_constraints()
        select_queries = [
            query["sql"]
            for query in ctx.captured_queries
            if "select" in query["sql"].lower()
        ]
        self.assertEqual(len(select_queries), 1)

    @skipUnlessDBFeature("supports_table_check_constraints")
    def test_validate_constraints_excluding_foreign_object(self):
        customer_tab = CustomerTab(customer_id=150)
        customer_tab.validate_constraints(exclude={"customer"})


Review comment (suggestion):
If a composite member of the _virtual_ field is specified the validation should also be skipped. That is if `exclude={"customuser_id"}` the validation for `customuser` should also be skipped.

Code after:
        self.assertIn("reverse_path_infos", foreign_object_restored.__dict__)


class ForeignObjectModelValidationTests(TestCase):
    @skipUnlessDBFeature("supports_table_check_constraints")
    def test_validate_constraints_with_foreign_object(self):
        customer_tab = CustomerTab(customer_id=1500)
        with self.assertRaisesMessage(ValidationError, "customer_id_limit"):
            customer_tab.validate_constraints()

    @skipUnlessDBFeature("supports_table_check_constraints")
    def test_validate_constraints_success_case_single_query(self):
        customer_tab = CustomerTab(customer_id=500)
        with CaptureQueriesContext(connection) as ctx:
            customer_tab.validate_constraints()
        select_queries = [
            query["sql"]
            for query in ctx.captured_queries
            if "select" in query["sql"].lower()
        ]
        self.assertEqual(len(select_queries), 1)

    @skipUnlessDBFeature("supports_table_check_constraints")
    def test_validate_constraints_excluding_foreign_object(self):
        customer_tab = CustomerTab(customer_id=150)
        customer_tab.validate_constraints(exclude={"customer"})

    @skipUnlessDBFeature("supports_table_check_constraints")
    def test_validate_constraints_excluding_foreign_object_member(self):
        customer_tab = CustomerTab(customer_id=150)
        customer_tab.validate_constraints(exclude={"customer_id"})
Repository: django/django
File: tests/test_runner/test_parallel.py (Python)

Code before:
        test = TestCase()
        event = ("addSuccess", 0)
        pts.handle_event(result, tests=[test], event=event)

        self.assertEqual(len(result.errors), 0)
        self.assertEqual(len(result.failures), 0)


class ParallelBufferTest(DjangoTestCase):
    def test_parallel_buffer(self):
        """
        Test that ParallelTestSuite.run() properly handles buffered TestResult
        to prevent AttributeError when setUpTestData raises exceptions.
        Before the fix, running tests with
        --parallel --buffer would crash with AttributeError when setUpTestData
        raised an exception because the main process TestResult had buffer=True
        but _setupStdout() was never called.
        """

        class AbortingTest(DjangoTestCase):
            @classmethod
            def setUpTestData(cls):
                raise RuntimeError("Boo!")

            def test_pass(self):
                pass

        test = AbortingTest("test_pass")
        remote_result = RemoteTestResult()
        suite = TestSuite([test])
        suite.run(remote_result)

        pts = ParallelTestSuite([suite], processes=2, buffer=True)
        pts.serialized_aliases = set()
        test_result = TestResult()
        test_result.buffer = True

        with unittest.mock.patch("multiprocessing.Pool") as mock_pool:

            def fake_next(*args, **kwargs):
                test_result.shouldStop = True
                return (0, remote_result.events)

            mock_pool.return_value.imap_unordered.return_value = unittest.mock.Mock(
                next=fake_next
            )
            pts.run(test_result)


Review comment (suggestion):
Is there any way to test without `mock.patch()`? I don't understand why you'd need to trigger `shouldStop`.

Code after:
        event = ("addSuccess", 0)
        pts.handle_event(result, tests=[test], event=event)

        self.assertEqual(len(result.errors), 0)
        self.assertEqual(len(result.failures), 0)

    def test_buffer_mode_reports_setupclass_failure(self):
        test = SampleErrorTest("dummy_test")
        remote_result = RemoteTestResult()
        suite = TestSuite([test])
        suite.run(remote_result)

        pts = ParallelTestSuite([suite], processes=2, buffer=True)
        pts.serialized_aliases = set()
        test_result = TestResult()
        test_result.buffer = True

        with unittest.mock.patch("multiprocessing.Pool") as mock_pool:

            def fake_next(*args, **kwargs):
                test_result.shouldStop = True
                return (0, remote_result.events)

            mock_pool.return_value.imap_unordered.return_value = unittest.mock.Mock(
                next=fake_next
            )
            pts.run(test_result)

        self.assertIn("ValueError: woops", test_result.errors[0][1])
Repository: django/django
File: tests/test_runner/test_parallel.py (Python)

Code before:
        event = ("addSuccess", 0)
        pts.handle_event(result, tests=[test], event=event)

        self.assertEqual(len(result.errors), 0)
        self.assertEqual(len(result.failures), 0)


class ParallelBufferTest(DjangoTestCase):
    def test_parallel_buffer(self):
        """
        Test that ParallelTestSuite.run() properly handles buffered TestResult
        to prevent AttributeError when setUpTestData raises exceptions.
        Before the fix, running tests with
        --parallel --buffer would crash with AttributeError when setUpTestData
        raised an exception because the main process TestResult had buffer=True
        but _setupStdout() was never called.
        """

        class AbortingTest(DjangoTestCase):
            @classmethod
            def setUpTestData(cls):
                raise RuntimeError("Boo!")

            def test_pass(self):
                pass

        test = AbortingTest("test_pass")
        remote_result = RemoteTestResult()
        suite = TestSuite([test])
        suite.run(remote_result)

        pts = ParallelTestSuite([suite], processes=2, buffer=True)
        pts.serialized_aliases = set()
        test_result = TestResult()
        test_result.buffer = True

        with unittest.mock.patch("multiprocessing.Pool") as mock_pool:

            def fake_next(*args, **kwargs):
                test_result.shouldStop = True
                return (0, remote_result.events)

            mock_pool.return_value.imap_unordered.return_value = unittest.mock.Mock(
                next=fake_next
            )
            pts.run(test_result)


Review comment (suggestion):
Often when a test makes no clear assertions, we add a comment, like:

```suggestion
            pts.run(test_result)  # Does not raise AttributeError.
```

Code after:
        pts.handle_event(result, tests=[test], event=event)

        self.assertEqual(len(result.errors), 0)
        self.assertEqual(len(result.failures), 0)

    def test_buffer_mode_reports_setupclass_failure(self):
        test = SampleErrorTest("dummy_test")
        remote_result = RemoteTestResult()
        suite = TestSuite([test])
        suite.run(remote_result)

        pts = ParallelTestSuite([suite], processes=2, buffer=True)
        pts.serialized_aliases = set()
        test_result = TestResult()
        test_result.buffer = True

        with unittest.mock.patch("multiprocessing.Pool") as mock_pool:

            def fake_next(*args, **kwargs):
                test_result.shouldStop = True
                return (0, remote_result.events)

            mock_pool.return_value.imap_unordered.return_value = unittest.mock.Mock(
                next=fake_next
            )
            pts.run(test_result)

        self.assertIn("ValueError: woops", test_result.errors[0][1])
Repository: django/django
File: django/conf/locale/ht/formats.py (Python)

Code before:
    "%b %d, %Y",  # 'Oct 25, 2006'
    "%d %b %Y",  # '25 Oct 2006'
    "%d %b, %Y",  # '25 Oct, 2006'
    "%B %d %Y",  # 'October 25 2006'
    "%B %d, %Y",  # 'October 25, 2006'
    "%d %B %Y",  # '25 October 2006'
    "%d %B, %Y",  # '25 October, 2006'
]
DATETIME_INPUT_FORMATS = [
    "%Y-%m-%d %H:%M:%S",  # '2006-10-25 14:30:59'
    "%Y-%m-%d %H:%M:%S.%f",  # '2006-10-25 14:30:59.000200'
    "%Y-%m-%d %H:%M",  # '2006-10-25 14:30'
    "%m/%d/%Y %H:%M:%S",  # '10/25/2006 14:30:59'
    "%m/%d/%Y %H:%M:%S.%f",  # '10/25/2006 14:30:59.000200'
    "%m/%d/%Y %H:%M",  # '10/25/2006 14:30'
    "%m/%d/%y %H:%M:%S",  # '10/25/06 14:30:59'
    "%m/%d/%y %H:%M:%S.%f",  # '10/25/06 14:30:59.000200'
    "%m/%d/%y %H:%M",  # '10/25/06 14:30'
]
TIME_INPUT_FORMATS = [
    "%H:%M:%S",  # '14:30:59'
    "%H:%M:%S.%f",  # '14:30:59.000200'
    "%H:%M",  # '14:30'
]

DECIMAL_SEPARATOR = "."
THOUSAND_SEPARATOR = ","
NUMBER_GROUPING = 3


Review comment (suggestion):
The decimal separator in Haiti is a comma. I looked it up to check my memory. https://www.localeplanet.com/icu/fr-HT/index.html Interestingly, a lot of "comprehensive" lists of countries that use the comma exclude Haiti. 

Code after:
    "%b %d, %Y",  # 'Oct 25, 2006'
    "%d %b %Y",  # '25 Oct 2006'
    "%d %b, %Y",  # '25 Oct, 2006'
    "%B %d %Y",  # 'October 25 2006'
    "%B %d, %Y",  # 'October 25, 2006'
    "%d %B %Y",  # '25 October 2006'
    "%d %B, %Y",  # '25 October, 2006'
]
DATETIME_INPUT_FORMATS = [
    "%Y-%m-%d %H:%M:%S",  # '2006-10-25 14:30:59'
    "%Y-%m-%d %H:%M:%S.%f",  # '2006-10-25 14:30:59.000200'
    "%Y-%m-%d %H:%M",  # '2006-10-25 14:30'
    "%m/%d/%Y %H:%M:%S",  # '10/25/2006 14:30:59'
    "%m/%d/%Y %H:%M:%S.%f",  # '10/25/2006 14:30:59.000200'
    "%m/%d/%Y %H:%M",  # '10/25/2006 14:30'
    "%m/%d/%y %H:%M:%S",  # '10/25/06 14:30:59'
    "%m/%d/%y %H:%M:%S.%f",  # '10/25/06 14:30:59.000200'
    "%m/%d/%y %H:%M",  # '10/25/06 14:30'
]
TIME_INPUT_FORMATS = [
    "%H:%M:%S",  # '14:30:59'
    "%H:%M:%S.%f",  # '14:30:59.000200'
    "%H:%M",  # '14:30'
]

DECIMAL_SEPARATOR = ","
THOUSAND_SEPARATOR = "\xa0"
NUMBER_GROUPING = 3
Repository: django/django
File: tests/queries/test_db_returning.py (Python)

Code before:
    def test_insert_returning_multiple(self):
        with CaptureQueriesContext(connection) as captured_queries:
            obj = ReturningModel.objects.create()
        table_name = connection.ops.quote_name(ReturningModel._meta.db_table)
        self.assertIn(
            "RETURNING %s.%s, %s.%s"
            % (
                table_name,
                connection.ops.quote_name(ReturningModel._meta.get_field("id").column),
                table_name,
                connection.ops.quote_name(
                    ReturningModel._meta.get_field("created").column
                ),
            ),
            captured_queries[-1]["sql"],
        )
        self.assertTrue(obj.pk)
        self.assertIsInstance(obj.created, datetime.datetime)

    @skipUnlessDBFeature("can_return_rows_from_bulk_insert")
    def test_bulk_insert(self):
        objs = [ReturningModel(), ReturningModel(pk=2**11), ReturningModel()]
        ReturningModel.objects.bulk_create(objs)
        for obj in objs:
            with self.subTest(obj=obj):
                self.assertTrue(obj.pk)
                self.assertIsInstance(obj.created, datetime.datetime)


Review comment (suggestion):
@timgraham if this is an issue for MongoDB we could use `connection.ops.returning_columns([..., ReturningModel._meta.get_field("created")])` instead.

Code after:
    def test_insert_returning_multiple(self):
        with CaptureQueriesContext(connection) as captured_queries:
            obj = ReturningModel.objects.create()
        table_name = connection.ops.quote_name(ReturningModel._meta.db_table)
        self.assertIn(
            "RETURNING %s.%s, %s.%s"
            % (
                table_name,
                connection.ops.quote_name(ReturningModel._meta.get_field("id").column),
                table_name,
                connection.ops.quote_name(
                    ReturningModel._meta.get_field("created").column
                ),
            ),
            captured_queries[-1]["sql"],
        )
        self.assertEqual(
            captured_queries[-1]["sql"]
            .split("RETURNING ")[1]
            .count(
                connection.ops.quote_name(
                    ReturningModel._meta.get_field("created").column
                ),
            ),
            1,
        )
        self.assertTrue(obj.pk)
        self.assertIsInstance(obj.created, datetime.datetime)

    @skipUnlessDBFeature("can_return_rows_from_bulk_insert")
    def test_bulk_insert(self):
        objs = [ReturningModel(), ReturningModel(pk=2**11), ReturningModel()]
        ReturningModel.objects.bulk_create(objs)
        for obj in objs:
            with self.subTest(obj=obj):
                self.assertTrue(obj.pk)
                self.assertIsInstance(obj.created, datetime.datetime)
Repository: django/django
File: tests/serializers/models/natural.py (Python)

Code before:
        def get_by_natural_key(self, name):
            return self.get(name=name)

    objects = Manager()

    def natural_key(self):
        return (self.name,)


class FKAsPKNoNaturalKeyManager(models.Manager):
    def get_by_natural_key(self, *args, **kwargs):
        return super().get_by_natural_key(*args, **kwargs)


class FKAsPKNoNaturalKey(models.Model):
    pk_fk = models.ForeignKey(
        NaturalKeyAnchor, on_delete=models.CASCADE, primary_key=True
    )

    objects = FKAsPKNoNaturalKeyManager()

    def natural_key(self):
        raise NotImplementedError("This method was not expected to be called.")


class PKTupleOptOutUser(AbstractBaseUser):
    email = models.EmailField(unique=False, null=True, blank=True)
    USERNAME_FIELD = "email"
    REQUIRED_FIELDS = []

    class Manager(models.Manager):
        def get_by_natural_key(self, pk_value):
            return self.get(pk=pk_value)

    objects = Manager()

    def natural_key(self):
        return (self.pk,)


class NoneOptOutUser(AbstractBaseUser):
    email = models.EmailField(unique=False, null=True, blank=True)
    USERNAME_FIELD = "email"
    REQUIRED_FIELDS = []

    class Manager(models.Manager):
        def get_by_natural_key(self, pk_value):
            return self.get(pk=pk_value)

    objects = Manager()


Review comment (suggestion):
Let's replace these models with a single one using a 0-tuple. Suggest renaming the model to mention subclass, not the tuple detail.

Code after:
        def get_by_natural_key(self, name):
            return self.get(name=name)

    objects = Manager()

    def natural_key(self):
        return (self.name,)


class FKAsPKNoNaturalKeyManager(models.Manager):
    def get_by_natural_key(self, *args, **kwargs):
        return super().get_by_natural_key(*args, **kwargs)


class FKAsPKNoNaturalKey(models.Model):
    pk_fk = models.ForeignKey(
        NaturalKeyAnchor, on_delete=models.CASCADE, primary_key=True
    )

    objects = FKAsPKNoNaturalKeyManager()

    def natural_key(self):
        raise NotImplementedError("This method was not expected to be called.")


class SubclassNaturalKeyOptOutUser(AbstractBaseUser):
    email = models.EmailField(unique=False, null=True, blank=True)
    USERNAME_FIELD = "email"

    def natural_key(self):
        return ()


class PostToOptOutSubclassUser(models.Model):
    author = models.ForeignKey(SubclassNaturalKeyOptOutUser, on_delete=models.CASCADE)
    title = models.CharField(max_length=100)
    subscribers = models.ManyToManyField(
        SubclassNaturalKeyOptOutUser, related_name="subscribed_posts", blank=True
    )
Repository: django/django
File: tests/serializers/test_natural.py (Python)

Code before:
            )

        post_data_str = serializers.serialize(
            format,
            [post],
            indent=2,
            use_natural_foreign_keys=True,
        )

        if is_json:
            post_data = json.loads(post_data_str)
            serialized_post = post_data[0]

            self.assertEqual(
                serialized_post["fields"]["author"],
                user_pk,
                f"[{user.__class__.__name__}] FK must be serialized as PK integer"
                f", not a natural key list.",
            )

        try:
            list(serializers.deserialize(format, user_data_str))
        except Exception as e:
            self.fail(
                f"Deserialization of {user.__class__.__name__} failed in {format}: {e}"
            )

        try:
            list(serializers.deserialize(format, post_data_str))
        except Exception as e:
            self.fail(
                f"Deserialization of {post.__class__.__name__} failed in {format}: {e}"
            )


# Dynamically register tests for each serializer
register_tests(
    NaturalKeySerializerTests,
    "test_%s_natural_key_serializer",
    natural_key_serializer_test,
)
register_tests(
    NaturalKeySerializerTests, "test_%s_serializer_natural_keys", natural_key_test
)
register_tests(
    NaturalKeySerializerTests, "test_%s_serializer_natural_pks_mti", natural_pk_mti_test
)
register_tests(
    NaturalKeySerializerTests, "test_%s_forward_references_fks", forward_ref_fk_test
)
register_tests(

Review comment (bug):
We let exceptions escape naturally instead of `self.fail`ing. If you need additional messaging, use the last argument to `self.assert...`

Code after:
    NaturalKeySerializerTests, "test_%s_forward_references_fks", forward_ref_fk_test
)
register_tests(
    NaturalKeySerializerTests,
    "test_%s_forward_references_fk_errors",
    forward_ref_fk_with_error_test,
)
register_tests(
    NaturalKeySerializerTests, "test_%s_forward_references_m2ms", forward_ref_m2m_test
)
register_tests(
    NaturalKeySerializerTests,
    "test_%s_forward_references_m2m_errors",
    forward_ref_m2m_with_error_test,
)
register_tests(NaturalKeySerializerTests, "test_%s_pk_with_default", pk_with_default)
register_tests(
    NaturalKeySerializerTests,
    "test_%s_fk_as_pk_natural_key_not_called",
    fk_as_pk_natural_key_not_called,
)
register_tests(
    NaturalKeySerializerTests,
    "test_%s_natural_key_opt_out",
    natural_key_opt_out_test,
)