# 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,
)
        assert not self.is_locked("default", Alphabet._meta.db_table)
        with TableLock(write=[Alphabet._meta.db_table]):
            assert self.is_locked("default", Alphabet._meta.db_table)

    def test_write_with_using(self):
        Alphabet.objects.using("other").create(a=878787)
        assert not self.is_locked("other", Alphabet._meta.db_table)

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

            Alphabet.objects.using("other").all().delete()
            assert Alphabet.objects.using("other").count() == 0

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

    def test_write_fails_touching_other_table(self):
        with pytest.raises(OperationalError) as excinfo:
            with TableLock(write=[Alphabet]):
                Customer.objects.create(name="Lizzy")

        assert excinfo.value.args[0] == 1100  # ER_TABLE_NOT_LOCKED

    def test_read_and_write(self):
        Customer.objects.create(name="Fred")
        with TableLock(read=[Customer], write=[Alphabet]):
            assert self.is_locked("default", Alphabet._meta.db_table)
            assert self.is_locked("default", Customer._meta.db_table)

# 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,
)

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

    def test_write_with_using(self):
        Alphabet.objects.using("other").create(a=878787)
        assert not self.is_locked("other", Alphabet._meta.db_table)

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

            Alphabet.objects.using("other").all().delete()
            assert Alphabet.objects.using("other").count() == 0

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

    def test_write_fails_touching_other_table(self):
        with pytest.raises(OperationalError) as excinfo:
            with TableLock(write=[Alphabet]):
                Customer.objects.create(name="Lizzy")

        assert excinfo.value.args[0] == 1100  # ER_TABLE_NOT_LOCKED

    def test_read_and_write(self):
        Customer.objects.create(name="Fred")
        with TableLock(read=[Customer], write=[Alphabet]):
            assert self.is_locked("default", Alphabet._meta.db_table)
            assert self.is_locked("default", Customer._meta.db_table)

            ab = Alphabet.objects.create(a=Customer.objects.count())