OBU OTA Server FQDN Fix

docker-compose-obu-ota-server.yml

@@ -25,6 +25,7 @@ services:
       PG_DB_PASS: ${PG_DB_PASS}
 
       MAX_COUNT: ${MAX_COUNT}
+      NGINX_ENCRYPTION: ${NGINX_ENCRYPTION}
     volumes:
       - ./resources/ota/firmwares:/firmwares
     logging:

resources/kubernetes/obu-ota-server.yaml

@@ -170,6 +170,8 @@ spec:
                 secretKeyRef:
                   name: some-postgres-secret-password
                   key: some-postgres-secret-key
+            - name: NGINX_ENCRYPTION
+              value: 'ssl'
           volumeMounts:
             - name: cv-manager-service-key
               mountPath: /home/secret

services/Dockerfile.obu_ota_server

@@ -3,9 +3,9 @@ FROM python:3.12-alpine
 WORKDIR /home
 
 ADD addons/images/obu_ota_server/requirements.txt .
+RUN pip install --no-cache-dir --upgrade -r requirements.txt
+
 ADD addons/images/obu_ota_server/*.py .
 ADD common/*.py ./common/
 
-RUN pip install --no-cache-dir --upgrade -r requirements.txt
-
 CMD ["uvicorn", "obu_ota_server:app", "--host", "0.0.0.0", "--port", "8085"]

services/addons/images/obu_ota_server/obu_ota_server.py

@@ -52,12 +52,22 @@ def get_firmware_list() -> list:
     return files
 
 
+def get_host_name() -> str:
+    host_name = os.getenv("SERVER_HOST", "localhost")
+    tls_enabled = os.getenv("NGINX_ENCRYPTION", "plain")
+    if tls_enabled.lower() == "ssl":
+        host_name = "https://" + host_name
+    else:
+        host_name = "http://" + host_name
+    return host_name
+
+
 @app.get("/firmwares/commsignia", dependencies=[Depends(authenticate_user)])
 async def get_manifest(request: Request) -> dict[str, Any]:
     try:
         files = get_firmware_list()
         logging.debug(f"get_manifest :: Files: {files}")
-        host_name = os.getenv("SERVER_HOST", "localhost")
+        host_name = get_host_name()
         response_manifest = commsignia_manifest.add_contents(host_name, files)
         return response_manifest
     except Exception as e:

services/addons/images/obu_ota_server/sample.env

@@ -21,6 +21,7 @@ OTA_PASSWORD = "admin"
 
 # Nginx encryption options: "plain", "ssl"
 # Note that this just changes the config file attached as a volume to the Nginx container
+# This is also used to generate the proper FQDN in the manifest response
 NGINX_ENCRYPTION="plain"
 
 # SSL file name in path /docker/nginx/ssl/

services/addons/tests/obu_ota_server/test_obu_ota_server.py

@@ -318,5 +318,73 @@ def test_removed_old_logs_with_removal(mock_pgquery):
     )
 
 
+@patch.dict("os.environ", {"OTA_USERNAME": "username", "OTA_PASSWORD": "password"})
+@pytest.mark.anyio
+@patch("addons.images.obu_ota_server.obu_ota_server.get_firmware_list")
+@patch("addons.images.obu_ota_server.obu_ota_server.commsignia_manifest.add_contents")
+async def test_get_manifest(mock_commsignia_manifest, mock_get_firmware_list):
+    mock_get_firmware_list.return_value = [
+        "/firmwares/test1.tar.sig",
+        "/firmwares/test2.tar.sig",
+    ]
+    mock_commsignia_manifest.return_value = {"json": "data"}
+    async with AsyncClient(app=app, base_url="http://test") as ac:
+        response = await ac.get(
+            "/firmwares/commsignia", auth=BasicAuth("username", "password")
+        )
+    assert response.status_code == 200
+    assert response.json() == {"json": "data"}
+
+
+@patch.dict(
+    "os.environ",
+    {
+        "OTA_USERNAME": "username",
+        "OTA_PASSWORD": "password",
+        "NGINX_ENCRYPTION": "plain",
+    },
+)
+@pytest.mark.anyio
+@patch("addons.images.obu_ota_server.obu_ota_server.get_firmware_list")
+@patch("addons.images.obu_ota_server.obu_ota_server.commsignia_manifest.add_contents")
+async def test_fqdn_response_plain(mock_commsignia_manifest, mock_get_firmware_list):
+    mock_get_firmware_list.return_value = []
+    expected_hostname = "http://localhost"
+    mock_commsignia_manifest.return_value = {"json": "data"}
+
+    async with AsyncClient(app=app, base_url="http://test") as ac:
+        response = await ac.get(
+            "/firmwares/commsignia", auth=BasicAuth("username", "password")
+        )
+
+    assert response.status_code == 200
+    mock_commsignia_manifest.assert_called_once_with(expected_hostname, [])
+
+
+@patch.dict(
+    "os.environ",
+    {
+        "OTA_USERNAME": "username",
+        "OTA_PASSWORD": "password",
+        "NGINX_ENCRYPTION": "SSL",
+    },
+)
+@pytest.mark.anyio
+@patch("addons.images.obu_ota_server.obu_ota_server.get_firmware_list")
+@patch("addons.images.obu_ota_server.obu_ota_server.commsignia_manifest.add_contents")
+async def test_fqdn_response_ssl(mock_commsignia_manifest, mock_get_firmware_list):
+    mock_get_firmware_list.return_value = []
+    expected_hostname = "https://localhost"
+    mock_commsignia_manifest.return_value = {"json": "data"}
+
+    async with AsyncClient(app=app, base_url="http://test") as ac:
+        response = await ac.get(
+            "/firmwares/commsignia", auth=BasicAuth("username", "password")
+        )
+
+    assert response.status_code == 200
+    mock_commsignia_manifest.assert_called_once_with(expected_hostname, [])
+
+
 if __name__ == "__main__":
     pytest.main()