service_keywords.py

# -*- coding: utf-8 -
import operator
from .local_time import get_now, TZ
from copy import deepcopy
from datetime import timedelta
from dateutil.parser import parse
from dpath.util import new as xpathnew
from haversine import haversine
from iso8601 import parse_date
from json import load
from jsonpath_rw import parse as parse_path
from munch import fromYAML, Munch, munchify
from robot.api import logger
from robot.errors import ExecutionFailed
from robot.libraries.BuiltIn import BuiltIn
# These imports are not pointless. Robot's resource and testsuite files
# can access them by simply importing library "service_keywords".
# Please ignore the warning given by Flake8 or other linter.
from .initial_data import (
    create_fake_amount,
    create_fake_doc,
    create_fake_guarantee,
    create_fake_image,
    create_fake_minimal_step,
    create_fake_sentence,
    fake,
    test_bid_data,
    test_bid_value,
    test_claim_answer_data,
    test_claim_data,
    test_complaint_data,
    test_complaint_reply_data,
    test_confirm_data,
    test_feature_data,
    test_invalid_features_data,
    test_item_data,
    test_item_data_financial,
    test_related_question,
    test_question_answer_data,
    test_question_data,
    test_supplier_data,
    test_tender_data,
    test_tender_data_competitive_dialogue,
    test_tender_data_dgf_financial,
    test_tender_data_dgf_other,
    test_tender_data_limited,
    test_tender_data_openeu,
    test_tender_data_openua,
)
from barbecue import chef
from restkit import request
# End of non-pointless imports
import os
import re

NUM_TYPES = (int, long, float)


def get_current_tzdate():
    return get_now().strftime('%Y-%m-%d %H:%M:%S.%f')


def add_minutes_to_date(date, minutes):
    return (parse(date) + timedelta(minutes=float(minutes))).isoformat()


def compare_objects(first, second, msg=None, values=True,
                    inequal=False, allow_nonetype=False):
    """Test two objects for (in)equality.

    :param first:          First object to be compared
    :param second:         Second object to be compared
    :param msg:            Custom error message to be printed
    :type msg:             str or unicode
    :param values:         If set to True, the values will be printed as well
                           as the error message
    :type values:          bool
    :param inequal:        If set to True, the values should not be equal
    :type inequal:         bool
    :param allow_nonetype: If set to False, the comparison will fail
                           if at least one of the two objects is None
    :type allow_nonetype:  bool
    """
    logger.info('first: ' + unicode(first))
    logger.info('second: ' + unicode(second))
    if not allow_nonetype:
        BuiltIn().should_not_be_equal(first,
                                      None,
                                      msg="First object is None",
                                      values=False)
        BuiltIn().should_not_be_equal(second,
                                      None,
                                      msg="Second object is None",
                                      values=False)
    if msg is None:
        if inequal:
            msg="Objects are equal"
        else:
            msg="Objects are not equal"
    if inequal:
        BuiltIn().should_not_be_equal(first, second, msg=msg, values=values)
    else:
        BuiltIn().should_be_equal(first, second, msg=msg, values=values)


def compare_date(left, right, accuracy="minute", absolute_delta=True):
    '''Compares dates with specified accuracy

    Before comparison dates are parsed into datetime.datetime format
    and localized.

    :param left:            First date
    :param right:           Second date
    :param accuracy:        Max difference between dates to consider them equal
                            Default value   - "minute"
                            Possible values - "day", "hour", "minute" or float value
                            of seconds
    :param absolute_delta:  Type of comparison. If set to True, then no matter which date order. If set to
                            False then right must be lower then left for accuracy value.
                            Default value   - True
                            Possible values - True and False or something what can be casted into them
    :returns:               Boolean value

    :error:                 ValueError when there is problem with converting accuracy
                            into float value. When it will be catched warning will be
                            given and accuracy will be set to 60.

    '''
    left = parse(left)
    right = parse(right)

    if left.tzinfo is None:
        left = TZ.localize(left)
    if right.tzinfo is None:
        right = TZ.localize(right)

    delta = (left - right).total_seconds()

    if accuracy == "day":
        accuracy = 24 * 60 * 60 - 1
    elif accuracy == "hour":
        accuracy = 60 * 60 - 1
    elif accuracy == "minute":
        accuracy = 60 - 1
    else:
        try:
            accuracy = float(accuracy)
        except ValueError:
            logger.warn("Could not convert from {} to float. "
                        "Accuracy is set to 60 seconds.".format(accuracy))
            accuracy = 60
    if absolute_delta:
        delta = abs(delta)
    if delta > accuracy:
        return False
    return True


def compare_coordinates(left_lat, left_lon, right_lat, right_lon, accuracy=0.1):
    '''Compares coordinates with specified accuracy

    :param left_lat:        First coordinate latitude
    :param left_lon:        First coordinate longitude
    :param right_lat:       Second coordinate latitude
    :param right_lon:       Second coordinate longitude
    :param accuracy:        Max difference between coordinates to consider them equal
                            Default value   - 0.1
                            Possible values - float or integer value of kilometers

    :returns:               Boolean value

    :error:                 ValueError when there is problem with converting accuracy
                            into float value. When it will be catched warning will be
                            given and accuracy will be set to 0.1.
    '''
    for key, value in {'left_lat': left_lat, 'left_lon': left_lon, 'right_lat': right_lat, 'right_lon': right_lon}.iteritems():
        if not isinstance(value, NUM_TYPES):
            raise TypeError("Invalid type for coordinate '{0}'. "
                            "Expected one of {1}, got {2}".format(
                                key, str(NUM_TYPES), str(type(value))))
    distance = haversine((left_lat, left_lon), (right_lat, right_lon))
    if distance > accuracy:
        return False
    return True


def log_object_data(data, file_name=None, format="yaml", update=False, artifact=False):
    """Log object data in pretty format (JSON or YAML)

    Two output formats are supported: "yaml" and "json".

    If a file name is specified, the output is written into that file.

    If you would like to get similar output everywhere,
    use the following snippet somewhere in your code
    before actually using Munch. For instance,
    put it into your __init__.py, or, if you use zc.buildout,
    specify it in "initialization" setting of zc.recipe.egg.

    from munch import Munch
    Munch.__str__ = lambda self: Munch.toYAML(self, allow_unicode=True,
                                              default_flow_style=False)
    Munch.__repr__ = Munch.__str__
    """
    if not isinstance(data, Munch):
        data = munchify(data)
    if file_name:
        if artifact:
            file_path = os.path.join(os.path.dirname(__file__), 'data', file_name + '.' + format)
        else:
            output_dir = BuiltIn().get_variable_value("${OUTPUT_DIR}")
            file_path = os.path.join(output_dir, file_name + '.' + format)
        if update:
            try:
                with open(file_path, "r+") as file_obj:
                    new_data = data.copy()
                    data = munch_from_object(file_obj.read(), format)
                    data.update(new_data)
                    file_obj.seek(0)
                    file_obj.truncate()
            except IOError as e:
                logger.info(e)
                logger.info("Nothing to update, creating new file.")
        data_obj = munch_to_object(data, format)
        with open(file_path, "w") as file_obj:
            file_obj.write(data_obj)
    data_obj = munch_to_object(data, format)
    logger.info(data_obj.decode('utf-8'))


def munch_from_object(data, format="yaml"):
    if format.lower() == 'json':
        return Munch.fromJSON(data)
    else:
        return Munch.fromYAML(data)


def munch_to_object(data, format="yaml"):
    if format.lower() == 'json':
        return data.toJSON(indent=2)
    else:
        return data.toYAML(allow_unicode=True, default_flow_style=False)


def load_data_from(file_name, mode=None):
    if not os.path.exists(file_name):
        file_name = os.path.join(os.path.dirname(__file__), 'data', file_name)
    with open(file_name) as file_obj:
        if file_name.endswith(".json"):
            file_data = Munch.fromDict(load(file_obj))
        elif file_name.endswith(".yaml"):
            file_data = fromYAML(file_obj)
    if mode == "brokers":
        default = file_data.pop('Default')
        brokers = {}
        for k, v in file_data.iteritems():
            brokers[k] = merge_dicts(default, v)
        return brokers
    else:
        return file_data


def compute_intrs(brokers_data, used_brokers):
    """Compute optimal values for period intervals.

    Notice: This function is maximally effective if ``brokers_data``
    does not contain ``Default`` entry.
    Using `load_data_from` with ``mode='brokers'`` is recommended.
    """
    def recur(l, r):
        l, r = deepcopy(l), deepcopy(r)
        if isinstance(l, list) and isinstance(r, list) and len(l) == len(r):
            lst = []
            for ll, rr in zip(l, r):
                lst.append(recur(ll, rr))
            return lst
        elif isinstance(l, NUM_TYPES) and isinstance(r, NUM_TYPES):
            if l == r:
                return l
            if l > r:
                return l
            if l < r:
                return r
        elif isinstance(l, dict) and isinstance(r, dict):
            for k, v in r.iteritems():
                if k not in l.keys():
                    l[k] = v
                else:
                    l[k] = recur(l[k], v)
            return l
        else:
            raise TypeError("Couldn't recur({0}, {1})".format(
                str(type(l)), str(type(r))))

    intrs = []
    for i in used_brokers:
        intrs.append(brokers_data[i]['intervals'])
    result = intrs.pop(0)
    for i in intrs:
        result = recur(result, i)
    return result


def prepare_test_tender_data(procedure_intervals, tender_parameters):
    # Get actual intervals by mode name
    mode = tender_parameters['mode']
    if mode in procedure_intervals:
        intervals = procedure_intervals[mode]
    else:
        intervals = procedure_intervals['default']
    logger.info(intervals)
    tender_parameters['intervals'] = intervals

    # Set acceleration value for certain modes
    assert isinstance(intervals['accelerator'], int), \
        "Accelerator should be an 'int', " \
        "not '{}'".format(type(intervals['accelerator']).__name__)
    assert intervals['accelerator'] >= 0, \
        "Accelerator should not be less than 0"
    if mode == 'belowThreshold':
        return munchify({'data': test_tender_data(tender_parameters)})
    elif mode == 'dgfFinancialAssets':
        return munchify({'data': test_tender_data_dgf_financial(tender_parameters)})
    elif mode == 'dgfOtherAssets':
        return munchify({'data': test_tender_data_dgf_other(tender_parameters)})
    elif mode == 'negotiation':
        return munchify({'data': test_tender_data_limited(tender_parameters)})
    elif mode == 'negotiation.quick':
        return munchify({'data': test_tender_data_limited(tender_parameters)})
    elif mode == 'openeu':
        return munchify({'data': test_tender_data_openeu(tender_parameters)})
    elif mode == 'openua':
        return munchify({'data': test_tender_data_openua(tender_parameters)})
    elif mode == 'open_competitive_dialogue':
        return munchify({'data': test_tender_data_competitive_dialogue(tender_parameters)})
    elif mode == 'reporting':
        return munchify({'data': test_tender_data_limited(tender_parameters)})
    raise ValueError("Invalid mode for prepare_test_tender_data")


def run_keyword_and_ignore_keyword_definitions(name, *args, **kwargs):
    """This keyword is pretty similar to `Run Keyword And Ignore Error`,
    which, unfortunately, does not suppress the error when you try
    to use it to run a keyword which is not defined.
    As a result, the execution of its parent keyword / test case is aborted.

    How this works:

    This is a simple wrapper for `Run Keyword And Ignore Error`.
    It handles the error mentioned above and additionally provides
    a meaningful error message.
    """
    try:
        status, _ = BuiltIn().run_keyword_and_ignore_error(name, *args, **kwargs)
    except ExecutionFailed as e:
        status, _ = "FAIL", e.message
    return status, _


def set_access_key(tender, access_token):
    tender.access = munchify({"token": access_token})
    return tender


def get_from_object(obj, attribute):
    """Gets data from a dictionary using a dotted accessor-string"""
    jsonpath_expr = parse_path(attribute)
    return_list = [i.value for i in jsonpath_expr.find(obj)]
    if return_list:
        return return_list[0]
    else:
        raise AttributeError('Attribute not found: {0}'.format(attribute))


def set_to_object(obj, attribute, value):
    # Search the list index in path to value
    list_index = re.search('\d+', attribute)
    if list_index:
        list_index = list_index.group(0)
        parent, child = attribute.split('[' + list_index + '].')[:2]
        # Split attribute to path to lits (parent) and path to value in list element (child)
        try:
            # Get list from parent
            listing = get_from_object(obj, parent)
            # Create object with list_index if he don`t exist
            if len(listing) < int(list_index) + 1:
                listing.append({})
        except AttributeError:
            # Create list if he don`t exist
            listing = [{}]
        # Update list in parent
        xpathnew(obj, parent, listing, separator='.')
        # Set value in obj
        xpathnew(obj, '.'.join([parent, list_index,  child]), value, separator='.')
    else:
        xpathnew(obj, attribute, value, separator='.')
    return munchify(obj)


def wait_to_date(date_stamp):
    date = parse(date_stamp)
    logger.info("date: {}".format(date.isoformat()))
    now = get_now()
    logger.info("now: {}".format(now.isoformat()))
    wait_seconds = (date - now).total_seconds()
    wait_seconds += 2
    if wait_seconds < 0:
        return 0
    return wait_seconds


def merge_dicts(a, b):
    """Merge dicts recursively.

    Origin: https://www.xormedia.com/recursively-merge-dictionaries-in-python/
    """
    if not isinstance(b, dict):
        return b
    result = deepcopy(a)
    for k, v in b.iteritems():
        if k in result and isinstance(result[k], dict):
                result[k] = merge_dicts(result[k], v)
        else:
            result[k] = deepcopy(v)
    return munchify(result)


def create_data_dict(path_to_value=None, value=None):
    data_dict = munchify({'data': {}})
    if isinstance(path_to_value, basestring) and value:
        list_items = re.search('\d+', path_to_value)
        if list_items:
            list_items = list_items.group(0)
            path_to_value = path_to_value.split('[' + list_items + ']')
            path_to_value.insert(1, '.' + list_items)
            set_to_object(data_dict, path_to_value[0], [])
            set_to_object(data_dict, ''.join(path_to_value[:2]), {})
            set_to_object(data_dict, ''.join(path_to_value), value)
        else:
            data_dict = set_to_object(data_dict, path_to_value, value)
    return data_dict


def munch_dict(arg=None, data=False):
    if arg is None:
        arg = {}
    if data:
        arg['data'] = {}
    return munchify(arg)


def _get_id_from_object(obj, key):
    logger.debug('obj["%s"]: "%s"' % (key, unicode(obj[key])))
    if obj[key] is None:
        return None
    else:
        return re.match(r'(^[filq]-[0-9a-fA-F]{8}): ', obj[key])


def get_id_from_object(obj):
    """Extract the identifier string from an object.

    :param obj: The dictionary-like object (mapping) with at least one
                mandatory key, either "title" or "description".
                The value mapped to that key should begin with a prefix which
                holds an automatically generated identifier of the object
    :type obj:  dict
    :returns:   object ID
    :rtype:     str
    """
    if not isinstance(obj, dict):
        raise TypeError('Object is not an instance of class "dict"')

    if not ('title' in obj.keys() or 'description' in obj.keys()):
        raise KeyError('Unable to get object ID. '
                       'Object does not contain any keys called '
                       '"title" or "description"')

    title_id = None
    description_id = None

    if 'title' in obj.keys():
        title_id = _get_id_from_object(obj, 'title')

    if 'description' in obj.keys():
        description_id = _get_id_from_object(obj, 'description')

    if not (title_id or description_id):
        raise ValueError('Object ID can not be found'
                         'in title or description')

    if title_id and description_id and title_id != description_id:
        # This is a rare case, yet it may happen
        raise ValueError('IDs in title and description are not equal')

    # Returning group 1 instead of group 0 because the only part we need
    # is that one in the parentheses
    return (title_id or description_id).group(1)


def get_id_from_doc_name(name):
    return re.match(r'd\-[0-9a-fA-F]{8}', name).group(0)


def get_object_type_by_id(object_id):
    prefixes = {'q': 'questions', 'f': 'features', 'i': 'items', 'l': 'lots'}
    return prefixes.get(object_id[0])


def get_object_index_by_id(data, object_id):
    if not data:
        return 0
    for index, element in enumerate(data):
        element_id = get_id_from_object(element)
        if element_id == object_id:
            break
        else:
            index += 1
    return index


def get_complaint_index_by_complaintID(data, complaintID):
    if not data:
        return 0
    for index, element in enumerate(data):
        if element['complaintID'] == complaintID:
            break
        else:
            index += 1
    return index


def generate_test_bid_data(tender_data):
    bid = test_bid_data()
    if 'aboveThreshold' in tender_data.get('procurementMethodType', '') or 'competitiveDialogue' in tender_data.get('procurementMethodType', ''):
        bid.data.selfEligible = True
        bid.data.selfQualified = True
    if 'lots' in tender_data:
        bid.data.lotValues = []
        for lot in tender_data['lots']:
            value = test_bid_value(lot['value']['amount'])
            value['relatedLot'] = lot.get('id', '')
            bid.data.lotValues.append(value)
    else:
        bid.data.update(test_bid_value(tender_data['value']['amount']))
    if 'features' in tender_data:
        bid.data.parameters = []
        for feature in tender_data['features']:
            parameter = {"value": fake.random_element(elements=(0.05, 0.01, 0)), "code": feature.get('code', '')}
            bid.data.parameters.append(parameter)
    if 'dgfOtherAssets' in tender_data.get('procurementMethodType', ''):
        bid.data.qualified = True
    if 'dgfFinancialAssets' in tender_data.get('procurementMethodType', ''):
        bid.data.eligible = True
        bid.data.qualified = True
        bid.data.tenderers[0]["additionalIdentifiers"] = [fake.additionalIdentifier()]

    return bid


def mult_and_round(*args, **kwargs):
    return round(reduce(operator.mul, args), kwargs.get('precision', 2))


# GUI Frontends common
def add_data_for_gui_frontends(tender_data):
    now = get_now()
    # tender_data.data.enquiryPeriod['startDate'] = (now + timedelta(minutes=2)).isoformat()
    tender_data.data.enquiryPeriod['endDate'] = (now + timedelta(minutes=6)).isoformat()
    tender_data.data.tenderPeriod['startDate'] = (now + timedelta(minutes=7)).isoformat()
    tender_data.data.tenderPeriod['endDate'] = (now + timedelta(minutes=11)).isoformat()
    return tender_data


def convert_date_to_slash_format(isodate):
    iso_dt = parse_date(isodate)
    date_string = iso_dt.strftime("%d/%m/%Y")
    return date_string


def convert_datetime_to_dot_format(isodate):
    iso_dt = parse_date(isodate)
    day_string = iso_dt.strftime("%d.%m.%Y %H:%M")
    return day_string


def local_path_to_file(file_name):
    return os.path.join(os.path.dirname(__file__), 'documents', file_name)