test

software/proftpd/test/setup.py

@@ -34,7 +34,7 @@ long_description = open("README.md").read()
 
 setup(name=name,
       version=version,
-      description="SlapOS recipes.",
+      description="Test for SlapOS' ProFTPd",
       long_description=long_description,
       long_description_content_type='text/markdown',
       maintainer="Nexedi",

software/proftpd/test/test.py

@@ -43,20 +43,72 @@ import logging
 
 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 TestSFTPFeature(utils.SlapOSInstanceTestCase):
   software_url_list = (software_url, )
-  def test_a(self):
-    print self.id()
-  def test_b(self):
+
+  def test_simple_sftp_session(self):
+    # we will not verify host key
+    cnopts = pysftp.CnOpts()
+    cnopts.hostkeys = None
+
+    parameter_dict = self.computer_partition.getConnectionParameterDict()
+    sftp_url = urlparse.urlparse(parameter_dict['url'])
+
+    with pysftp.Connection(
+        sftp_url.hostname,
+        port=sftp_url.port,
+        cnopts=cnopts,
+        username=parameter_dict['user'],
+        password=parameter_dict['password']) as sftp:
+      with tempfile.NamedTemporyFile() as f:
+        f.write("Hello FTP !")
+        f.close()
+        sftp.put(f.name)
+        retrieved = sftp.get(f.name)
+        self.assertEqual(retrieved, "Hello FTP :)")
+
     print self.id()
+    import pdb; pdb.set_trace()
+
+  def test_connect_on_ipv4(self):
+    pass
+
+  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'])
+
+  def test_uploaded_file_not_visible_until_fully_uploaded(self):
+    pass
+
+  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, )
-  intance_arguments = {
+  instance_parameter_dict = {
       'port': 10022
   }
   def test_c(self):

software/proftpd/test/utils.py

@@ -39,7 +39,9 @@ global config
 
 def setUpModule():
   # XXX 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/'
   if not os.path.exists(working_directory):
     os.mkdir(working_directory)
 
@@ -57,7 +59,7 @@ def setUpModule():
   # TODO
   ipv4_address = '127.0.0.1'
   config['proxy_host'] = config['ipv4_address'] = ipv4_address
-  config['ipv6_address'] = '::1'
+  config['ipv6_address'] = '2001:67c:1254:26::3318'
 
   config['master_url'] = 'http://{proxy_host}:{proxy_port}'.format(**config)
   print config
@@ -74,10 +76,14 @@ 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 = {}
+
   @classmethod
   def setUpClass(cls):
     print "Setting up", cls, cls.software_url_list
@@ -94,75 +100,22 @@ class SlapOSInstanceTestCase(unittest.TestCase):
     process_manager.supervisord_pid_file = os.path.join(\
            slapos_controler.instance_root, 'var', 'run', 'supervisord.pid')
 
-    method_list= ["runSoftwareRelease"]
-    create_partition = 1 # XXX
-    if create_partition:
-      method_list.append("runComputerPartition")
-    for method_name in method_list:
-      slapos_method = getattr(slapos_controler, method_name)
-      logger.debug("Before status_dict = slapos_method(...)")
-      status_dict = slapos_method(config,
-          environment=os.environ, # XXX
-                                  )
-                                 # **kw)
-      logger.info(status_dict)
-      logger.debug("After status_dict = slapos_method(...)")
-      if status_dict['status_code'] != 0:
-        # slapos_instance.retry = True
-        # slapos_instance.retry_software_count += 1
-        print "ERROR ?", status_dict
-        import pdb; pdb.set_trace()
-
-       # raise SubprocessError(status_dict)
-      #else:
-     #    slapos_instance.retry_software_count = 0
-
-    def _prepareSlapOS(self, working_directory, slapos_instance,
-            create_partition=1, software_path_list=None, **kw):
-      """
-      Launch slapos to build software and partitions
-      """
-      slapproxy_log = os.path.join(self.testnode.config['log_directory'],
-                                    'slapproxy.log')
-      logger.debug('Configured slapproxy log to %r', slapproxy_log)
-      reset_software = slapos_instance.retry_software_count > 10
-      if reset_software:
-        slapos_instance.retry_software_count = 0
-      logger.info('testnode, retry_software_count: %r',
-               slapos_instance.retry_software_count)
-
-      # XXX Create a new controler because working_directory can be
-      #     Diferent depending of the preparation
-      slapos_controler = self._getSlapOSControler(working_directory)
-
-      slapos_controler.initializeSlapOSControler(slapproxy_log=slapproxy_log,
-         process_manager=self.testnode.process_manager, reset_software=reset_software,
-         software_path_list=software_path_list)
-      self.testnode.process_manager.supervisord_pid_file = os.path.join(\
-           slapos_controler.instance_root, 'var', 'run', 'supervisord.pid')
-      method_list= ["runSoftwareRelease"]
-      if create_partition:
-        method_list.append("runComputerPartition")
-      for method_name in method_list:
-        slapos_method = getattr(slapos_controler, method_name)
-        logger.debug("Before status_dict = slapos_method(...)")
-        status_dict = slapos_method(self.testnode.config,
-                                    environment=self.testnode.config['environment'],
-                                    **kw)
-        logger.info(status_dict)
-        logger.debug("After status_dict = slapos_method(...)")
-        if status_dict['status_code'] != 0:
-           slapos_instance.retry = True
-           slapos_instance.retry_software_count += 1
-           raise SubprocessError(status_dict)
-        else:
-           slapos_instance.retry_software_count = 0
-      return status_dict
+    software_status_dict = slapos_controler.runSoftwareRelease(config, environment=os.environ)
+    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
+    cls.slapos_controler = slapos_controler
+    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)
 
   @classmethod
   def tearDownClass(cls):
     print "Teardown", cls
 
 
-