test: improved test coverage for gbfs endpoints (#1145)

Author: cka-y

api/src/scripts/populate_db_test_data.py

@@ -15,6 +15,9 @@
     t_feedsearch,
     Location,
     Officialstatushistory,
+    Gbfsversion,
+    Gbfsendpoint,
+    Gbfsfeed,
 )
 from scripts.populate_db import set_up_configs, DatabasePopulateHelper
 from shared.common.logging_utils import Logger
@@ -47,7 +50,6 @@ def populate_test_datasets(self, filepath, db_session: "Session"):
         """
         Populate the database with the test datasets
         """
-        # TODO: parse GBFS versions
         # Load the JSON file
         with open(filepath) as f:
             data = json.load(f)
@@ -125,6 +127,27 @@ def populate_test_datasets(self, filepath, db_session: "Session"):
                     db_session.query(Feature).filter(Feature.name == report_features["feature_name"]).first()
                 )
 
+        # GBFS version
+        if "gbfs_versions" in data:
+            for version in data["gbfs_versions"]:
+                gbfs_feed = db_session.query(Gbfsfeed).filter(Gbfsfeed.stable_id == version["feed_id"]).one_or_none()
+                if not gbfs_feed:
+                    self.logger.error(f"No feed found with stable_id: {version['feed_id']}")
+                    continue
+                gbfs_version = Gbfsversion(
+                    id=version["id"], version=version["version"], url=version["url"], latest=version["latest"]
+                )
+                if version.get("endpoints"):
+                    for endpoint in version["endpoints"]:
+                        gbfs_endpoint = Gbfsendpoint(
+                            id=endpoint["id"],
+                            url=endpoint["url"],
+                            language=endpoint.get("language"),
+                            name=endpoint["name"],
+                        )
+                        gbfs_version.gbfsendpoints.append(gbfs_endpoint)
+                gbfs_feed.gbfsversions.append(gbfs_version)
+
         db_session.commit()
         db_session.execute(text(f"REFRESH MATERIALIZED VIEW CONCURRENTLY {t_feedsearch.name}"))
 

api/tests/integration/test_feeds_api.py

@@ -842,3 +842,104 @@ def test_hard_limits(client, monkeypatch, values):
         params={"limit": hard_limit + 1},
     )
     assert response.status_code != 200
+
+
+@pytest.mark.parametrize(
+    "values",
+    [
+        {"response_code": 200, "expected_feed_ids": ["gbfs-system_id_1", "gbfs-system_id_2", "gbfs-system_id_3"]},
+        {
+            "provider": "Provider Name 1",
+            "response_code": 200,
+            "expected_feed_ids": ["gbfs-system_id_1", "gbfs-system_id_2"],
+        },
+        {"provider": "Provider Name 2", "response_code": 200, "expected_feed_ids": ["gbfs-system_id_3"]},
+        {
+            "country_code": "CA",
+            "response_code": 200,
+            "expected_feed_ids": ["gbfs-system_id_1", "gbfs-system_id_2", "gbfs-system_id_3"],
+        },
+        {"country_code": "US", "response_code": 200, "expected_feed_ids": []},
+        {"municipality": "Laval", "response_code": 200, "expected_feed_ids": ["gbfs-system_id_2"]},
+        {
+            "producer_url": "https://www.example.com/gbfs_feed_3/",
+            "response_code": 200,
+            "expected_feed_ids": ["gbfs-system_id_3"],
+        },
+        {"system_id": "system_id_1", "response_code": 200, "expected_feed_ids": ["gbfs-system_id_1"]},
+        {"version": "3.0", "response_code": 200, "expected_feed_ids": ["gbfs-system_id_1"]},
+        {"version": "2.3", "response_code": 200, "expected_feed_ids": ["gbfs-system_id_1", "gbfs-system_id_2"]},
+        {"version": "1.0", "response_code": 200, "expected_feed_ids": []},
+        {"system_id": "system_id_1", "response_code": 200, "expected_feed_ids": ["gbfs-system_id_1"]},
+        {"system_id": "doesnt_exist", "response_code": 200, "expected_feed_ids": []},
+    ],
+    ids=[
+        "all_none",
+        "provider_name_1",
+        "provider_name_2",
+        "country_code_ca",
+        "country_code_us",
+        "municipality_laval",
+        "producer_url",
+        "system_id",
+        "version_3.0",
+        "version_2.3",
+        "version_1.0",
+        "system_id_1",
+        "system_id_doesnt_exist",
+    ],
+)
+def test_gbfs_filters(client, values):
+    """Test /v1/gbfs_feeds filters by system_id"""
+    params = {
+        "provider": values["provider"] if "provider" in values else None,
+        "country_code": values["country_code"] if "country_code" in values else None,
+        "municipality": values["municipality"] if "municipality" in values else None,
+        "producer_url": values["producer_url"] if "producer_url" in values else None,
+        "system_id": values["system_id"] if "system_id" in values else None,
+        "version": values["version"] if "version" in values else None,
+    }
+    params = {k: v for k, v in params.items() if v is not None}
+
+    response = client.request(
+        "GET",
+        "/v1/gbfs_feeds",
+        headers=authHeaders,
+        params=params,
+    )
+    assert response.status_code == values["response_code"]
+    if values["response_code"] != 200:
+        return
+
+    feeds = response.json()
+    assert isinstance(feeds, list), "Response should be a list."
+    assert len(feeds) == len(values["expected_feed_ids"]), (
+        f"Expected {len(values['expected_feed_ids'])} feeds, " f"got {len(feeds)}."
+    )
+    if len(values["expected_feed_ids"]) != 0:
+        assert any(feed["id"] in values["expected_feed_ids"] for feed in feeds)
+
+
+@pytest.mark.parametrize(
+    "values",
+    [
+        {"response_code": 200, "expected_feed_ids": ["gbfs-system_id_1"]},
+        {"response_code": 200, "expected_feed_ids": ["gbfs-system_id_2"]},
+        {"response_code": 200, "expected_feed_ids": ["gbfs-system_id_3"]},
+        {"response_code": 404},
+    ],
+    ids=["valid_id", "valid_id_2", "valid_id_3", "invalid_id"],
+)
+def test_gbfs_feed_id_get(client: TestClient, values):
+    """Test case for gbfs_feed_id_get"""
+    test_id = values.get("expected_feed_ids", ["dummy_id"])[0]
+    response = client.request(
+        "GET",
+        "/v1/gbfs_feeds/{id}".format(id=test_id),
+        headers=authHeaders,
+    )
+
+    assert response.status_code == values["response_code"]
+    if values["response_code"] != 200:
+        return
+    assert response.json()["id"] == test_id

api/tests/test_data/extra_test_data.json

@@ -759,5 +759,74 @@
       "severity": "ERROR",
       "total_notices": 2
     }
+  ],
+  "gbfs_versions": [
+    {
+      "feed_id": "gbfs-system_id_1",
+      "id": "gbfs-system_id_1-2.3",
+      "version": "2.3",
+      "url": "https://www.example.com/gbfs_feed_1/2.3/gbfs.json",
+      "latest": false,
+      "endpoints": [
+        {
+          "id": "gbfs-system_id_1-2.3-system_information",
+          "url": "https://www.example.com/gbfs_feed_1/2.3/system_information.json",
+          "language": "en",
+          "name": "system_information",
+          "is_feature": false
+        },
+        {
+          "id": "gbfs-system_id_1-2.3-vehicle_position",
+          "url": "https://www.example.com/gbfs_feed_1/2.3/vehicule_position.json",
+          "language": "en",
+          "name": "vehicle_position",
+          "is_feature": false
+        }
+      ]
+    },
+    {
+      "feed_id": "gbfs-system_id_1",
+      "id": "gbfs-system_id_1-3.0",
+      "version": "3.0",
+      "url": "https://www.example.com/gbfs_feed_1/3.0/gbfs.json",
+      "latest": true,
+      "endpoints": [
+        {
+          "id": "gbfs-system_id_1-3.0-system_information",
+          "url": "https://www.example.com/gbfs_feed_1/3.0/system_information.json",
+          "name": "system_information",
+          "is_feature": false
+        },
+        {
+          "id": "gbfs-system_id_1-3.0-vehicle_position",
+          "url": "https://www.example.com/gbfs_feed_1/3.0/vehicule_position.json",
+          "name": "vehicle_position",
+          "is_feature": false
+        }
+      ]
+    },
+    {
+      "feed_id": "gbfs-system_id_2",
+      "id": "gbfs-system_id_2-2.3",
+      "version": "2.3",
+      "url": "https://www.example.com/gbfs_feed_1/2.3/gbfs.json",
+      "latest": false,
+      "endpoints": [
+        {
+          "id": "gbfs-system_id_2-2.3-system_information",
+          "url": "https://www.example.com/gbfs_feed_2/2.3/system_information.json",
+          "language": "en",
+          "name": "system_information",
+          "is_feature": false
+        },
+        {
+          "id": "gbfs-system_id_2-2.3-vehicle_position",
+          "url": "https://www.example.com/gbfs_feed_2/2.3/vehicule_position.json",
+          "language": "en",
+          "name": "vehicle_position",
+          "is_feature": false
+        }
+      ]
+    }
   ]
 }

integration-tests/src/endpoints/gbfs_feeds.py

@@ -0,0 +1,77 @@
+from endpoints.integration_tests import IntegrationTests
+
+
+class GBFSFeedsEndpointTests(IntegrationTests):
+    def __init__(self, file_path, access_token, url, progress):
+        super().__init__(file_path, access_token, url, progress=progress)
+
+    def test_gbfs_feeds_filter_by_provider(self):
+        """Test retrieval of GBFS feeds filtered by provider"""
+        providers = ["BIXI Montréal", "Bird Laval", "Lime Ottawa"]
+        task_id = self.progress.add_task(
+            "[yellow]Validating GBFS feeds by provider...[/yellow]",
+            total=len(providers),
+        )
+        for i, provider_id in enumerate(providers):
+            self._test_filter_by_provider(
+                provider_id,
+                "v1/gbfs_feeds",
+                task_id=task_id,
+                index=f"{i + 1}/{len(providers)}",
+            )
+
+    def test_gbfs_feeds_filter_by_version(self):
+        """Test retrieval of GBFS feeds filtered by version"""
+        versions = ["1.0", "2.3", "3.0"]
+        task_id = self.progress.add_task(
+            "[yellow]Validating GBFS feeds by version...[/yellow]",
+            total=len(versions),
+        )
+        total_returned = 0
+        for i, version in enumerate(versions):
+            response = self.get_response(
+                "v1/gbfs_feeds",
+                params={"version": version},
+            )
+            assert (
+                response.status_code == 200
+            ), f"Expected 200 status code for version '{version}', got {response.status_code}."
+            gbfs_feeds = response.json()
+            total_returned += len(gbfs_feeds)
+            self.console.log(
+                ""
+                f"Number of feeds returned for version '{version}': {len(gbfs_feeds)}"
+            )
+            self._update_progression(
+                task_id,
+                f"Retrieved feeds version {version}",
+                f"{(i + 1)} / {len(versions)}",
+            )
+        assert (
+            total_returned > 0
+        ), f"No feeds returned for the specified versions: {versions}."
+
+    def test_gbfs_feeds_filter_by_system_id(self):
+        """Test retrieval of GBFS feeds filtered by system_id"""
+        system_ids = ["bird-edmonton", "Bixi_MTL", "lime_ottawa"]
+        task_id = self.progress.add_task(
+            "[yellow]Validating GBFS feeds by system_id...[/yellow]",
+            total=len(system_ids),
+        )
+        for i, system_id in enumerate(system_ids):
+            response = self.get_response(
+                "v1/gbfs_feeds",
+                params={"system_id": system_id},
+            )
+            assert (
+                response.status_code == 200
+            ), f"Expected 200 status code for system_id '{system_id}', got {response.status_code}."
+            gbfs_feeds = response.json()
+            assert (
+                len(gbfs_feeds) > 0
+            ), f"No feeds returned for system_id '{system_id}'."
+            self._update_progression(
+                task_id,
+                f"Retrieved feeds system_id {system_id}",
+                f"{(i + 1)} / {len(system_ids)}",
+            )