jasquat 18600189c8 Feature/background proc with celery (#788)
* WIP: some initial test code to test out celery w/ burnettk

* some cleanup for celery and added base model to put tasks waiting on timers

* removed dup bpmn file

* some more cleanup and added strategy to queue instructions

* some minor code changes w/ burnettk

* remove the unused next_task key from api calls since nobody uses it w/ burnettk essweine

* added migration for future tasks and added test to make sure we are inserting into it w/ burnettk essweine

* ensure future task run at time can be updated w/ burnettk

* added table to queue instructions for end user in w/ burnettk

* added test to ensure we are storing instructions for end users w/ burnettk

* added progress page to display new instructions to user

* ignore dup instructions on db insert w/ burnettk

* some more updates for celery w/ burnettk

* some pyl and test fixes w/ burnettk

* fixed tests w/ burnettk

* WIP: added in page to show instructions on pi show page w/ burnettk

* pi show page is fully using not interstitial now w/ burnettk

* fixed broken test w/ burnettk

* moved background processing items to own module w/ burnettk

* fixed apscheduler start script

* updated celery task queue to handle future tasks and upgraded black and set its line-length to match ruff w/ burnettk

* added support to run future tasks using countdown w/ burnettk

* build image for celery branch w/ burnettk

* poet does not exist in the image w/ burnettk

* start blocking scheduler should always start the scheduler w/ burnettk

* add init and stuff for this branch

* make this work not just on my mac

* send other args to only

* added running status for process instance and use that on fe to go to show page and added additional identifier to locking system to isolate celery workers better w/ burnettk

* fixed typing error that typeguard found, not sure why mypy did not w/ burnettk

* do not check for no instructions on interstitial page for cypress tests on frontend w/ burnettk

* do not queue process instances twice w/ burnettk

* removed bad file w/ burnettk

* queue tasks using strings to avoid circular imports when attmepting to queue w/ burnettk

* only queue imminent new timer events and mock celery

* some keyboard shortcut support on frontend and added ability to force run a process instance over the api w/ burnettk

* some styles added for the shortcut menu w/ burnettk

* pyl w/ burnettk

* fixed test w/ burnettk

* removed temporary celery script and added support for celery worker in run server locally w/ burnettk

* cleaned up migrations w/ burnettk

* created new migration to clean up old migrations

---------

Co-authored-by: jasquat <jasquat@users.noreply.github.com>
Co-authored-by: burnettk <burnettk@users.noreply.github.com>
2023-12-05 11:41:59 -05:00

82 lines
3.1 KiB
Python

from flask import Flask
from flask.testing import FlaskClient
from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
from spiffworkflow_backend.models.user import UserModel
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
class TestOnboarding(BaseTest):
def test_returns_nothing_if_no_onboarding_model(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
results = client.get(
"/v1.0/onboarding",
headers=self.logged_in_headers(with_super_admin_user),
)
assert results.status_code == 200
assert results.json == {}
def set_up_onboarding(self, client: FlaskClient, with_super_admin_user: UserModel, file_location: str) -> None:
process_group_id = "site-administration"
process_model_id = "onboarding"
bpmn_file_location = file_location
self.create_group_and_model_with_bpmn(
client,
with_super_admin_user,
process_group_id=process_group_id,
process_model_id=process_model_id,
bpmn_file_location=bpmn_file_location,
)
def test_returns_onboarding_if_onboarding_model(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
self.set_up_onboarding(client, with_super_admin_user, "onboarding")
results = client.get(
"/v1.0/onboarding",
headers=self.logged_in_headers(with_super_admin_user),
)
assert results.status_code == 200
assert len(results.json.keys()) == 4
assert results.json["type"] == "default_view"
assert results.json["value"] == "my_tasks"
assert results.json["instructions"] == ""
assert results.json["task_id"] is not None
# Assure no residual process model is left behind if it executes and completes without additinal user tasks
assert len(ProcessInstanceModel.query.all()) == 0
def skip_test_persists_if_user_task_encountered(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
with_super_admin_user: UserModel,
) -> None:
"""We are moving towards replacing the onboarding with Extensions
so disabling this test, and the ability to start a person off on
a workflow instantly on arrival."""
self.set_up_onboarding(client, with_super_admin_user, "onboarding_with_user_task")
results = client.get(
"/v1.0/onboarding",
headers=self.logged_in_headers(with_super_admin_user),
)
assert results.status_code == 200
assert len(results.json.keys()) == 4
assert results.json["type"] == "user_input_required"
assert results.json["process_instance_id"] is not None
instance = ProcessInstanceModel.query.filter(ProcessInstanceModel.id == results.json["process_instance_id"]).first()
assert instance is not None