Adding fail cases for get_table and upload_file

Author: samyarpotlapalli

src/sasctl/_services/score_definitions.py

@@ -77,14 +77,14 @@ def create_score_definition(
             raise HTTPError({
                 f"This model may not exist in a project or the model may not exist at all. See error: {model.json()}"}
             )
-        model_project_id = model.json()["projectId"]
-        model_project_version_id = model.json()["projectVersionId"]
-        model_name = model.json()["name"]
+        model_project_id = model.get("projectId")
+        model_project_version_id = model.get("projectVersionId")
+        model_name = model.get("name")
         # Checking if the model exists and if it's in a project
 
         try:
             inputMapping = []
-            for input_item in model.json()["inputVariables"]:
+            for input_item in model.get("inputVariables"):
                 var = {
                     "mappingValue": input_item["name"],
                     "mappingType": "datasource",

src/sasctl/_services/score_execution.py

@@ -57,11 +57,11 @@ def create_score_execution(
         score_definition = sd.get_definition(score_definition_id)
         if score_definition.status_code >= 400:
             raise HTTPError(score_definition.json())
-        score_exec_name = score_definition.json()["name"]
-        model_uri = score_definition.json()["objectDescriptor"]["uri"]
-        model_name = score_definition.json()["objectDescriptor"]["name"]
-        model_input_library = score_definition.json()["inputData"]["libraryName"]
-        model_table_name = score_definition.json()["inputData"]["tableName"]
+        score_exec_name = score_definition.get("name")
+        model_uri = score_definition.get("objectDescriptor", ["uri"])
+        model_name = score_definition.get("objectDescriptor", ["name"])
+        model_input_library = score_definition.get("inputData", ["libraryName"])
+        model_table_name = score_definition.get("inputData", ["tableName"])
 
         # Defining a default output table name if none is provided
         if not output_table_name:
@@ -72,11 +72,11 @@ def create_score_execution(
             score_execution = cls.get(
                 f"/executions?filter=eq(scoreExecutionRequest.scoreDefinitionId,%27{score_definition_id}%27)"
             ) 
-            execution_count = score_execution.json()[
+            execution_count = score_execution.get(
                 "count"
-            ]  # Exception catch location
+              )  # Exception catch location
             if execution_count == 1:
-                execution_id = score_execution.json()["items"][0]["id"]
+                execution_id = score_execution.get("items", [0], ["id"])
                 deleted_score_execution = cls.delete_execution(execution_id)
                 print(deleted_score_execution)
         except KeyError:

tests/unit/test_score_definitions.py

@@ -56,6 +56,24 @@ def test_create_score_definition():
                         model_id = "12345",
                         table_name = "test_table"
                     )
+                # get_model.return_value.status_code = 200
+                # get_model.return_value.json.return_value = {
+                #     "id": "12345",
+                #     "projectId": "54321",
+                #     "projectVersionId": "67890",
+                #     "name": "test_model"
+                #     "inputVariables": {"A", "B", "C"} #look at syntax combine square brackets and curly bracket
+                # }
+                #     sd.create_score_definition(
+                #         score_def_name = "test_create_sd",
+                #         model_id = "12345",
+                #         table_name = "test_table",
+                #         mappings = {
+                #             {"mappingVariable": "A", "datasource": "datasource", "variableName": "A"},
+                #             {"mappingVariable": "B", "datasource": "datasource", "variableName": "B"},
+                #             {"mappingVariable": "C", "datasource": "datasource", "variableName": "C"}
+                #         }
+                #     )
 
                 get_model.return_value.status_code = 200
                 get_model.return_value.json.return_value = {
@@ -71,6 +89,45 @@ def test_create_score_definition():
                             model_id = "12345",
                             table_name = "test_table"
                     )
+                get_table.return_value.status_code = 404
+                upload_file.return_value = None
+                get_table.return_value.status_code = 404
+                with pytest.raises(HTTPError):
+                    sd.create_score_definition(
+                            score_def_name = "test_create_sd",
+                            model_id = "12345",
+                            table_name = "test_table",
+                            table_file = "test_path"
+                        )
+                
+                get_table.return_value.status_code = 404
+                upload_file.return_value = RestObj
+                get_table.return_value.status_code = 200
+                response = sd.create_score_definition(
+                        score_def_name = "test_create_sd",
+                        model_id = "12345",
+                        table_name = "test_table",
+                        table_file = "test_path"
+                )
+                assert response
+                
+                
+                
+                
+                # get_table.return_value.status_code = 200
+                # get_table.return_value.json.return_value = {
+                #     "tableName": "test_table"
+                # }
+                # sd.create_score_definition(
+                #     score_def_name = "test_create_sd",
+                #     model_id = "12345",
+                #     table_name = "test_table"
+                # )
+
+                
+                
+                
+                
 
                 # # Test for valid model id, no input mapping, get_table succeeded, and post succeeds