# Copyright 2009-present MongoDB, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations

import datetime
import random
import re
import sys
import traceback

sys.path[0:0] = [""]

from bson.dbref import DBRef
from bson.objectid import ObjectId
from bson.son import SON

_IS_SYNC = False

gen_target = 100
reduction_attempts = 10
examples = 5


def lift(value):
    return lambda: value


def choose_lifted(generator_list):
    return lambda: random.choice(generator_list)


def my_map(generator, function):
    return lambda: function(generator())


def choose(list):
    return lambda: random.choice(list)()


def gen_range(start, stop):
    return lambda: random.randint(start, stop)


def gen_int():
    max_int = 2147483647
    return lambda: random.randint(-max_int - 1, max_int)


def gen_float():
    return lambda: (random.random() - 0.5) * sys.maxsize


def gen_boolean():
    return lambda: random.choice([True, False])


def gen_printable_char():
    return lambda: chr(random.randint(32, 126))


def gen_printable_string(gen_length):
    return lambda: "".join(gen_list(gen_printable_char(), gen_length)())


def gen_char(set=None):
    return lambda: bytes([random.randint(0, 255)])


def gen_string(gen_length):
    return lambda: b"".join(gen_list(gen_char(), gen_length)())


def gen_unichar():
    return lambda: chr(random.randint(1, 0xFFF))


def gen_unicode(gen_length):
    return lambda: "".join([x for x in gen_list(gen_unichar(), gen_length)() if x not in ".$"])


def gen_list(generator, gen_length):
    return lambda: [generator() for _ in range(gen_length())]


def gen_datetime():
    return lambda: datetime.datetime(
        random.randint(1970, 2037),
        random.randint(1, 12),
        random.randint(1, 28),
        random.randint(0, 23),
        random.randint(0, 59),
        random.randint(0, 59),
        random.randint(0, 999) * 1000,
    )


def gen_dict(gen_key, gen_value, gen_length):
    def a_dict(gen_key, gen_value, length):
        result = {}
        for _ in range(length):
            result[gen_key()] = gen_value()
        return result

    return lambda: a_dict(gen_key, gen_value, gen_length())


def gen_regexp(gen_length):
    # TODO our patterns only consist of one letter.
    # this is because of a bug in CPython's regex equality testing,
    # which I haven't quite tracked down, so I'm just ignoring it...
    def pattern():
        return "".join(gen_list(choose_lifted("a"), gen_length)())

    def gen_flags():
        flags = 0
        if random.random() > 0.5:
            flags = flags | re.IGNORECASE
        if random.random() > 0.5:
            flags = flags | re.MULTILINE
        if random.random() > 0.5:
            flags = flags | re.VERBOSE

        return flags

    return lambda: re.compile(pattern(), gen_flags())


def gen_objectid():
    return lambda: ObjectId()


def gen_dbref():
    collection = gen_unicode(gen_range(0, 20))
    return lambda: DBRef(collection(), gen_mongo_value(1, True)())


def gen_mongo_value(depth, ref):
    choices = [
        gen_unicode(gen_range(0, 50)),
        gen_printable_string(gen_range(0, 50)),
        my_map(gen_string(gen_range(0, 1000)), bytes),
        gen_int(),
        gen_float(),
        gen_boolean(),
        gen_datetime(),
        gen_objectid(),
        lift(None),
    ]
    if ref:
        choices.append(gen_dbref())
    if depth > 0:
        choices.append(gen_mongo_list(depth, ref))
        choices.append(gen_mongo_dict(depth, ref))
    return choose(choices)


def gen_mongo_list(depth, ref):
    return gen_list(gen_mongo_value(depth - 1, ref), gen_range(0, 10))


def gen_mongo_dict(depth, ref=True):
    return my_map(
        gen_dict(gen_unicode(gen_range(0, 20)), gen_mongo_value(depth - 1, ref), gen_range(0, 10)),
        SON,
    )


def simplify(case):  # TODO this is a hack
    if isinstance(case, SON) and "$ref" not in case:
        simplified = SON(case)  # make a copy!
        if random.choice([True, False]):
            # delete
            simplified_keys = list(simplified)
            if not len(simplified_keys):
                return (False, case)
            simplified.pop(random.choice(simplified_keys))
            return (True, simplified)
        else:
            # simplify a value
            simplified_items = list(simplified.items())
            if not len(simplified_items):
                return (False, case)
            (key, value) = random.choice(simplified_items)
            (success, value) = simplify(value)
            simplified[key] = value
            return (success, success and simplified or case)
    if isinstance(case, list):
        simplified = list(case)
        if random.choice([True, False]):
            # delete
            if not len(simplified):
                return (False, case)
            simplified.pop(random.randrange(len(simplified)))
            return (True, simplified)
        else:
            # simplify an item
            if not len(simplified):
                return (False, case)
            index = random.randrange(len(simplified))
            (success, value) = simplify(simplified[index])
            simplified[index] = value
            return (success, success and simplified or case)
    return (False, case)


async def reduce(case, predicate, reductions=0):
    for _ in range(reduction_attempts):
        (reduced, simplified) = simplify(case)
        if reduced and not await predicate(simplified):
            return await reduce(simplified, predicate, reductions + 1)
    return (reductions, case)


async def isnt(predicate):
    async def is_not(x):
        return not await predicate(x)

    return is_not


async def check(predicate, generator):
    counter_examples = []
    for _ in range(gen_target):
        case = generator()
        try:
            if not await predicate(case):
                reduction = await reduce(case, predicate)
                counter_examples.append("after {} reductions: {!r}".format(*reduction))
        except:
            counter_examples.append(f"{case!r} : {traceback.format_exc()}")
    return counter_examples


async def check_unittest(test, predicate, generator):
    counter_examples = await check(predicate, generator)
    if counter_examples:
        failures = len(counter_examples)
        message = "\n".join(["    -> %s" % f for f in counter_examples[:examples]])
        message = "found %d counter examples, displaying first %d:\n%s" % (
            failures,
            min(failures, examples),
            message,
        )
        test.fail(message)