Merge pull request #4234 from hove-io/deactivate_opg_scenario

_deactivate_opg_scenario journeys arg added

Author: xlqian

source/jormungandr/jormungandr/interfaces/v1/journey_common.py

@@ -556,6 +556,13 @@ def __init__(self, output_type_serializer):
             default=None,
             help="Show natural opg journeys",
         )
+        parser_get.add_argument(
+            "_deactivate_opg_scenario",
+            type=BooleanType(),
+            hidden=True,
+            default=False,
+            help="Deactivate opg scenario",
+        )
         # Advanced parameters for valhalla bike
         parser_get.add_argument(
             "bike_use_roads",

source/jormungandr/jormungandr/olympic_site_params_manager.py

@@ -225,6 +225,10 @@ def build(self, pt_object_origin, pt_object_destination, api_request):
         # Warning, the order of functions is important
         # Order 1 : get_olympic_site_params
         # Order 2 : build_api_request
+        if api_request.get("_deactivate_opg_scenario", False):
+            logging.getLogger(__name__).warning("OPG scenario decativated.")
+            return
+
         api_request["olympic_site_params"] = self.get_olympic_site_params(
             pt_object_origin, pt_object_destination, api_request
         )

source/jormungandr/jormungandr/tests/olympic_site_params_manager_tests.py

@@ -705,6 +705,34 @@ def test_build_departure_and_arrival_poi_jo_add_forbidden_uris():
     assert not api_request["olympic_site_params"]['show_natural_opg_journeys']
 
 
+def test_build_departure_and_arrival_poi_jo_add_forbidden_uris_deactivate_opg_scenario():
+    instance = FakeInstance(olympics_forbidden_uris=DEFAULT_OLYMPICS_FORBIDDEN_URIS)
+    osp = FakeOlympicSiteParamsManager(instance, DEFAULT_OLYMPIC_SITE_PARAMS_BUCKET)
+    osp.fill_olympic_site_params_from_s3()
+
+    pt_origin_detail = make_pt_object(type_pb2.POI, 1, 2, "poi:EFG")
+    property = pt_origin_detail.poi.properties.add()
+    property.type = DEFAULT_OLYMPICS_FORBIDDEN_URIS["poi_property_key"]
+    property.value = DEFAULT_OLYMPICS_FORBIDDEN_URIS["poi_property_value"]
+
+    pt_destination_detail = make_pt_object(type_pb2.POI, 2, 3, "poi:BCD")
+    property = pt_destination_detail.poi.properties.add()
+    property.type = DEFAULT_OLYMPICS_FORBIDDEN_URIS["poi_property_key"]
+    property.value = DEFAULT_OLYMPICS_FORBIDDEN_URIS["poi_property_value"]
+
+    api_request = {"forbidden_uris[]": ["uri1", "uri2"]}
+
+    assert not api_request.get("criteria")
+    assert not api_request.get("max_walking_duration_to_pt")
+    assert "_keep_olympics_journeys" not in api_request
+    api_request["datetime"] = osp.get_timestamp('20230715T110000')
+    api_request["_deactivate_opg_scenario"] = True
+    osp.build(pt_origin_detail, pt_destination_detail, api_request)
+    assert "olympic_site_params" not in api_request
+    assert len(api_request["forbidden_uris[]"]) == 2
+    assert "_keep_olympics_journeys" not in api_request
+
+
 def test_get_dict_scenario_invalid_scanario_name():
     instance = FakeInstance(olympics_forbidden_uris=DEFAULT_OLYMPICS_FORBIDDEN_URIS)
     osp = FakeOlympicSiteParamsManager(instance, DEFAULT_OLYMPIC_SITE_PARAMS_BUCKET)