conftest.py

# Alternatives:
# - https://factoryboy.readthedocs.io/en/stable/
# - https://github.com/klen/mixer

@pytest.fixture(scope="function")
def baker():
    """Fixture to create MongoEngine model instances while patching dependencies and disabling signals."""
    created_instances = []

    def _bake(document_class, _quantity=1, **kwargs):
        """Creates and saves one or more instances of a MongoEngine document."""
        if not issubclass(document_class, Document):
            raise ValueError("The document must be a subclass of mongoengine.Document")

        dependencies_to_patch = ["KafkaProducer", "ChatStoreDispatcher"]
        patch_dependencies = {}
        module_name = document_class.__module__

        if module_name in sys.modules:
            module = sys.modules[module_name]

            source_lines = inspect.getsource(module).splitlines()
            for dep in dependencies_to_patch:
                if any(f" {dep}" in line or f"{dep} " in line for line in source_lines):
                    patch_dependencies[dep] = patch(f"{module_name}.{dep}", new=MagicMock())


        # Temporarily disable signals
        if hasattr(document_class, "post_save"):
            signals.post_save.disconnect(document_class.post_save, sender=document_class)

        instances = []

        with ExitStack() as stack:

            for mock in patch_dependencies.values():
                stack.enter_context(mock) 

            for _ in range(_quantity):
                instance_data = {}

                for field_name, field in document_class._fields.items():
                    if field_name in kwargs or field_name == "id":
                        continue  # Use manually provided values

                    # Skip optional fields (required=False)
                    if not field.required:
                        continue

                    # Generate mock data
                    if isinstance(field, fields.StringField):
                        instance_data[field_name] = faker.word()
                    elif isinstance(field, fields.IntField):
                        instance_data[field_name] = faker.random_int(min=0, max=100)
                    elif isinstance(field, fields.FloatField):
                        instance_data[field_name] = faker.pyfloat(min_value=0.1, max_value=1000)
                    elif isinstance(field, fields.BooleanField):
                        instance_data[field_name] = faker.boolean()
                    elif isinstance(field, fields.DateTimeField):
                        instance_data[field_name] = faker.date_time_this_decade()
                    elif isinstance(field, fields.ListField):
                        instance_data[field_name] = [faker.word() for _ in range(2)]
                    elif isinstance(field, fields.DictField):
                        instance_data[field_name] = {"key": faker.word(), "value": faker.word()}
                    elif isinstance(field, fields.ObjectIdField):
                        from bson import ObjectId
                        instance_data[field_name] = ObjectId()
                    elif isinstance(field, fields.EmbeddedDocumentField):
                        instance_data[field_name] = _bake(field.document_type)  # Embedded object
                    elif isinstance(field, fields.ReferenceField):
                        ref_instance = _bake(field.document_type)  # Valid reference
                        instance_data[field_name] = ref_instance

                instance_data.update(kwargs)  # Override with manually provided values

                instance = document_class(**instance_data)
                instance.save()
                created_instances.append(instance)
                instances.append(instance)

        # Reconnect the signal after creating the instances
        if hasattr(document_class, "post_save"):
            signals.post_save.connect(document_class.post_save, sender=document_class)

        return instances if _quantity > 1 else instances[0]

    yield _bake

    # Cleanup: Delete created instances
    for instance in created_instances:
        instance.delete()

tests_create_models.py

def test_must_return_accounts_active_inactive_metrics(client, baker):

    # Dado 30 Messages
    dt = datetime.now()
    stats_date = dt.replace(year=2024, month=11, day=1)

    user1 = baker(User, name="normalu1", email="[email protected]")
    accountUser1 = AccountUser(
        user=user1,
        type='normal',
        easy_mode=True,
    )

    user2 = baker(User, name="normalu2", email="[email protected]")
    accountUser2 = AccountUser(
        user=user2,
        type='normal',
        easy_mode=True,
    )

    conta1 = baker(
        Account,
        name="conta1",
        users=[accountUser1, accountUser2],
    )

    conta1_phone1 = baker(Phone, description="f1", account=conta1, on_off=True, is_official=True)
    conta1_phone2 = baker(Phone, description="f2", account=conta1, on_off=True, is_official=True)

    chat1 = baker(
        Chat,
        phone=conta1_phone1,
        account=conta1,
        wa_chat_id="553196248849",
        name="Ronivaldo",
        whatsapp_name="Não Disponível",
        kind="chat",
        archived=False,
        status="RESOLVIDO",
    )

    # Mensagem recebida do cliente (nao entra no stats) - NAO ENTRA NA STATS
    message1 = baker(
        Message,
        chat=chat1,
        phone=conta1_phone1,
        account_id=conta1.id,
        author=user1,
        created=stats_date,
        timestamp=stats_date,
        send_date=stats_date,
        wa_message_id="[email protected]_BCB9EF341D814EE094C621B63BF7B821",
        wa_sender_id="[email protected]",
        status="new",
        is_approved=True,
        type="chat",
        is_out=False,
        ack=1,
        text="Oi",
    )