KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use linesep_bytes method in tox
Best Python code snippet using tox_python
test_hgpoller.py
Source:test_hgpoller.py 
1# This file is part of Buildbot. Buildbot is free software: you can2# redistribute it and/or modify it under the terms of the GNU General Public3# License as published by the Free Software Foundation, version 2.4#5# This program is distributed in the hope that it will be useful, but WITHOUT6# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS7# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more8# details.9#10# You should have received a copy of the GNU General Public License along with11# this program; if not, write to the Free Software Foundation, Inc., 5112# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.13#14# Copyright Buildbot Team Members15import os16from twisted.internet import defer17from twisted.trial import unittest18from buildbot.changes import hgpoller19from buildbot.test.util import changesource20from buildbot.test.util import gpo21from buildbot.test.util.misc import TestReactorMixin22ENVIRON_2116_KEY = 'TEST_THAT_ENVIRONMENT_GETS_PASSED_TO_SUBPROCESSES'23LINESEP_BYTES = os.linesep.encode("ascii")24PATHSEP_BYTES = os.pathsep.encode("ascii")25class TestHgPollerBase(gpo.GetProcessOutputMixin,26 changesource.ChangeSourceMixin,27 TestReactorMixin,28 unittest.TestCase):29 usetimestamps = True30 branches = None31 bookmarks = None32 @defer.inlineCallbacks33 def setUp(self):34 self.setUpTestReactor()35 # To test that environment variables get propagated to subprocesses36 # (See #2116)37 os.environ[ENVIRON_2116_KEY] = 'TRUE'38 self.setUpGetProcessOutput()39 yield self.setUpChangeSource()40 self.remote_repo = 'ssh://example.com/foo/baz'41 self.remote_hgweb = 'http://example.com/foo/baz/rev/{}'42 self.repo_ready = True43 def _isRepositoryReady():44 return self.repo_ready45 self.poller = hgpoller.HgPoller(self.remote_repo,46 usetimestamps=self.usetimestamps,47 workdir='/some/dir',48 branches=self.branches,49 bookmarks=self.bookmarks,50 revlink=lambda branch, revision:51 self.remote_hgweb.format(revision))52 yield self.poller.setServiceParent(self.master)53 self.poller._isRepositoryReady = _isRepositoryReady54 yield self.master.db.setup()55 @defer.inlineCallbacks56 def check_current_rev(self, wished, branch='default'):57 rev = yield self.poller._getCurrentRev(branch)58 self.assertEqual(rev, str(wished))59class TestHgPollerBranches(TestHgPollerBase):60 branches = ['one', 'two']61 @defer.inlineCallbacks62 def test_poll_initial(self):63 self.expectCommands(64 gpo.Expect('hg', 'pull', '-b', 'one', '-b', 'two',65 'ssh://example.com/foo/baz')66 .path('/some/dir'),67 gpo.Expect(68 'hg', 'heads', 'one', '--template={rev}' + os.linesep)69 .path('/some/dir').stdout(b"73591"),70 gpo.Expect(71 'hg', 'heads', 'two', '--template={rev}' + os.linesep)72 .path('/some/dir').stdout(b"22341"),73 )74 # do the poll75 yield self.poller.poll()76 # check the results77 self.assertEqual(len(self.master.data.updates.changesAdded), 0)78 yield self.check_current_rev(73591, 'one')79 yield self.check_current_rev(22341, 'two')80 @defer.inlineCallbacks81 def test_poll_regular(self):82 # normal operation. There's a previous revision, we get a new one.83 # Let's say there was an intervening commit on an untracked branch, to84 # make it more interesting.85 self.expectCommands(86 gpo.Expect('hg', 'pull', '-b', 'one', '-b', 'two',87 'ssh://example.com/foo/baz')88 .path('/some/dir'),89 gpo.Expect(90 'hg', 'heads', 'one', '--template={rev}' + os.linesep)91 .path('/some/dir').stdout(b'6' + LINESEP_BYTES),92 gpo.Expect('hg', 'log', '-r', '4::6',93 '--template={rev}:{node}\\n')94 .path('/some/dir').stdout(LINESEP_BYTES.join([95 b'4:1aaa5',96 b'6:784bd',97 ])),98 gpo.Expect('hg', 'log', '-r', '784bd',99 '--template={date|hgdate}' + os.linesep +100 '{author}' + os.linesep +101 "{files % '{file}" +102 os.pathsep + "'}" +103 os.linesep + '{desc|strip}')104 .path('/some/dir').stdout(LINESEP_BYTES.join([105 b'1273258009.0 -7200',106 b'Joe Test <joetest@example.org>',107 b'file1 file2',108 b'Comment',109 b''])),110 gpo.Expect(111 'hg', 'heads', 'two', '--template={rev}' + os.linesep)112 .path('/some/dir').stdout(b'3' + LINESEP_BYTES),113 )114 yield self.poller._setCurrentRev(3, 'two')115 yield self.poller._setCurrentRev(4, 'one')116 yield self.poller.poll()117 yield self.check_current_rev(6, 'one')118 self.assertEqual(len(self.master.data.updates.changesAdded), 1)119 change = self.master.data.updates.changesAdded[0]120 self.assertEqual(change['revision'], '784bd')121 self.assertEqual(change['revlink'], 'http://example.com/foo/baz/rev/784bd')122 self.assertEqual(change['comments'], 'Comment')123class TestHgPollerBookmarks(TestHgPollerBase):124 bookmarks = ['one', 'two']125 @defer.inlineCallbacks126 def test_poll_initial(self):127 self.expectCommands(128 gpo.Expect('hg', 'pull', '-B', 'one', '-B', 'two',129 'ssh://example.com/foo/baz')130 .path('/some/dir'),131 gpo.Expect(132 'hg', 'heads', 'one', '--template={rev}' + os.linesep)133 .path('/some/dir').stdout(b"73591"),134 gpo.Expect(135 'hg', 'heads', 'two', '--template={rev}' + os.linesep)136 .path('/some/dir').stdout(b"22341"),137 )138 # do the poll139 yield self.poller.poll()140 # check the results141 self.assertEqual(len(self.master.data.updates.changesAdded), 0)142 yield self.check_current_rev(73591, 'one')143 yield self.check_current_rev(22341, 'two')144 @defer.inlineCallbacks145 def test_poll_regular(self):146 # normal operation. There's a previous revision, we get a new one.147 # Let's say there was an intervening commit on an untracked branch, to148 # make it more interesting.149 self.expectCommands(150 gpo.Expect('hg', 'pull', '-B', 'one', '-B', 'two',151 'ssh://example.com/foo/baz')152 .path('/some/dir'),153 gpo.Expect(154 'hg', 'heads', 'one', '--template={rev}' + os.linesep)155 .path('/some/dir').stdout(b'6' + LINESEP_BYTES),156 gpo.Expect('hg', 'log', '-r', '4::6',157 '--template={rev}:{node}\\n')158 .path('/some/dir').stdout(LINESEP_BYTES.join([159 b'4:1aaa5',160 b'6:784bd',161 ])),162 gpo.Expect('hg', 'log', '-r', '784bd',163 '--template={date|hgdate}' + os.linesep +164 '{author}' + os.linesep +165 "{files % '{file}" +166 os.pathsep + "'}" +167 os.linesep + '{desc|strip}')168 .path('/some/dir').stdout(LINESEP_BYTES.join([169 b'1273258009.0 -7200',170 b'Joe Test <joetest@example.org>',171 b'file1 file2',172 b'Comment',173 b''])),174 gpo.Expect(175 'hg', 'heads', 'two', '--template={rev}' + os.linesep)176 .path('/some/dir').stdout(b'3' + LINESEP_BYTES),177 )178 yield self.poller._setCurrentRev(3, 'two')179 yield self.poller._setCurrentRev(4, 'one')180 yield self.poller.poll()181 yield self.check_current_rev(6, 'one')182 self.assertEqual(len(self.master.data.updates.changesAdded), 1)183 change = self.master.data.updates.changesAdded[0]184 self.assertEqual(change['revision'], '784bd')185 self.assertEqual(change['comments'], 'Comment')186class TestHgPoller(TestHgPollerBase):187 def tearDown(self):188 del os.environ[ENVIRON_2116_KEY]189 return self.tearDownChangeSource()190 def gpoFullcommandPattern(self, commandName, *expected_args):191 """Match if the command is commandName and arg list start as expected.192 This allows to test a bit more if expected GPO are issued, be it193 by obscure failures due to the result not being given.194 """195 def matchesSubcommand(bin, given_args, **kwargs):196 return bin == commandName and tuple(197 given_args[:len(expected_args)]) == expected_args198 return matchesSubcommand199 def test_describe(self):200 self.assertSubstring("HgPoller", self.poller.describe())201 def test_name(self):202 self.assertEqual(self.remote_repo, self.poller.name)203 # and one with explicit name...204 other = hgpoller.HgPoller(205 self.remote_repo, name="MyName", workdir='/some/dir')206 self.assertEqual("MyName", other.name)207 # and one with explicit branches...208 other = hgpoller.HgPoller(209 self.remote_repo, branches=["b1", "b2"], workdir='/some/dir')210 self.assertEqual(self.remote_repo + "_b1_b2", other.name)211 def test_hgbin_default(self):212 self.assertEqual(self.poller.hgbin, "hg")213 @defer.inlineCallbacks214 def test_poll_initial(self):215 self.repo_ready = False216 # Test that environment variables get propagated to subprocesses217 # (See #2116)218 expected_env = {ENVIRON_2116_KEY: 'TRUE'}219 self.addGetProcessOutputExpectEnv(expected_env)220 self.expectCommands(221 gpo.Expect('hg', 'init', '/some/dir'),222 gpo.Expect('hg', 'pull', '-b', 'default',223 'ssh://example.com/foo/baz')224 .path('/some/dir'),225 gpo.Expect(226 'hg', 'heads', 'default', '--template={rev}' + os.linesep)227 .path('/some/dir').stdout(b"73591"),228 )229 # do the poll230 yield self.poller.poll()231 # check the results232 self.assertEqual(len(self.master.data.updates.changesAdded), 0)233 yield self.check_current_rev(73591)234 @defer.inlineCallbacks235 def test_poll_several_heads(self):236 # If there are several heads on the named branch, the poller mustn't237 # climb (good enough for now, ideally it should even go to the common238 # ancestor)239 self.expectCommands(240 gpo.Expect('hg', 'pull', '-b', 'default',241 'ssh://example.com/foo/baz')242 .path('/some/dir'),243 gpo.Expect(244 'hg', 'heads', 'default', '--template={rev}' + os.linesep)245 .path('/some/dir').stdout(b'5' + LINESEP_BYTES + b'6' + LINESEP_BYTES)246 )247 yield self.poller._setCurrentRev(3)248 # do the poll: we must stay at rev 3249 yield self.poller.poll()250 yield self.check_current_rev(3)251 @defer.inlineCallbacks252 def test_poll_regular(self):253 # normal operation. There's a previous revision, we get a new one.254 self.expectCommands(255 gpo.Expect('hg', 'pull', '-b', 'default',256 'ssh://example.com/foo/baz')257 .path('/some/dir'),258 gpo.Expect(259 'hg', 'heads', 'default', '--template={rev}' + os.linesep)260 .path('/some/dir').stdout(b'5' + LINESEP_BYTES),261 gpo.Expect('hg', 'log', '-r', '4::5',262 '--template={rev}:{node}\\n')263 .path('/some/dir').stdout(LINESEP_BYTES.join([264 b'4:1aaa5',265 b'5:784bd',266 ])),267 gpo.Expect('hg', 'log', '-r', '784bd',268 '--template={date|hgdate}' + os.linesep +269 '{author}' + os.linesep +270 "{files % '{file}" +271 os.pathsep + "'}" +272 os.linesep + '{desc|strip}')273 .path('/some/dir').stdout(LINESEP_BYTES.join([274 b'1273258009.0 -7200',275 b'Joe Test <joetest@example.org>',276 b'file1 file2',277 b'Comment for rev 5',278 b''])),279 )280 yield self.poller._setCurrentRev(4)281 yield self.poller.poll()282 yield self.check_current_rev(5)283 self.assertEqual(len(self.master.data.updates.changesAdded), 1)284 change = self.master.data.updates.changesAdded[0]285 self.assertEqual(change['revision'], '784bd')286 self.assertEqual(change['comments'], 'Comment for rev 5')287 @defer.inlineCallbacks288 def test_poll_force_push(self):289 # There's a previous revision, but not linked with new rev290 self.expectCommands(291 gpo.Expect('hg', 'pull', '-b', 'default',292 'ssh://example.com/foo/baz')293 .path('/some/dir'),294 gpo.Expect(295 'hg', 'heads', 'default', '--template={rev}' + os.linesep)296 .path('/some/dir').stdout(b'5' + LINESEP_BYTES),297 gpo.Expect('hg', 'log', '-r', '4::5',298 '--template={rev}:{node}\\n')299 .path('/some/dir').stdout(b""),300 gpo.Expect('hg', 'log', '-r', '5',301 '--template={rev}:{node}\\n')302 .path('/some/dir').stdout(LINESEP_BYTES.join([303 b'5:784bd',304 ])),305 gpo.Expect('hg', 'log', '-r', '784bd',306 '--template={date|hgdate}' + os.linesep +307 '{author}' + os.linesep +308 "{files % '{file}" +309 os.pathsep + "'}" +310 os.linesep + '{desc|strip}')311 .path('/some/dir').stdout(LINESEP_BYTES.join([312 b'1273258009.0 -7200',313 b'Joe Test <joetest@example.org>',314 b'file1 file2',315 b'Comment for rev 5',316 b''])),317 )318 yield self.poller._setCurrentRev(4)319 yield self.poller.poll()320 yield self.check_current_rev(5)321 self.assertEqual(len(self.master.data.updates.changesAdded), 1)322 change = self.master.data.updates.changesAdded[0]323 self.assertEqual(change['revision'], '784bd')324 self.assertEqual(change['comments'], 'Comment for rev 5')325class HgPollerNoTimestamp(TestHgPoller):326 """ Test HgPoller() without parsing revision commit timestamp """...
9912_test_changes_hgpoller.py
Source:9912_test_changes_hgpoller.py
1# This file is part of Buildbot. Buildbot is free software: you can2# redistribute it and/or modify it under the terms of the GNU General Public3# License as published by the Free Software Foundation, version 2.4#5# This program is distributed in the hope that it will be useful, but WITHOUT6# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS7# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more8# details.9#10# You should have received a copy of the GNU General Public License along with11# this program; if not, write to the Free Software Foundation, Inc., 5112# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.13#14# Copyright Buildbot Team Members15from __future__ import absolute_import16from __future__ import print_function17import os18from twisted.internet import defer19from twisted.trial import unittest20from buildbot.changes import hgpoller21from buildbot.test.util import changesource22from buildbot.test.util import gpo23ENVIRON_2116_KEY = 'TEST_THAT_ENVIRONMENT_GETS_PASSED_TO_SUBPROCESSES'24LINESEP_BYTES = os.linesep.encode("ascii")25PATHSEP_BYTES = os.pathsep.encode("ascii")26class TestHgPoller(gpo.GetProcessOutputMixin,27 changesource.ChangeSourceMixin,28 unittest.TestCase):29 usetimestamps = True30 def setUp(self):31 # To test that environment variables get propagated to subprocesses32 # (See #2116)33 os.environ[ENVIRON_2116_KEY] = 'TRUE'34 self.setUpGetProcessOutput()35 d = self.setUpChangeSource()36 self.remote_repo = 'ssh://example.com/foo/baz'37 self.branch = 'default'38 self.repo_ready = True39 def _isRepositoryReady():40 return self.repo_ready41 def create_poller(_):42 self.poller = hgpoller.HgPoller(self.remote_repo,43 usetimestamps=self.usetimestamps,44 workdir='/some/dir')45 self.poller.setServiceParent(self.master)46 self.poller._isRepositoryReady = _isRepositoryReady47 d.addCallback(create_poller)48 def create_db(_):49 return self.master.db.setup()50 d.addCallback(create_db)51 return d52 def tearDown(self):53 del os.environ[ENVIRON_2116_KEY]54 return self.tearDownChangeSource()55 def gpoFullcommandPattern(self, commandName, *expected_args):56 """Match if the command is commandName and arg list start as expected.57 This allows to test a bit more if expected GPO are issued, be it58 by obscure failures due to the result not being given.59 """60 def matchesSubcommand(bin, given_args, **kwargs):61 return bin == commandName and tuple(62 given_args[:len(expected_args)]) == expected_args63 return matchesSubcommand64 def test_describe(self):65 self.assertSubstring("HgPoller", self.poller.describe())66 def test_name(self):67 self.assertEqual(68 "%s[%s]" % (self.remote_repo, self.branch), self.poller.name)69 # and one with explicit name...70 other = hgpoller.HgPoller(71 self.remote_repo, name="MyName", workdir='/some/dir')72 self.assertEqual("MyName", other.name)73 def test_hgbin_default(self):74 self.assertEqual(self.poller.hgbin, "hg")75 def test_poll_initial(self):76 self.repo_ready = False77 # Test that environment variables get propagated to subprocesses78 # (See #2116)79 expected_env = {ENVIRON_2116_KEY: 'TRUE'}80 self.addGetProcessOutputExpectEnv(expected_env)81 self.expectCommands(82 gpo.Expect('hg', 'init', '/some/dir'),83 gpo.Expect('hg', 'pull', '-b', 'default',84 'ssh://example.com/foo/baz')85 .path('/some/dir'),86 gpo.Expect(87 'hg', 'heads', 'default', '--template={rev}' + os.linesep)88 .path('/some/dir').stdout(b"73591"),89 gpo.Expect('hg', 'log', '-b', 'default', '-r', '73591:73591', # only fetches that head90 '--template={rev}:{node}\\n')91 .path('/some/dir').stdout(LINESEP_BYTES.join([b'73591:4423cdb'])),92 gpo.Expect('hg', 'log', '-r', '4423cdb',93 '--template={date|hgdate}' + os.linesep + '{author}' + os.linesep + "{files % '{file}" + os.pathsep + "'}" + os.linesep + '{desc|strip}')94 .path('/some/dir').stdout(LINESEP_BYTES.join([95 b'1273258100.0 -7200',96 b'Bob Test <bobtest@example.org>',97 b'file1 with spaces' + PATHSEP_BYTES +98 os.path.join(b'dir with spaces', b'file2') + PATHSEP_BYTES,99 b'This is rev 73591',100 b''])),101 )102 # do the poll103 d = self.poller.poll()104 # check the results105 def check_changes(_):106 self.assertEqual(len(self.master.data.updates.changesAdded), 1)107 change = self.master.data.updates.changesAdded[0]108 self.assertEqual(change['revision'], '4423cdb')109 self.assertEqual(change['author'],110 'Bob Test <bobtest@example.org>')111 if self.usetimestamps:112 self.assertEqual(change['when_timestamp'], 1273258100)113 else:114 self.assertEqual(change['when_timestamp'], None)115 self.assertEqual(116 change['files'], ['file1 with spaces', os.path.join('dir with spaces', 'file2')])117 self.assertEqual(change['src'], 'hg')118 self.assertEqual(change['branch'], 'default')119 self.assertEqual(change['comments'], 'This is rev 73591')120 d.addCallback(check_changes)121 d.addCallback(self.check_current_rev(73591))122 return d123 def check_current_rev(self, wished):124 def check_on_rev(_):125 d = self.poller._getCurrentRev()126 d.addCallback(lambda oid_rev: self.assertEqual(oid_rev[1], wished))127 return check_on_rev128 @defer.inlineCallbacks129 def test_poll_several_heads(self):130 # If there are several heads on the named branch, the poller mustn't131 # climb (good enough for now, ideally it should even go to the common132 # ancestor)133 self.expectCommands(134 gpo.Expect('hg', 'pull', '-b', 'default',135 'ssh://example.com/foo/baz')136 .path('/some/dir'),137 gpo.Expect(138 'hg', 'heads', 'default', '--template={rev}' + os.linesep)139 .path('/some/dir').stdout(b'5' + LINESEP_BYTES + b'6' + LINESEP_BYTES)140 )141 yield self.poller._setCurrentRev(3)142 # do the poll: we must stay at rev 3143 d = self.poller.poll()144 d.addCallback(self.check_current_rev(3))145 @defer.inlineCallbacks146 def test_poll_regular(self):147 # normal operation. There's a previous revision, we get a new one.148 self.expectCommands(149 gpo.Expect('hg', 'pull', '-b', 'default',150 'ssh://example.com/foo/baz')151 .path('/some/dir'),152 gpo.Expect(153 'hg', 'heads', 'default', '--template={rev}' + os.linesep)154 .path('/some/dir').stdout(b'5' + LINESEP_BYTES),155 gpo.Expect('hg', 'log', '-b', 'default', '-r', '5:5',156 '--template={rev}:{node}\\n')157 .path('/some/dir').stdout(b'5:784bd' + LINESEP_BYTES),158 gpo.Expect('hg', 'log', '-r', '784bd',159 '--template={date|hgdate}' + os.linesep + '{author}' + os.linesep + "{files % '{file}" + os.pathsep + "'}" + os.linesep + '{desc|strip}')160 .path('/some/dir').stdout(LINESEP_BYTES.join([161 b'1273258009.0 -7200',162 b'Joe Test <joetest@example.org>',163 b'file1 file2',164 b'Comment for rev 5',165 b''])),166 )167 yield self.poller._setCurrentRev(4)168 d = self.poller.poll()169 d.addCallback(self.check_current_rev(5))170 def check_changes(_):171 self.assertEqual(len(self.master.data.updates.changesAdded), 1)172 change = self.master.data.updates.changesAdded[0]173 self.assertEqual(change['revision'], u'784bd')174 self.assertEqual(change['comments'], u'Comment for rev 5')175 d.addCallback(check_changes)176class HgPollerNoTimestamp(TestHgPoller):177 """ Test HgPoller() without parsing revision commit timestamp """...
test_changes_hgpoller.py
Source:test_changes_hgpoller.py
1# This file is part of Buildbot. Buildbot is free software: you can2# redistribute it and/or modify it under the terms of the GNU General Public3# License as published by the Free Software Foundation, version 2.4#5# This program is distributed in the hope that it will be useful, but WITHOUT6# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS7# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more8# details.9#10# You should have received a copy of the GNU General Public License along with11# this program; if not, write to the Free Software Foundation, Inc., 5112# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.13#14# Copyright Buildbot Team Members15from __future__ import absolute_import16from __future__ import print_function17import os18from twisted.internet import defer19from twisted.trial import unittest20from buildbot.changes import hgpoller21from buildbot.test.util import changesource22from buildbot.test.util import gpo23ENVIRON_2116_KEY = 'TEST_THAT_ENVIRONMENT_GETS_PASSED_TO_SUBPROCESSES'24LINESEP_BYTES = os.linesep.encode("ascii")25PATHSEP_BYTES = os.pathsep.encode("ascii")26class TestHgPoller(gpo.GetProcessOutputMixin,27 changesource.ChangeSourceMixin,28 unittest.TestCase):29 usetimestamps = True30 def setUp(self):31 # To test that environment variables get propagated to subprocesses32 # (See #2116)33 os.environ[ENVIRON_2116_KEY] = 'TRUE'34 self.setUpGetProcessOutput()35 d = self.setUpChangeSource()36 self.remote_repo = 'ssh://example.com/foo/baz'37 self.branch = 'default'38 self.repo_ready = True39 def _isRepositoryReady():40 return self.repo_ready41 def create_poller(_):42 self.poller = hgpoller.HgPoller(self.remote_repo,43 usetimestamps=self.usetimestamps,44 workdir='/some/dir')45 self.poller.setServiceParent(self.master)46 self.poller._isRepositoryReady = _isRepositoryReady47 d.addCallback(create_poller)48 def create_db(_):49 return self.master.db.setup()50 d.addCallback(create_db)51 return d52 def tearDown(self):53 del os.environ[ENVIRON_2116_KEY]54 return self.tearDownChangeSource()55 def gpoFullcommandPattern(self, commandName, *expected_args):56 """Match if the command is commandName and arg list start as expected.57 This allows to test a bit more if expected GPO are issued, be it58 by obscure failures due to the result not being given.59 """60 def matchesSubcommand(bin, given_args, **kwargs):61 return bin == commandName and tuple(62 given_args[:len(expected_args)]) == expected_args63 return matchesSubcommand64 def test_describe(self):65 self.assertSubstring("HgPoller", self.poller.describe())66 def test_name(self):67 self.assertEqual(68 "%s[%s]" % (self.remote_repo, self.branch), self.poller.name)69 # and one with explicit name...70 other = hgpoller.HgPoller(71 self.remote_repo, name="MyName", workdir='/some/dir')72 self.assertEqual("MyName", other.name)73 def test_hgbin_default(self):74 self.assertEqual(self.poller.hgbin, "hg")75 def test_poll_initial(self):76 self.repo_ready = False77 # Test that environment variables get propagated to subprocesses78 # (See #2116)79 expected_env = {ENVIRON_2116_KEY: 'TRUE'}80 self.addGetProcessOutputExpectEnv(expected_env)81 self.expectCommands(82 gpo.Expect('hg', 'init', '/some/dir'),83 gpo.Expect('hg', 'pull', '-b', 'default',84 'ssh://example.com/foo/baz')85 .path('/some/dir'),86 gpo.Expect(87 'hg', 'heads', 'default', '--template={rev}' + os.linesep)88 .path('/some/dir').stdout(b"73591"),89 gpo.Expect('hg', 'log', '-b', 'default', '-r', '73591:73591', # only fetches that head90 '--template={rev}:{node}\\n')91 .path('/some/dir').stdout(LINESEP_BYTES.join([b'73591:4423cdb'])),92 gpo.Expect('hg', 'log', '-r', '4423cdb',93 '--template={date|hgdate}' + os.linesep + '{author}' + os.linesep + "{files % '{file}" + os.pathsep + "'}" + os.linesep + '{desc|strip}')94 .path('/some/dir').stdout(LINESEP_BYTES.join([95 b'1273258100.0 -7200',96 b'Bob Test <bobtest@example.org>',97 b'file1 with spaces' + PATHSEP_BYTES +98 os.path.join(b'dir with spaces', b'file2') + PATHSEP_BYTES,99 b'This is rev 73591',100 b''])),101 )102 # do the poll103 d = self.poller.poll()104 # check the results105 def check_changes(_):106 self.assertEqual(len(self.master.data.updates.changesAdded), 1)107 change = self.master.data.updates.changesAdded[0]108 self.assertEqual(change['revision'], '4423cdb')109 self.assertEqual(change['author'],110 'Bob Test <bobtest@example.org>')111 if self.usetimestamps:112 self.assertEqual(change['when_timestamp'], 1273258100)113 else:114 self.assertEqual(change['when_timestamp'], None)115 self.assertEqual(116 change['files'], ['file1 with spaces', os.path.join('dir with spaces', 'file2')])117 self.assertEqual(change['src'], 'hg')118 self.assertEqual(change['branch'], 'default')119 self.assertEqual(change['comments'], 'This is rev 73591')120 d.addCallback(check_changes)121 d.addCallback(self.check_current_rev(73591))122 return d123 def check_current_rev(self, wished):124 def check_on_rev(_):125 d = self.poller._getCurrentRev()126 d.addCallback(lambda oid_rev: self.assertEqual(oid_rev[1], wished))127 return check_on_rev128 @defer.inlineCallbacks129 def test_poll_several_heads(self):130 # If there are several heads on the named branch, the poller mustn't131 # climb (good enough for now, ideally it should even go to the common132 # ancestor)133 self.expectCommands(134 gpo.Expect('hg', 'pull', '-b', 'default',135 'ssh://example.com/foo/baz')136 .path('/some/dir'),137 gpo.Expect(138 'hg', 'heads', 'default', '--template={rev}' + os.linesep)139 .path('/some/dir').stdout(b'5' + LINESEP_BYTES + b'6' + LINESEP_BYTES)140 )141 yield self.poller._setCurrentRev(3)142 # do the poll: we must stay at rev 3143 d = self.poller.poll()144 d.addCallback(self.check_current_rev(3))145 @defer.inlineCallbacks146 def test_poll_regular(self):147 # normal operation. There's a previous revision, we get a new one.148 self.expectCommands(149 gpo.Expect('hg', 'pull', '-b', 'default',150 'ssh://example.com/foo/baz')151 .path('/some/dir'),152 gpo.Expect(153 'hg', 'heads', 'default', '--template={rev}' + os.linesep)154 .path('/some/dir').stdout(b'5' + LINESEP_BYTES),155 gpo.Expect('hg', 'log', '-b', 'default', '-r', '5:5',156 '--template={rev}:{node}\\n')157 .path('/some/dir').stdout(b'5:784bd' + LINESEP_BYTES),158 gpo.Expect('hg', 'log', '-r', '784bd',159 '--template={date|hgdate}' + os.linesep + '{author}' + os.linesep + "{files % '{file}" + os.pathsep + "'}" + os.linesep + '{desc|strip}')160 .path('/some/dir').stdout(LINESEP_BYTES.join([161 b'1273258009.0 -7200',162 b'Joe Test <joetest@example.org>',163 b'file1 file2',164 b'Comment for rev 5',165 b''])),166 )167 yield self.poller._setCurrentRev(4)168 d = self.poller.poll()169 d.addCallback(self.check_current_rev(5))170 def check_changes(_):171 self.assertEqual(len(self.master.data.updates.changesAdded), 1)172 change = self.master.data.updates.changesAdded[0]173 self.assertEqual(change['revision'], u'784bd')174 self.assertEqual(change['comments'], u'Comment for rev 5')175 d.addCallback(check_changes)176class HgPollerNoTimestamp(TestHgPoller):177 """ Test HgPoller() without parsing revision commit timestamp """...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run tox automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
