flowserv.tests.model module

Helper method for creating database objects.

flowserv.tests.model.create_group(session, workflow_id, users)

Create a new workflow group in the database. Expects a workflow identifier and a list of user identifier. Returns the identifier for the created group.

Parameters

  • session (sqlalchemy.orm.session.Session) – Database session.

  • workflow_id (string) – Unique workflow identifier.

  • users (list) – List of unique user identifier.

Return type

string

flowserv.tests.model.create_run(session, workflow_id, group_id)

Create a new group run. Returns the run identifier.

Parameters

  • session (sqlalchemy.orm.session.Session) – Database session.

  • workflow_id (string) – Unique workflow identifier.

  • group_id (string) – Unique group identifier.

Return type

string

flowserv.tests.model.create_user(session, active=True)

Create a new user in the database. User identifier, name and password are all the same UUID. Returns the user identifier.

Parameters

  • session (sqlalchemy.orm.session.Session) – Database session.

  • active (bool, default=True) – User activation flag.

Return type

string

flowserv.tests.model.create_workflow(session, workflow_spec={}, result_schema=None)

Create a new workflow handle for a given workflow specification. Returns the workflow identifier.

Parameters

  • session (sqlalchemy.orm.session.Session) – Database session.

  • workflow_spec (dict, default=dict()) – Optional workflow specification.

  • result_schema (dict, default=None) – Optional result schema.

Return type

string

flowserv.tests.model.success_run(database: flowserv.model.database.DB, fs: flowserv.volume.base.StorageVolume, basedir: str) → Tuple[str, str, str, str]

Create a successful run with two result files:

  • A.json

  • results/B.json

Returns the identifier of the created workflow, group, run, and user.