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