Coverage for gws-app/gws/base/database/_test/connection_test.py: 100%

1import gws.test.util as u 

2 

3 

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

5def db(): 

6 u.pg.create('tab', {'id': 'int primary key', 'a': 'text'}) 

7 root = u.gws_root() 

8 yield u.get_db(root) 

9 

10 

11## 

12 

13 

14def test_connect(db): 

15 with db.connect() as conn: 

16 assert conn.fetch_int('select 123') == 123 

17 

18 

19def test_connection_closed(db): 

20 with db.connect() as conn: 

21 conn.fetch_int('select 123') 

22 assert db._sa_connection() is None 

23 

24 

25def test_nested_connection(db): 

26 with db.connect() as conn: 

27 assert conn.fetch_int('select 123') == 123 

28 assert db._sa_connection() is not None 

29 with db.connect() as conn: 

30 assert conn.fetch_int('select 234') == 234 

31 assert db._sa_connection() is not None 

32 

33 assert db._sa_connection() is None 

34 

35 

36def test_commit(db): 

37 with db.connect() as conn: 

38 conn.exec_commit("insert into tab (id, a) values (1, 'X')") 

39 assert conn.fetch_int("select id from tab where a = 'X'") == 1 

40 

41 with db.connect() as conn: 

42 conn.exec("insert into tab (id, a) values (2, 'Y')") 

43 conn.commit() 

44 

45 with db.connect() as conn: 

46 assert conn.fetch_int("select id from tab where a = 'Y'") == 2 

47 

48 

49def test_error_rollback(db): 

50 with db.connect() as conn: 

51 cnt_1 = conn.fetch_int("select count(*) from tab") 

52 

53 with u.raises(Exception): 

54 with db.connect() as conn: 

55 conn.exec("insert into tab (id, a) values (100, 'X')") 

56 conn.exec("insert into tab (id, a) values (NULL, 'X')") 

57 

58 with db.connect() as conn: 

59 cnt_2 = conn.fetch_int("select count(*) from tab") 

60 

61 assert cnt_1 == cnt_2 

62 

63# @TODO other connection methods