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_cmdagent_test.py 12.5 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
import mock
import unittest

6 7 8 9
from mercurial import (
    pycompat,
)

Yuya Nishihara's avatar
Yuya Nishihara committed
10
from tortoisehg.hgqt.qtcore import (
11
    QBuffer,
Yuya Nishihara's avatar
Yuya Nishihara committed
12 13
    QCoreApplication,
    QEventLoop,
14
    QIODevice,
Yuya Nishihara's avatar
Yuya Nishihara committed
15 16 17 18
    QObject,
    QTimer,
    pyqtSlot,
)
19 20

from tortoisehg.hgqt import cmdcore
21
from tortoisehg.util import hglib
22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42

import helpers

class CmdWaiter(QObject):
    def __init__(self, session):
        super(CmdWaiter, self).__init__()
        self._session = session
        self._outputs = []
        self._session.outputReceived.connect(self._captureOutput)

    def outputString(self):
        return ''.join(self._outputs).rstrip()

    def wait(self, timeout=5000):
        if self._session.isFinished():
            return
        loop = QEventLoop()
        self._session.commandFinished.connect(loop.quit)
        QTimer.singleShot(timeout, loop.quit)
        loop.exec_()

43
    @pyqtSlot(str, str)
44 45
    def _captureOutput(self, msg, label):
        if not label:
46
            self._outputs.append(pycompat.unicode(msg))
47 48


49 50 51 52 53 54 55 56 57
def waitForCmdStarted(session, timeout=5000):
    if session.isRunning() or session.isFinished():
        return
    loop = QEventLoop()
    session.controlMessage.connect(loop.quit)  # wait for initial banner
    QTimer.singleShot(timeout, loop.quit)
    loop.exec_()


58
class _CmdAgentTestBase(object):
59 60 61 62 63 64 65 66 67
    @classmethod
    def setUpClass(cls):
        tmpdir = helpers.mktmpdir(cls.__name__)
        cls.hg = hg = helpers.HgClient(tmpdir)
        hg.init()
        hg.ftouch('foo')
        hg.commit('-Am', 'add foo')

    def setUp(self):
68
        ui = hglib.loadui()
69 70
        self.agent = agent = cmdcore.CmdAgent(ui, cwd=self.hg.path,
                                              worker=self.workername)
71
        self.busyChanged = mock.Mock()
72
        self.commandFinished = mock.Mock()
73
        agent.busyChanged.connect(self.busyChanged)
74
        agent.commandFinished.connect(self.commandFinished)
75

76 77 78 79 80 81 82 83
    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_()

84 85 86 87 88 89 90 91 92 93 94 95
    def test_runcommand(self):
        sess = self.agent.runCommand(['root'])
        self._check_runcommand(sess, self.hg.path)

    def test_runcommandseq(self):
        sess = self.agent.runCommandSequence([['root'], ['id', '-i']])
        self._check_runcommand(sess, '\n'.join([self.hg.path, '53245c60e682']))

    def test_runcommandseq_firsterror(self):
        sess = self.agent.runCommandSequence([['id', '-r100'], ['root']])
        self._check_runcommand(sess, '', 255)

96 97 98
    def test_runcommand_delayedstart(self):
        sess = self.agent.runCommand(['root'])
        self.assertFalse(sess.isRunning())
99
        waitForCmdStarted(sess)
100 101 102
        self.assertTrue(sess.isRunning())
        self._check_runcommand(sess, self.hg.path)

103 104
    def test_runcommand_queued(self):
        sess1 = self.agent.runCommand(['id', '-i'])
105
        waitForCmdStarted(sess1)
106 107 108 109 110
        self.assertTrue(sess1.isRunning())
        sess2 = self.agent.runCommand(['id', '-n'])
        self.assertFalse(sess2.isRunning())

        self._check_runcommand(sess1, '53245c60e682')
111
        QCoreApplication.processEvents()  # should start in next loop
112 113 114 115 116
        self.assertTrue(sess2.isRunning())
        self._check_runcommand(sess2, '0')

    def test_runcommand_signal_chain(self):
        sess = self.agent.runCommand(['id', '-i'])
117 118 119
        chainedsessq = []
        sess.commandFinished.connect(
            lambda: chainedsessq.append(self.agent.runCommand(['id', '-n'])))
120
        self._check_runcommand(sess, '53245c60e682')
121 122
        self.assertTrue(chainedsessq)
        sess = chainedsessq.pop(0)
123
        QCoreApplication.processEvents()  # should start in next loop
124 125 126
        self.assertTrue(sess.isRunning())
        self._check_runcommand(sess, '0')

127 128 129 130 131 132 133 134 135 136 137 138 139
    def test_captureout(self):
        sess = self.agent.runCommandSequence(
            [['root'], ['id', '-i'], ['cat', 'unknown']])
        sess.setCaptureOutput(True)
        readyRead = mock.Mock()
        sess.readyRead.connect(readyRead)
        self._check_runcommand(sess, '', 1)
        self.assertTrue('no such file' in sess.warningString())  # not captured
        self.assertTrue(readyRead.called)
        self.assertTrue(sess.canReadLine())
        self.assertEqual(self.hg.path + '\n', sess.readLine())
        self.assertEqual('5', sess.read(1))
        self.assertEqual('3', str(sess.peek(1)))
140
        self.assertEqual(b'3245c60e682\n', bytes(sess.readAll()))
141

142 143 144 145 146 147 148 149
    def _check_runcommand(self, sess, expectedout, expectedcode=0):
        self.assertFalse(sess.isFinished())
        waiter = CmdWaiter(sess)
        waiter.wait()
        self.assertTrue(sess.isFinished())
        self.assertFalse(sess.isRunning())
        self.assertEqual(expectedcode, sess.exitCode())
        self.assertEqual(expectedout, waiter.outputString())
150
        self.commandFinished.assert_any_call(sess)
151 152 153 154 155

    def test_abort_session(self):
        sess = self.agent.runCommand(['log'])
        finished = mock.Mock()
        sess.commandFinished.connect(finished)
156 157
        waitForCmdStarted(sess)
        self.assertTrue(sess.isRunning())
158
        sess.abort()
159 160 161 162 163 164 165 166 167 168
        self._check_abort_session(sess)
        self.assertEqual(1, finished.call_count)

    def test_abort_session_not_running(self):
        sess1 = self.agent.runCommand(['id', '-i'])
        sess2 = self.agent.runCommand(['id', '-n'])
        finished = mock.Mock()
        sess1.commandFinished.connect(finished.sess1)
        sess2.commandFinished.connect(finished.sess2)

169
        # waiting session aborts immediately
170
        sess2.abort()
171 172
        self.assertTrue(sess2.isAborted())
        self.assertTrue(sess2.isFinished())
173
        CmdWaiter(sess1).wait()
174 175 176
        # "finished" signal of waiting session should be emitted, because the
        # session object is known to the client
        self.assertEqual(['sess2', 'sess1'],
177
                         [x[0] for x in finished.method_calls])
178 179 180 181 182
        # but agent's signal should be emitted only for the active session
        self.assertEqual([mock.call(sess1)],
                         self.commandFinished.call_args_list)

    def test_abort_all_waiting_sessions(self):
183 184
        sessions = pycompat.maplist(self.agent.runCommand,
                                    [['id', '-i'], ['id', '-n'], ['root']])
185 186 187 188 189 190 191 192 193 194
        self.assertTrue(self.agent.isBusy())
        self.busyChanged.reset_mock()
        # abort from waiting one
        for sess in sessions[1:] + [sessions[0]]:
            sess.abort()
            self.assertTrue(sess.isAborted())
            self.assertTrue(sess.isFinished())
        # sessions should be deleted from queue immediately
        self.assertFalse(self.agent.isBusy())
        self.busyChanged.assert_called_once_with(False)
195

196
    def test_abortcommands(self):
197 198
        sessions = pycompat.maplist(self.agent.runCommand,
                                    [['id', '-i'], ['id', '-n'], ['root']])
199 200
        waitForCmdStarted(sessions[0])
        # sess0 (running), sess1 (waiting), ...
201 202 203 204 205
        self.agent.abortCommands()
        for sess in sessions:
            self._check_abort_session(sess)
        self.assertFalse(self.agent.isBusy())

206
    def _check_abort_session(self, sess):
207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232
        waiter = CmdWaiter(sess)
        waiter.wait()
        self.assertTrue(sess.isAborted())
        self.assertTrue(sess.isFinished())
        self.assertFalse(sess.isRunning())

    def test_busystate(self):
        self.assertFalse(self.agent.isBusy())
        sess = self.agent.runCommand(['id'])
        self.assertTrue(self.agent.isBusy())
        self.busyChanged.assert_called_once_with(True)
        self.busyChanged.reset_mock()

        CmdWaiter(sess).wait()
        self.assertFalse(self.agent.isBusy())
        self.busyChanged.assert_called_once_with(False)

    def test_busystate_queued(self):
        sess1 = self.agent.runCommand(['id'])
        self.busyChanged.assert_called_once_with(True)
        self.busyChanged.reset_mock()

        sess2 = self.agent.runCommand(['id'])
        self.assertTrue(self.agent.isBusy())

        CmdWaiter(sess1).wait()
233 234
        self.assertTrue(self.agent.isBusy())
        self.assertFalse(self.busyChanged.called)
235 236 237 238

        CmdWaiter(sess2).wait()
        self.assertFalse(self.agent.isBusy())
        self.busyChanged.assert_called_once_with(False)
239

240 241 242 243 244 245 246 247
    def test_busystate_on_finished(self):
        self.agent.commandFinished.connect(self._check_busystate_on_finished)
        sess = self.agent.runCommand(['id'])
        CmdWaiter(sess).wait()

    def _check_busystate_on_finished(self):
        # busy state should be off when the last commandFinished emitted
        self.assertFalse(self.agent.isBusy())
248 249


250
class CmdAgentProcTest(_CmdAgentTestBase, unittest.TestCase):
251
    workername = 'proc'
252 253


254
class CmdAgentServerTest(_CmdAgentTestBase, unittest.TestCase):
255 256
    workername = 'server'

257 258 259 260
    @classmethod
    def setUpClass(cls):
        super(CmdAgentServerTest, cls).setUpClass()
        cls.hg.fwrite('testext.py',
261
                      'from binascii import unhexlify\n'
262
                      'from mercurial import registrar\n'
263
                      'cmdtable = {}\n'
264
                      'command = registrar.command(cmdtable)\n'
265
                      '@command(b"echoback")\n'
266 267
                      'def echoback(ui, repo):\n'
                      '    ui.write(ui.fin.read())\n'
268
                      '@command(b"writestdout")\n'
269
                      'def writestdout(ui, repo, *data):\n'
270
                      '    stdout = ui.fout.out\n'
271
                      '    stdout.write("".join(map(unhexlify, data)))\n'
272
                      '    stdout.flush()\n')
273 274
        cls.hg.fappend('.hg/hgrc', '[extensions]\ntestext = testext.py\n')

275 276 277 278 279 280 281 282 283
    def test_abort_session_waiting_for_worker(self):
        sess = self.agent.runCommand(['id', '-i'])
        QCoreApplication.processEvents()  # start session
        self.assertTrue(sess._worker)
        self.assertNotEqual(cmdcore.CmdWorker.Ready,
                            sess._worker.serviceState())
        sess.abort()
        CmdWaiter(sess).wait()
        self.assertTrue(sess.isAborted())
284

285 286 287
    def test_runcommand_while_service_stopping(self):
        sess1 = self.agent.runCommand(['id', '-i'])  # start server
        CmdWaiter(sess1).wait()
288
        worker = self.agent._worker
289 290 291 292 293 294 295 296 297
        self.assertEqual(cmdcore.CmdWorker.Ready, worker.serviceState())
        self.agent.stopService()
        self.assertEqual(cmdcore.CmdWorker.Stopping, worker.serviceState())
        # start new session while server is shutting down
        sess2 = self.agent.runCommand(['id', '-n'])
        CmdWaiter(sess2).wait()
        self.assertTrue(sess2.isFinished())
        self.assertEqual(0, sess2.exitCode())  # should not be canceled

298 299 300 301 302 303 304
    @mock.patch('tortoisehg.util.paths.get_hg_command',
                return_value=['/inexistent'])
    def test_server_failed_to_start(self, m):
        sess = self.agent.runCommand(['id', '-i'])
        CmdWaiter(sess).wait()
        self.assertTrue(sess.isFinished())
        self.assertEqual(-1, sess.exitCode())
305 306 307

    def test_stop_server_by_data_timeout(self):
        CmdWaiter(self.agent.runCommand(['id'])).wait()  # start server
308
        worker = self.agent._worker
309
        worker._readtimer.setInterval(100)  # avoid long wait
310 311
        sess = self.agent.runCommand(['writestdout',
                                      'o'.encode('hex'), '00000001'])
312 313
        CmdWaiter(sess).wait()
        self.assertEqual(cmdcore.CmdWorker.NotRunning, worker.serviceState())
314

315 316 317 318 319 320 321 322 323
    def test_stop_server_by_unexpected_required_channel(self):
        CmdWaiter(self.agent.runCommand(['id'])).wait()  # start server
        worker = self.agent._worker
        sess = self.agent.runCommand(['writestdout',
                                      'X'.encode('hex'), '00000000'])
        CmdWaiter(sess).wait()
        failstates = (cmdcore.CmdWorker.Stopping, cmdcore.CmdWorker.NotRunning)
        self.assertTrue(worker.serviceState() in failstates)

324 325 326 327 328
    def test_unicode_error_in_runcommand(self):
        sess = self.agent.runCommand(['files', '-X', u'\u3000'])
        CmdWaiter(sess).wait()
        self.assertEqual(-1, sess.exitCode())
        self.assertTrue('failed to encode command' in sess.errorString())
329 330 331 332 333 334 335 336 337 338 339 340

    def test_data_input(self):
        buf = QBuffer()
        buf.setData(b'foo')
        buf.open(QIODevice.ReadOnly)
        sess = self.agent.runCommand(['echoback'])
        sess.setInputDevice(buf)
        self._check_runcommand(sess, b'foo')

    def test_data_input_unset(self):
        sess = self.agent.runCommand(['echoback'])
        self._check_runcommand(sess, b'')