# Path: tests/testapp/models.py
# class AgedCustomer(Customer):
#     age = IntegerField(default=21)
# 
# class Alphabet(Model):
#     a = IntegerField(default=1)
#     b = IntegerField(default=2)
#     c = IntegerField(default=3)
#     d = CharField(max_length=32)
#     e = CharField(max_length=32, null=True)
#     f = IntegerField(null=True)
#     g = DecimalField(default=0, decimal_places=2, max_digits=10)
# 
# class Customer(Model):
#     name = CharField(max_length=32)
# 
# class ProxyAlphabet(Alphabet):
#     class Meta:
#         proxy = True
# 
#     @property
#     def a_times_b(self):
#         return self.a * self.b
# 
# class TitledAgedCustomer(AgedCustomer):
#     class Meta:
#         db_table = "titled aged customer"  # Test name quoting
# 
#     title = CharField(max_length=16)


import queue
import pytest
from threading import Thread
from typing import TYPE_CHECKING
from django.db import OperationalError, connection, connections
from django.db.transaction import TransactionManagementError, atomic
from django.test import TestCase, TransactionTestCase
from django_mysql.exceptions import TimeoutError
from django_mysql.locks import Lock, TableLock
from django_mysql.models import Model
from django_mysql.utils import connection_is_mariadb
from tests.testapp.models import (
    AgedCustomer,
    Alphabet,
    Customer,
    ProxyAlphabet,
    TitledAgedCustomer,
)
        conn = connections[connection_name]
        with conn.cursor() as cursor:
            cursor.execute(
                "SHOW OPEN TABLES FROM {} LIKE %s".format(conn.settings_dict["NAME"]),
                [table_name],
            )
            rows = cursor.fetchall()
            if rows:
                assert len(rows) == 1
                return rows[0][2] > 0
            else:
                # MySQL 8+ closes the table really quickly. If it's closed,
                # it's not locked.
                return False

    def test_write(self):
        Alphabet.objects.create(a=12345)
        assert not self.is_locked("default", Alphabet._meta.db_table)

        with TableLock(write=[Alphabet]):
            assert self.is_locked("default", Alphabet._meta.db_table)
            assert Alphabet.objects.count() == 1

            Alphabet.objects.all().delete()
            assert Alphabet.objects.count() == 0

        assert not self.is_locked("default", Alphabet._meta.db_table)
        assert Alphabet.objects.count() == 0

    def test_write_with_table_name(self):

# Path: tests/testapp/models.py
# class AgedCustomer(Customer):
#     age = IntegerField(default=21)
# 
# class Alphabet(Model):
#     a = IntegerField(default=1)
#     b = IntegerField(default=2)
#     c = IntegerField(default=3)
#     d = CharField(max_length=32)
#     e = CharField(max_length=32, null=True)
#     f = IntegerField(null=True)
#     g = DecimalField(default=0, decimal_places=2, max_digits=10)
# 
# class Customer(Model):
#     name = CharField(max_length=32)
# 
# class ProxyAlphabet(Alphabet):
#     class Meta:
#         proxy = True
# 
#     @property
#     def a_times_b(self):
#         return self.a * self.b
# 
# class TitledAgedCustomer(AgedCustomer):
#     class Meta:
#         db_table = "titled aged customer"  # Test name quoting
# 
#     title = CharField(max_length=16)


# Path: tests/testapp/test_locks.py
import queue
import pytest
from threading import Thread
from typing import TYPE_CHECKING
from django.db import OperationalError, connection, connections
from django.db.transaction import TransactionManagementError, atomic
from django.test import TestCase, TransactionTestCase
from django_mysql.exceptions import TimeoutError
from django_mysql.locks import Lock, TableLock
from django_mysql.models import Model
from django_mysql.utils import connection_is_mariadb
from tests.testapp.models import (
    AgedCustomer,
    Alphabet,
    Customer,
    ProxyAlphabet,
    TitledAgedCustomer,
)

        conn = connections[connection_name]
        with conn.cursor() as cursor:
            cursor.execute(
                "SHOW OPEN TABLES FROM {} LIKE %s".format(conn.settings_dict["NAME"]),
                [table_name],
            )
            rows = cursor.fetchall()
            if rows:
                assert len(rows) == 1
                return rows[0][2] > 0
            else:
                # MySQL 8+ closes the table really quickly. If it's closed,
                # it's not locked.
                return False

    def test_write(self):
        Alphabet.objects.create(a=12345)
        assert not self.is_locked("default", Alphabet._meta.db_table)

        with TableLock(write=[Alphabet]):
            assert self.is_locked("default", Alphabet._meta.db_table)
            assert Alphabet.objects.count() == 1

            Alphabet.objects.all().delete()
            assert Alphabet.objects.count() == 0

        assert not self.is_locked("default", Alphabet._meta.db_table)
        assert Alphabet.objects.count() == 0

    def test_write_with_table_name(self):

        assert not self.is_locked("default", Alphabet._meta.db_table)