test_factory_boy.py

from typing import Dict

import factory
import flask
import flask_sqlalchemy
import pydantic
import pytest

# Conforme: https://flask-sqlalchemy.palletsprojects.com/en/2.x/api/?highlight=init_app#flask_sqlalchemy.SQLAlchemy.init_app
db = flask_sqlalchemy.SQLAlchemy()


@pytest.fixture(name="app", autouse=False, scope="function")
def create_app_context():
    app = flask.Flask(__name__)
    app.config["SQLALCHEMY_DATABASE_URI"] = "mysql+pymysql://root:1234@127.0.0.1:3307/biops_test"
    # app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///"
    with app.app_context():
        yield app


@pytest.fixture(name="db_", autouse=False, scope="function")
def db_(app):
    db.init_app(app)
    db.create_all()
    yield db
    db.session.rollback()
    db.session.remove()
    db.drop_all()


@pytest.fixture(name="example_factory", autouse=False, scope="function")
def _workflow_factory(db_):
    class ExampleFactory(factory.alchemy.SQLAlchemyModelFactory):
        class Meta:
            model = ExampleDB
            sqlalchemy_session = db_.session

        bioinfo_operation = factory.Sequence(lambda n: f"uuid-{n}")
        outputs: Dict[str, str] = {}

    yield ExampleFactory


class ExampleDB(db.Model):
    __tablename__ = "examples"
    bioinfo_operation = db.Column(db.String(128), primary_key=True)
    outputs = db.Column(db.JSON, nullable=False)


class BaseWorkflow(pydantic.BaseModel):
    class Config:
        orm_mode = True

    bioinfo_operation: str
    outputs: dict


def test_should_fill_metrics(example_factory):
    # Arrange
    rec = example_factory()
    wf = BaseWorkflow.from_orm(rec)

    # Act
    wf.outputs["metrics"] = {"a": 1}
    rec = db.session.query(ExampleDB).filter(ExampleDB.bioinfo_operation == wf.bioinfo_operation).first()
    rec.outputs = wf.outputs

    # Assert
    assert rec.outputs["metrics"] == {'a': 1}


def test_should_not_contain_metrics(example_factory):
    # Arrange
    rec = example_factory()
    assert "metrics" not in rec.outputs