proftpd: instance test suite

software/proftpd/test/README.md

+Tests for ProFTPd software release

software/proftpd/test/setup.py

+##############################################################################
+#
+# Copyright (c) 2018 Nexedi SA and Contributors. All Rights Reserved.
+#
+# WARNING: This program as such is intended to be used by professional
+# programmers who take the whole responsibility of assessing all potential
+# consequences resulting from its eventual inadequacies and bugs
+# End users who are looking for a ready-to-use solution with commercial
+# guarantees and support are strongly adviced to contract a Free Software
+# Service Company
+#
+# This program is Free Software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 3
+# of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
+#
+##############################################################################
+from setuptools import setup, find_packages
+import glob
+import os
+
+version = '0.0.1.dev0'
+name = 'slapos.test.proftpd'
+long_description = open("README.md").read()
+
+setup(name=name,
+      version=version,
+      description="SlapOS recipes.",
+      long_description=long_description,
+      long_description_content_type='text/markdown',
+      maintainer="Nexedi",
+      maintainer_email="info@nexedi.com",
+      url="https://lab.nexedi.com/nexedi/slapos",
+      packages=find_packages(),
+      install_requires=[
+        'slapos.core',
+        'erp5.util',
+        'pysftp',
+        ],
+      zip_safe=True,
+      test_suite='test',
+    )

software/proftpd/test/test.py

+##############################################################################
+#
+# Copyright (c) 2018 Nexedi SA and Contributors. All Rights Reserved.
+#
+# WARNING: This program as such is intended to be used by professional
+# programmers who take the whole responsibility of assessing all potential
+# consequences resulting from its eventual inadequacies and bugs
+# End users who are looking for a ready-to-use solution with commercial
+# guarantees and support are strongly adviced to contract a Free Software
+# Service Company
+#
+# This program is Free Software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 3
+# of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
+#
+##############################################################################
+
+import pysftp
+import os
+import utils
+
+def setUpModule():
+  utils.setUpModule()
+
+def tearDownModule():
+  utils.tearDownModule()
+
+
+import unittest
+unittest.installHandler()
+
+import logging
+
+logging.basicConfig(level=logging.DEBUG)
+
+
+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):
+    print self.id()
+
+
+class TestPortParameter(utils.SlapOSInstanceTestCase):
+  software_url_list = (software_url, )
+  intance_arguments = {
+      'port': 10022
+  }
+  def test_c(self):
+    print self.id()
+
+

software/proftpd/test/utils.py

+##############################################################################
+#
+# Copyright (c) 2018 Nexedi SA and Contributors. All Rights Reserved.
+#
+# WARNING: This program as such is intended to be used by professional
+# programmers who take the whole responsibility of assessing all potential
+# consequences resulting from its eventual inadequacies and bugs
+# End users who are looking for a ready-to-use solution with commercial
+# guarantees and support are strongly adviced to contract a Free Software
+# Service Company
+#
+# This program is Free Software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 3
+# of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
+#
+##############################################################################
+
+import unittest
+import os
+
+import erp5.util.testnode.SlapOSControler
+from erp5.util.testnode.ProcessManager import ProcessManager
+import slapos
+
+process_manager = ProcessManager()
+
+global slapos_controler
+global config
+
+def setUpModule():
+  # XXX read from environ
+  working_directory = os.path.join(os.path.dirname(__file__), '.slapos')
+  if not os.path.exists(working_directory):
+    os.mkdir(working_directory)
+
+  global config
+  config={
+    "working_directory": working_directory,
+    "slapos_directory": working_directory,
+    "log_directory": working_directory,
+    "computer_id": 'TODO',
+    'proxy_database': os.path.join(working_directory, 'proxy.db'),
+    'proxy_port': 5050,
+    'partition_reference': 'test', # XXX shouldn't this by by testclass ?
+    'slapos_binary': '/opt/slapgrid/b0bcd9831b23f334bc85e4a193c748a0/bin/slapos',
+  }
+  # TODO
+  ipv4_address = '127.0.0.1'
+  config['proxy_host'] = config['ipv4_address'] = ipv4_address
+  config['ipv6_address'] = '::1'
+
+  config['master_url'] = 'http://{proxy_host}:{proxy_port}'.format(**config)
+  print config
+
+  global slapos_controler
+  slapos_controler = erp5.util.testnode.SlapOSControler.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):
+  @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)
+
+    slapos_controler.initializeSlapOSControler(
+        slapproxy_log=slapproxy_log,
+        process_manager=process_manager,
+        reset_software=False,
+        software_path_list=cls.software_url_list)
+   
+    print "initialized"
+    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
+
+
+  @classmethod
+  def tearDownClass(cls):
+    print "Teardown", cls
+
+
+