get rid of the 150 or so easiest type hint errors ever
This commit is contained in:
burnettk
parent
49b59136a4
commit
44117a41b8
|
|
@ -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/<process_group_id>", 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/<process_model_id>", 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/<process_model_id>/<file_name>", 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/<process_model_id>/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/<process_model_id>/edit/<file_name>", 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/<process_model_id>/save/<file_name>", 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/<process_model_id>/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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in New Issue