KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use execute_transaction method in localstack

Best Python code snippet using localstack_python

utils.py

Source:utils.py

...54        proj_sig.setdefault(app,{})[name] = signature.create_model_sig(model)55    56    return proj_sig57    58def execute_transaction(sql, output=False):59    "A transaction wrapper for executing a list of SQL statements"60    try:61        # Begin Transaction62        transaction.enter_transaction_management()63        transaction.managed(True)64        cursor = connection.cursor()65        66        # Perform the SQL67        if output:68            write_sql(sql)69        execute_sql(cursor, sql)70        71        transaction.commit()72        transaction.leave_transaction_management()73    except Exception, ex:74        transaction.rollback()75        raise ex76def execute_test_sql(start, end, sql, debug=False):77    """78    Execute a test SQL sequence. This method also creates and destroys the 79    database tables required by the models registered against the test application.80    81    start and end are the start- and end-point states of the application cache.82    83    sql is the list of sql statements to execute.84    85    cleanup is a list of extra sql statements required to clean up. This is86    primarily for any extra m2m tables that were added during a test that won't 87    be cleaned up by Django's sql_delete() implementation.88    89    debug is a helper flag. It displays the ALL the SQL that would be executed,90    (including setup and teardown SQL), and executes the Django-derived setup/teardown91    SQL.92    """    93    # Set up the initial state of the app cache94    cache.app_models['tests'] = copy.deepcopy(start)95        96    # Install the initial tables and indicies97    style = no_style()    98    execute_transaction(sql_create(evo_test, style), output=debug)99    execute_transaction(sql_indexes(evo_test, style), output=debug)100    create_test_data(models.get_models(evo_test))101            102    # Set the app cache to the end state103    cache.app_models['tests'] = copy.deepcopy(end)104    105    try:106        # Execute the test sql107        if debug:108            write_sql(sql)109        else:110            execute_transaction(sql, output=True)111    finally:112        # Cleanup the apps.113        if debug:114            print sql_delete(evo_test, style)115        else:116            execute_transaction(sql_delete(evo_test, style), output=debug)117    118def create_test_data(app_models):119    deferred_models = []120    deferred_fields = {}121    for model in app_models:122        params = {}123        deferred = False124        for field in model._meta.fields:125            if not deferred:126                if type(field) == models.ForeignKey or type(field) == models.ManyToManyField:127                    related_model = field.rel.to128                    if related_model.objects.count():129                        related_instance = related_model.objects.all()[0]130                    else:...

specific_tests.py

Source:specific_tests.py

1from neo4j.v1 import GraphDatabase2import os3def execute_transaction(query):4    host = os.environ['NEO4J_NQC_HOST']5    port = os.environ['NEO4J_NQC_PORT']6    uri = "bolt://" + host + ":" + port7    graph = GraphDatabase.driver(uri, auth=("neo4j", "neo4j"))8    result = None9    with graph.session() as session:10        result = session.run(query)11    return result12def test_fgf8a_exists():13    query = "MATCH (g:Gene) WHERE g.symbol = 'fgf8a' RETURN count(g) AS count"14    result = execute_transaction(query)15    for record in result:16        assert record["count"] > 017def test_hip1_exists():18    query = "MATCH (g:Gene) WHERE g.symbol = 'Hip1' RETURN count(g) AS count"19    result = execute_transaction(query)20    for record in result:21        assert record["count"] > 022def test_doterm_exists():23    query = "MATCH(n:DOTerm) where n.primaryKey = 'DOID:0001816' RETURN count(n) AS count"24    result = execute_transaction(query)25    for record in result:26        assert record["count"] == 127def test_isobsolete_false():28    query = "MATCH(n:DOTerm) where n.is_obsolete = 'false' RETURN count(n) AS count"29    result = execute_transaction(query)30    for record in result:31        assert record["count"] > 032def test_species_disease_pub_gene_exists():33    query = "MATCH (s:Species)--(g:Gene)--(dg:DiseaseEntityJoin)--(p:Publication) RETURN COUNT(p) AS count"34    result = execute_transaction(query)35    for record in result:36        assert record["count"] > 037def test_species_disease_pub_allele_exists():38    query = "MATCH (s:Species)--(f:Feature)--(dg:DiseaseEntityJoin)--(p:Publication) RETURN COUNT(p) AS count"39    result = execute_transaction(query)40    for record in result:41        assert record["count"] > 042def test_uuid_is_not_duplicated():43    query = "MATCH (g) WITH g.uuid AS uuid, count(*) AS counter WHERE counter > 0 AND g.uuid IS NOT NULL RETURN uuid, counter"44    result = execute_transaction(query)45    for record in result:...

pm.py

Source:pm.py

...3if __name__ == '__main__':4    5    acct2 = Account(2)6    acct2.fund_account(10000)7    acct2.execute_transaction('BUY', 'AAPL', 10)8    acct2.execute_transaction('BUY', 'W', 20)9    acct2.execute_transaction('BUY', 'W', 30)10    acct2.execute_transaction('SELL', 'AAPL', 5)11    acct2.view_portfolio()12    acct2.execute_transaction('SELL', 'AAPL', 5)13    acct2.fund_account(5000)14    acct2.view_portfolio()15    target_buys = {'W':85, 'AAPL': 126}16    target_sells = {'W': 17, 'AAPL': 29}17    for ticker, quantity in target_buys.items():18        acct2.execute_transaction('BUY', ticker, quantity)19    for ticker, quantity in target_sells.items():...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.