feat: Add endpoint for serial logs

Author: steverydz

requirements.txt

@@ -6,7 +6,7 @@ canonicalwebteam.discourse==7.1.1
 canonicalwebteam.blog==6.8.4
 canonicalwebteam.search==2.1.2
 canonicalwebteam.image-template==1.9.0
-canonicalwebteam.store-api==7.3.10
+canonicalwebteam.store-api==7.4.10
 canonicalwebteam.launchpad==0.9.0
 django-openid-auth==0.17
 Flask-OpenID==1.3.1

tests/endpoints/tests_models.py

@@ -342,3 +342,109 @@ def test_create_remodel_allowlist_general_error(
         self.assertEqual(response.status_code, 500)
         self.assertFalse(data["success"])
         self.assertEqual(data["message"], "An error occurred")
+
+
+class TestGetSerialLog(TestModelServiceEndpoints):
+    @patch(
+        "canonicalwebteam.store_api.publishergw.PublisherGW"
+        + ".get_store_model_serial_log"
+    )
+    @patch("canonicalwebteam.store_api.dashboard.Dashboard.get_store")
+    def test_get_serial_log_success(self, mock_get_store, mock_get_serial_log):
+        mock_serial_log = [
+            {
+                "brand-id": "test-brand-id",
+                "created-at": "2026-03-23T04:00:23.875000",
+                "model-name": "test-model",
+                "serial": "test-serial",
+            }
+        ]
+        mock_get_serial_log.return_value = {"items": mock_serial_log}
+        mock_get_store.return_value = {"brand-id": "test-brand-id"}
+
+        response = self.client.get("/api/store/1/models/test-model/serial-log")
+        data = response.json
+
+        self.assertEqual(response.status_code, 200)
+        self.assertTrue(data["success"])
+        self.assertEqual(data["data"], mock_serial_log)
+
+    @patch(
+        "canonicalwebteam.store_api.publishergw.PublisherGW"
+        + ".get_store_model_serial_log"
+    )
+    @patch("canonicalwebteam.store_api.dashboard.Dashboard.get_store")
+    def test_get_serial_log_empty(self, mock_get_store, mock_get_serial_log):
+        mock_get_serial_log.return_value = {"items": []}
+        mock_get_store.return_value = {"brand-id": "test-brand-id"}
+
+        response = self.client.get("/api/store/1/models/test-model/serial-log")
+        data = response.json
+
+        self.assertEqual(response.status_code, 200)
+        self.assertTrue(data["success"])
+        self.assertEqual(data["data"], [])
+
+    @patch(
+        "canonicalwebteam.store_api.publishergw.PublisherGW"
+        + ".get_store_model_serial_log"
+    )
+    @patch("canonicalwebteam.store_api.dashboard.Dashboard.get_store")
+    def test_get_serial_log_unauthorized(
+        self, mock_get_store, mock_get_serial_log
+    ):
+        mock_get_serial_log.side_effect = StoreApiResponseErrorList(
+            "unauthorized", 401, [{"message": "unauthorized"}]
+        )
+        mock_get_store.return_value = {"brand-id": "test-brand-id"}
+
+        response = self.client.get("/api/store/1/models/test-model/serial-log")
+        data = response.json
+
+        self.assertEqual(response.status_code, 500)
+        self.assertFalse(data["success"])
+        self.assertEqual(data["message"], "Store not found")
+
+    @patch(
+        "canonicalwebteam.store_api.publishergw.PublisherGW"
+        + ".get_store_model_serial_log"
+    )
+    @patch("canonicalwebteam.store_api.dashboard.Dashboard.get_store")
+    def test_get_serial_log_store_not_found(
+        self, mock_get_store, mock_get_serial_log
+    ):
+        mock_get_serial_log.side_effect = StoreApiResponseErrorList(
+            "Store not found", 404, [{"message": "Store not found"}]
+        )
+        mock_get_store.return_value = {"brand-id": "test-brand-id"}
+
+        response = self.client.get(
+            "/api/store/999/models/test-model/serial-log"
+        )
+        data = response.json
+
+        self.assertEqual(response.status_code, 500)
+        self.assertFalse(data["success"])
+        self.assertEqual(data["message"], "Store not found")
+
+    @patch(
+        "canonicalwebteam.store_api.publishergw.PublisherGW"
+        + ".get_store_model_serial_log"
+    )
+    @patch("canonicalwebteam.store_api.dashboard.Dashboard.get_store")
+    def test_get_serial_log_general_error(
+        self, mock_get_store, mock_get_serial_log
+    ):
+        mock_get_serial_log.side_effect = StoreApiResponseErrorList(
+            "Internal server error",
+            500,
+            [{"message": "Internal server error"}],
+        )
+        mock_get_store.return_value = {"brand-id": "test-brand-id"}
+
+        response = self.client.get("/api/store/1/models/test-model/serial-log")
+        data = response.json
+
+        self.assertEqual(response.status_code, 500)
+        self.assertFalse(data["success"])
+        self.assertEqual(data["message"], "Internal server error")

webapp/endpoints/models.py

@@ -347,3 +347,34 @@ def create_remodel_allowlist(store_id: str):
         res["message"] = "An error occurred"
 
     return make_response(res, 500)
+
+
+@models.route("/api/store/<store_id>/models/<model_name>/serial-log")
+@login_required
+@exchange_required
+def get_serial_log(store_id: str, model_name: str):
+    res = {}
+
+    try:
+        brand_id = get_brand_id(flask.session, store_id)
+        logs = publisher_gateway.get_store_model_serial_log(
+            flask.session,
+            brand_id,
+            model_name,
+        )
+        res["data"] = logs["items"]
+        res["success"] = True
+        response = make_response(res, 200)
+    except StoreApiResponseErrorList as error_list:
+        error_messages = [
+            f"{error.get('message', 'An error occurred')}"
+            for error in error_list.errors
+        ]
+        if "unauthorized" in error_messages:
+            res["message"] = "Store not found"
+        else:
+            res["message"] = " ".join(error_messages)
+        res["success"] = False
+        response = make_response(res, 500)
+
+    return response