test: use raise_for_status() when making http requests in integrati…

…on tests (#364)
canvas-medical · Jan 27, 2025 · df947cc · df947cc
1 parent 518c62c
commit df947cc
Show file tree

Hide file tree

Showing 2 changed files with 61 additions and 37 deletions.
diff --git a/canvas_sdk/commands/tests/test_utils.py b/canvas_sdk/commands/tests/test_utils.py
@@ -187,17 +187,18 @@ def compute(self):
 def install_plugin(plugin_name: str, token: MaskedValue) -> None:
     """Install a plugin."""
     with open(_build_package(Path(f"./custom-plugins/{plugin_name}")), "rb") as package:
-        requests.post(
+        response = requests.post(
             plugin_url(cast(str, settings.INTEGRATION_TEST_URL)),
             data={"is_enabled": True},
             files={"package": package},
             headers={"Authorization": f"Bearer {token.value}"},
         )
+        response.raise_for_status()
 
 
 def trigger_plugin_event(token: MaskedValue) -> None:
     """Trigger a plugin event."""
-    requests.post(
+    response = requests.post(
         f"{settings.INTEGRATION_TEST_URL}/api/Note/",
         headers={
             "Authorization": f"Bearer {token.value}",
@@ -212,18 +213,22 @@ def trigger_plugin_event(token: MaskedValue) -> None:
             "lastModifiedBySessionKey": "8fee3c03a525cebee1d8a6b8e63dd4dg",
         },
     )
+    response.raise_for_status()
 
 
 def get_original_note_body_commands(new_note_id: int, token: MaskedValue) -> list[str]:
     """Get the commands from the original note body."""
-    original_note = requests.get(
+    response = requests.get(
         f"{settings.INTEGRATION_TEST_URL}/api/Note/{new_note_id}",
         headers={
             "Authorization": f"Bearer {token.value}",
             "Content-Type": "application/json",
             "Accept": "application/json",
         },
-    ).json()
+    )
+    response.raise_for_status()
+
+    original_note = response.json()
 
     body = original_note["body"]
     return [
@@ -243,18 +248,21 @@ def clean_up_files_and_plugins(plugin_name: str, token: MaskedValue) -> None:
         shutil.rmtree(Path(f"./custom-plugins/{plugin_name}"))
 
     # disable
-    requests.patch(
+    response = requests.patch(
         plugin_url(cast(str, settings.INTEGRATION_TEST_URL), plugin_name),
         data={"is_enabled": False},
         headers={
             "Authorization": f"Bearer {token.value}",
         },
     )
+    response.raise_for_status()
+
     # delete
-    requests.delete(
+    response = requests.delete(
         plugin_url(cast(str, settings.INTEGRATION_TEST_URL), plugin_name),
         headers={"Authorization": f"Bearer {token.value}"},
     )
+    response.raise_for_status()
 
 
 # For reuse with the protocol code
@@ -287,21 +295,24 @@ def create_new_note(token: MaskedValue) -> dict:
         "note_type_version": 1,
         "lastModifiedBySessionKey": "8fee3c03a525cebee1d8a6b8e63dd4dg",
     }
-    return requests.post(
+    response = requests.post(
         f"{settings.INTEGRATION_TEST_URL}/api/Note/", headers=headers, json=data
-    ).json()
+    )
+    response.raise_for_status()
+    return response.json()
 
 
 def get_token() -> MaskedValue:
     """Get a valid token."""
-    return MaskedValue(
-        requests.post(
-            f"{settings.INTEGRATION_TEST_URL}/auth/token/",
-            headers={"Content-Type": "application/x-www-form-urlencoded"},
-            data={
-                "grant_type": "client_credentials",
-                "client_id": settings.INTEGRATION_TEST_CLIENT_ID,
-                "client_secret": settings.INTEGRATION_TEST_CLIENT_SECRET,
-            },
-        ).json()["access_token"]
+    response = requests.post(
+        f"{settings.INTEGRATION_TEST_URL}/auth/token/",
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+        data={
+            "grant_type": "client_credentials",
+            "client_id": settings.INTEGRATION_TEST_CLIENT_ID,
+            "client_secret": settings.INTEGRATION_TEST_CLIENT_SECRET,
+        },
     )
+    response.raise_for_status()
+
+    return MaskedValue(response.json()["access_token"])
diff --git a/canvas_sdk/effects/banner_alert/tests.py b/canvas_sdk/effects/banner_alert/tests.py
@@ -23,17 +23,18 @@
 @pytest.fixture(scope="session")
 def token() -> MaskedValue:
     """Get a valid token."""
-    return MaskedValue(
-        requests.post(
-            f"{settings.INTEGRATION_TEST_URL}/auth/token/",
-            headers={"Content-Type": "application/x-www-form-urlencoded"},
-            data={
-                "grant_type": "client_credentials",
-                "client_id": settings.INTEGRATION_TEST_CLIENT_ID,
-                "client_secret": settings.INTEGRATION_TEST_CLIENT_SECRET,
-            },
-        ).json()["access_token"]
+    response = requests.post(
+        f"{settings.INTEGRATION_TEST_URL}/auth/token/",
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+        data={
+            "grant_type": "client_credentials",
+            "client_id": settings.INTEGRATION_TEST_CLIENT_ID,
+            "client_secret": settings.INTEGRATION_TEST_CLIENT_SECRET,
+        },
     )
+    response.raise_for_status()
+
+    return MaskedValue(response.json()["access_token"])
 
 
 @pytest.fixture(scope="session")
@@ -44,7 +45,9 @@ def first_patient_id(token: MaskedValue) -> dict:
         "Content-Type": "application/json",
         "Accept": "application/json",
     }
-    patients = requests.get(f"{settings.INTEGRATION_TEST_URL}/api/Patient", headers=headers).json()
+    response = requests.get(f"{settings.INTEGRATION_TEST_URL}/api/Patient", headers=headers)
+    response.raise_for_status()
+    patients = response.json()
     return patients["entry"][0]["resource"]["key"]
 
 
@@ -90,12 +93,13 @@ def compute(self):
 
     with open(_build_package(Path(f"./custom-plugins/{plugin_name}")), "rb") as package:
         # install the plugin
-        requests.post(
+        response = requests.post(
             plugin_url(settings.INTEGRATION_TEST_URL),
             data={"is_enabled": True},
             files={"package": package},
             headers={"Authorization": f"Bearer {token.value}"},
         )
+        response.raise_for_status()
 
     yield
 
@@ -104,26 +108,31 @@ def compute(self):
         shutil.rmtree(Path(f"./custom-plugins/{plugin_name}"))
 
     # disable
-    requests.patch(
+    response = requests.patch(
         plugin_url(settings.INTEGRATION_TEST_URL, plugin_name),
         data={"is_enabled": False},
         headers={
             "Authorization": f"Bearer {token.value}",
         },
     )
+    response.raise_for_status()
+
     # delete
-    requests.delete(
+    response = requests.delete(
         plugin_url(settings.INTEGRATION_TEST_URL, plugin_name),
         headers={"Authorization": f"Bearer {token.value}"},
     )
+    response.raise_for_status()
 
     # confirm no more banner
-    patient_banners_none = requests.get(
+    response = requests.get(
         f"{settings.INTEGRATION_TEST_URL}/api/BannerAlert/?patient__key={first_patient_id}",
         headers={
             "Authorization": f"Bearer {token.value}",
         },
-    ).json()
+    )
+    response.raise_for_status()
+    patient_banners_none = response.json()
     patient_banner = next(
         (b for b in patient_banners_none["results"] if b["key"] == plugin_name), None
     )
@@ -139,7 +148,7 @@ def test_protocol_that_adds_banner_alert(
 ) -> None:
     """Test that the protocol adds a banner alert."""
     # trigger the event
-    requests.post(
+    response = requests.post(
         f"{settings.INTEGRATION_TEST_URL}/api/Note/",
         headers={
             "Authorization": f"Bearer {token.value}",
@@ -154,13 +163,17 @@ def test_protocol_that_adds_banner_alert(
             "lastModifiedBySessionKey": "8fee3c03a525cebee1d8a6b8e63dd4dg",
         },
     )
+    response.raise_for_status()
 
-    patient_banners = requests.get(
+    response = requests.get(
         f"{settings.INTEGRATION_TEST_URL}/api/BannerAlert/?patient__key={first_patient_id}",
         headers={
             "Authorization": f"Bearer {token.value}",
         },
-    ).json()
+    )
+    response.raise_for_status()
+
+    patient_banners = response.json()
     assert patient_banners["count"] > 0
 
     patient_banner = next(b for b in patient_banners["results"] if b["key"] == plugin_name)