1# -*- coding: utf-8 -*- 2"""QGIS Unit tests for postgres transaction groups. 3 4.. note:: This program is free software; you can redistribute it and/or modify 5it under the terms of the GNU General Public License as published by 6the Free Software Foundation; either version 2 of the License, or 7(at your option) any later version. 8""" 9__author__ = 'Nyall Dawson' 10__date__ = '11/06/2018' 11__copyright__ = 'Copyright 2018, The QGIS Project' 12 13import qgis # NOQA 14 15import os 16 17from qgis.core import ( 18 QgsVectorLayer, 19 QgsProject, 20 QgsTransaction, 21 QgsDataSourceUri 22) 23 24from qgis.testing import start_app, unittest 25 26start_app() 27 28 29class TestQgsPostgresTransaction(unittest.TestCase): 30 31 @classmethod 32 def setUpClass(cls): 33 """ 34 Setup the involved layers and relations for a n:m relation 35 :return: 36 """ 37 cls.dbconn = 'service=qgis_test' 38 if 'QGIS_PGTEST_DB' in os.environ: 39 cls.dbconn = os.environ['QGIS_PGTEST_DB'] 40 # Create test layer 41 cls.vl_b = QgsVectorLayer(cls.dbconn + ' sslmode=disable key=\'pk\' table="qgis_test"."books" sql=', 'books', 42 'postgres') 43 cls.vl_a = QgsVectorLayer(cls.dbconn + ' sslmode=disable key=\'pk\' table="qgis_test"."authors" sql=', 44 'authors', 'postgres') 45 46 QgsProject.instance().addMapLayer(cls.vl_b) 47 QgsProject.instance().addMapLayer(cls.vl_a) 48 49 cls.relMgr = QgsProject.instance().relationManager() 50 51 assert (cls.vl_a.isValid()) 52 assert (cls.vl_b.isValid()) 53 54 def startTransaction(self): 55 """ 56 Start a new transaction and set all layers into transaction mode. 57 58 :return: None 59 """ 60 lyrs = [self.vl_a, self.vl_b] 61 62 self.transaction = QgsTransaction.create(lyrs) 63 self.transaction.begin() 64 for l in lyrs: 65 l.startEditing() 66 67 def rollbackTransaction(self): 68 """ 69 Rollback all changes done in this transaction. 70 We always rollback and never commit to have the database in a pristine 71 state at the end of each test. 72 73 :return: None 74 """ 75 lyrs = [self.vl_a, self.vl_b] 76 for l in lyrs: 77 l.commitChanges() 78 self.transaction.rollback() 79 80 def test_transactionsGroup(self): 81 conn_string = QgsDataSourceUri(self.vl_b.source()).connectionInfo() 82 83 # No transaction group. 84 QgsProject.instance().setAutoTransaction(False) 85 noTg = QgsProject.instance().transactionGroup("postgres", conn_string) 86 self.assertIsNone(noTg) 87 88 # start transaction - no auto transaction 89 self.startTransaction() 90 noTg = QgsProject.instance().transactionGroup("postgres", conn_string) 91 self.assertIsNone(noTg) 92 self.rollbackTransaction() 93 94 # with auto transactions 95 QgsProject.instance().setAutoTransaction(True) 96 self.startTransaction() 97 noTg = QgsProject.instance().transactionGroup("postgres", conn_string) 98 self.assertIsNotNone(noTg) 99 self.rollbackTransaction() 100 101 # bad provider key 102 self.startTransaction() 103 noTg = QgsProject.instance().transactionGroup("xxxpostgres", conn_string) 104 self.assertIsNone(noTg) 105 self.rollbackTransaction() 106 107 def test_transactionGroupEditingStatus(self): 108 """Not particularly related to PG but it fits here nicely: test GH #39282""" 109 110 project = QgsProject() 111 project.setAutoTransaction(True) 112 113 vl_b = QgsVectorLayer(self.dbconn + ' sslmode=disable key=\'pk\' table="qgis_test"."books" sql=', 'books', 114 'postgres') 115 vl_a = QgsVectorLayer(self.dbconn + ' sslmode=disable key=\'pk\' table="qgis_test"."authors" sql=', 116 'authors', 'postgres') 117 118 project.addMapLayers([vl_a, vl_b]) 119 120 vl_a.startEditing() 121 self.assertTrue(vl_a.isEditable()) 122 self.assertTrue(vl_b.isEditable()) 123 124 self.assertTrue(vl_a.commitChanges(False)) 125 self.assertTrue(vl_a.isEditable()) 126 self.assertTrue(vl_b.isEditable()) 127 128 129if __name__ == '__main__': 130 unittest.main() 131