test_user.py

# -*- coding: utf-8 -*-

# -----------------------------------------------------------------------------
# Copyright (c) 2014--, The Qiita Development Team.
#
# Distributed under the terms of the BSD 3-clause License.
#
# The full license is in the file LICENSE, distributed with this software.
# -----------------------------------------------------------------------------

from unittest import TestCase, main
from datetime import datetime, timedelta

from qiita_core.exceptions import (IncorrectEmailError, IncorrectPasswordError,
                                   IncompetentQiitaDeveloperError)
from qiita_core.util import qiita_test_checker
from qiita_core.qiita_settings import qiita_config
import qiita_db as qdb


class SupportTests(TestCase):
    def test_validate_password(self):
        valid1 = 'abcdefgh'
        valid2 = 'abcdefgh1234'
        valid3 = 'abcdefgh!@#$'
        valid4 = 'aBC123!@#{}'
        invalid1 = 'abc'
        invalid2 = u'øabcdefghi'
        invalid3 = 'abcd   efgh'

        self.assertTrue(qdb.user.validate_password(valid1))
        self.assertTrue(qdb.user.validate_password(valid2))
        self.assertTrue(qdb.user.validate_password(valid3))
        self.assertTrue(qdb.user.validate_password(valid4))
        self.assertFalse(qdb.user.validate_password(invalid1))
        self.assertFalse(qdb.user.validate_password(invalid2))
        self.assertFalse(qdb.user.validate_password(invalid3))

    def test_validate_email(self):
        valid1 = 'foo@bar.com'
        valid2 = 'asdasd.asdasd.asd123asd@stuff.edu'
        valid3 = 'w00t@123.456.789.com'
        valid4 = 'name@a.b-c.d'
        invalid1 = '@stuff.com'
        invalid2 = 'asdasdásd@things.com'
        invalid3 = '.asdas@com'
        invalid4 = 'name@a.b-c.d-'

        self.assertTrue(qdb.user.validate_email(valid1))
        self.assertTrue(qdb.user.validate_email(valid2))
        self.assertTrue(qdb.user.validate_email(valid3))
        self.assertTrue(qdb.user.validate_email(valid4))
        self.assertFalse(qdb.user.validate_email(invalid1))
        self.assertFalse(qdb.user.validate_email(invalid2))
        self.assertFalse(qdb.user.validate_email(invalid3))
        self.assertFalse(qdb.user.validate_email(invalid4))


@qiita_test_checker()
class UserTest(TestCase):
    """Tests the User object and all properties/methods"""

    def setUp(self):
        self.user = qdb.user.User('admin@foo.bar')
        self.portal = qiita_config.portal

        self.userinfo = {
            'name': 'Dude',
            'affiliation': 'Nowhere University',
            'address': '123 fake st, Apt 0, Faketown, CO 80302',
            'phone': '111-222-3344',
            'pass_reset_code': None,
            'pass_reset_timestamp': None,
            'user_verify_code': None,
            'receive_processing_job_emails': True,
            'social_orcid': None,
            'social_researchgate': None,
            'social_googlescholar': None,
            'creation_timestamp': None
        }

    def tearDown(self):
        qiita_config.portal = self.portal

    def test_instantiate_user(self):
        qdb.user.User('admin@foo.bar')

    def test_instantiate_unknown_user(self):
        with self.assertRaises(qdb.exceptions.QiitaDBUnknownIDError):
            qdb.user.User('FAIL@OMG.bar')

    def _check_correct_info(self, obs, exp, ts_before=None):
        """Compares info dict of user with special handling of specific keys.

        Parameters
        ----------
        obs : dict
            Observed user info dictionary.
        exp : dict
            Expected user info dictionary.
        ts_before : datetime.datetime or None
            User.create records the creation timestamp through SQL's NOW().
            Since it is set by the database to the microsecond, we can't
            predict it a priori and therefore simply record timestamp before
            execution of user.create() and compare the relation.
            The DB creation_timestamp column is optional, i.e. can be None.
        """
        self.assertEqual(set(exp.keys()), set(obs.keys()))
        for key in exp:
            # user_verify_code and password seed randomly generated so just
            # making sure they exist and is correct length
            if key == 'user_verify_code':
                self.assertEqual(len(obs[key]), 20)
            elif key == "password":
                self.assertEqual(len(obs[key]), 60)
            elif key == "creation_timestamp":
                self.assertTrue(((exp[key] is None) and (obs[key] is None))
                                or (ts_before <= exp[key]))
            else:
                self.assertEqual(obs[key], exp[key])

    def test_create_user(self):
        before = datetime.now()
        user = qdb.user.User.create('testcreateuser@test.bar', 'password')

        # adding a couple of messages
        qdb.util.add_system_message("TESTMESSAGE_OLD", datetime.now())
        qdb.util.add_system_message(
            "TESTMESSAGE", datetime.now() + timedelta(milliseconds=1))

        self.assertEqual(user.id, 'testcreateuser@test.bar')
        sql = """SELECT *
                 FROM qiita.qiita_user
                 WHERE email = 'testcreateuser@test.bar'"""
        with qdb.sql_connection.TRN:
            qdb.sql_connection.TRN.add(sql)
            obs = qdb.sql_connection.TRN.execute_fetchindex()
        self.assertEqual(len(obs), 1)
        obs = dict(obs[0])
        exp = {
            'password': '',
            'name': None,
            'pass_reset_timestamp': None,
            'affiliation': None,
            'pass_reset_code': None,
            'phone': None,
            'user_verify_code': '',
            'address': None,
            'user_level_id': 5,
            'receive_processing_job_emails': False,
            'email': 'testcreateuser@test.bar',
            'social_orcid': None,
            'social_researchgate': None,
            'social_googlescholar': None,
            'creation_timestamp': datetime.now()}
        self._check_correct_info(obs, exp, before)

        # Make sure new system messages are linked to user
        sql = """SELECT message_id FROM qiita.message_user
                 WHERE email = 'testcreateuser@test.bar'"""
        m_id = qdb.util.get_count('qiita.message')
        # the user should have the latest message (m_id) and the one before
        with qdb.sql_connection.TRN:
            qdb.sql_connection.TRN.add(sql)
            obs = qdb.sql_connection.TRN.execute_fetchindex()
        self.assertEqual(obs, [[m_id-1], [m_id]])
        qdb.util.clear_system_messages()

    def test_create_user_info(self):
        before = datetime.now()
        user = qdb.user.User.create('testcreateuserinfo@test.bar', 'password',
                                    self.userinfo)
        self.assertEqual(user.id, 'testcreateuserinfo@test.bar')
        sql = """SELECT *
                 FROM qiita.qiita_user
                 WHERE email = 'testcreateuserinfo@test.bar'"""
        with qdb.sql_connection.TRN:
            qdb.sql_connection.TRN.add(sql)
            obs = qdb.sql_connection.TRN.execute_fetchindex()
        self.assertEqual(len(obs), 1)
        obs = dict(obs[0])
        exp = {
            'password': '',
            'name': 'Dude',
            'affiliation': 'Nowhere University',
            'address': '123 fake st, Apt 0, Faketown, CO 80302',
            'phone': '111-222-3344',
            'pass_reset_timestamp': None,
            'pass_reset_code': None,
            'user_verify_code': '',
            'user_level_id': 5,
            'receive_processing_job_emails': True,
            'email': 'testcreateuserinfo@test.bar',
            'social_orcid': None,
            'social_researchgate': None,
            'social_googlescholar': None,
            'creation_timestamp': datetime.now()}
        self._check_correct_info(obs, exp, before)

    def test_create_user_column_not_allowed(self):
        self.userinfo["email"] = "FAIL"
        with self.assertRaises(qdb.exceptions.QiitaDBColumnError):
            qdb.user.User.create('new@test.bar', 'password', self.userinfo)

    def test_create_user_non_existent_column(self):
        self.userinfo["BADTHING"] = "FAIL"
        with self.assertRaises(qdb.exceptions.QiitaDBColumnError):
            qdb.user.User.create('new@test.bar', 'password', self.userinfo)

    def test_create_user_duplicate(self):
        with self.assertRaises(qdb.exceptions.QiitaDBDuplicateError):
            qdb.user.User.create('test@foo.bar', 'password')

    def test_create_user_bad_email(self):
        with self.assertRaises(IncorrectEmailError):
            qdb.user.User.create('notanemail', 'password')

    def test_create_user_bad_password(self):
        with self.assertRaises(IncorrectPasswordError):
            qdb.user.User.create('new@test.com', '')

    def test_login(self):
        self.assertEqual(qdb.user.User.login("test@foo.bar", "password"),
                         qdb.user.User("test@foo.bar"))

    def test_login_incorrect_user(self):
        with self.assertRaises(IncorrectEmailError):
            qdb.user.User.login("notexist@foo.bar", "password")

    def test_login_incorrect_password(self):
        with self.assertRaises(IncorrectPasswordError):
            qdb.user.User.login("test@foo.bar", "WRONGPASSWORD")

    def test_login_invalid_password(self):
        with self.assertRaises(IncorrectPasswordError):
            qdb.user.User.login("test@foo.bar", "SHORT")

    def test_exists(self):
        self.assertTrue(qdb.user.User.exists("test@foo.bar"))

    def test_exists_notindb(self):
        self.assertFalse(qdb.user.User.exists("notexist@foo.bar"))

    def test_exists_invalid_email(self):
        with self.assertRaises(IncorrectEmailError):
            qdb.user.User.exists("notanemail.@badformat")

    def test_get_email(self):
        self.assertEqual(self.user.email, 'admin@foo.bar')

    def test_get_level(self):
        self.assertEqual(self.user.level, "admin")

    def test_get_info(self):
        expinfo = {
            'name': 'Admin',
            'affiliation': 'Owner University',
            'address': '312 noname st, Apt K, Nonexistantown, CO 80302',
            'phone': '222-444-6789',
            'pass_reset_code': None,
            'pass_reset_timestamp': None,
            'user_verify_code': None,
            'receive_processing_job_emails': False,
            'phone': '222-444-6789',
            'social_orcid': None,
            'social_researchgate': None,
            'social_googlescholar': None,
            'creation_timestamp': None
        }
        self.assertEqual(self.user.info, expinfo)

    def test_set_info(self):
        self.user.info = self.userinfo
        self.assertEqual(self.user.info, self.userinfo)

    def test_set_info_not_info(self):
        """Tests setting info with a non-allowed column"""
        self.userinfo["email"] = "FAIL"
        with self.assertRaises(qdb.exceptions.QiitaDBColumnError):
            self.user.info = self.userinfo

    def test_set_info_bad_info(self):
        """Test setting info with a key not in the table"""
        self.userinfo["BADTHING"] = "FAIL"
        with self.assertRaises(qdb.exceptions.QiitaDBColumnError):
            self.user.info = self.userinfo

    def test_default_analysis(self):
        qiita_config.portal = "QIITA"
        obs = self.user.default_analysis
        self.assertEqual(obs, qdb.analysis.Analysis(4))

        qiita_config.portal = "EMP"
        obs = self.user.default_analysis
        self.assertEqual(obs, qdb.analysis.Analysis(8))

    def test_get_user_studies(self):
        user = qdb.user.User('test@foo.bar')
        qiita_config.portal = "QIITA"
        self.assertEqual(user.user_studies, {qdb.study.Study(1)})

        qiita_config.portal = "EMP"
        self.assertEqual(user.user_studies, set())

    def test_get_shared_studies(self):
        user = qdb.user.User('shared@foo.bar')
        qiita_config.portal = "QIITA"
        self.assertEqual(user.shared_studies, {qdb.study.Study(1)})

        qiita_config.portal = "EMP"
        self.assertEqual(user.shared_studies, set())

    def test_get_private_analyses(self):
        user = qdb.user.User('test@foo.bar')
        qiita_config.portal = "QIITA"
        exp = {qdb.analysis.Analysis(1)}
        self.assertEqual(user.private_analyses, exp)

        qiita_config.portal = "EMP"
        self.assertEqual(user.private_analyses, set())

    def test_get_shared_analyses(self):
        user = qdb.user.User('shared@foo.bar')
        qiita_config.portal = "QIITA"
        self.assertEqual(user.shared_analyses, {qdb.analysis.Analysis(1)})

        qiita_config.portal = "EMP"
        self.assertEqual(user.shared_analyses, set())

    def test_verify_code(self):
        email = 'testverifycode@test.bar'
        qdb.user.User.create(email, 'password')
        # making sure that we know the user codes
        sql = """UPDATE qiita.qiita_user SET
                        user_verify_code='verifycode',
                        pass_reset_code='resetcode'
                    WHERE email=%s"""
        qdb.sql_connection.perform_as_transaction(sql, [email])

        self.assertFalse(
            qdb.user.User.verify_code(email, 'wrongcode', 'create'))
        self.assertFalse(
            qdb.user.User.verify_code(email, 'wrongcode', 'reset'))

        self.assertTrue(
            qdb.user.User.verify_code(email, 'verifycode', 'create'))
        self.assertTrue(
            qdb.user.User.verify_code(email, 'resetcode', 'reset'))

        # make sure errors raised if code already used or wrong type
        with self.assertRaises(qdb.exceptions.QiitaDBError):
            qdb.user.User.verify_code(email, 'verifycode', 'create')
        with self.assertRaises(qdb.exceptions.QiitaDBError):
            qdb.user.User.verify_code(email, 'resetcode', 'reset')

        with self.assertRaises(IncompetentQiitaDeveloperError):
            qdb.user.User.verify_code(email, 'fakecode', 'badtype')

        # make sure default analyses created
        sql = ("SELECT email, name, description, dflt FROM qiita.analysis "
               "WHERE email = %s")
        with qdb.sql_connection.TRN:
            qdb.sql_connection.TRN.add(sql, [email])
            obs = qdb.sql_connection.TRN.execute_fetchindex()
        exp = [[email, 'testverifycode@test.bar-dflt-2', 'dflt', True],
               [email, 'testverifycode@test.bar-dflt-1', 'dflt', True]]
        self.assertEqual(obs, exp)

        # Make sure default analyses are linked with the portal
        sql = """SELECT COUNT(1)
                 FROM qiita.analysis
                    JOIN qiita.analysis_portal USING (analysis_id)
                    JOIN qiita.portal_type USING (portal_type_id)
                 WHERE email = 'testverifycode@test.bar' AND dflt = true"""
        with qdb.sql_connection.TRN:
            qdb.sql_connection.TRN.add(sql)
            obs = qdb.sql_connection.TRN.execute_fetchflatten()[0]
        self.assertEqual(obs, 2)

    def _check_pass(self, user, passwd):
        self.assertEqual(qdb.util.hash_password(passwd, user.password),
                         user.password)

    def test_password(self):
        user = qdb.user.User('shared@foo.bar')
        self.assertEqual(user.password, '$2a$12$gnUi8Qg.0tvW243v889BhOBhWLIHy'
                         'IJjjgaG6dxuRJkUM8nXG9Efe')

    def test_change_pass(self):
        user = qdb.user.User.create('testchangepass@test.bar', 'password')
        user._change_pass("newpassword")
        self._check_pass(user, "newpassword")
        self.assertIsNone(user.info["pass_reset_code"])

    def test_change_pass_short(self):
        with self.assertRaises(IncorrectPasswordError):
            self.user._change_pass("newpass")
        self._check_pass(self.user, "password")

    def test_change_password(self):
        self.user.change_password("password", "newpassword")
        self._check_pass(self.user, "newpassword")

    def test_change_password_wrong_oldpass(self):
        user = qdb.user.User.create('changepasswrongold@test.bar', 'password')
        user.change_password("WRONG", "newpass")
        self._check_pass(user, "password")

    def test_generate_reset_code(self):
        user = qdb.user.User.create('new@test.bar', 'password')
        sql = "SELECT LOCALTIMESTAMP"
        with qdb.sql_connection.TRN:
            qdb.sql_connection.TRN.add(sql)
            before = qdb.sql_connection.TRN.execute_fetchflatten()[0]
        user.generate_reset_code()
        with qdb.sql_connection.TRN:
            qdb.sql_connection.TRN.add(sql)
            after = qdb.sql_connection.TRN.execute_fetchflatten()[0]
            sql = ("SELECT pass_reset_code, pass_reset_timestamp FROM "
                   "qiita.qiita_user WHERE email = %s")
            qdb.sql_connection.TRN.add(sql, ('new@test.bar',))
            obscode, obstime = qdb.sql_connection.TRN.execute_fetchindex()[0]
        self.assertEqual(len(obscode), 20)
        self.assertTrue(before < obstime < after)

    def test_change_forgot_password(self):
        user = qdb.user.User.create(
            'changeforgotpassword@test.bar', 'password')
        user.generate_reset_code()
        code = user.info["pass_reset_code"]
        obsbool = user.change_forgot_password(code, "newpassword")
        self.assertEqual(obsbool, True)
        self._check_pass(user, "newpassword")

    def test_change_forgot_password_bad_code(self):
        user = qdb.user.User.create('badcode@test.bar', 'password')
        user.generate_reset_code()
        code = "AAAAAAA"
        obsbool = user.change_forgot_password(code, "newpassword")
        self.assertEqual(obsbool, False)
        self._check_pass(user, "password")

    def test_messages(self):
        qdb.util.add_system_message('SYS MESSAGE', datetime.now())
        user = qdb.user.User('test@foo.bar')
        obs = user.messages()
        exp_msg = [
            'SYS MESSAGE', 'message 1',
            'Lorem ipsum dolor sit amet, consectetur adipiscing elit. '
            'Pellentesque sed auctor ex, non placerat sapien. Vestibulum '
            'vestibulum massa ut sapien condimentum, cursus consequat diam'
            ' sodales. Nulla aliquam arcu ut massa auctor, et vehicula '
            'mauris tempor. In lacinia viverra ante quis pellentesque. '
            'Nunc vel mi accumsan, porttitor eros ut, pharetra elit. Nulla'
            ' ac nisi quis dui egestas malesuada vitae ut mauris. Morbi '
            'blandit non nisl a finibus. In erat velit, congue at ipsum '
            'sit amet, venenatis bibendum sem. Curabitur vel odio sed est '
            'rutrum rutrum. Quisque efficitur ut purus in ultrices. '
            'Pellentesque eu auctor justo.', 'message <a href="#">3</a>']
        self.assertCountEqual([(x[1]) for x in obs], exp_msg)
        self.assertTrue(all(x[2] < datetime.now() for x in obs))
        self.assertFalse(all(x[3] for x in obs))
        self.assertEqual([x[4] for x in obs], [True, False, False, False])

        obs = user.messages(1)
        exp_msg = ['SYS MESSAGE']
        self.assertEqual([x[1] for x in obs], exp_msg)

    def test_mark_messages(self):
        user = qdb.user.User('test@foo.bar')
        user.mark_messages([1, 2])
        obs = user.messages()
        exp = [True, True, False]
        self.assertEqual([x[3] for x in obs], exp)

        user.mark_messages([1], read=False)
        obs = user.messages()
        exp = [False, True, False]
        self.assertEqual([x[3] for x in obs], exp)

    def test_delete_messages(self):
        user = qdb.user.User.create('deletemsg@test.bar', 'password')
        self.assertEqual(user.messages(), [])
        qdb.util.add_message("New message", [user])
        user_msgs = user.messages()
        # Magic number 1: the actual message
        self.assertEqual([msg[1] for msg in user_msgs], ["New message"])
        # Magic numbers [0][0] - there is only one message and the first
        # element of that message is the message id
        user.delete_messages([user_msgs[0][0]])
        self.assertEqual([msg[1] for msg in user.messages()], [])

    def test_user_artifacts(self):
        user = qdb.user.User('test@foo.bar')
        obs = user.user_artifacts()
        exp = {qdb.study.Study(1): [qdb.artifact.Artifact(1),
                                    qdb.artifact.Artifact(2),
                                    qdb.artifact.Artifact(3),
                                    qdb.artifact.Artifact(4),
                                    qdb.artifact.Artifact(5),
                                    qdb.artifact.Artifact(6),
                                    qdb.artifact.Artifact(7)]}
        self.assertEqual(obs, exp)
        obs = user.user_artifacts(artifact_type='BIOM')
        exp = {qdb.study.Study(1): [qdb.artifact.Artifact(4),
                                    qdb.artifact.Artifact(5),
                                    qdb.artifact.Artifact(6),
                                    qdb.artifact.Artifact(7)]}
        self.assertEqual(obs, exp)

    def test_jobs(self):
        PJ = qdb.processing_job.ProcessingJob
        ignore_status = []
        # generates expected jobs
        jobs = qdb.user.User('shared@foo.bar').jobs(
            ignore_status=ignore_status)
        self.assertEqual(jobs, [PJ('b72369f9-a886-4193-8d3d-f7b504168e75')])

        jobs = qdb.user.User('shared@foo.bar').jobs(
            ignore_status=ignore_status, show_hidden=True)
        self.assertEqual(jobs, [
            PJ('d19f76ee-274e-4c1b-b3a2-a12d73507c55'),
            PJ('b72369f9-a886-4193-8d3d-f7b504168e75')])

        # just one job
        self.assertEqual(qdb.user.User('shared@foo.bar').jobs(
            limit=1, ignore_status=ignore_status), [
                PJ('b72369f9-a886-4193-8d3d-f7b504168e75')])

        # generates expected jobs
        jobs = qdb.user.User('shared@foo.bar').jobs()
        self.assertEqual(jobs, [])

    def test_update_email(self):
        user = qdb.user.User('shared@foo.bar')
        with self.assertRaisesRegex(IncorrectEmailError, 'Bad email given:'):
            user.update_email('bladfa.adferqerq@$EWE')

        with self.assertRaisesRegex(IncorrectEmailError,
                                    'This email already exists'):
            user.update_email('test@foo.bar')

        user.update_email('bla@ble.bli')

    def test_slurm_parameters(self):
        self.assertEqual(qdb.user.User('shared@foo.bar').slurm_parameters,
                         '--nice=10000')
        self.assertEqual(qdb.user.User('admin@foo.bar').slurm_parameters,
                         '--nice=5000')


@qiita_test_checker()
class DeleteUser(TestCase):
    def test_delete_users(self):
        # let's start with the errors
        error = 'This email does not exist: x@y.z'
        with self.assertRaisesRegex(IncorrectEmailError, error):
            qdb.user.User.delete('x@y.z')

        with self.assertRaises(ValueError):
            qdb.user.User.delete('shared@foo.bar')

        qdb.user.User.delete('shared@foo.bar', True)
        # verify that the user doesn't exist any more
        with self.assertRaises(qdb.exceptions.QiitaDBUnknownIDError):
            qdb.user.User('shared@foo.bar')


if __name__ == "__main__":
    main()