<s> from __future__ import unicode_literals <EOL> from django import VERSION as django_version <EOL> from django . contrib . auth . models import User <EOL> from django . db import connection , transaction <EOL> from django . test import TransactionTestCase <EOL> from . models import Test <EOL> class AtomicTestCase ( TransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> if connection . vendor == '<STR_LIT>' : <EOL> connection . cursor ( ) <EOL> def test_successful_read_atomic ( self ) : <EOL> is_sqlite = connection . vendor == '<STR_LIT>' <EOL> with self . assertNumQueries ( <NUM_LIT:2> if is_sqlite else <NUM_LIT:1> ) : <EOL> with transaction . atomic ( ) : <EOL> data1 = list ( Test . objects . all ( ) ) <EOL> self . assertListEqual ( data1 , [ ] ) <EOL> with self . assertNumQueries ( <NUM_LIT:0> ) : <EOL> data2 = list ( Test . objects . all ( ) ) <EOL> self . assertListEqual ( data2 , [ ] ) <EOL> def test_unsuccessful_read_atomic ( self ) : <EOL> is_sqlite = connection . vendor == '<STR_LIT>' <EOL> with self . assertNumQueries ( <NUM_LIT:2> if is_sqlite else <NUM_LIT:1> ) : <EOL> try : <EOL> with transaction . atomic ( ) : <EOL> data1 = list ( Test . objects . all ( ) ) <EOL> raise ZeroDivisionError <EOL> except ZeroDivisionError : <EOL> pass <EOL> self . assertListEqual ( data1 , [ ] ) <EOL> with self . assertNumQueries ( <NUM_LIT:1> ) : <EOL> data2 = list ( Test . objects . all ( ) ) <EOL> self . assertListEqual ( data2 , [ ] ) <EOL> def test_successful_write_atomic ( self ) : <EOL> with self . assertNumQueries ( <NUM_LIT:1> ) : <EOL> data1 = list ( Test . objects . all ( ) ) <EOL> self . assertListEqual ( data1 , [ ] ) <EOL> is_sqlite = connection . vendor == '<STR_LIT>' <EOL> with self . assertNumQueries ( <NUM_LIT:2> if is_sqlite else <NUM_LIT:1> ) : <EOL> with transaction . atomic ( ) : <EOL> t1 = Test . objects . create ( name = '<STR_LIT>' ) <EOL> with self . assertNumQueries ( <NUM_LIT:1> ) : <EOL> data2 = list ( Test . objects . all ( ) ) <EOL> self . assertListEqual ( data2 , [ t1 ] ) <EOL> with self . assertNumQueries ( <NUM_LIT:2> if is_sqlite else <NUM_LIT:1> ) : <EOL> with transaction . atomic ( ) : <EOL> t2 = Test . objects . create ( name = '<STR_LIT>' ) <EOL> with self . assertNumQueries ( <NUM_LIT:1> ) : <EOL> data3 = list ( Test . objects . all ( ) ) <EOL> self . assertListEqual ( data3 , [ t1 , t2 ] ) <EOL> with self . assertNumQueries ( <NUM_LIT:4> if is_sqlite else <NUM_LIT:3> ) : <EOL> with transaction . atomic ( ) : <EOL> data4 = list ( Test . objects . all ( ) ) <EOL> t3 = Test . objects . create ( name = '<STR_LIT>' ) <EOL> t4 = Test . objects . create ( name = '<STR_LIT>' ) <EOL> data5 = list ( Test . objects . all ( ) ) <EOL> self . assertListEqual ( data4 , [ t1 , t2 ] ) <EOL> self . assertListEqual ( data5 , [ t1 , t2 , t3 , t4 ] ) <EOL> self . assertNotEqual ( t4 , t3 ) <EOL> def test_unsuccessful_write_atomic ( self ) : <EOL> with self . assertNumQueries ( <NUM_LIT:1> ) : <EOL> data1 = list ( Test . objects . all ( ) ) <EOL> self . assertListEqual ( data1 , [ ] ) <EOL> is_sqlite = connection . vendor == '<STR_LIT>' <EOL> with self . assertNumQueries ( <NUM_LIT:2> if is_sqlite else <NUM_LIT:1> ) : <EOL> try : <EOL> with transaction . atomic ( ) : <EOL> Test . objects . create ( name = '<STR_LIT:test>' ) <EOL> raise ZeroDivisionError <EOL> except ZeroDivisionError : <EOL> pass <EOL> with self . assertNumQueries ( <NUM_LIT:0> ) : <EOL> data2 = list ( Test . objects . all ( ) ) <EOL> self . assertListEqual ( data2 , [ ] ) <EOL> with self . assertNumQueries ( <NUM_LIT:1> ) : <EOL> with self . assertRaises ( Test . DoesNotExist ) : <EOL> Test . objects . get ( name = '<STR_LIT:test>' ) <EOL> def test_cache_inside_atomic ( self ) : <EOL> is_sqlite = connection . vendor == '<STR_LIT>' <EOL> with self . assertNumQueries ( <NUM_LIT:2> if is_sqlite else <NUM_LIT:1> ) : <EOL> with transaction . atomic ( ) : <EOL> data1 = list ( Test . objects . all ( ) ) <EOL> data2 = list ( Test . objects . all ( ) ) <EOL> self . assertListEqual ( data2 , data1 ) <EOL> self . assertListEqual ( data2 , [ ] ) <EOL> def test_invalidation_inside_atomic ( self ) : <EOL> is_sqlite = connection . vendor == '<STR_LIT>' <EOL> with self . assertNumQueries ( <NUM_LIT:4> if is_sqlite else <NUM_LIT:3> ) : <EOL> with transaction . atomic ( ) : <EOL> data1 = list ( Test . objects . all ( ) ) <EOL> t = Test . objects . create ( name = '<STR_LIT:test>' ) <EOL> data2 = list ( Test . objects . all ( ) ) <EOL> self . assertListEqual ( data1 , [ ] ) <EOL> self . assertListEqual ( data2 , [ t ] ) <EOL> def test_successful_nested_read_atomic ( self ) : <EOL> is_sqlite = connection . vendor == '<STR_LIT>' <EOL> with self . assertNumQueries ( <NUM_LIT:7> if is_sqlite else <NUM_LIT:6> ) : <EOL> with transaction . atomic ( ) : <EOL> list ( Test . objects . all ( ) ) <EOL> with transaction . atomic ( ) : <EOL> list ( User . objects . all ( ) ) <EOL> with self . assertNumQueries ( <NUM_LIT:2> ) : <EOL> with transaction . atomic ( ) : <EOL> list ( User . objects . all ( ) ) <EOL> with self . assertNumQueries ( <NUM_LIT:0> ) : <EOL> list ( User . objects . all ( ) ) <EOL> with self . assertNumQueries ( <NUM_LIT:0> ) : <EOL> list ( Test . objects . all ( ) ) <EOL> list ( User . objects . all ( ) ) <EOL> def test_unsuccessful_nested_read_atomic ( self ) : <EOL> is_sqlite = connection . vendor == '<STR_LIT>' <EOL> if django_version [ : <NUM_LIT:2> ] == ( <NUM_LIT:1> , <NUM_LIT:7> ) : <EOL> num_queries = <NUM_LIT:5> if is_sqlite else <NUM_LIT:4> <EOL> else : <EOL> num_queries = <NUM_LIT:6> if is_sqlite else <NUM_LIT:5> <EOL> with self . assertNumQueries ( num_queries ) : <EOL> with transaction . atomic ( ) : <EOL> try : <EOL> with transaction . atomic ( ) : <EOL> with self . assertNumQueries ( <NUM_LIT:1> ) : <EOL> list ( Test . objects . all ( ) ) <EOL> raise ZeroDivisionError <EOL> except ZeroDivisionError : <EOL> pass <EOL> with self . assertNumQueries ( <NUM_LIT:1> ) : <EOL> list ( Test . objects . all ( ) ) <EOL> def test_successful_nested_write_atomic ( self ) : <EOL> is_sqlite = connection . vendor == '<STR_LIT>' <EOL> with self . assertNumQueries ( <NUM_LIT> if is_sqlite else <NUM_LIT:12> ) : <EOL> with transaction . atomic ( ) : <EOL> t1 = Test . objects . create ( name = '<STR_LIT>' ) <EOL> with transaction . atomic ( ) : <EOL> t2 = Test . objects . create ( name = '<STR_LIT>' ) <EOL> data1 = list ( Test . objects . all ( ) ) <EOL> self . assertListEqual ( data1 , [ t1 , t2 ] ) <EOL> with transaction . atomic ( ) : <EOL> t3 = Test . objects . create ( name = '<STR_LIT>' ) <EOL> with transaction . atomic ( ) : <EOL> data2 = list ( Test . objects . all ( ) ) <EOL> self . assertListEqual ( data2 , [ t1 , t2 , t3 ] ) <EOL> t4 = Test . objects . create ( name = '<STR_LIT>' ) <EOL> data3 = list ( Test . objects . all ( ) ) <EOL> self . assertListEqual ( data3 , [ t1 , t2 , t3 , t4 ] ) <EOL> def test_unsuccessful_nested_write_atomic ( self ) : <EOL> is_sqlite = connection . vendor == '<STR_LIT>' <EOL> if django_version [ : <NUM_LIT:2> ] == ( <NUM_LIT:1> , <NUM_LIT:7> ) : <EOL> num_queries = <NUM_LIT> if is_sqlite else <NUM_LIT:12> <EOL> else : <EOL> num_queries = <NUM_LIT:16> if is_sqlite else <NUM_LIT:15> <EOL> with self . assertNumQueries ( num_queries ) : <EOL> with transaction . atomic ( ) : <EOL> t1 = Test . objects . create ( name = '<STR_LIT>' ) <EOL> try : <EOL> with transaction . atomic ( ) : <EOL> t2 = Test . objects . create ( name = '<STR_LIT>' ) <EOL> data1 = list ( Test . objects . all ( ) ) <EOL> self . assertListEqual ( data1 , [ t1 , t2 ] ) <EOL> raise ZeroDivisionError <EOL> except ZeroDivisionError : <EOL> pass <EOL> data2 = list ( Test . objects . all ( ) ) <EOL> self . assertListEqual ( data2 , [ t1 ] ) <EOL> try : <EOL> with transaction . atomic ( ) : <EOL> t3 = Test . objects . create ( name = '<STR_LIT>' ) <EOL> with transaction . atomic ( ) : <EOL> data2 = list ( Test . objects . all ( ) ) <EOL> self . assertListEqual ( data2 , [ t1 , t3 ] ) <EOL> raise ZeroDivisionError <EOL> except ZeroDivisionError : <EOL> pass <EOL> with self . assertNumQueries ( <NUM_LIT:1> ) : <EOL> data3 = list ( Test . objects . all ( ) ) <EOL> self . assertListEqual ( data3 , [ t1 ] ) </s>
