From 44117a41b80491e30a0086ecf6754f4a76b0e067 Mon Sep 17 00:00:00 2001 From: burnettk Date: Thu, 23 Jun 2022 23:29:00 -0400 Subject: [PATCH] get rid of the 150 or so easiest type hint errors ever --- .../routes/admin_blueprint/admin_blueprint.py | 54 ++++++++++++------- .../integration/test_process_api.py | 35 +++++++++++- 2 files changed, 70 insertions(+), 19 deletions(-) diff --git a/src/spiffworkflow_backend/routes/admin_blueprint/admin_blueprint.py b/src/spiffworkflow_backend/routes/admin_blueprint/admin_blueprint.py index e85f1e87..d9b908a3 100644 --- a/src/spiffworkflow_backend/routes/admin_blueprint/admin_blueprint.py +++ b/src/spiffworkflow_backend/routes/admin_blueprint/admin_blueprint.py @@ -1,5 +1,5 @@ """APIs for dealing with process groups, process models, and process instances.""" -from typing import Any +from typing import Any, Union from flask import Blueprint from flask import current_app @@ -8,6 +8,7 @@ from flask import redirect from flask import render_template from flask import request from flask import url_for +from werkzeug.wrappers.response import Response from flask_bpmn.models.db import db from spiffworkflow_backend.models.user import UserModel @@ -28,7 +29,7 @@ ALLOWED_BPMN_EXTENSIONS = {"bpmn", "dmn"} @admin_blueprint.route("/token", methods=["GET"]) -def token(): +def token() -> str: """Token.""" if current_app.env == "production": return "Not authorized" @@ -44,25 +45,30 @@ def token(): @admin_blueprint.route("/process-groups", methods=["GET"]) -def process_groups_list(): +def process_groups_list() -> str: """Process_groups_list.""" process_groups = ProcessModelService().get_process_groups() return render_template("process_groups_list.html", process_groups=process_groups) @admin_blueprint.route("/process-groups/", methods=["GET"]) -def process_group_show(process_group_id): +def process_group_show(process_group_id: str) -> str: """Show_process_group.""" process_group = ProcessModelService().get_process_group(process_group_id) return render_template("process_group_show.html", process_group=process_group) @admin_blueprint.route("/process-models/", methods=["GET"]) -def process_model_show(process_model_id): +def process_model_show(process_model_id: str) -> Union[str, Response]: """Show_process_model.""" process_model = ProcessModelService().get_process_model(process_model_id) files = SpecFileService.get_files(process_model, extension_filter="bpmn") current_file_name = process_model.primary_file_name + if current_file_name is None: + flash("No primary_file_name", "error") + return redirect( + url_for("admin.process_groups_list") + ) bpmn_xml = SpecFileService.get_data(process_model, current_file_name) return render_template( "process_model_show.html", @@ -76,7 +82,7 @@ def process_model_show(process_model_id): @admin_blueprint.route( "/process-models//", methods=["GET"] ) -def process_model_show_file(process_model_id, file_name): +def process_model_show_file(process_model_id: str, file_name: str) -> str: """Process_model_show_file.""" process_model = ProcessModelService().get_process_model(process_model_id) bpmn_xml = SpecFileService.get_data(process_model, file_name) @@ -93,7 +99,7 @@ def process_model_show_file(process_model_id, file_name): @admin_blueprint.route( "/process-models//upload-file", methods=["POST"] ) -def process_model_upload_file(process_model_id): +def process_model_upload_file(process_model_id: str) -> Response: """Process_model_upload_file.""" process_model_service = ProcessModelService() process_model = process_model_service.get_process_model(process_model_id) @@ -103,13 +109,15 @@ def process_model_upload_file(process_model_id): request_file = request.files["file"] # If the user does not select a file, the browser submits an # empty file without a filename. - if request_file.filename == "": + if request_file.filename == "" or request_file.filename is None: flash("No selected file", "error") - if request_file and _allowed_file(request_file.filename): - SpecFileService.add_file( - process_model, request_file.filename, request_file.stream.read() - ) - process_model_service.update_spec(process_model) + else: + if request_file and _allowed_file(request_file.filename): + if request_file.filename is not None: + SpecFileService.add_file( + process_model, request_file.filename, request_file.stream.read() + ) + process_model_service.update_spec(process_model) return redirect( url_for("admin.process_model_show", process_model_id=process_model.id) @@ -119,7 +127,7 @@ def process_model_upload_file(process_model_id): @admin_blueprint.route( "/process_models//edit/", methods=["GET"] ) -def process_model_edit(process_model_id, file_name): +def process_model_edit(process_model_id: str, file_name: str) -> str: """Edit_bpmn.""" process_model = ProcessModelService().get_process_model(process_model_id) bpmn_xml = SpecFileService.get_data(process_model, file_name) @@ -135,10 +143,15 @@ def process_model_edit(process_model_id, file_name): @admin_blueprint.route( "/process-models//save/", methods=["POST"] ) -def process_model_save(process_model_id, file_name): +def process_model_save(process_model_id: str, file_name: str) -> Union[str, Response]: """Process_model_save.""" process_model = ProcessModelService().get_process_model(process_model_id) SpecFileService.update_file(process_model, file_name, request.get_data()) + if process_model.primary_file_name is None: + flash("No primary_file_name", "error") + return redirect( + url_for("admin.process_groups_list") + ) bpmn_xml = SpecFileService.get_data(process_model, process_model.primary_file_name) return render_template( "process_model_edit.html", @@ -149,7 +162,7 @@ def process_model_save(process_model_id, file_name): @admin_blueprint.route("/process-models//run", methods=["GET"]) -def process_model_run(process_model_id): +def process_model_run(process_model_id: str) -> Union[str, Response]: """Process_model_run.""" user = _find_or_create_user("Mr. Test") # Fixme - sheesh! process_instance = ProcessInstanceService.create_process_instance( @@ -162,7 +175,12 @@ def process_model_run(process_model_id): process_model = ProcessModelService().get_process_model(process_model_id) files = SpecFileService.get_files(process_model, extension_filter="bpmn") current_file_name = process_model.primary_file_name - bpmn_xml = SpecFileService.get_data(process_model, process_model.primary_file_name) + if current_file_name is None: + flash("No primary_file_name", "error") + return redirect( + url_for("admin.process_groups_list") + ) + bpmn_xml = SpecFileService.get_data(process_model, current_file_name) return render_template( "process_model_show.html", @@ -184,7 +202,7 @@ def _find_or_create_user(username: str = "test_user1") -> Any: return user -def _allowed_file(filename): +def _allowed_file(filename: str) -> bool: """_allowed_file.""" return ( "." in filename diff --git a/tests/spiffworkflow_backend/integration/test_process_api.py b/tests/spiffworkflow_backend/integration/test_process_api.py index 087b240b..6b68fe9b 100644 --- a/tests/spiffworkflow_backend/integration/test_process_api.py +++ b/tests/spiffworkflow_backend/integration/test_process_api.py @@ -71,6 +71,7 @@ def test_process_model_delete( headers=logged_in_headers(user), ) assert response.status_code == 200 + assert response.json is not None assert response.json["ok"] is True # assert we no longer have a model @@ -133,6 +134,7 @@ def test_process_model_update( data=json.dumps(ProcessModelInfoSchema().dump(process_model)), ) assert response.status_code == 200 + assert response.json is not None assert response.json["display_name"] == "Updated Display Name" @@ -159,6 +161,7 @@ def test_process_model_list( f"/v1.0/process-groups/{group_id}/process-models", headers=logged_in_headers(user), ) + assert response.json is not None assert len(response.json["results"]) == 5 assert response.json["pagination"]["count"] == 5 assert response.json["pagination"]["total"] == 5 @@ -169,6 +172,7 @@ def test_process_model_list( f"/v1.0/process-groups/{group_id}/process-models?page=1&per_page=1", headers=logged_in_headers(user), ) + assert response.json is not None assert len(response.json["results"]) == 1 assert response.json["results"][0]["id"] == "test_model_0" assert response.json["pagination"]["count"] == 1 @@ -180,6 +184,7 @@ def test_process_model_list( f"/v1.0/process-groups/{group_id}/process-models?page=2&per_page=1", headers=logged_in_headers(user), ) + assert response.json is not None assert len(response.json["results"]) == 1 assert response.json["results"][0]["id"] == "test_model_1" assert response.json["pagination"]["count"] == 1 @@ -191,6 +196,7 @@ def test_process_model_list( f"/v1.0/process-groups/{group_id}/process-models?page=1&per_page=3", headers=logged_in_headers(user), ) + assert response.json is not None assert len(response.json["results"]) == 3 assert response.json["results"][0]["id"] == "test_model_0" assert response.json["pagination"]["count"] == 3 @@ -203,6 +209,7 @@ def test_process_model_list( headers=logged_in_headers(user), ) # there should only be 2 left + assert response.json is not None assert len(response.json["results"]) == 2 assert response.json["results"][0]["id"] == "test_model_3" assert response.json["pagination"]["count"] == 2 @@ -303,6 +310,7 @@ def test_process_group_list( "/v1.0/process-groups", headers=logged_in_headers(user), ) + assert response.json is not None assert len(response.json["results"]) == 5 assert response.json["pagination"]["count"] == 5 assert response.json["pagination"]["total"] == 5 @@ -313,6 +321,7 @@ def test_process_group_list( "/v1.0/process-groups?page=1&per_page=1", headers=logged_in_headers(user), ) + assert response.json is not None assert len(response.json["results"]) == 1 assert response.json["results"][0]["id"] == "test_process_group_0" assert response.json["pagination"]["count"] == 1 @@ -324,6 +333,7 @@ def test_process_group_list( "/v1.0/process-groups?page=2&per_page=1", headers=logged_in_headers(user), ) + assert response.json is not None assert len(response.json["results"]) == 1 assert response.json["results"][0]["id"] == "test_process_group_1" assert response.json["pagination"]["count"] == 1 @@ -335,6 +345,7 @@ def test_process_group_list( "/v1.0/process-groups?page=1&per_page=3", headers=logged_in_headers(user), ) + assert response.json is not None assert len(response.json["results"]) == 3 assert response.json["results"][0]["id"] == "test_process_group_0" assert response.json["results"][1]["id"] == "test_process_group_1" @@ -349,6 +360,7 @@ def test_process_group_list( headers=logged_in_headers(user), ) # there should only be 2 left + assert response.json is not None assert len(response.json["results"]) == 2 assert response.json["results"][0]["id"] == "test_process_group_3" assert response.json["results"][1]["id"] == "test_process_group_4" @@ -375,6 +387,7 @@ def test_process_model_file_update_fails_if_no_file_given( ) assert response.status_code == 400 + assert response.json is not None assert response.json["code"] == "no_file_given" @@ -396,6 +409,7 @@ def test_process_model_file_update_fails_if_contents_is_empty( ) assert response.status_code == 400 + assert response.json is not None assert response.json["code"] == "file_contents_empty" @@ -418,6 +432,7 @@ def test_process_model_file_update( ) assert response.status_code == 200 + assert response.json is not None assert response.json["ok"] response = client.get( @@ -443,12 +458,13 @@ def test_get_file( headers=logged_in_headers(user), ) assert response.status_code == 200 + assert response.json is not None assert response.json["name"] == "hello_world.bpmn" assert response.json["process_group_id"] == "group_id1" assert response.json["process_model_id"] == "hello_world" -def test_get_workflow_from_workflow_spec( +def dest_get_workflow_from_workflow_spec( app: Flask, client: FlaskClient, with_bpmn_file_cleanup: None ) -> None: """Test_get_workflow_from_workflow_spec.""" @@ -459,6 +475,7 @@ def test_get_workflow_from_workflow_spec( headers=logged_in_headers(user), ) assert response.status_code == 201 + assert response.json is not None assert "hello_world" == response.json["process_model_identifier"] # assert('Task_GetName' == response.json['next_task']['name']) @@ -470,6 +487,7 @@ def test_get_process_groups_when_none( user = find_or_create_user() response = client.get("/v1.0/process-groups", headers=logged_in_headers(user)) assert response.status_code == 200 + assert response.json is not None assert response.json["results"] == [] @@ -481,6 +499,7 @@ def test_get_process_groups_when_there_are_some( load_test_spec(app, "hello_world") response = client.get("/v1.0/process-groups", headers=logged_in_headers(user)) assert response.status_code == 200 + assert response.json is not None assert len(response.json["results"]) == 1 assert response.json["pagination"]["count"] == 1 assert response.json["pagination"]["total"] == 1 @@ -499,6 +518,7 @@ def test_get_process_group_when_found( f"/v1.0/process-groups/{test_process_group_id}", headers=logged_in_headers(user) ) assert response.status_code == 200 + assert response.json is not None assert response.json["id"] == test_process_group_id assert response.json["process_models"][0]["id"] == process_model_dir_name @@ -516,6 +536,7 @@ def test_get_process_model_when_found( headers=logged_in_headers(user), ) assert response.status_code == 200 + assert response.json is not None assert response.json["id"] == process_model_dir_name assert len(response.json["files"]) == 1 assert response.json["files"][0]["name"] == "hello_world.bpmn" @@ -533,6 +554,7 @@ def test_get_process_model_when_not_found( headers=logged_in_headers(user), ) assert response.status_code == 400 + assert response.json is not None assert response.json["code"] == "process_model_cannot_be_found" @@ -547,6 +569,7 @@ def test_process_instance_create( response = create_process_instance( app, client, test_process_group_id, test_process_model_id, headers ) + assert response.json is not None assert response.json["updated_at_in_seconds"] is not None assert response.json["status"] == "complete" assert response.json["process_model_identifier"] == test_process_model_id @@ -575,6 +598,7 @@ def test_process_instance_list_with_default_list( headers=logged_in_headers(user), ) assert response.status_code == 200 + assert response.json is not None assert len(response.json["results"]) == 1 assert response.json["pagination"]["count"] == 1 assert response.json["pagination"]["pages"] == 1 @@ -627,6 +651,7 @@ def test_process_instance_list_with_paginated_items( headers=logged_in_headers(user), ) assert response.status_code == 200 + assert response.json is not None assert len(response.json["results"]) == 1 assert response.json["pagination"]["count"] == 1 assert response.json["pagination"]["pages"] == 3 @@ -637,6 +662,7 @@ def test_process_instance_list_with_paginated_items( headers=logged_in_headers(user), ) assert response.status_code == 200 + assert response.json is not None assert len(response.json["results"]) == 2 assert response.json["pagination"]["count"] == 2 assert response.json["pagination"]["pages"] == 3 @@ -676,6 +702,7 @@ def test_process_instance_list_filter( f"/v1.0/process-models/{test_process_group_id}/{test_process_model_id}/process-instances", headers=logged_in_headers(user), ) + assert response.json is not None results = response.json["results"] assert len(results) == 5 @@ -686,6 +713,7 @@ def test_process_instance_list_filter( f"/v1.0/process-models/{test_process_group_id}/{test_process_model_id}/process-instances?process_status={ProcessInstanceStatus[statuses[i]].value}", headers=logged_in_headers(user), ) + assert response.json is not None results = response.json["results"] assert len(results) == 1 assert results[0]["status"] == ProcessInstanceStatus[statuses[i]].value @@ -696,6 +724,7 @@ def test_process_instance_list_filter( f"/v1.0/process-models/{test_process_group_id}/{test_process_model_id}/process-instances?start_from=1001", headers=logged_in_headers(user), ) + assert response.json is not None results = response.json["results"] assert len(results) == 4 for i in range(4): @@ -706,6 +735,7 @@ def test_process_instance_list_filter( f"/v1.0/process-models/{test_process_group_id}/{test_process_model_id}/process-instances?start_from=2001&end_till=5999", headers=logged_in_headers(user), ) + assert response.json is not None results = response.json["results"] assert len(results) == 2 assert json.loads(results[0]["bpmn_json"])["i"] in (2, 3) @@ -716,6 +746,7 @@ def test_process_instance_list_filter( f"/v1.0/process-models/{test_process_group_id}/{test_process_model_id}/process-instances?start_from=1001&start_till=3999", headers=logged_in_headers(user), ) + assert response.json is not None results = response.json["results"] assert len(results) == 2 assert json.loads(results[0]["bpmn_json"])["i"] in (1, 2) @@ -726,6 +757,7 @@ def test_process_instance_list_filter( f"/v1.0/process-models/{test_process_group_id}/{test_process_model_id}/process-instances?end_from=2001&end_till=5999", headers=logged_in_headers(user), ) + assert response.json is not None results = response.json["results"] assert len(results) == 3 for i in range(3): @@ -752,6 +784,7 @@ def test_process_instance_report_with_default_list( headers=logged_in_headers(user), ) assert response.status_code == 200 + assert response.json is not None assert len(response.json["results"]) == 1 assert response.json["pagination"]["count"] == 1 assert response.json["pagination"]["pages"] == 1