<commit_before>from test.integration.base import DBTIntegrationTest, use_profile
import threading
from dbt.adapters.factory import get_adapter

class BaseTestConcurrentTransaction(DBTIntegrationTest):

    def reset(self):
        self.query_state = {
            'view_model': 'wait',
            'model_1': 'wait',
        }

    def setUp(self):
        super(BaseTestConcurrentTransaction, self).setUp()
        self.reset()

    @property
    def schema(self):
        return "concurrent_transaction_032"

    @property
    def project_config(self):
        return {
            "macro-paths": ["test/integration/032_concurrent_transaction_test/macros"],
            "on-run-start": [
                "{{ create_udfs() }}",
            ],
        }

    def run_select_and_check(self, rel, sql):
        connection_name = '__test_{}'.format(id(threading.current_thread()))
        try:
            with get_adapter(self.config).connection_named(connection_name) as conn:
                res = self.run_sql_common(self.transform_sql(sql), 'one', conn)

            # The result is the output of f_sleep(), which is True
            if res[0] == True:
                self.query_state[rel] = 'good'
            else:
                self.query_state[rel] = 'bad'

        except Exception as e:
            if 'concurrent transaction' in str(e):
                self.query_state[rel] = 'error: {}'.format(e)
            else:
                self.query_state[rel] = 'error: {}'.format(e)

    def async_select(self, rel, sleep=10):<commit_msg>in the concurrent transactions test, use a completely separate adapter for our goofy sql running<commit_after>from test.integration.base import DBTIntegrationTest, use_profile
import threading
from dbt.adapters.factory import ADAPTER_TYPES


def get_adapter_standalone(config):
    cls = ADAPTER_TYPES[config.credentials.type]
    return cls(config)


class BaseTestConcurrentTransaction(DBTIntegrationTest):

    def reset(self):
        self.query_state = {
            'view_model': 'wait',
            'model_1': 'wait',
        }

    def setUp(self):
        super(BaseTestConcurrentTransaction, self).setUp()
        self._secret_adapter = get_adapter_standalone(self.config)
        self.reset()

    def tearDown(self):
        self._secret_adapter.cleanup_connections()
        super(BaseTestConcurrentTransaction, self).tearDown()

    @property
    def schema(self):
        return "concurrent_transaction_032"

    @property
    def project_config(self):
        return {
            "macro-paths": ["test/integration/032_concurrent_transaction_test/macros"],
            "on-run-start": [
                "{{ create_udfs() }}",
            ],
        }

    def run_select_and_check(self, rel, sql):
        connection_name = '__test_{}'.format(id(threading.current_thread()))
        try:
            with self._secret_adapter.connection_named(connection_name) as conn:
                res = self.run_sql_common(self.transform_sql(sql), 'one', conn)

            # The result is the output of f_sleep(), which is True
            if res[0] == True:
                self.query_state[rel] = 'good'
            else:
                self.query_state[rel] = 'bad'

        except Exception as e:
            if 'concurrent transaction' in str(e):
                self.query_state[rel] = 'error: {}'.format(e)
            else:
                self.query_state[rel] = 'error: {}'.format(e)

    def async_select(self, rel, sleep=10):
{'fraction_non_alphanumeric': 0.11034680424190313, 'fraction_numerical': 0.005732301519059902, 'mean_word_length': 2.3589990375360923, 'pattern_counts': {'":': 4, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2280151', 'n_tokens_mistral': 1038, 'n_tokens_neox': 989, 'n_words': 208}