source: peerreviewplugin/trunk/codereview/tests/review_model.py

Last change on this file was 17453, checked in by Cinc-th, 4 years ago

PeerReviewPlugin: fix failing test on windows. The time stamp resolution (on Windows) seems not fine enough when calling save_changes() in fast succession causing an IntegrityError during db write.
File size: 10.6 KB
Line 
1# -*- coding: utf-8 -*-
2
3import unittest
4from datetime import datetime
5from trac.util.datefmt import to_datetime, to_utimestamp
6from trac.web import Href
7from trac.test import EnvironmentStub, Mock, MockPerm
8from codereview.model import PeerReviewModel, PeerReviewModelProvider, ReviewFileModel
9from codereview.tracgenericworkflow.api import  ResourceWorkflowSystem
10from codereview.tracgenericworkflow.model import GenericWorkflowModelProvider
11
12
13__author__ = 'Cinc'
14__copyright__ = "Copyright 2016"
15__license__ = "BSD"
16
17
18def _prepare_review_data(env):
19    # owner, status, name, notes, parent_id
20    revs = [
21        ['Rev1', 'bar', to_utimestamp(to_datetime(datetime(2019, 2, 4))), 'name1', 'note1', 0],
22        ['Rev1', 'closed', to_utimestamp(to_datetime(datetime(2019, 3, 4))), 'name2', 'note2', 0],
23        ['Rev2', 'bar', to_utimestamp(to_datetime(datetime(2019, 3, 14))), 'name3', 'note3', 1],
24        ['Rev3', 'foo', to_utimestamp(to_datetime(datetime(2019, 4, 4))), 'name4', 'note4', 2]
25    ]
26
27    with env.db_transaction as db:
28        cursor = db.cursor()
29        for rev in revs:
30            cursor.execute("INSERT INTO peerreview (owner, status, created, name, notes, parent_id) "
31                           "VALUES (%s,%s,%s,%s,%s,%s)", rev)
32
33def _prepare_file_data(env):
34    # review_id, path, start, end, revision, status
35    files = [
36        [1, '/foo/bar', 5, 100, '1234', 'new', None],
37        [1, '/foo/bar2', 6, 101, '1234', 'new', None],
38        [2, '/foo/bar', 5, 100, '1234', 'closed', None],
39        [2, '/foo/bar2', 6, 101, '12346', 'closed', None],
40        [2, '/foo/bar3', 7, 102, '12347', 'closed', None],
41        [3, '/foo/bar2', 6, 101, '1234', 'new', None],
42        [4, '/foo/bar', 5, 100, '1234', 'new', None],
43        [4, '/foo/bar2', 6, 101, '1234', 'new', None],
44        # File list data for several projects
45        [0, '/foo/bar', 5, 100, '1234', 'new', 'PrjFoo'],
46        [0, '/foo/bar2', 6, 101, '1234', 'new', 'PrjFoo'],
47        [0, '/foo/bar', 5, 100, '1234', 'new', 'PrjBar'],
48        [0, '/foo/bar2', 6, 101, '12346', 'new', 'PrjBar'],
49        [0, '/foo/bar3', 7, 102, '12347', 'new', 'PrjFoo'],
50        [0, '/foo/bar/baz', 6, 101, '1234', 'new', 'PrjBar'],
51        [0, '/foo/bar', 5, 100, '1234', 'new', 'PrjBaz'],
52        [0, '/foo/bar2', 6, 101, '1234', 'new', 'PrjBaz'],
53    ]
54    for f in files:
55        rfm = ReviewFileModel(env)
56        rfm['review_id'] = f[0]
57        rfm['path'] = f[1]
58        rfm['line_start'] = f[2]
59        rfm['line_end'] = f[3]
60        rfm['revision'] = f[4]
61        rfm['status'] = f[5]
62        rfm['project'] = f[6]
63        rfm.insert()
64
65
66class TestReviewModel(unittest.TestCase):
67
68    def setUp(self):
69        self.env = EnvironmentStub(default_data=True, enable=['trac.*', 'codereview.*'])
70        PeerReviewModelProvider(self.env).environment_created()
71        GenericWorkflowModelProvider(self.env).environment_created()  # Needed for creating the workflow table
72        _prepare_review_data(self.env)
73        _prepare_file_data(self.env)
74        self.plugin = ResourceWorkflowSystem(self.env)
75        self.req = Mock(href=Href(''), perm=MockPerm(), redirect=lambda x: x)
76        self.req.authname = 'Tester'
77        self.req.path_info = '/workflowtransition'
78
79    def tearDown(self):
80        self.env.shutdown()
81
82    def test_get_review(self):
83        rm = PeerReviewModel(self.env, 1)
84        self.assertTrue(isinstance(rm, PeerReviewModel))
85        self.assertTrue(rm.exists)
86        self.assertEqual('name1', rm['name'])
87        self.assertEqual('bar', rm['status'])
88        self.assertEqual('note1', rm['notes'])
89        self.assertEqual('Rev1', rm['owner'])
90
91    def test_query_by_status(self):
92        rm = PeerReviewModel(self.env)
93        rm.clear_props()
94        rm['status'] = 'bar'
95        revs = list(rm.list_matching_objects())
96        self.assertEqual(2, len(revs))
97        rm = PeerReviewModel(self.env)
98        rm.clear_props()
99        rm['status'] = 'closed'
100        revs = list(rm.list_matching_objects())
101        self.assertEqual(1, len(revs))
102        rev = revs[0]
103        self.assertEqual('name2', rev['name'])
104        self.assertEqual('closed', rev['status'])
105        self.assertEqual('note2', rev['notes'])
106        self.assertEqual('Rev1', rev['owner'])
107
108    def test_query_by_owner(self):
109        rm = PeerReviewModel(self.env)
110        rm.clear_props()
111        rm['owner'] = 'Rev1'
112        revs = list(rm.list_matching_objects())
113        self.assertEqual(2, len(revs))
114
115    def test_change_note(self):
116        rm = PeerReviewModel(self.env, 1)
117        self.assertEqual(u'note1', rm['notes'])
118        rm['notes'] = u'note1_changed'
119        rm.save_changes()
120        rm = PeerReviewModel(self.env, 1)
121        self.assertEqual(u'note1_changed', rm['notes'])
122
123    def test_change_status(self):
124        from time import sleep
125        self.env.config.set("peerreview", "terminal_review_states", "closed, removed, obsolete")
126        self.env.config.save()
127        rev_pre = PeerReviewModel(self.env, 4)
128        self.assertEqual('Rev3', rev_pre['owner'])
129        self.assertEqual('foo', rev_pre['status'])
130        files = list(ReviewFileModel.select_by_review(self.env, rev_pre['review_id']))
131        self.assertEqual(2, len(files))
132        for item in files:
133            self.assertEqual('new', item['status'])
134
135        # Change review to 'closed' which is a terminal state
136
137        # On windows the timestamp resolution for 'when' in save_changes() is not microseconds
138        # (only updated in greater intervals).
139        # To prevent IntegrityError when writing to the database add some delay so we are sure
140        # that the time will be different.
141        sleep(0.02)
142        rev_pre.change_status('closed')
143        rev = PeerReviewModel(self.env, 4)
144        self.assertEqual('Rev3', rev['owner'])
145        self.assertEqual('closed', rev['status'])
146        # files must be closed now
147        files = list(ReviewFileModel.select_by_review(self.env, rev['review_id']))
148        self.assertEqual(2, len(files))
149        for item in files:
150            self.assertEqual('closed', item['status'])
151
152        # Do it again to check if something toggles the status
153        sleep(0.02)
154        rev_pre = PeerReviewModel(self.env, 4)
155        rev_pre.change_status('closed')
156        rev = PeerReviewModel(self.env, 4)
157        self.assertEqual('Rev3', rev['owner'])
158        self.assertEqual('closed', rev['status'])
159        # files must be closed now
160        files = list(ReviewFileModel.select_by_review(self.env, rev['review_id']))
161        self.assertEqual(2, len(files))
162        for item in files:
163            self.assertEqual('closed', item['status'])
164
165        # Change to another terminal state
166        sleep(0.02)
167        rev_pre = PeerReviewModel(self.env, 4)
168        rev_pre.change_status('removed')
169        rev = PeerReviewModel(self.env, 4)
170        self.assertEqual('Rev3', rev['owner'])
171        self.assertEqual('removed', rev['status'])
172        # files must have status 'removed' now
173        files = list(ReviewFileModel.select_by_review(self.env, rev['review_id']))
174        self.assertEqual(2, len(files))
175        for item in files:
176            self.assertEqual('removed', item['status'])
177
178        # Change to a non-terminal state
179        sleep(0.02)
180        rev_pre = PeerReviewModel(self.env, 4)
181        rev_pre.change_status('Baz')
182        rev = PeerReviewModel(self.env, 4)
183        self.assertEqual('Rev3', rev['owner'])
184        self.assertEqual('Baz', rev['status'])
185        # files must be 'new' now
186        files = list(ReviewFileModel.select_by_review(self.env, rev['review_id']))
187        self.assertEqual(2, len(files))
188        for item in files:
189            self.assertEqual('new', item['status'])
190
191
192    def test_review_workflow(self):
193        self.req.args = {
194            'id': '4',
195            'res_realm': 'peerreview',
196            'selected_action': 'approve'
197        }
198        rm = PeerReviewModel(self.env, 4)
199        self.assertEqual('foo', rm['status'])
200        # Do workflow transition
201        self.plugin.process_request(self.req)
202        rm_after = PeerReviewModel(self.env, 4)
203        self.assertEqual('approved', rm_after['status'])
204
205    def test_insert_new_review(self):
206        rev = PeerReviewModel(self.env)
207        rev['name'] = 'insert'
208        rev['owner'] = 'Tester'
209        rev.insert()
210        self.assertEqual(5, rev['review_id'])
211        all_revs = list(PeerReviewModel.select_all_reviews(self.env))
212        self.assertEqual(5, len(all_revs))
213
214    def test_select_all_reviews(self):
215        all_revs = list(PeerReviewModel.select_all_reviews(self.env))
216        self.assertEqual(4, len(all_revs))
217        # we have 4 different notes. This is an easy check if we got different objects.
218        seen = {}
219        for rev in all_revs:
220            seen[rev['notes']] = rev
221        self.assertEqual(4, len(seen))
222        # Same for name
223        seen = {}
224        for rev in all_revs:
225            seen[rev['name']] = rev
226        self.assertEqual(4, len(seen))
227        self.assertEqual('note2', seen['name2']['notes'])
228        self.assertEqual('Rev1', seen['name2']['owner'])
229        self.assertEqual('closed', seen['name2']['status'])
230
231    def test_reviews_by_period(self):
232        revs = PeerReviewModel.reviews_by_period(self.env,
233                                                 to_utimestamp(to_datetime(datetime(2019, 3, 4))),
234                                                 to_utimestamp(to_datetime(datetime(2019, 3, 14))))
235        self.assertEqual(2, len(revs))
236        # Reviews are ordered by 'created'
237        self.assertEqual('Rev1', revs[0]['owner'])
238        self.assertEqual('Rev2', revs[1]['owner'])
239
240        revs = PeerReviewModel.reviews_by_period(self.env,
241                                                 to_utimestamp(to_datetime(datetime(2019, 3, 4))),
242                                                 to_utimestamp(to_datetime(datetime(2019, 3, 13))))
243        self.assertEqual(1, len(revs))
244        self.assertEqual('Rev1', revs[0]['owner'])
245        revs = PeerReviewModel.reviews_by_period(self.env,
246                                                 to_utimestamp(to_datetime(datetime(2019, 3, 4))),
247                                                 to_utimestamp(to_datetime(datetime(2019, 4, 5))))
248        self.assertEqual(3, len(revs))
249        # Reviews are ordered by 'created'
250        self.assertEqual('Rev1', revs[0]['owner'])
251        self.assertEqual('Rev2', revs[1]['owner'])
252        self.assertEqual('Rev3', revs[2]['owner'])
253
254def review_model_suite():
255    suite = unittest.TestSuite()
256
257    suite.addTest(unittest.makeSuite(TestReviewModel))
258
259    return suite
Note: See TracBrowser for help on using the repository browser.