<commit_before># -*- coding: utf-8 -*-
# Copyright (c) 2001-2014, Canal TP and/or its affiliates. All rights reserved.
#
# This file is part of Navitia,
#     the software to build cool stuff with public transport.
#
# Hope you'll enjoy and contribute to this project,
#     powered by Canal TP (www.canaltp.fr).
# Help us simplify mobility and open public transport:
#     a non ending quest to the responsive locomotion way of traveling!
#
# LICENCE: This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# Stay tuned using
# twitter @navitia
# IRC #navitia on freenode
# https://groups.google.com/d/forum/navitia
# www.navitia.io

from __future__ import absolute_import, print_function, unicode_literals, division

from six.moves.urllib.parse import quote
from .tests_mechanism import dataset
from jormungandr import i_manager
from .check_utils import *
import mock
from os import getenv
from pytest import approx


def check_best(resp):
    assert not resp.get('journeys') or sum((1 for j in resp['journeys'] if j['type'] == "best")) == 1


@dataset({"main_routing_test": {}})
class JourneyCommon(object):

    """
    Test the structure of the journeys response
    """

    def test_journeys(self):
        # NOTE: we query /v1/coverage/main_routing_test/journeys and not directly /v1/journeys
        # not to use the jormungandr database
        response = self.query_region(journey_basic_query, display=True)
        check_best(response)
        self.is_valid_journey_response(response, journey_basic_query)

        feed_publishers = get_not_null(response, "feed_publishers")
        assert len(feed_publishers) == 1
        feed_publisher = feed_publishers[0]
        assert feed_publisher["id"] == "builder"
        assert feed_publisher["name"] == 'routing api data'
        assert feed_publisher["license"] == "ODBL"
        assert feed_publisher["url"] == "www.canaltp.fr"

        self.check_context(response)

    def test_error_on_journeys_out_of_bounds(self):
        """ if we got an error with kraken, an error should be returned"""

        query_out_of_production_bound = "journeys?from={from_coord}&to={to_coord}&datetime={datetime}".format(
            from_coord="0.0000898312;0.0000898312",  # coordinate of S in the dataset
            to_coord="0.00188646;0.00071865",  # coordinate of R in the dataset
            datetime="20110614T080000",
        )  # 2011 should not be in the production period

        response, status = self.query_region(query_out_of_production_bound, check=False)

        assert status != 200, "the response should not be valid"
        check_best(response)
        assert response['error']['id'] == "date_out_of_bounds"
        assert response['error']['message'] == "date is not in data production period"

        # and no journey is to be provided
        assert 'journeys' not in response or len(response['journeys']) == 0

    def test_error_on_journeys_too_early(self):
        """ datetime is > 1970y """

        query_out_of_production_bound = "journeys?from={from_coord}&to={to_coord}&datetime={datetime}".format(
            from_coord="0.0000898312;0.0000898312",
            to_coord="0.00188646;0.00071865",
            datetime="19700101T000000",
        )

        response, status = self.query_region(query_out_of_production_bound, check=False)

        assert status != 200, "the response should not be valid"
        check_best(response)
        assert response['message'] == 'parameter \"datetime\" invalid: Unable to parse datetime, date is too early!\ndatetime description: Date and time to go/arrive (see `datetime_represents`).\nNote: the datetime must be in the coverage’s publication period.'

        # and no journey is to be provided
        assert 'journeys' not in response or len(response['journeys']) == 0

    def test_missing_params(self):
        """we should at least provide a from or a to on the /journeys api"""
        query = "journeys?datetime=20120614T080000"

        response, status = self.query_no_assert("v1/coverage/main_routing_test/" + query)

        assert status == 400
        get_not_null(response, 'message')

    def test_best_filtering(self):
        """Filter to get the best journey, we should have only one journey, the best one"""
        query = "{query}&type=best".format(query=journey_basic_query)
        response = self.query_region(query)
        check_best(response)
        self.is_valid_journey_response(response, query)
        assert len(response['journeys']) == 1

        assert response['journeys'][0]["type"] == "best"
        assert response['journeys'][0]['durations']['total'] == 99
        assert response['journeys'][0]['durations']['walking'] == 97
        assert response['journeys'][0]['durations']['bike'] == 0
        assert response['journeys'][0]['durations']['car'] == 0

    def test_other_filtering(self):<commit_msg>black<commit_after># -*- coding: utf-8 -*-
# Copyright (c) 2001-2014, Canal TP and/or its affiliates. All rights reserved.
#
# This file is part of Navitia,
#     the software to build cool stuff with public transport.
#
# Hope you'll enjoy and contribute to this project,
#     powered by Canal TP (www.canaltp.fr).
# Help us simplify mobility and open public transport:
#     a non ending quest to the responsive locomotion way of traveling!
#
# LICENCE: This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# Stay tuned using
# twitter @navitia
# IRC #navitia on freenode
# https://groups.google.com/d/forum/navitia
# www.navitia.io

from __future__ import absolute_import, print_function, unicode_literals, division

from six.moves.urllib.parse import quote
from .tests_mechanism import dataset
from jormungandr import i_manager
from .check_utils import *
import mock
from os import getenv
from pytest import approx


def check_best(resp):
    assert not resp.get('journeys') or sum((1 for j in resp['journeys'] if j['type'] == "best")) == 1


@dataset({"main_routing_test": {}})
class JourneyCommon(object):

    """
    Test the structure of the journeys response
    """

    def test_journeys(self):
        # NOTE: we query /v1/coverage/main_routing_test/journeys and not directly /v1/journeys
        # not to use the jormungandr database
        response = self.query_region(journey_basic_query, display=True)
        check_best(response)
        self.is_valid_journey_response(response, journey_basic_query)

        feed_publishers = get_not_null(response, "feed_publishers")
        assert len(feed_publishers) == 1
        feed_publisher = feed_publishers[0]
        assert feed_publisher["id"] == "builder"
        assert feed_publisher["name"] == 'routing api data'
        assert feed_publisher["license"] == "ODBL"
        assert feed_publisher["url"] == "www.canaltp.fr"

        self.check_context(response)

    def test_error_on_journeys_out_of_bounds(self):
        """ if we got an error with kraken, an error should be returned"""

        query_out_of_production_bound = "journeys?from={from_coord}&to={to_coord}&datetime={datetime}".format(
            from_coord="0.0000898312;0.0000898312",  # coordinate of S in the dataset
            to_coord="0.00188646;0.00071865",  # coordinate of R in the dataset
            datetime="20110614T080000",
        )  # 2011 should not be in the production period

        response, status = self.query_region(query_out_of_production_bound, check=False)

        assert status != 200, "the response should not be valid"
        check_best(response)
        assert response['error']['id'] == "date_out_of_bounds"
        assert response['error']['message'] == "date is not in data production period"

        # and no journey is to be provided
        assert 'journeys' not in response or len(response['journeys']) == 0

    def test_error_on_journeys_too_early(self):
        """ datetime is > 1970y """

        query_out_of_production_bound = "journeys?from={from_coord}&to={to_coord}&datetime={datetime}".format(
            from_coord="0.0000898312;0.0000898312", to_coord="0.00188646;0.00071865", datetime="19700101T000000"
        )

        response, status = self.query_region(query_out_of_production_bound, check=False)

        assert status != 200, "the response should not be valid"
        check_best(response)
        assert (
            response['message']
            == 'parameter \"datetime\" invalid: Unable to parse datetime, date is too early!\ndatetime description: Date and time to go/arrive (see `datetime_represents`).\nNote: the datetime must be in the coverage’s publication period.'
        )

        # and no journey is to be provided
        assert 'journeys' not in response or len(response['journeys']) == 0

    def test_missing_params(self):
        """we should at least provide a from or a to on the /journeys api"""
        query = "journeys?datetime=20120614T080000"

        response, status = self.query_no_assert("v1/coverage/main_routing_test/" + query)

        assert status == 400
        get_not_null(response, 'message')

    def test_best_filtering(self):
        """Filter to get the best journey, we should have only one journey, the best one"""
        query = "{query}&type=best".format(query=journey_basic_query)
        response = self.query_region(query)
        check_best(response)
        self.is_valid_journey_response(response, query)
        assert len(response['journeys']) == 1

        assert response['journeys'][0]["type"] == "best"
        assert response['journeys'][0]['durations']['total'] == 99
        assert response['journeys'][0]['durations']['walking'] == 97
        assert response['journeys'][0]['durations']['bike'] == 0
        assert response['journeys'][0]['durations']['car'] == 0

    def test_other_filtering(self):
{'fraction_non_alphanumeric': 0.10469050013714913, 'fraction_numerical': 0.031087135411904544, 'mean_word_length': 3.570831592143753, 'pattern_counts': {'":': 2, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 2, 'lorem ipsum': 0, 'www.': 8, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1555239', 'n_tokens_mistral': 3373, 'n_tokens_neox': 3102, 'n_words': 993}