1. Yuya Nishihara
  2. hgext-workarounds

Commits

Yuya Nishihara  committed 76bbdbc

thgbackdoor: make client specify backdoor host and port

  • Participants
  • Parent commits 1e91d5f
  • Branches default

Comments (0)

Files changed (1)

File tests/thgbackdoor_test.py

View file
 HGEXTDIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..',
                                         'hgext'))
 THG = os.path.abspath(os.environ.get('THG', 'thg'))
+BACKDOOR_ADDRESS = ('localhost', 8000)
 
 def setup():
     global _tempdir, _tempenv
     _tempdir = tempfile.mkdtemp('', 'thgbackdoortest-')
     _tempenv = os.environ.copy()
     _tempenv['HOME'] = _tempdir
-    _tempenv['THGBACKDOOR'] = 'localhost:8000'
     f = open(os.path.join(_tempdir, '.hgrc'), 'w')
     try:
         f.write('[extensions]\n')
     return lines[0] + '\n' + '\n'.join(l[margin:] for l in lines[1:])
 
 class ThgController(object):
-    def __init__(self, timeout=None, connect=True):
+    def __init__(self, address, timeout=None, connect=True):
+        self._address = address
         self._timeout = timeout
         self._sock = None
         if connect:
         # dirty code to wait server wake up
         while True:  # TODO timeout here?
             try:
-                self._sock.connect(('localhost', 8000))
+                self._sock.connect(self._address)
                 break
             except socket.error, err:
                 if err.errno not in (errno.ECONNREFUSED, errno.ECONNABORTED):
             raise ValueError('unexpected response type: %r' % resp)
 
 class ThgClient(object):
-    def __init__(self, cmdargs=None, timeout=None, connect=True):
+    def __init__(self, address, cmdargs=None, timeout=None, connect=True):
+        self._address = address
         self._cmdargs = list(cmdargs or [])
         self._proc = None
-        self._controller = ThgController(timeout=timeout, connect=False)
+        self._controller = ThgController(address, timeout=timeout,
+                                         connect=False)
         if connect:
             self.open()
 
     def open(self):
         if self._proc:
             raise ValueError('already open')
+        env = _tempenv.copy()
+        env['THGBACKDOOR'] = '%s:%d' % self._address
         self._proc = subprocess.Popen([THG, '--nofork'] + self._cmdargs,
-                                      cwd=_tempdir, env=_tempenv)
+                                      cwd=_tempdir, env=env)
         self._controller.open()
 
     def close(self):
         return getattr(self._controller, name)
 
 def test_run_workbench():
-    thg = ThgClient(timeout=5)
+    thg = ThgClient(BACKDOOR_ADDRESS, timeout=5)
     try:
         assert thg.eval('dlg').startswith('<tortoisehg.hgqt.workbench')
     finally:
         thg.close()
 
 def test_control_by_qtest():
-    thg = ThgClient(timeout=5)
+    thg = ThgClient(BACKDOOR_ADDRESS, timeout=5)
     try:
         thg.exec_('''
         from PyQt4.QtTest import QTest
 
 @timed(5)
 def test_timeout():
-    thg = ThgClient(timeout=1)
+    thg = ThgClient(BACKDOOR_ADDRESS, timeout=1)
     try:
         assert_raises(socket.timeout, lambda: thg.wait())
     finally:
         thg.close()
 
 def test_raise_exception():
-    thg = ThgClient(timeout=5)
+    thg = ThgClient(BACKDOOR_ADDRESS, timeout=5)
     try:
         assert_raises(RemoteError,
                       lambda: thg.exec_('raise Exception("oops")'))
         thg.close()
 
 def test_raise_delayed_exception():
-    thg = ThgClient()
+    thg = ThgClient(BACKDOOR_ADDRESS, timeout=5)
     try:
         thg.exec_('QTimer.singleShot(0, lambda: _unknown_attribute)')
         assert_raises(RemoteError, lambda: thg.wait())