Coverage for gws-app/gws/base/database/_test/provider_test.py: 0%

1import gws.base.database 

2import gws.lib.sa as sa 

3import gws.test.util as u 

4 

5 

6@u.fixture(scope='module') 

7def root(): 

8 yield u.gws_root() 

9 

10 

11## 

12 

13def test_error(root: gws.Root): 

14 db = u.get_db(root) 

15 

16 err = '' 

17 with db.connect() as conn: 

18 try: 

19 conn.execute('not an object') 

20 except sa.Error as e: 

21 err = str(e) 

22 assert 'Not an executable object' in err 

23 

24 err = '' 

25 with db.connect() as conn: 

26 try: 

27 conn.execute(sa.text('select * from not_a_table')) 

28 except sa.Error as e: 

29 err = str(e) 

30 assert 'UndefinedTable' in err 

31 

32 

33def test_error_rollback(root: gws.Root): 

34 db = u.get_db(root) 

35 

36 err = '' 

37 with db.connect() as conn: 

38 try: 

39 conn.execute(sa.text('select * from not_a_table')) 

40 except sa.Error as e: 

41 err = str(e) 

42 conn.rollback() 

43 try: 

44 conn.execute(sa.text('select 1')) 

45 except sa.Error as e: 

46 err = 'FAILED' 

47 assert 'UndefinedTable' in err 

48 

49 

50def test_error_no_rollback(root: gws.Root): 

51 db = u.get_db(root) 

52 

53 err = '' 

54 with db.connect() as conn: 

55 try: 

56 conn.execute(sa.text('select * from not_a_table')) 

57 except sa.Error as e: 

58 err = str(e) 

59 # no rollback 

60 try: 

61 conn.execute(sa.text('select 1')) 

62 except sa.Error as e: 

63 # ERROR: current transaction is aborted... 

64 err = 'FAILED' 

65 assert 'FAILED' in err 

66 

67 

68def test_nested_connections(root: gws.Root): 

69 db = u.get_db(root) 

70 db_name = u.option('service.postgres.database') 

71 

72 def _num_conn(c): 

73 return ( 

74 c.execute( 

75 sa.text(f"SELECT numbackends FROM pg_stat_database WHERE datname='{db_name}' ")) 

76 ).scalar_one() 

77 

78 with db.connect() as c1: 

79 n1 = _num_conn(c1) 

80 with db.connect() as c2: 

81 n2 = _num_conn(c2) 

82 with db.connect() as c3: 

83 n3 = _num_conn(c3) 

84 

85 assert n1 == n2 == n3 

86 

87 

88def test_nested_connection_opens_once(root: gws.Root): 

89 db = u.get_db(root) 

90 

91 log = [] 

92 

93 class MockConn: 

94 def __init__(self): 

95 log.append('OPEN') 

96 

97 def execute(self, s): 

98 log.append(f'EXEC {s}') 

99 

100 def close(self): 

101 log.append('CLOSE') 

102 

103 with u.monkey_patch() as mp: 

104 mp.setattr(sa.Engine, 'connect', lambda *args: MockConn()) 

105 

106 with db.connect() as conn: 

107 conn.execute('1') 

108 with db.connect() as conn: 

109 conn.execute('2') 

110 with db.connect() as conn: 

111 conn.execute('3') 

112 

113 assert log == ['OPEN', 'EXEC 1', 'EXEC 2', 'EXEC 3', 'CLOSE']