Add backend test for invoke url mapping

Author: hgreebe

api/PclusterApiHandler.py

@@ -242,9 +242,9 @@ def ec2_action():
 def get_cluster_config_text(cluster_name, region=None):
     url = f"/v3/clusters/{cluster_name}"
     if region:
-        info_resp = sigv4_request("GET", API_BASE_URL_MAPPING[_get_version(request)], url, params={"region": region})
+        info_resp = sigv4_request("GET", get_base_url(request.args.get("version")), url, params={"region": region})
     else:
-        info_resp = sigv4_request("GET", API_BASE_URL_MAPPING[_get_version(request)], url)
+        info_resp = sigv4_request("GET", get_base_url(request.args.get("version")), url)
     if info_resp.status_code != 200:
         abort(info_resp.status_code)
 
@@ -493,7 +493,7 @@ def get_dcv_session():
 
 
 def get_custom_image_config():
-    image_info = sigv4_request("GET", API_BASE_URL_MAPPING[_get_version(request)], f"/v3/images/custom/{request.args.get('image_id')}").json()
+    image_info = sigv4_request("GET", get_base_url(request.args.get("version")), f"/v3/images/custom/{request.args.get('image_id')}").json()
     configuration = requests.get(image_info["imageConfiguration"]["url"])
     return configuration.text
 
@@ -744,10 +744,10 @@ def _get_params(_request):
     params.pop("path")
     return params
 
-def _get_version(v):
+def get_base_url(v):
     if v and str(v) in API_VERSION:
-        return str(v)
-    return DEFAULT_API_VERSION
+        return API_BASE_URL_MAPPING[str(v)]
+    return API_BASE_URL_MAPPING[DEFAULT_API_VERSION]
 
 
 pc = Blueprint('pc', __name__)
@@ -756,7 +756,7 @@ def _get_version(v):
 @authenticated({'admin'})
 @validated(params=PCProxyArgs)
 def pc_proxy_get():
-    response = sigv4_request(request.method, API_BASE_URL_MAPPING[_get_version(request.args.get("version"))], request.args.get("path"), _get_params(request))
+    response = sigv4_request(request.method, get_base_url(request.args.get("version")), request.args.get("path"), _get_params(request))
     return response.json(), response.status_code
 
 @pc.route('/', methods=['POST','PUT','PATCH','DELETE'], strict_slashes=False)
@@ -770,5 +770,5 @@ def pc_proxy():
     except:
         pass
 
-    response = sigv4_request(request.method, API_BASE_URL_MAPPING[_get_version(request.args.get("version"))], request.args.get("path"), _get_params(request), body=body)
+    response = sigv4_request(request.method, get_base_url(request.args.get("version")), request.args.get("path"), _get_params(request), body=body)
     return response.json(), response.status_code

api/tests/test_pcluster_api_handler.py

@@ -1,5 +1,5 @@
 from unittest import mock
-from api.PclusterApiHandler import login
+from api.PclusterApiHandler import login, get_base_url
 
 
 @mock.patch("api.PclusterApiHandler.requests.post")
@@ -27,3 +27,11 @@ def test_login_with_no_access_token_returns_401(mocker, app):
         login()
 
         mock_abort.assert_called_once_with(401)
+
+def test_get_base_url(monkeypatch):
+    monkeypatch.setattr('api.PclusterApiHandler.API_VERSION', ['3.12.0', '3.11.0'])
+    monkeypatch.setattr('api.PclusterApiHandler.API_BASE_URL', '3.12.0=https://example.com,3.11.0=https://example1.com,')
+    monkeypatch.setattr('api.PclusterApiHandler.API_BASE_URL_MAPPING', {'3.12.0': 'https://example.com', '3.11.0': 'https://example1.com'})
+
+    assert 'https://example.com' == get_base_url('3.12.0')
+

infrastructure/parallelcluster-ui.yaml

@@ -1008,7 +1008,6 @@ Resources:
               - execute-api:Invoke
             Effect: Allow
             Resource: !Sub "arn:${AWS::Partition}:execute-api:${AWS::Region}:${AWS::AccountId}:*/*/*"
-#              - { PCApiGateway: !Select [2, !Split ['/', !Select [0, !Split ['.', !GetAtt [ ParallelClusterApi3120, Outputs.ParallelClusterApiInvokeUrl ]]]]] }
 
   CognitoPolicy:
     Type: AWS::IAM::ManagedPolicy