Simple working fakessh_test.

tests/fakessh_test.py

 
-import mitogen.utils
-import mitogen.fakessh
-
-
-@mitogen.utils.run_with_router
-def main(router):
-    import logging
-    mitogen.utils.log_to_file(level=logging.INFO)
-    router.enable_debug()
-    #mitogen.fakessh.run_with_fake_ssh(router, ['bash', '-c', 'echo $PATH'])
-    #mitogen.fakessh.run_with_fake_ssh(router, ['bash', '-c', 'ls -Gl $SSH_PATH'])
-    # k3 = router.ssh(hostname='k3')
-    k3 = router.local()
-    print 'GOT HERE'
-    import os
-    k3.call(os.system, 'hostname')
-    return
-    #sud = router.sudo(via=k3, username='root')
-    sud = k3
-
-    mitogen.fakessh.run_with_fake_ssh(sud, router, ['rsync', '--progress', '-vvvr', 'h:/var/lib/docker'])
-    #mitogen.fakessh.run_with_fake_ssh(router, ['bash', '-c', 'ssh h rsync --server . foo'])
-    # mitogen.fakessh.run_with_fake_ssh(router, ['bash', '-c', 'ssh k3.botanicus.net -t screen -dr'])
-    print 'end of t.py'
-
+import os
+import shutil
+import unittest
 
-import logging
 import mitogen.utils
 import mitogen.fakessh
 
-@mitogen.utils.run_with_router
-def main(router):
-    mitogen.utils.log_to_file(level=logging.DEBUG)
-    #router.enable_debug()
-    #router.enable_debug()
+import testlib
 
-    k3 = router.ssh(hostname='k3')
-    sudo = router.sudo(via=k3, username='root')
 
-    mitogen.fakessh.run(sudo, router, ['rsync', '--progress', '-r', 'h:/var/lib/docker'])
+class RsyncTest(testlib.DockerMixin, unittest.TestCase):
+    def test_rsync_from_master(self):
+        context = self.docker_ssh_any()
+        context.call(shutil.rmtree, '/tmp/data', ignore_errors=True)
+        mitogen.fakessh.run(context, self.router, [
+            'rsync', '--progress', '-vvva',
+            testlib.data_path('.'), 'target:/tmp/data'
+        ])
+        assert context.call(os.path.exists, '/tmp/data')
+        assert context.call(os.path.exists, '/tmp/data/simple_pkg/a.py')
 

tests/ssh_test.py

@@ -26,10 +26,7 @@ class SshTest(testlib.DockerMixin, unittest.TestCase):
 
     def test_password_required(self):
         try:
-            context = self.router.ssh(
-                hostname=self.dockerized_ssh.host,
-                port=self.dockerized_ssh.port,
-                check_host_keys=False,
+            context = self.docker_ssh(
                 username='has-sudo',
             )
             assert 0, 'exception not thrown'
@@ -40,10 +37,7 @@ class SshTest(testlib.DockerMixin, unittest.TestCase):
 
     def test_password_incorrect(self):
         try:
-            context = self.router.ssh(
-                hostname=self.dockerized_ssh.host,
-                port=self.dockerized_ssh.port,
-                check_host_keys=False,
+            context = self.docker_ssh(
                 username='has-sudo',
                 password='badpw',
             )
@@ -54,10 +48,7 @@ class SshTest(testlib.DockerMixin, unittest.TestCase):
         assert e[0] == self.stream_class.password_incorrect_msg
 
     def test_password_specified(self):
-        context = self.router.ssh(
-            hostname=self.dockerized_ssh.host,
-            port=self.dockerized_ssh.port,
-            check_host_keys=False,
+        context = self.docker_ssh(
             username='has-sudo',
             password='y',
         )
@@ -67,10 +58,7 @@ class SshTest(testlib.DockerMixin, unittest.TestCase):
 
     def test_pubkey_required(self):
         try:
-            context = self.router.ssh(
-                hostname=self.dockerized_ssh.host,
-                port=self.dockerized_ssh.port,
-                check_host_keys=False,
+            context = self.docker_ssh(
                 username='has-sudo-pubkey',
             )
             assert 0, 'exception not thrown'
@@ -80,13 +68,9 @@ class SshTest(testlib.DockerMixin, unittest.TestCase):
         assert e[0] == self.stream_class.password_required_msg
 
     def test_pubkey_specified(self):
-        context = self.router.ssh(
-            hostname=self.dockerized_ssh.host,
-            port=self.dockerized_ssh.port,
-            check_host_keys=False,
+        context = self.docker_ssh(
             username='has-sudo-pubkey',
             identity_file=testlib.data_path('docker/has-sudo-pubkey.key'),
         )
-
         sentinel = 'i-am-mitogen-test-docker-image\n'
         assert sentinel == context.call(plain_old_module.get_sentinel_value)

tests/testlib.py

@@ -76,3 +76,15 @@ class DockerMixin(RouterMixin):
     def tearDownClass(cls):
         cls.dockerized_ssh.close()
         super(DockerMixin, cls).tearDownClass()
+
+    def docker_ssh(self, **kwargs):
+        kwargs.setdefault('hostname', self.dockerized_ssh.host)
+        kwargs.setdefault('port', self.dockerized_ssh.port)
+        kwargs.setdefault('check_host_keys', False)
+        return self.router.ssh(**kwargs)
+
+    def docker_ssh_any(self, **kwargs):
+        return self.docker_ssh(
+            username='has-sudo-nopw',
+            password='y',
+        )