tests

Author: giancarloromeo

services/web/server/tests/unit/isolated/test_utils_aiohttp.py

@@ -5,14 +5,18 @@
 
 import json
 from typing import Any
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 from uuid import UUID
 
 import pytest
 from aiohttp import web
 from faker import Faker
 from pydantic import BaseModel
-from simcore_service_webserver.utils_aiohttp import envelope_json_response, iter_origins
+from simcore_service_webserver.utils_aiohttp import (
+    envelope_json_response,
+    get_api_base_url,
+    iter_origins,
+)
 
 
 @pytest.fixture
@@ -31,6 +35,18 @@ class Point(BaseModel):
     }
 
 
+@pytest.fixture
+def make_request():
+    def _make_request(headers=None, scheme="http", host="example.com"):
+        req = Mock(spec=web.Request)
+        req.headers = headers or {}
+        req.scheme = scheme
+        req.host = host
+        return req
+
+    return _make_request
+
+
 def test_enveloped_successful_response(data: dict):
     resp = envelope_json_response(data, web.HTTPCreated)
     assert resp.text is not None
@@ -111,3 +127,52 @@ def test_iter_origins(headers, expected_output):
     results = list(iter_origins(request))
 
     assert results == expected_output
+
+
+def test_no_forwarded_headers_regular_host(make_request):
+    req = make_request()
+    with patch("common_library.network.is_ip_address", return_value=False):
+        url = get_api_base_url(req)
+    assert url == "http://api.example.com"
+
+
+def test_no_forwarded_headers_ip_host(make_request):
+    req = make_request(host="192.168.1.2")
+    with patch("common_library.network.is_ip_address", return_value=True):
+        url = get_api_base_url(req)
+    assert url == "http://192.168.1.2"
+
+
+def test_with_forwarded_headers(make_request):
+    headers = {"X-Forwarded-Proto": "https", "X-Forwarded-Host": "mydomain.com"}
+    req = make_request(headers=headers, scheme="http", host="example.com")
+    with patch("common_library.network.is_ip_address", return_value=False):
+        url = get_api_base_url(req)
+    assert url == "https://api.mydomain.com"
+
+
+def test_with_multiple_forwarded_headers(make_request):
+    headers = {
+        "X-Forwarded-Proto": "https, http",
+        "X-Forwarded-Host": "api1.com, api2.com",
+    }
+    req = make_request(headers=headers, scheme="http", host="example.com")
+    with patch("common_library.network.is_ip_address", side_effect=[False, False]):
+        url = get_api_base_url(req)
+    assert url == "https://api.api1.com"
+
+
+def test_forwarded_host_with_port(make_request):
+    headers = {"X-Forwarded-Proto": "https", "X-Forwarded-Host": "mydomain.com:8080"}
+    req = make_request(headers=headers, scheme="http", host="example.com:8080")
+    with patch("common_library.network.is_ip_address", return_value=False):
+        url = get_api_base_url(req)
+    assert url == "https://api.mydomain.com"
+
+
+def test_empty_forwarded_headers_fallback(make_request):
+    headers = {"X-Forwarded-Proto": "", "X-Forwarded-Host": ""}
+    req = make_request(headers=headers, scheme="https", host="example.com")
+    with patch("common_library.network.is_ip_address", return_value=False):
+        url = get_api_base_url(req)
+    assert url == "https://api.example.com"