APM-2488 Update deployment tests to use SP

Author: sophieclayton12-nhs

scripts/test_pull_request_deployments.py

@@ -57,9 +57,7 @@ def main():
     for process in jobs:
         if process.exitcode != 0:
             print("A job failed")
-            # uncomment this line once authentication to azure is working
-            # so the job fails if there is a problem with deployment actions
-            # sys.exit(1)
+            sys.exit(1)
     sys.exit(0)
 
 

scripts/trigger_pipelines.py

@@ -1,19 +1,23 @@
 import os
 import json
 import time
+from json import JSONDecodeError
+from typing import Dict, Any
+
 import requests
 
 
 class AzureDevOps:
 
     def __init__(self):
         self.base_url = "https://dev.azure.com/NHSD-APIM/API Platform/_apis/pipelines"
-        self.token = os.environ["AZURE_TOKEN"]
-        self.auth = requests.auth.HTTPBasicAuth("", self.token)
+        self.client_id = os.environ["AZ_CLIENT_ID"]
+        self.client_secret = os.environ["AZ_CLIENT_SECRET"]
+        self.client_tenant = os.environ["AZ_CLIENT_TENANT"]
+        self.access_token = self._get_access_token()
         self.notify_commit_sha = os.environ["NOTIFY_COMMIT_SHA"]
         self.utils_pr_number = os.environ["UTILS_PR_NUMBER"]
         self.notify_github_repo = "NHSDigital/api-management-utils"
-        self.api_params = {"api-version": "6.0-preview.1"}
         self.api_request_delay = 60
 
     @staticmethod
@@ -34,16 +38,17 @@ def run_pipeline(self,
         run_url = self.base_url + f"/{pipeline_id}/runs"
         request_body = self._build_request_body(pipeline_branch)
 
-        response = requests.post(run_url, auth=self.auth, params=self.api_params, json=request_body)
+        response = self.api_request(
+            run_url,
+            json=request_body,
+            method='post',
+        )
         self.print_response(response, f"Initial request to {run_url}")
 
         result = "failed"
         if response.status_code == 200:
             result = self._check_pipeline_response(response)
             print(f"Result of {service} {pipeline_type} pipeline: {result}")
-        elif response.status_code == 203 or response.status_code == 401:
-            print(f"{response.status_code}: Invalid or expired PAT (Personal Access Token),"
-                  f" please verify or renew token")
         else:
             print(f"Triggering pipeline: {service} {pipeline_type} failed, status code: {response.status_code}")
         return result
@@ -54,8 +59,8 @@ def _check_pipeline_response(self, response: requests.Response):
         while response.status_code == 200 and response.json()["state"] == "inProgress":
             time.sleep(self.api_request_delay)
             delay = delay + self.api_request_delay
-            response = requests.get(state_url, auth=self.auth, params=self.api_params)
-            self.print_response(response, f"Response from {state_url} after {delay} seconds")
+            state_response = self.api_request(state_url)
+            self.print_response(state_response, f"Response from {state_url} after {delay} seconds")
         return response.json()["result"]
 
     def _build_request_body(self, pipeline_branch: str):
@@ -87,3 +92,54 @@ def _build_request_body(self, pipeline_branch: str):
                 }
             }
         }
+
+    def _get_access_token(self):
+        url = f"https://login.microsoftonline.com/{self.client_tenant}/oauth2/v2.0/token"
+        data = {
+            "client_id": self.client_id,
+            "client_secret": self.client_secret,
+            "grant_type": "client_credentials",
+            "scope": "https://app.vssps.visualstudio.com/.default",
+        }
+        headers = {"Content-Type": "application/x-www-form-urlencoded"}
+        res = requests.post(url=url, data=data, headers=headers)
+        res.raise_for_status()
+
+        return res.json()["access_token"]
+
+    def api_request(
+        self,
+        uri,
+        params: Dict[str, Any] = None,
+        headers: Dict[str, Any] = None,
+        api_version: str = "6.0-preview.1",
+        method: str = "get",
+        max_tries: int = 5,
+        **kwargs,
+    ):
+        def get_headers():
+
+            _headers = {"Accept": "application/json", "Authorization": f"Bearer {self.access_token}"}
+            _headers.update(headers or {})
+            return _headers
+
+        _params = {"api-version": api_version}
+        _params.update(params or {})
+        action = getattr(requests, method)
+
+        result = action(uri, params=_params, headers=get_headers(), **kwargs)
+        tries = 0
+        while result.status_code not in (200, 201, 202, 204):
+            tries += 1
+
+            if tries > max_tries:
+                break
+
+            if result.status_code in (203, 401):
+                print("REFRESHING ACCESS TOKEN...")
+                self.access_token = self._get_access_token()
+
+            time.sleep(0.5 * tries)
+            result = action(uri, params=_params, headers=get_headers(), **kwargs)
+
+        return result