This instance will be upgraded to Heptapod 0.21.0rc2 on 2021-04-16 at 11:00 UTC+2 (a few minutes of down time)

qt_repoagent_test.py 8.47 KB
Newer Older
Yuya Nishihara's avatar
Yuya Nishihara committed
1
from __future__ import absolute_import
2

Yuya Nishihara's avatar
Yuya Nishihara committed
3 4 5 6
import mock
import os
import unittest

7 8 9 10
from mercurial import (
    pycompat,
)

Yuya Nishihara's avatar
Yuya Nishihara committed
11 12 13 14
from tortoisehg.hgqt.qtcore import (
    QEventLoop,
    QTimer,
)
15 16

from tortoisehg.hgqt import thgrepo
17
from tortoisehg.util import hglib
18 19 20

import helpers

21 22 23 24 25 26 27 28 29 30 31 32 33 34 35
class RepoAgentAttributeTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        tmpdir = helpers.mktmpdir(cls.__name__)
        cls.hg = hg = helpers.HgClient(tmpdir)
        hg.init()
        hg.fappend('foo', '0\n')
        hg.commit('-Am', 'append 0')

        hg.clone('.', '2')
        cls.hg2 = hg2 = helpers.HgClient(os.path.join(tmpdir, '2'))
        hg2.fappend('foo', '1\n')
        hg2.commit('-m', 'append 1')

    def setUp(self):
36
        ui = hglib.loadui()
37
        ui.setconfig(b'tortoisehg', b'monitorrepo', b'never')
38 39 40 41 42 43 44 45 46 47 48 49 50 51 52
        repo = thgrepo.repository(ui, self.hg.path)
        self.agent = thgrepo.RepoAgent(repo)

    def tearDown(self):
        thgrepo._repocache.clear()

    def test_keep_hiddenfilter_on_overlay_change(self):
        self.agent.setHiddenRevsIncluded(True)
        self.agent.setOverlay('union:%s' % self.hg2.path)
        self.assertTrue(self.agent.hiddenRevsIncluded())
        self.agent.setHiddenRevsIncluded(False)
        self.agent.clearOverlay()
        self.assertFalse(self.agent.hiddenRevsIncluded())


53
class _RepoAgentChangedSignalTestBase(object):
54 55 56
    @classmethod
    def setUpClass(cls):
        tmpdir = helpers.mktmpdir(cls.__name__)
57
        cls.srchg = hg = helpers.HgClient(tmpdir)
58
        hg.init()
59
        for i in pycompat.xrange(2):
60 61 62
            hg.fappend('foo', '%d\n' % i)
            hg.commit('-Am', 'append %d' % i)

63 64 65 66 67
        hg.clone('.', '2')
        cls.hg2 = hg2 = helpers.HgClient(os.path.join(tmpdir, '2'))
        hg2.fappend('foo', '2\n')
        hg2.commit('-m', 'append 2')

68 69
        cls.mtimedelay = helpers.guessmtimedelay(tmpdir)

70
    def setUp(self):
71
        testname = self.id().split('.')[-1]
72
        self.srchg.clone('--pull', '.', testname)  # --pull for rollback
73
        self.hg = helpers.HgClient(os.path.join(self.srchg.path, testname))
74 75
        self.hg.fappend('.hg/hgrc',
                        '[debug]\ndelaylock = %d\n' % self.mtimedelay)
76
        ui = hglib.loadui()
77
        ui.setconfig(b'tortoisehg', b'monitorrepo', self.monitorrepo)
78 79 80 81 82 83 84 85 86 87 88 89 90 91 92
        repo = thgrepo.repository(ui, self.hg.path)
        self.agent = agent = thgrepo.RepoAgent(repo)
        self.repositoryChanged = mock.Mock()
        agent.repositoryChanged.connect(self.repositoryChanged)
        agent.startMonitoringIfEnabled()

    def tearDown(self):
        self.agent.stopService()
        if self.agent.isServiceRunning():
            loop = QEventLoop()
            self.agent.serviceStopped.connect(loop.quit)
            QTimer.singleShot(5000, loop.quit)  # timeout
            loop.exec_()
        thgrepo._repocache.clear()

93 94 95 96 97 98 99 100 101 102
    def wait(self, timeout=5000):
        loop = QEventLoop()
        self.agent.busyChanged.connect(loop.quit)
        timer = QTimer(interval=timeout, singleShot=True)
        timer.timeout.connect(loop.quit)
        timer.start()
        while self.agent.isBusy() and timer.isActive():
            loop.exec_()
        self.assertFalse(self.agent.isBusy(), 'timeout while busy')

103
    def wait_changed(self, flags, timeout=5000):
104
        self.repositoryChanged.reset_mock()
105 106 107 108
        loop = QEventLoop()
        self.agent.repositoryChanged.connect(loop.quit)
        QTimer.singleShot(timeout, loop.quit)
        loop.exec_()
109
        self.repositoryChanged.assert_called_once_with(flags)
110 111 112
        self.assertFalse(self.agent.isBusy(),
                         'repositoryChanged emitted while busy')

113 114 115 116 117 118 119
    def test_no_changed_signal_while_command_running(self):
        self.agent.runCommand(['update', '0',
                               '--config', 'debug.delaylock=2000'])
        self.hg.update('0')
        self.wait_changed(thgrepo.WorkingParentChanged
                          | thgrepo.WorkingStateChanged)

120
    def test_poll_on_command_finished(self):
121
        self.agent.runCommandSequence([['update', '0'], ['root']])
122 123
        self.wait_changed(thgrepo.WorkingParentChanged
                          | thgrepo.WorkingStateChanged)
124

125 126 127
    def test_rollback_to_null(self):
        self.agent.runCommand(['rollback'])
        # it seems "rollback" does not touch dirstate
128 129 130 131 132 133 134 135 136 137 138 139 140 141
        self.wait_changed(thgrepo.LogChanged
                          | thgrepo.WorkingParentChanged
                          | thgrepo.WorkingStateChanged)

    def test_rollback_commit(self):
        self.hg.ftouch('bar')
        self.agent.runCommand(['commit', '-Am', 'change'])
        self.wait_changed(thgrepo.LogChanged
                          | thgrepo.WorkingParentChanged
                          | thgrepo.WorkingStateChanged)
        self.agent.runCommand(['rollback'])  # mtime of dirstate will go back
        self.wait_changed(thgrepo.LogChanged
                          | thgrepo.WorkingParentChanged
                          | thgrepo.WorkingStateChanged)
142

143
    def test_branch_changed(self):
144 145
        self.agent.runCommand(['branch', 'foo'])
        self.wait_changed(thgrepo.WorkingBranchChanged)
146 147 148 149

    def test_branch_and_parent_changed(self):
        self.agent.runCommandSequence([['update', '0'],
                                       ['branch', 'foo']])
150 151
        self.wait_changed(thgrepo.WorkingBranchChanged
                          | thgrepo.WorkingParentChanged
152
                          | thgrepo.WorkingStateChanged)
153

154 155 156 157 158 159 160
    def test_listen_to_new_queue_changes(self):
        self.agent.runCommand(['qqueue', '-c', 'foo'])
        self.wait_changed(thgrepo.LogChanged)
        self.agent.runCommandSequence([['export', 'tip', '-o', 'patch'],
                                       ['qimport', 'patch']])
        self.wait_changed(thgrepo.LogChanged)

161 162
    def test_invalidate_on_dirstate_changed(self):
        repo = self.agent.rawRepo()
163
        self.assertEqual({'foo'}, set(repo.dirstate))  # preload cache
164 165 166
        self.hg.ftouch('bar')
        self.agent.runCommand(['add', 'bar'])
        self.wait()
167
        self.assertEqual({'bar', 'foo'}, set(repo.dirstate))
168

169 170
    def test_immediate_hiddenfilter(self):
        self.agent.setHiddenRevsIncluded(not self.agent.hiddenRevsIncluded())
171
        self.repositoryChanged.assert_called_once_with(thgrepo.LogChanged)
172 173

    def test_hiddenfilter_while_command_running(self):
174
        self.agent.runCommand(['root'])
175 176
        self.agent.setHiddenRevsIncluded(not self.agent.hiddenRevsIncluded())
        self.assertFalse(self.repositoryChanged.called)
177
        self.wait_changed(thgrepo.LogChanged)
178

179 180
    def test_immediate_overlay(self):
        self.agent.setOverlay('union:%s' % self.hg2.path)
181
        self.repositoryChanged.assert_called_once_with(thgrepo.LogChanged)
182 183
        self.repositoryChanged.reset_mock()
        self.agent.clearOverlay()
184
        self.repositoryChanged.assert_called_once_with(thgrepo.LogChanged)
185 186

    def test_overlay_while_command_running(self):
187
        self.agent.runCommand(['root'])
188 189
        self.agent.setOverlay('union:%s' % self.hg2.path)
        self.assertFalse(self.repositoryChanged.called)
190
        self.wait_changed(thgrepo.LogChanged)
191 192 193 194

    def test_clearoverlay_while_command_running(self):
        self.agent.setOverlay('union:%s' % self.hg2.path)
        self.repositoryChanged.reset_mock()
195
        self.agent.runCommand(['root'])
196 197
        self.agent.clearOverlay()
        self.assertFalse(self.repositoryChanged.called)
198
        self.wait_changed(thgrepo.LogChanged)
199 200


201
class RepoAgentChangedSignalWithMonitorTest(_RepoAgentChangedSignalTestBase, unittest.TestCase):
202
    monitorrepo = b'always'
203 204 205 206 207 208

    def test_filesystem_watcher(self):
        self.hg.update('0')
        self.wait_changed(thgrepo.WorkingParentChanged
                          | thgrepo.WorkingStateChanged)

209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226
    def test_resume_monitoring_while_command_running(self):
        self.agent.suspendMonitoring()
        self.agent.runCommand(['root'])
        self.agent.resumeMonitoring()
        self.wait()
        self.hg.update('0')
        self.wait_changed(thgrepo.WorkingParentChanged
                          | thgrepo.WorkingStateChanged)

    def test_resume_monitoring_with_overlay(self):
        self.agent.suspendMonitoring()
        self.agent.setOverlay('union:%s' % self.hg2.path)
        self.agent.resumeMonitoring()
        self.agent.clearOverlay()
        self.hg.update('0')
        self.wait_changed(thgrepo.WorkingParentChanged
                          | thgrepo.WorkingStateChanged)

227

228
class RepoAgentChangedSignalWithoutMonitorTest(_RepoAgentChangedSignalTestBase, unittest.TestCase):
229
    monitorrepo = b'never'