from __future__ import (absolute_import, division, print_function)

__metaclass__ = type

import unittest
import sys
import os

path = os.path.dirname(os.path.realpath(__file__))
path = "{0}/../../plugins/module_utils".format(path)
sys.path.append(path)
import mongodb_common
from pymongo import MongoClient
from bson.timestamp import Timestamp
import datetime
from bson import ObjectId


class FakeAnsibleModule:
    params = {
        "login_host": "localhost",
        "login_port": 27017,
        "replica_set": "rs0",
        "reconfigure": True,
        "ssl": False,
        "ssl_cert_reqs": "CERT_REQUIRED",
        "ssl_ca_certs": "/tmp/ca.crt",
        "ssl_crlfile": "/tmp/tmp.crl",
        "ssl_certfile": "/tmp/tls.key",
        "ssl_keyfile": "/tmp/tls.key",
        "ssl_pem_passphrase": "secret",
        "auth_mechanism": None,
        "atlas_auth": False,
        "connection_options": [{"one": 1},
                               {"two": 2},
                               {"three": 3},
                               {"four": 4},
                               {"five": 5}]
    }

    def __init__(self):
        self.msg = ""
        self.warning = ""

    def fail_json(self, msg):
        self.msg = msg

    def warn(self, warning):
        self.warning = warning

    def get_msg(self):
        return self.msg

    def get_warn(self):
        return self.warning


class TestMongoDBCommonMethods(unittest.TestCase):
    member_config_defaults = {
        "arbiterOnly": False,
        "buildIndexes": True,
        "hidden": False,
        "priority": 1,
        "tags": {},
        "horizons": {},
        "secondardDelaySecs": 0,
        "votes": 1
    }

    def test_check_compatibility_strict_compatibility_True(self):
        tests = [
            ('4.4.4', '3.12.1', False),  # srv, drv, empty msg expected
            ('5.0.1', '3.12.1', False),
            ('4.4.4', '4.0.1', True),
            ('5.0.1', '4.0.1', True),
            ('2.4.1', '3.12.1', False),
            ('2.4.1', '4.0.1', False),
            ('2.4.1', '2.3.4', False),
        ]
        for t in tests:
            fake_module = FakeAnsibleModule()
            fake_module.params['strict_compatibility'] = True
            mongodb_common.check_compatibility(fake_module, t[0], t[1])
            msg = fake_module.get_msg()
            if t[2]:
                assert msg == ""
            else:
                assert "This version of MongoDB is pretty old" in msg \
                       or 'You must use pymongo 4+' in msg

    def test_check_compatibility_strict_compatibility_False(self):
        tests = [
            ('4.4.4', '3.12.1', False),  # srv, drv, empty msg expected
            ('5.0.1', '3.12.1', False),
            ('4.4.4', '4.0.1', True),
            ('5.0.1', '4.0.1', True),
            ('2.4.1', '3.12.1', False),
            ('2.4.1', '4.0.1', False),
            ('2.4.1', '2.3.4', False),
        ]
        for t in tests:
            fake_module = FakeAnsibleModule()
            fake_module.params['strict_compatibility'] = False
            mongodb_common.check_compatibility(fake_module, t[0], t[1])
            warn = fake_module.get_warn()
            if t[2]:
                assert warn == ""
            else:
                assert 'You must use pymongo 4+' in warn \
                       or 'This version of MongoDB is pretty old' in warn

    def test_load_mongocnf(self):
        with open(os.path.expanduser("~/.mongodb.cnf"), "w+") as w:
            w.write("""
            [client]
            user = mongo_user
            pass = 123456
            """)
        creds = mongodb_common.load_mongocnf()
        assert creds['user'] == "mongo_user"
        assert creds['password'] == "123456"

    def test_mongodb_common_argument_spec(self):
        mongo_dict = mongodb_common.mongodb_common_argument_spec()
        assert "login_user" in mongo_dict
        assert "login_password" in mongo_dict
        assert "login_host" in mongo_dict
        assert "login_port" in mongo_dict
        assert "ssl" in mongo_dict
        assert "ssl_cert_reqs" in mongo_dict
        assert "ssl_ca_certs" in mongo_dict
        assert "ssl_crlfile" in mongo_dict
        assert "ssl_certfile" in mongo_dict
        assert "ssl_keyfile" in mongo_dict
        assert "ssl_pem_passphrase" in mongo_dict
        assert "auth_mechanism" in mongo_dict
        assert mongo_dict["login_port"]["default"] == 27017
        assert mongo_dict["login_host"]["default"] == "localhost"
        assert mongo_dict["login_database"]["default"] == "admin"

    def test_ssl_connection_options(self):
        connection_params = dict()
        fake_module = FakeAnsibleModule()
        ssl_dict = mongodb_common.ssl_connection_options(connection_params, fake_module)
        assert isinstance(ssl_dict, dict)
        assert ssl_dict["ssl"] is True
        assert "ssl_cert_reqs" in ssl_dict
        assert "ssl_ca_certs" in ssl_dict
        assert "ssl_crlfile" in ssl_dict
        assert "ssl_certfile" in ssl_dict
        assert "ssl_keyfile" in ssl_dict
        assert "ssl_pem_passphrase" in ssl_dict
        assert "authMechanism" not in ssl_dict

    def test_ssl_connection_options_auth_mechanism(self):
        connection_params = dict()
        fake_module = FakeAnsibleModule()
        fake_module.params["auth_mechanism"] = "MONGODB-X509"
        ssl_dict = mongodb_common.ssl_connection_options(connection_params, fake_module)
        assert isinstance(ssl_dict, dict)
        assert ssl_dict["ssl"] is True
        assert "ssl_cert_reqs" in ssl_dict
        assert "ssl_ca_certs" in ssl_dict
        assert "ssl_crlfile" in ssl_dict
        assert "ssl_certfile" in ssl_dict
        assert "ssl_keyfile" in ssl_dict
        assert "ssl_pem_passphrase" in ssl_dict
        assert "authMechanism" in ssl_dict
        assert ssl_dict["one"] == 1
        assert ssl_dict["two"] == 2
        assert ssl_dict["three"] == 3
        assert ssl_dict["four"] == 4
        assert ssl_dict["five"] == 5
        assert ssl_dict["authMechanism"] == "MONGODB-X509"

    def test_ssl_connection_options_auth_mechanism_strings(self):
        connection_params = dict()
        fake_module = FakeAnsibleModule()
        fake_module.params["auth_mechanism"] = "MONGODB-X509"
        fake_module.params["connection_options"] = [
            "one=1",
            "two=2",
            "three=3",
            "four=4",
            "five=5"
        ]
        ssl_dict = mongodb_common.ssl_connection_options(connection_params, fake_module)
        assert isinstance(ssl_dict, dict)
        assert ssl_dict["ssl"] is True
        assert "ssl_cert_reqs" in ssl_dict
        assert "ssl_ca_certs" in ssl_dict
        assert "ssl_crlfile" in ssl_dict
        assert "ssl_certfile" in ssl_dict
        assert "ssl_keyfile" in ssl_dict
        assert "ssl_pem_passphrase" in ssl_dict
        assert "authMechanism" in ssl_dict
        assert ssl_dict["one"] == '1'
        assert ssl_dict["two"] == '2'
        assert ssl_dict["three"] == '3'
        assert ssl_dict["four"] == '4'
        assert ssl_dict["five"] == '5'
        assert ssl_dict["authMechanism"] == "MONGODB-X509"

    def test_member_state(self):
        client = MongoClient(host=['localhost:27017'],
                             username='user',
                             password='password',
                             replicaSet='replset')
        ms = mongodb_common.member_state(client)
        assert ms == "PRIMARY" or ms == "SECONDARY"

    def test_index_exists(self):
        client = MongoClient(host=['localhost:27017'],
                             username='user',
                             password='password',
                             replicaSet='replset')
        mongodb_common.create_index(client,
                                    'test',
                                    'rhys',
                                    {'username': 1},
                                    {"name": "test_index"})
        index_exists = mongodb_common.index_exists(client,
                                                   'test',
                                                   'rhys',
                                                   'test_index')
        assert isinstance(index_exists, bool) and index_exists
        mongodb_common.drop_index(client,
                                  'test',
                                  'rhys',
                                  'test_index')
        index_exists = mongodb_common.index_exists(client,
                                                   'test',
                                                   'rhys',
                                                   'test_index')
        assert isinstance(index_exists, bool) and not index_exists

    def test_mongodb_auth(self):
        fake_module = FakeAnsibleModule()
        fake_module.params["login_user"] = "user"
        fake_module.params["login_password"] = "password"
        fake_module.params["login_database"] = "test"
        fake_module.params["strict_compatibility"] = False

        client = MongoClient(host=['localhost:27017'],
                             username=None,
                             password=None,
                             replicaSet='replset')

        success = mongodb_common.mongo_auth(fake_module, client)
        assert success

        with open(os.path.expanduser("~/.mongodb.cnf"), "w+") as w:
            w.write("""
            [client]
            user = mongo_user
            pass = 123456
            """)
            fake_module.params["login_user"] = "user"
            fake_module.params["login_password"] = "password"
            fake_module.params["login_database"] = "test"
            client = MongoClient(host=['localhost:27017'],
                                 username=None,
                                 password=None,
                                 replicaSet='replset')

            success = mongodb_common.mongo_auth(fake_module, client)
            assert success

        with open(os.path.expanduser("~/.mongodb.cnf"), "w+") as w:
            w.write("""
            [client]
            user = mongo_user
            pass = 123456
            """)
            fake_module.params["login_user"] = None
            fake_module.params["login_password"] = "password"
            fake_module.params["login_database"] = "test"
            client = MongoClient(host=['localhost:27017'],
                                 username=None,
                                 password=None,
                                 replicaSet='replset')

            success = mongodb_common.mongo_auth(fake_module, client)
            assert success

    def test_mongo_auth(self):
        client = MongoClient(host=['localhost:27017'],
                             username='user',
                             password='password',
                             directConnection=True)
        fake_module = FakeAnsibleModule()
        fake_module.params["login_user"] = "dummy"
        fake_module.params["login_password"] = None
        fake_module.params["login_database"] = "test"
        fake_module.params["replica_set"] = None
        fake_module.params["strict_compatibility"] = False
        mongodb_common.mongo_auth(fake_module, client)
        msg = fake_module.get_msg()
        assert "When supplying login arguments" in msg

        fake_module.params["login_password"] = "password"
        fake_module.params["login_database"] = "test"
        client = mongodb_common.mongo_auth(fake_module, client)
        assert "MongoClient" in str(client)

        fake_module.params["'create_for_localhost_exception"] = None
        client = mongodb_common.mongo_auth(fake_module, client)
        assert "MongoClient" in str(client)

        if os.path.exists(os.path.expanduser('~/.mongodb.cnf')):
            os.remove(os.path.expanduser('~/.mongodb.cnf'))
        fake_module.params["login_user"] = None
        fake_module.params["login_password"] = None
        client = mongodb_common.mongo_auth(fake_module, client)
        fail_msg = fake_module.get_msg()
        self.assertTrue('No credentials to authenticate' in fail_msg, msg='{0}'.format(fail_msg))

        fake_module.params['create_for_localhost_exception'] = None
        fake_module.params["login_user"] = None
        fake_module.params["login_password"] = None
        fake_module.params["login_database"] = "test"
        fake_module.params["database"] = "test"
        client = mongodb_common.mongo_auth(fake_module, client)
        fail_msg = fake_module.get_msg()
        print(fail_msg)
        self.assertTrue('The localhost login exception only allows the first admin account to be created' in fail_msg)

        fake_module.params['create_for_localhost_exception'] = None
        fake_module.params["login_user"] = None
        fake_module.params["login_password"] = None
        fake_module.params["login_database"] = "admin"
        fake_module.params["database"] = "admin"
        client = mongodb_common.mongo_auth(fake_module, client)
        assert "MongoClient" in str(client)

        client = MongoClient(host=['localhost:27017'],
                             replicaSet='replset')
        fake_module.params['create_for_localhost_exception'] = None
        fake_module.params["login_user"] = "user"
        fake_module.params["login_password"] = "password"
        fake_module.params["login_database"] = "test"
        try:
            client = mongodb_common.mongo_auth(fake_module, client)
        except Exception as excep:
            assert 'Authentication failed' in str(excep)
            assert "MongoClient" in str(client)

    def test_member_dicts_different_1(self):
        # mongodb replicaset config document format
        conf = {
            "members": [
                {"_id": 1, "host": "localhost:3001"},
                {"_id": 2, "host": "localhost:3002"},
                {"_id": 3, "host": "localhost:3003"}
            ]
        }
        for member in conf["members"]:
            member.update(self.member_config_defaults)
        # list of dicts
        members = [{"host": "localhost:3001"},
                   {"host": "localhost:3002"},
                   {"host": "localhost:3003"}]
        self.assertFalse(mongodb_common.member_dicts_different(conf, members))

    def test_member_dicts_different_2(self):
        # mongodb replicaset config document format
        conf = {
            "members": [
                {"_id": 1, "host": "localhost:3001"},
                {"_id": 2, "host": "localhost:3002"}
            ]
        }
        for member in conf["members"]:
            member.update(self.member_config_defaults)
        # list of dicts
        members = [{"host": "localhost:3001"},
                   {"host": "localhost:3002"},
                   {"host": "localhost:3004"}]
        self.assertTrue(mongodb_common.member_dicts_different(conf, members))

    def test_member_dicts_different_3(self):
        # mongodb replicaset config document format
        conf = {
            "members": [{"_id": 1, "host": "localhost:3001"},
                        {"_id": 2, "host": "localhost:3002"},
                        {"_id": 3, "host": "localhost:3003"}]
        }
        for member in conf["members"]:
            member.update(self.member_config_defaults)
        # list of dicts
        members = [{"host": "localhost:3001"},
                   {"host": "localhost:3002"}]
        self.assertTrue(mongodb_common.member_dicts_different(conf, members))

    def test_member_dicts_different_4(self):
        # mongodb replicaset config document format
        conf = {
            "members": [
                {"_id": 1, "host": "localhost:3001"},
                {"_id": 2, "host": "localhost:3002"}
            ]
        }
        for member in conf["members"]:
            member.update(self.member_config_defaults)
        # list of dicts
        members = [{"host": "localhost:3001"},
                   {"host": "localhost:3002"},
                   {"host": "localhost:3004", "votes": 0, "priority": 0, "hidden": True}]
        # 3004 using non-default values
        self.assertTrue(mongodb_common.member_dicts_different(conf, members))

    def test_member_dicts_different_5(self):
        # mongodb replicaset config document format
        conf = {
            "members": [
                {"_id": 1, "host": "localhost:3001"},
                {"_id": 2, "host": "localhost:3002"},
                {"_id": 3, "host": "localhost:3004"}
            ]
        }
        for member in conf["members"]:
            member.update(self.member_config_defaults)
        # list of dicts
        members = [{"host": "localhost:3001"},
                   {"host": "localhost:3002"},
                   {"host": "localhost:3004", "votes": 1, "priority": 1, "hidden": False}]
        # Should return false as the additonal dict keys are default values
        self.assertFalse(mongodb_common.member_dicts_different(conf, members))

    def test_member_dicts_different_6(self):
        # mongodb replicaset config document format
        conf = {
            "members": [
                {"_id": 1, "host": "localhost:3001"},
                {"_id": 2, "host": "localhost:3002"},
                {"_id": 3, "host": "localhost:3004"}
            ]
        }
        for member in conf["members"]:
            member.update(self.member_config_defaults)
        # list of dicts
        members = [{"host": "localhost:3001", "horizons": {"test": "testhost:3001"}},
                   {"host": "localhost:3002", "horizons": {"test": "testhost:3002"}},
                   {"host": "localhost:3004", "horizons": {"test": "testhost:3004"}}]
        # Using non default values
        self.assertTrue(mongodb_common.member_dicts_different(conf, members))

    def test_lists_are_different1(self):
        l1 = [
            "localhost:3001",
            "localhost:3002",
            "localhost:3003"
        ]
        l2 = [
            "localhost:3001",
            "localhost:3002",
            "localhost:3003"
        ]
        self.assertFalse(mongodb_common.lists_are_different(l1, l2))

    def test_lists_are_different2(self):
        l1 = [
            "localhost:3001",
            "localhost:3002"
        ]
        l2 = [
            "localhost:3001",
            "localhost:3002",
            "localhost:3003"
        ]
        self.assertTrue(mongodb_common.lists_are_different(l1, l2))

    def test_lists_are_different1(self):
        l1 = [
            "localhost:3001",
            "localhost:3002",
            "localhost:3003"
        ]
        l2 = [
            "localhost:3002",
            "localhost:3003"
        ]
        self.assertTrue(mongodb_common.lists_are_different(l1, l2))

    def test_get_mongodb_client_1(self):
        fake_module = FakeAnsibleModule()
        client = mongodb_common.get_mongodb_client(fake_module)
        assert "MongoClient" in str(client)

    # def test_get_mongodb_client_2(self):
    #     fake_module = FakeAnsibleModule()
    #     fake_module.params['ssl'] = True
    #     client = mongodb_common.get_mongodb_client(fake_module)
    #     assert "MongoClient" in str(client)

    def test_get_mongodb_client_3(self):
        fake_module = FakeAnsibleModule()
        del fake_module.params['reconfigure']
        client = mongodb_common.get_mongodb_client(fake_module)
        assert "MongoClient" in str(client)

    def test_is_auth_enabled(self):
        fake_module = FakeAnsibleModule()
        fake_module.params['replica_set'] = 'replset'
        result = mongodb_common.is_auth_enabled(fake_module)
        print("result = {0}".format(result))
        assert result

    def test_is_auth_enabled_no_auth(self):
        fake_module = FakeAnsibleModule()
        fake_module.params['login_port'] = 27999
        fake_module.params['replica_set'] = None
        result = mongodb_common.is_auth_enabled(fake_module)
        assert result is False

    def test_convert_to_supported_timestamp(self):
        dt = Timestamp(datetime.datetime.now(), 0)
        assert isinstance(dt, Timestamp)
        dt = mongodb_common.convert_to_supported(dt)
        assert isinstance(dt, str)

    def test_convert_to_supported_objectid(self):
        o = ObjectId()
        assert isinstance(o, ObjectId)
        o = mongodb_common.convert_to_supported(o)
        assert isinstance(o, str)

    def test_convert_to_supported_int(self):
        i = 1
        assert isinstance(i, int)
        i = mongodb_common.convert_to_supported(i)
        assert isinstance(i, int)

    def test_convert_to_supported_str(self):
        s = "hello world!"
        assert isinstance(s, str)
        s = mongodb_common.convert_to_supported(s)
        assert isinstance(s, str)

    def test_convert_bson_values_recur(self):
        d = {
            "id": ObjectId(),
            "dt": Timestamp(datetime.datetime.now(), 1),
            "nested": {
                "id": ObjectId(),
                "dt": Timestamp(datetime.datetime.now(), 1),
                "i": 1,
                "s": "hello world!"
            },
            "i": 1,
            "s": "hello world!"
        }
        d = mongodb_common.convert_bson_values_recur(d)
        assert isinstance(d["id"], str)
        assert isinstance(d["dt"], str)
        assert isinstance(d["nested"]["id"], str)
        assert isinstance(d["nested"]["dt"], str)
        assert isinstance(d["nested"]["i"], int)
        assert isinstance(d["nested"]["s"], str)
        assert isinstance(d["i"], int)
        assert isinstance(d["s"], str)


if __name__ == '__main__':
    unittest.main()
