add test for pgtest argument of TemporaryProfileManager

.ci/test_profile_manager.py

@@ -0,0 +1,94 @@
+# -*- coding: utf-8 -*-
+###########################################################################
+# Copyright (c), The AiiDA team. All rights reserved.                     #
+# This file is part of the AiiDA code.                                    #
+#                                                                         #
+# The code is hosted on GitHub at https://github.com/aiidateam/aiida-core #
+# For further information on the license, see the LICENSE.txt file        #
+# For further information please visit http://www.aiida.net               #
+###########################################################################
+"""Unittests for TestManager"""
+from __future__ import division
+from __future__ import print_function
+from __future__ import absolute_import
+import os
+import unittest
+import warnings
+import sys
+
+from pgtest import pgtest
+
+from aiida.manage.tests import TemporaryProfileManager, TestManagerError, get_test_backend_name
+from aiida.common.utils import Capturing
+
+
+class TemporaryProfileManagerTestCase(unittest.TestCase):
+    """Test the TemporaryProfileManager class"""
+
+    def setUp(self):
+        if sys.version_info[0] >= 3:
+            # tell unittest not to warn about running processes
+            warnings.simplefilter('ignore', ResourceWarning)  # pylint: disable=no-member,undefined-variable
+
+        self.backend = get_test_backend_name()
+        self.profile_manager = TemporaryProfileManager(backend=self.backend)
+
+    def tearDown(self):
+        self.profile_manager.destroy_all()
+
+    def test_create_db_cluster(self):
+        self.profile_manager.create_db_cluster()
+        self.assertTrue(pgtest.is_server_running(self.profile_manager.pg_cluster.cluster))
+
+    def test_create_aiida_db(self):
+        self.profile_manager.create_db_cluster()
+        self.profile_manager.create_aiida_db()
+        self.assertTrue(self.profile_manager.postgres.db_exists(self.profile_manager.profile_info['database_name']))
+
+    def test_create_use_destroy_profile2(self):
+        """
+        Test temporary test profile creation
+
+        * The profile gets created, the dbenv loaded
+        * Data can be stored in the db
+        * reset_db deletes all data added after profile creation
+        * destroy_all removes all traces of the test run
+
+        Note: This test function loads the dbenv - i.e. you cannot run similar test functions (that create profiles)
+        in the same test session. aiida.manage.configuration.reset_profile() was not yet enough, see
+        https://github.com/aiidateam/aiida-core/issues/3482
+        """
+        with Capturing() as output:
+            self.profile_manager.create_profile()
+
+        self.assertTrue(self.profile_manager.root_dir_ok, msg=output)
+        self.assertTrue(self.profile_manager.config_dir_ok, msg=output)
+        self.assertTrue(self.profile_manager.repo_ok, msg=output)
+        from aiida.manage.configuration.settings import AIIDA_CONFIG_FOLDER
+        self.assertEqual(AIIDA_CONFIG_FOLDER, self.profile_manager.config_dir, msg=output)
+
+        from aiida.orm import load_node
+        from aiida.plugins import DataFactory
+        data = DataFactory('dict')(dict={'key': 'value'})
+        data.store()
+        data_pk = data.pk
+        self.assertTrue(load_node(data_pk))
+
+        with self.assertRaises(TestManagerError):
+            self.test_create_aiida_db()
+
+        self.profile_manager.reset_db()
+        with self.assertRaises(Exception):
+            load_node(data_pk)
+
+        temp_dir = self.profile_manager.root_dir
+        self.profile_manager.destroy_all()
+        with self.assertRaises(Exception):
+            self.profile_manager.postgres.db_exists(self.profile_manager.dbinfo['db_name'])
+        self.assertFalse(os.path.exists(temp_dir))
+        self.assertIsNone(self.profile_manager.root_dir)
+        self.assertIsNone(self.profile_manager.pg_cluster)
+
+
+if __name__ == '__main__':
+    unittest.main()

.ci/test_script.sh

@@ -40,6 +40,7 @@ case "$TEST_TYPE" in
 
         # Run preliminary tests
         coverage run -a "${CI_DIR}/test_test_manager.py"
+        coverage run -a "${CI_DIR}/test_profile_manager.py"
         coverage run -a "${CI_DIR}/test_plugin_testcase.py"
         # append to coverage file, do not create final report
         pytest --cov=aiida --cov-append --cov-report= "${CI_DIR}/pytest"

.ci/test_test_manager.py

@@ -11,79 +11,41 @@
 from __future__ import division
 from __future__ import print_function
 from __future__ import absolute_import
-import os
 import unittest
 import warnings
 import sys
 
-from pgtest import pgtest
+from aiida.manage.tests import TestManager, get_test_backend_name
 
-from aiida.manage.tests import TemporaryProfileManager, TestManagerError, get_test_backend_name
-from aiida.common.utils import Capturing
 
-
-class TemporaryProfileManagerTestCase(unittest.TestCase):
-    """Test the TemporaryProfileManager class"""
+class TestManagerTestCase(unittest.TestCase):
+    """Test the TestManager class"""
 
     def setUp(self):
         if sys.version_info[0] >= 3:
             # tell unittest not to warn about running processes
             warnings.simplefilter('ignore', ResourceWarning)  # pylint: disable=no-member,undefined-variable
 
         self.backend = get_test_backend_name()
-        self.profile_manager = TemporaryProfileManager(backend=self.backend)
+        self.test_manager = TestManager()
 
     def tearDown(self):
-        self.profile_manager.destroy_all()
-
-    def test_create_db_cluster(self):
-        self.profile_manager.create_db_cluster()
-        self.assertTrue(pgtest.is_server_running(self.profile_manager.pg_cluster.cluster))
+        self.test_manager.destroy_all()
 
-    def test_create_aiida_db(self):
-        self.profile_manager.create_db_cluster()
-        self.profile_manager.create_aiida_db()
-        self.assertTrue(self.profile_manager.postgres.db_exists(self.profile_manager.profile_info['database_name']))
-
-    def test_create_use_destroy_profile(self):
+    def test_pgtest_argument(self):
         """
-        Test temporary test profile creation
-
-        * The profile gets created, the dbenv loaded
-        * Data can be stored in the db
-        * reset_db deletes all data added after profile creation
-        * destroy_all removes all traces of the test run
+        Create a temporary profile, passing the pgtest argument.
         """
-        with Capturing() as output:
-            self.profile_manager.create_profile()
-
-        self.assertTrue(self.profile_manager.root_dir_ok, msg=output)
-        self.assertTrue(self.profile_manager.config_dir_ok, msg=output)
-        self.assertTrue(self.profile_manager.repo_ok, msg=output)
-        from aiida.manage.configuration.settings import AIIDA_CONFIG_FOLDER
-        self.assertEqual(AIIDA_CONFIG_FOLDER, self.profile_manager.config_dir, msg=output)
-
-        from aiida.orm import load_node
-        from aiida.plugins import DataFactory
-        data = DataFactory('dict')(dict={'key': 'value'})
-        data.store()
-        data_pk = data.pk
-        self.assertTrue(load_node(data_pk))
-
-        with self.assertRaises(TestManagerError):
-            self.test_create_aiida_db()
+        from aiida.common.files import which
 
-        self.profile_manager.reset_db()
-        with self.assertRaises(Exception):
-            load_node(data_pk)
+        # this should fail
+        pgtest = {'pg_ctl': 'notapath'}
+        with self.assertRaises(AssertionError):
+            self.test_manager.use_temporary_profile(backend=self.backend, pgtest=pgtest)
 
-        temp_dir = self.profile_manager.root_dir
-        self.profile_manager.destroy_all()
-        with self.assertRaises(Exception):
-            self.profile_manager.postgres.db_exists(self.profile_manager.dbinfo['db_name'])
-        self.assertFalse(os.path.exists(temp_dir))
-        self.assertIsNone(self.profile_manager.root_dir)
-        self.assertIsNone(self.profile_manager.pg_cluster)
+        # pg_ctl is what PGTest also looks for (although it might be more clever)
+        pgtest = {'pg_ctl': which('pg_ctl')}
+        self.test_manager.use_temporary_profile(backend=self.backend, pgtest=pgtest)
 
 
 if __name__ == '__main__':

aiida/manage/tests/__init__.py

@@ -76,14 +76,18 @@ def use_temporary_profile(self, backend=None, pgtest=None):
          Uses :py:class:`aiida.manage.tests.TemporaryProfileManager` internally.
 
         :param backend: Backend to use.
-        :param pgtest: parameters for pgtest.pgtest.PGTest
+        :param pgtest: a dictionary of arguments to be passed to PGTest() for starting the postgresql cluster,
+           e.g. {'pg_ctl': '/somepath/pg_ctl'}. Should usually not be necessary.
+
         """
         if configuration.PROFILE is not None:
             raise TestManagerError('AiiDA dbenv must not be loaded before setting up a test profile.')
         if self._manager is not None:
             raise TestManagerError('Profile manager already loaded.')
-        self._manager = TemporaryProfileManager(backend=backend)
-        self._manager.create_profile(pgtest=pgtest)
+
+        mngr = TemporaryProfileManager(backend=backend, pgtest=pgtest)
+        mngr.create_profile()
+        self._manager = mngr  # don't assign before profile has actually been created!
 
     def use_profile(self, profile_name):
         """Set up Test manager to use existing profile.
@@ -109,7 +113,7 @@ def reset_db(self):
     def destroy_all(self):
         if self._manager:
             self._manager.destroy_all()
-        #self._manager = None
+        self._manager = None
 
 
 class ProfileManager(object):
@@ -232,12 +236,20 @@ class TemporaryProfileManager(ProfileManager):
 
     _test_case = None
 
-    def __init__(self, backend=BACKEND_DJANGO):  # pylint: disable=super-init-not-called
+    def __init__(self, backend=BACKEND_DJANGO, pgtest=None):  # pylint: disable=super-init-not-called
+        """Construct a TemporaryProfileManager
+
+        :param backend: a database backend
+        :param pgtest: a dictionary of arguments to be passed to PGTest() for starting the postgresql cluster,
+           e.g. {'pg_ctl': '/somepath/pg_ctl'}. Should usually not be necessary.
+
+        """
         from aiida.manage.configuration import settings
 
         self.dbinfo = {}
         self.profile_info = _DEFAULT_PROFILE_INFO
         self.profile_info['database_backend'] = backend
+        self._pgtest = pgtest or {}
 
         self.pg_cluster = None
         self.postgres = None
@@ -266,52 +278,43 @@ def profile_dictionary(self):
         }
         return dictionary
 
-    def create_db_cluster(self, pgtest=None):
+    def create_db_cluster(self):
         """
         Create the database cluster using PGTest.
-
-        :param pgtest: a dictionary containing input to PGTest()
         """
         from pgtest.pgtest import PGTest
 
         if self.pg_cluster is not None:
             raise TestManagerError(
                 'Running temporary postgresql cluster detected.' + 'Use destroy_all() before creating a new cluster.'
             )
-        if pgtest is None:
-            pgtest = {}
-        self.pg_cluster = PGTest(**pgtest)
+        self.pg_cluster = PGTest(**self._pgtest)
         self.dbinfo.update(self.pg_cluster.dsn)
 
-    def create_aiida_db(self, pgtest=None):
+    def create_aiida_db(self):
         """
         Create the necessary database on the temporary postgres instance.
-
-        By utilizing pgtest it is possible to forward initialization arguments to PGTest().
-
-        :param pgtest: a dictionary containing input to PGTest()
         """
         if configuration.PROFILE is not None:
             raise TestManagerError('AiiDA dbenv can not be loaded while creating a tests db environment')
         if self.pg_cluster is None:
-            self.create_db_cluster(pgtest)
+            self.create_db_cluster()
         self.postgres = Postgres(interactive=False, quiet=True, dbinfo=self.dbinfo)
         self.dbinfo = self.postgres.dbinfo.copy()
         self.postgres.create_dbuser(self.profile_info['database_username'], self.profile_info['database_password'])
         self.postgres.create_db(self.profile_info['database_username'], self.profile_info['database_name'])
         self._has_test_db = True
 
-    def create_profile(self, pgtest=None):
+    def create_profile(self):
         """
         Set AiiDA to use the tests config dir and create a default profile there
 
         Warning: the AiiDA dbenv must not be loaded when this is called!
-        :param pgtest: a dictionary containing input to PGTest()
         """
         from aiida.manage.configuration import settings, load_profile, Profile
 
         if not self._has_test_db:
-            self.create_aiida_db(pgtest)
+            self.create_aiida_db()
 
         if not self.root_dir:
             self.root_dir = tempfile.mkdtemp()