test_sync_tx.py

#
# This source file is part of the EdgeDB open source project.
#
# Copyright 2016-present MagicStack Inc. and the EdgeDB authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

import itertools
from concurrent.futures import ThreadPoolExecutor

import edgedb

from edgedb import _testbase as tb
from edgedb import TransactionOptions


class TestSyncTx(tb.SyncQueryTestCase):

    SETUP = '''
        CREATE TYPE test::TransactionTest EXTENDING std::Object {
            CREATE PROPERTY name -> std::str;
        };
    '''

    TEARDOWN_METHOD = '''
        DELETE test::TransactionTest;
    '''

    TEARDOWN = '''
        DROP TYPE test::TransactionTest;
    '''

    def test_sync_transaction_regular_01(self):
        tr = self.client.transaction()

        with self.assertRaises(ZeroDivisionError):
            for with_tr in tr:
                with with_tr:
                    with_tr.execute('''
                        INSERT test::TransactionTest {
                            name := 'Test Transaction'
                        };
                    ''')

                    1 / 0

        result = self.client.query('''
            SELECT
                test::TransactionTest
            FILTER
                test::TransactionTest.name = 'Test Transaction';
        ''')

        self.assertEqual(result, [])

    async def test_sync_transaction_kinds(self):
        isolations = [
            None,
            edgedb.IsolationLevel.Serializable,
        ]
        booleans = [None, True, False]
        all = itertools.product(isolations, booleans, booleans)
        for isolation, readonly, deferrable in all:
            opt = dict(
                isolation=isolation,
                readonly=readonly,
                deferrable=deferrable,
            )
            # skip None
            opt = {k: v for k, v in opt.items() if v is not None}
            client = self.client.with_transaction_options(
                TransactionOptions(**opt)
            )
            try:
                for tx in client.transaction():
                    with tx:
                        tx.execute(
                            'INSERT test::TransactionTest {name := "test"}')
            except edgedb.TransactionError:
                self.assertTrue(readonly)
            else:
                self.assertFalse(readonly)

            for tx in client.transaction():
                with tx:
                    pass

    def test_sync_transaction_commit_failure(self):
        with self.assertRaises(edgedb.errors.QueryError):
            for tx in self.client.transaction():
                with tx:
                    tx.execute("start migration to {};")
        self.assertEqual(self.client.query_single("select 42"), 42)

    def test_sync_transaction_exclusive(self):
        for tx in self.client.transaction():
            with tx:
                query = "select sys::_sleep(0.01)"
                with ThreadPoolExecutor(max_workers=2) as executor:
                    f1 = executor.submit(tx.execute, query)
                    f2 = executor.submit(tx.execute, query)
                    with self.assertRaisesRegex(
                        edgedb.InterfaceError,
                        "concurrent queries within the same transaction "
                        "are not allowed"
                    ):
                        f1.result(timeout=5)
                        f2.result(timeout=5)

    def test_sync_transaction_savepoint_1(self):
        for tx in self.client.transaction():
            with tx:
                sp1 = tx.savepoint()
                sp2 = tx.savepoint()
                tx.execute('''
                    INSERT test::TransactionTest { name := '1' };
                ''')
                sp2.release()
                with self.assertRaisesRegex(
                    edgedb.InterfaceError, "savepoint.*is no longer active"
                ):
                    sp2.release()
                sp1.release()

        result = self.client.query('SELECT test::TransactionTest.name')

        self.assertEqual(result, ["1"])

    def test_sync_transaction_savepoint_2(self):
        for tx in self.client.transaction():
            with tx:
                tx.execute('''
                    INSERT test::TransactionTest { name := '1' };
                ''')
                sp1 = tx.savepoint()
                tx.execute('''
                    INSERT test::TransactionTest { name := '2' };
                ''')
                sp2 = tx.savepoint()
                tx.execute('''
                    INSERT test::TransactionTest { name := '3' };
                ''')
                sp1.rollback()
                with self.assertRaisesRegex(
                    edgedb.InterfaceError, "savepoint.*is no longer active"
                ):
                    sp1.rollback()
                with self.assertRaisesRegex(
                    edgedb.InterfaceError, "savepoint.*is no longer active"
                ):
                    sp2.rollback()

        result = self.client.query('SELECT test::TransactionTest.name')

        self.assertEqual(result, ["1"])