test f

software/proftpd/test/test.py

@@ -25,8 +25,13 @@
 #
 ##############################################################################
 
-import pysftp
 import os
+import shutil
+import urlparse
+import tempfile
+
+import pysftp
+
 import utils
 
 def setUpModule():
@@ -36,62 +41,75 @@ def tearDownModule():
   utils.tearDownModule()
 
 
-import unittest
-unittest.installHandler()
-
-import logging
+# for development: log a lot and install Ctrl+C handler
+if 1:
+  import logging
+  logging.basicConfig(level=logging.DEBUG)
+  import unittest
+  unittest.installHandler()
 
-logging.basicConfig(level=logging.DEBUG)
 
-import urlparse
-import pysftp
-import tempfile
 
 
-software_url = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'software.cfg'))
+class ProFTPdTestCase(utils.SlapOSInstanceTestCase):
+  software_url_list = (
+      os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'software.cfg')), )
 
-class TestSFTPFeature(utils.SlapOSInstanceTestCase):
-  software_url_list = (software_url, )
+  def _getConnection(self, username=None, password=None, hostname=None):
+    """Returns a pysftp connection connected to the SFTP
 
-  def test_simple_sftp_session(self):
-    # we will not verify host key
+    username and password can be specified and default to the ones from
+    instance connection parameters.
+    another hostname can also be passed.
+    """
+    # this will not verify host key
     cnopts = pysftp.CnOpts()
     cnopts.hostkeys = None
 
     parameter_dict = self.computer_partition.getConnectionParameterDict()
     sftp_url = urlparse.urlparse(parameter_dict['url'])
+    print parameter_dict
 
-    with pysftp.Connection(
-        sftp_url.hostname,
+    return pysftp.Connection(
+        hostname or sftp_url.hostname,
         port=sftp_url.port,
         cnopts=cnopts,
-        username=parameter_dict['user'],
-        password=parameter_dict['password']) as sftp:
+        username=username or parameter_dict['user'], # XXX
+        password=password or parameter_dict['password'])
+
+
+class TestSFTPFeature(ProFTPdTestCase):
+  """Tests the features we expect in SFTP server.
+  """
+  def test_simple_sftp_session(self):
+    with self._getConnection() as sftp:
+      # put a file
       with tempfile.NamedTemporaryFile() as f:
         f.write("Hello FTP !")
-        f.close()
-        sftp.put(f.name)
-        retrieved = sftp.get(f.name)
-        self.assertEqual(retrieved, "Hello FTP :)")
+        f.flush()
+        sftp.put(f.name, remotepath='testfile')
 
-    print self.id()
-    import pdb; pdb.set_trace()
+      # it's visible in listdir()
+      self.assertEqual(['testfile'], sftp.listdir())
+
+      # download it, it should have same content
+      tempdir = tempfile.mkdtemp()
+      self.addCleanup(lambda : shutil.rmtree(tempdir))
+      local_file = os.path.join(tempdir, 'testfile')
+      retrieve_same_file = sftp.get('testfile', local_file)
+      with open(local_file) as f:
+        self.assertEqual(f.read(), "Hello FTP !")
 
   def test_connect_on_ipv4(self):
-    pass
+    self.assertEqual(
+        [],
+        self._getConnection(hostname=self.config['ipv4_address']).listdir())
 
   def test_proftpd_does_not_listen_on_all_ip(self):
-    parameter_dict = self.computer_partition.getConnectionParameterDict()
-    sftp_url = urlparse.urlparse(parameter_dict['url'])
-    cnopts = pysftp.CnOpts()
-    cnopts.hostkeys = None
-    with self.assertRaises(AttributeError):
-      pysftp.Connection(
-          '0.0.0.0',
-          port=sftp_url.port,
-          cnopts=cnopts,
-          username=parameter_dict['user'],
-          password=parameter_dict['password'])
+    from paramiko.ssh_exception import SSHException
+    with self.assertRaises(SSHException):
+      self._getConnection(hostname='127.0.0.1')
+    import pdb; pdb.set_trace()
 
   def test_uploaded_file_not_visible_until_fully_uploaded(self):
     pass
@@ -99,19 +117,36 @@ class TestSFTPFeature(utils.SlapOSInstanceTestCase):
   def test_partial_upload_are_deleted(self):
     pass
 
-  def test_client_are_banned_after_5_wrong_passwords(self):
-    pass
-
   def test_user_can_be_added(self):
     pass
 
 
-class TestPortParameter(utils.SlapOSInstanceTestCase):
-  software_url_list = (software_url, )
+class TestBan(ProFTPdTestCase):
+  def test_client_are_banned_after_5_wrong_passwords(self):
+    # Simulate 5 login errors
+    from paramiko.ssh_exception import AuthenticationException
+    for i in range(5):
+      with self.assertRaisesRegexp(AuthenticationException, 'Authentication failed'):
+        self._getConnection(password='wrong')
+
+    # after that, even with a valid password we cannot connect
+    from paramiko.ssh_exception import SSHException
+    with self.assertRaisesRegexp(SSHException, 'Connection reset by peer'):
+      self._getConnection()
+
+    # ban event is logged
+    with open(os.path.join(
+        self.computer_partition_root_path, 'var', 'log', 'proftpd-ban.log')) as ban_log_file:
+      self.assertRegexpMatches(
+        ban_log_file.readlines()[-1],
+        'login from host .* denied due to host bane')
+
+
+class TestPortParameter(ProFTPdTestCase):
   instance_parameter_dict = {
       'port': 10022
   }
-  def test_c(self):
-    print self.id()
+#  def test_c(self):
+#    print self.id()
 
 

software/proftpd/test/utils.py

@@ -38,7 +38,7 @@ global slapos_controler
 global config
 
 def setUpModule():
-  # XXX read from environ
+  # TODO read from environ
   # XXX beware of Error: Cannot open an HTTP server: socket.error reported AF_UNIX path too long
   working_directory = os.path.join(os.path.dirname(__file__), '.slapos')
   working_directory = '/tmp/slapotest/'
@@ -84,9 +84,11 @@ class SlapOSInstanceTestCase(unittest.TestCase):
   software_url_list = (NotImplemented, )
   instance_parameter_dict = {}
 
+  # This is set to the path of the instance, to inspect created files.
+  instance_root_path = None
+
   @classmethod
   def setUpClass(cls):
-    print "Setting up", cls, cls.software_url_list
     slapproxy_log = os.path.join(config['log_directory'], 'slapproxy.log')
     logger.debug('Configured slapproxy log to %r', slapproxy_log)
 
@@ -96,26 +98,36 @@ class SlapOSInstanceTestCase(unittest.TestCase):
         reset_software=False,
         software_path_list=cls.software_url_list)
    
-    print "initialized"
-    process_manager.supervisord_pid_file = os.path.join(\
+    process_manager.supervisord_pid_file = os.path.join(
            slapos_controler.instance_root, 'var', 'run', 'supervisord.pid')
 
     software_status_dict = slapos_controler.runSoftwareRelease(config, environment=os.environ)
+    # TODO: log more details in this case
     assert software_status_dict['status_code'] == 0
 
     instance_status_dict = slapos_controler.runComputerPartition(
         config,
         cluster_configuration=cls.instance_parameter_dict,
         environment=os.environ)
-    assert instance_status_dict['status_code'] == 0
+    # TODO: log more details in this case
+    assert software_status_dict['status_code'] == 0
+
     cls.slapos_controler = slapos_controler
+    cls.config = config
     cls.computer_partition = slapos_controler.slap.registerOpenOrder().request(
         cls.software_url_list[0],
         partition_reference='testing partition 0',
         partition_parameter_kw=cls.instance_parameter_dict)
+    cls.computer_partition_root_path = os.path.join(
+        config['working_directory'],
+        'inst',
+        cls.computer_partition.getId())
 
   @classmethod
   def tearDownClass(cls):
     print "Teardown", cls
-
-
+    #for reference in cls.slapos_controler.instance_config:
+    #  cls.slapos_controler.destroyInstance(reference)
+    cls.slapos_controler.process_manager.killPreviousRun()
+    import os
+    #os.waitpid(0, 0)