decisionengine.framework.dataspace.datasources.tests package

Submodules

decisionengine.framework.dataspace.datasources.tests.fixtures module

pytest fixtures/constants

decisionengine.framework.dataspace.datasources.tests.fixtures.PG_DE_DB_WITHOUT_SCHEMA(request: FixtureRequest) → Iterator[Connection]

Fixture factory for PostgreSQL.

Parameters:

request – fixture request object

Returns:

postgresql client

decisionengine.framework.dataspace.datasources.tests.fixtures.PG_PROG(request: FixtureRequest, tmp_path_factory: TempPathFactory) → Iterator[PostgreSQLExecutor]

Process fixture for PostgreSQL.

Parameters:

• request – fixture request object

• tmp_path_factory – temporary path object (fixture)

Returns:

tcp executor

decisionengine.framework.dataspace.datasources.tests.fixtures.SQLALCHEMY_PG_WITH_SCHEMA(PG_DE_DB_WITHOUT_SCHEMA)

Get a blank database from pytest_postgresql. Then setup the SQLAlchemy style URL with that DB. The SQLAlchemyDS will create the schema as needed.

decisionengine.framework.dataspace.datasources.tests.fixtures.SQLALCHEMY_TEMPFILE_SQLITE(tmp_path)

Setup an SQLite database with the pytest tmp_path fixture. Then setup the SQLAlchemy style URL with that DB. The SQLAlchemyDS will create the schema as needed.

decisionengine.framework.dataspace.datasources.tests.fixtures.datasource(request)

This parameterized fixture will setup up various datasources.

Add datasource objects to DATASOURCES_TO_TEST once they’ve got our basic schema loaded. And adjust our if statements here until we are SQLAlchemy only.

Pytest should take it from there and automatically run it through all the tests using this fixture.

decisionengine.framework.dataspace.datasources.tests.fixtures.mock_data_block()

This fixture replaces the standard datablock implementation.

The current DataBlock implementation does not own any data products but forwards them immediately to a backend datasource. The only implemented datasource requires Postgres, which is overkill when needing to test simple data-product communication between modules.

This mock datablock class directly owns the data products, thus avoiding the need for a datasource backend. It is anticipated that a future design of the DataBlock will own the data products, thus making this mock class unnecessary.

decisionengine.framework.dataspace.datasources.tests.test_datasource_api module

This test plan covers a generic dataspace object via pytest parameters.

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_create_tables(datasource)

create_tables() should be safe to call multiple times

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_delete_data_older_than_arg(datasource)

Can we delete old entries

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_duplicate_datablock(datasource)

Can we duplicate taskmanager1 and all its entries

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_get_datablock(datasource)

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_get_dataproduct(datasource)

Can we get the dataproduct by uuid with key

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_get_dataproduct_not_exist(datasource)

Does it error out if we ask for bogus information?

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_get_dataproducts(datasource)

Can we get the dataproducts by uuid and uuid with key

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_get_dataproducts_not_exist(datasource)

Does it error out if we ask for bogus information?

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_get_header(datasource)

Can we fetch a header?

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_get_header_not_exist(datasource)

Does it error out if we ask for a bogus header?

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_get_last_generation_id(datasource)

Can we get the last generation id by name or name and uuid

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_get_last_generation_id_not_exist(datasource)

Does it error out if we ask for a bogus taskmanager?

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_get_metadata(datasource)

Can we fetch a metadata element?

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_get_metadata_not_exist(datasource)

Does it error out if we ask for a bogus metadata element?

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_get_taskmanager_exists(datasource)

Can I get a taskmanager by name or name and uuid

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_get_taskmanager_not_exists(datasource)

This should error out

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_get_taskmanagers(datasource)

Can I get multimple task managers

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_get_taskmanagers_not_exist(datasource)

Do I error out when asking for garbage

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_has_config(datasource)

This should have a config dict we can pass to jsonnet

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_insert(datasource)

Can we insert new elements

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_reset_connections(datasource)

reset_connections() should be safe to call any time

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_store_taskmanager(datasource)

Can we make new entries

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_update(datasource)

Do updates work as expected

decisionengine.framework.dataspace.datasources.tests.test_datasource_api.test_update_bad(datasource)

Do updates fail to work on bogus taskmanager as expected

Module contents