test_persistence.py

import pytest

from schema import create_session
from workspaces.capability.services.capability_info import CapabilityInfo
from workspaces.capability.services.interfaces import CapabilityInfoIF
from workspaces.workflow.services.interfaces import WorkflowInfoIF
from workspaces.workflow.services.workflow_info import WorkflowInfo

SESSION = create_session("SDM")


def get_workflow_info() -> WorkflowInfoIF:
    """Return a WorkflowInfo based on a real database connection"""
    return WorkflowInfo(SESSION)


def get_capability_info() -> CapabilityInfoIF:
    """Return a CapabilityInfo based on a real database connection"""
    return CapabilityInfo(SESSION)


@pytest.mark.skip(
    reason="This test queries the database, making it an integration test and unsuitable for CI."
)
def test_get_requests_for_capability():
    """Ensure we can get all requests for a certain capability"""
    info = get_capability_info()
    info.create_capability("test_get_requests", None, 0)
    info.create_capability_request("test_get_requests")
    assert len(info.requests_for_capability("test_get_requests")) > 0


@pytest.mark.skip(
    reason="This test queries the database, making it an integration test and unsuitable for CI."
)
def test_persisting_workflow_request():
    """Ensure we can persist a workflow request"""
    info = get_workflow_info()
    null = info.lookup_workflow_definition("null")
    req = info.create_workflow_request(null, {})
    assert req.workflow_request_id is not None
    assert req.results_dir is not None


@pytest.mark.skip(
    reason="This test queries the database, making it an integration test and unsuitable for CI."
)
def test_reading_workflows():
    """Ensure we can get workflows"""
    info = get_workflow_info()
    workflows = info.all_workflows()
    assert len(workflows) > 0


@pytest.mark.skip(
    reason="This test queries the database, making it an integration test and unsuitable for CI."
)
def test_reading_workflow_requests():
    """Ensure we can get workflow requests"""
    info = get_workflow_info()
    workflow_requests = info.all_workflow_requests()
    assert len(workflow_requests) > 0


@pytest.mark.skip(
    reason="This test queries the database, making it an integration test and unsuitable for CI."
)
def test_reading_capability():
    """Ensure we can get a capability"""
    info = get_capability_info()
    null = info.lookup_capability("null")
    assert null is not None


@pytest.mark.skip(
    reason="This test queries the database, making it an integration test and unsuitable for CI."
)
def test_capability_associated_to_workflow():
    """
    Tests that the relationship between capability requests and workflows is sound
    FIXME: Needs an assertion to actually test something
    """
    cinfo = get_capability_info()
    winfo = get_workflow_info()
    req = cinfo.create_capability_request("null", {})
    exec = cinfo.create_execution(req)
    workflow_req = winfo.create_workflow_request(
        winfo.lookup_workflow_definition("null"), {}
    )
    exec.current_workflow_request = workflow_req
    SESSION.add(exec)


@pytest.mark.skip(
    reason="This test queries the database, making it an integration test and unsuitable for CI."
)
def test_download_capability_persisted():
    """
    Tests that the download capability is successfully added by its alembic upgrade
    """
    cinfo = get_capability_info()
    capability = cinfo.lookup_capability("download")
    assert capability.name == "download"


@pytest.mark.skip(
    reason="This test queries the database, making it an integration test and unsuitable for CI."
)
def test_download_workflow_persisted():
    """
    Tests that the download workflow is successfully added by its alembic upgrade
    """
    winfo = get_workflow_info()
    workflow = winfo.lookup_workflow_definition("download")
    templates = winfo.lookup_workflow_templates_for("download")
    assert workflow.workflow_name == "download"
    assert len(templates) == 3