test

software/proftpd/test/test.py

@@ -29,31 +29,25 @@ import os
 import shutil
 import urlparse
 import tempfile
+import StringIO
 
 import pysftp
 
 import utils
 
-def setUpModule():
-  utils.setUpModule()
 
-def tearDownModule():
-  utils.tearDownModule()
-
-
-# for development: log a lot and install Ctrl+C handler
-if 1:
+# for development: debugging logs and install Ctrl+C handler
+if os.environ.get('DEBUG'):
   import logging
   logging.basicConfig(level=logging.DEBUG)
   import unittest
   unittest.installHandler()
 
 
-
-
 class ProFTPdTestCase(utils.SlapOSInstanceTestCase):
-  software_url_list = (
-      os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'software.cfg')), )
+  @classmethod
+  def getSoftwareURLList(cls):
+    return (os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'software.cfg')), )
 
   def _getConnection(self, username=None, password=None, hostname=None):
     """Returns a pysftp connection connected to the SFTP
@@ -62,13 +56,12 @@ class ProFTPdTestCase(utils.SlapOSInstanceTestCase):
     instance connection parameters.
     another hostname can also be passed.
     """
-    # this will not verify host key
+    # this tells paramiko not to 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
 
     return pysftp.Connection(
         hostname or sftp_url.hostname,
@@ -78,9 +71,32 @@ class ProFTPdTestCase(utils.SlapOSInstanceTestCase):
         password=password or parameter_dict['password'])
 
 
-class TestSFTPFeature(ProFTPdTestCase):
-  """Tests the features we expect in SFTP server.
+class TestSFTPListen(ProFTPdTestCase):
+  def test_listen_on_ipv4(self):
+    self.assertTrue(self._getConnection(hostname=self.config['ipv4_address']))
+
+  def test_does_not_listen_on_all_ip(self):
+    from paramiko.ssh_exception import SSHException
+    with self.assertRaises(SSHException):
+      self._getConnection(hostname='127.0.0.1')
+    import pdb; pdb.set_trace()
+
+
+class TestSFTPOperations(ProFTPdTestCase):
+  """Tests upload / download features we expect in SFTP server.
   """
+  def setUp(self):
+    self.upload_dir = os.path.join(
+        self.computer_partition_root_path, 'srv', 'proftpd')
+
+  def tearDown(self):
+    for name in os.listdir(self.upload_dir):
+      path = os.path.join(self.upload_dir, name)
+      if os.path.isfile(path) or os.path.islink(path):
+        os.remove(path)
+      else:
+        shutil.rmtree(path)
+
   def test_simple_sftp_session(self):
     with self._getConnection() as sftp:
       # put a file
@@ -100,30 +116,41 @@ class TestSFTPFeature(ProFTPdTestCase):
       with open(local_file) as f:
         self.assertEqual(f.read(), "Hello FTP !")
 
-  def test_connect_on_ipv4(self):
-    self.assertEqual(
-        [],
-        self._getConnection(hostname=self.config['ipv4_address']).listdir())
+  def test_uploaded_file_not_visible_until_fully_uploaded(self):
+    test_self = self
+    class PartialFile(StringIO.StringIO):
+      def read(self, *args):
+        # file is not visible yet
+        test_self.assertNotIn('destination.', os.listdir(test_self.upload_dir))
+        # it's just a hidden file
+        test_self.assertEqual(['.in.destination.'], os.listdir(test_self.upload_dir))
+        return StringIO.StringIO.read(self, *args)
 
-  def test_proftpd_does_not_listen_on_all_ip(self):
-    from paramiko.ssh_exception import SSHException
-    with self.assertRaises(SSHException):
-      self._getConnection(hostname='127.0.0.1')
-    import pdb; pdb.set_trace()
+    with self._getConnection() as sftp:
+      sftp.sftp_client.putfo(PartialFile("content"), "destination") 
       
-  def test_uploaded_file_not_visible_until_fully_uploaded(self):
-    pass
+    # now file is visible
+    self.assertEqual(['destination'], os.listdir(self.upload_dir))
 
   def test_partial_upload_are_deleted(self):
-    pass
+    class ErrorFile(StringIO.StringIO):
+      def read(self, *args):
+        raise IOError("Something bad happened")
 
-  def test_user_can_be_added(self):
-    pass
+    with self._getConnection() as sftp:
+      with self.assertRaises(IOError):
+        sftp.sftp_client.putfo(ErrorFile(), "destination") 
+    self.assertEqual([], os.listdir(self.upload_dir))
+
+
+class TestUserManagement(ProFTPdTestCase):
+  def test_user_can_be_added_from_script(self):
+    raise NotImplementedError()
 
 
 class TestBan(ProFTPdTestCase):
   def test_client_are_banned_after_5_wrong_passwords(self):
-    # Simulate 5 login errors
+    # Simulate failed 5 login attempts
     from paramiko.ssh_exception import AuthenticationException
     for i in range(5):
       with self.assertRaisesRegexp(AuthenticationException, 'Authentication failed'):
@@ -139,14 +166,19 @@ class TestBan(ProFTPdTestCase):
         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')
+        'login from host .* denied due to host ban')
 
 
-class TestPortParameter(ProFTPdTestCase):
-  instance_parameter_dict = {
-      'port': 10022
-  }
-#  def test_c(self):
-#    print self.id()
+class TestInstanceParameterPort(ProFTPdTestCase):
+  @classmethod
+  def getInstanceParmeterDict(cls):
+    cls.free_port = utils.findFreeTCPPort(cls.config['ipv4_address'])
+    return {'port': cls.free_port, }
+
+  def test_instance_parameter_port(self):
+    parameter_dict = self.computer_partition.getConnectionParameterDict()
+    sftp_url = urlparse.urlparse(parameter_dict['url'])
+    self.assertEqual(self.free_port, sftp_url.port)
+    self.assertTrue(self._getConnection())
 
 

software/proftpd/test/utils.py

@@ -27,76 +27,89 @@
 
 import unittest
 import os
+import socket
+from contextlib import closing
+import logging
 
-import erp5.util.testnode.SlapOSControler
+from erp5.util.testnode.SlapOSControler import SlapOSControler
 from erp5.util.testnode.ProcessManager import ProcessManager
 import slapos
 
-process_manager = ProcessManager()
+logger = logging.getLogger(__name__)
+
+def findFreeTCPPort(ip=''):
+  """Find a free TCP port to listen to.
+  inspired by https://stackoverflow.com/a/45690594
+  """
+  with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
+    s.bind((ip, 0))
+    return s.getsockname()[1]
+
 
-global slapos_controler
-global config
+class SlapOSInstanceTestCase(unittest.TestCase):
+  @classmethod
+  def getSoftwareURLList(cls):
+    """Return URL of software releases to install.
+
+    To be defined by subclasses.
+    """
+    raise NotImplementedError()
+
+  @classmethod
+  def getInstanceParmeterDict(cls):
+    """Return instance parameters
+
+    To be defined by subclasses if needed
+    """
+    return {}
+
+  @classmethod
+  def setUpClass(cls):
 
-def setUpModule():
-  # TODO read from environ
     # XXX beware of Error: Cannot open an HTTP server: socket.error reported AF_UNIX path too long
+    # This `working_directory` should not be too deep.
     working_directory = os.path.join(os.path.dirname(__file__), '.slapos')
+    # TODO: allow using a specific directory from envionment variable, similar
+    # to --save/--load in erp5 tests.
     working_directory = '/tmp/slapotest/'
     if not os.path.exists(working_directory):
       os.mkdir(working_directory)
 
-  global config
-  config={
+    cls.config = config = {
       "working_directory": working_directory,
       "slapos_directory": working_directory,
       "log_directory": working_directory,
-    "computer_id": 'TODO',
+      "computer_id": 'slapos.test', # XXX
       'proxy_database': os.path.join(working_directory, 'proxy.db'),
-    'proxy_port': 5050,
-    'partition_reference': 'test', # XXX shouldn't this by by testclass ?
+      'partition_reference': cls.__name__,
+      # TODO
       'slapos_binary': '/opt/slapgrid/b0bcd9831b23f334bc85e4a193c748a0/bin/slapos',
     }
-  # TODO
+
+    # TODO: read from environment / guess ipv4 from buildout file
     ipv4_address = '127.0.0.1'
     config['proxy_host'] = config['ipv4_address'] = ipv4_address
     config['ipv6_address'] = '2001:67c:1254:26::3318'
-
+    config['proxy_port'] = findFreeTCPPort(ipv4_address)
     config['master_url'] = 'http://{proxy_host}:{proxy_port}'.format(**config)
-  print config
 
-  global slapos_controler
-  slapos_controler = erp5.util.testnode.SlapOSControler.SlapOSControler(
+    cls._process_manager = process_manager = ProcessManager()
+
+    # XXX this code is copied from testnode code
+    slapos_controler = SlapOSControler(
         working_directory,
         config
     )
-  print "setUpModule, starting slapos_controler", slapos_controler, "in directory", working_directory
-
 
-def tearDownModule():
-  print "tearDownModule, terminating processes"
-  process_manager.killPreviousRun()
-
-
-import logging
-logger = logging.getLogger(__name__)
-
-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):
     slapproxy_log = os.path.join(config['log_directory'], 'slapproxy.log')
     logger.debug('Configured slapproxy log to %r', slapproxy_log)
 
+    software_url_list = cls.getSoftwareURLList()
     slapos_controler.initializeSlapOSControler(
         slapproxy_log=slapproxy_log,
         process_manager=process_manager,
         reset_software=False,
-        software_path_list=cls.software_url_list)
+        software_path_list=software_url_list)
    
     process_manager.supervisord_pid_file = os.path.join(
            slapos_controler.instance_root, 'var', 'run', 'supervisord.pid')
@@ -105,29 +118,37 @@ class SlapOSInstanceTestCase(unittest.TestCase):
     # TODO: log more details in this case
     assert software_status_dict['status_code'] == 0
 
+    instance_parameter_dict = cls.getInstanceParmeterDict()
     instance_status_dict = slapos_controler.runComputerPartition(
         config,
-        cluster_configuration=cls.instance_parameter_dict,
+        cluster_configuration=instance_parameter_dict,
         environment=os.environ)
     # 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)
+    # FIXME: similar to test node, only one (root) partition is really supported for now.
+    computer_partition_list = []
+    for i in range(len(software_url_list)):
+      computer_partition_list.append(
+          slapos_controler.slap.registerOpenOrder().request(
+          software_url_list[i],
+          # XXX this is how testnode name created partitions
+          partition_reference='testing partition {i}'.format(i=i, **config),
+          partition_parameter_kw=instance_parameter_dict))
+     
+    # expose some class attributes so that tests can use them:
+    # the ComputerPartition instances, to getInstanceParmeterDict
+    cls.computer_partition = computer_partition_list[0]
+
+    # the path of the instance on the filesystem, for low level inspection
     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)
+    # FIXME: if setUpClass fail, this is not called and leaks zombie processes
+    cls._process_manager.killPreviousRun()