This is my example query and using UnifiedAlchemyMagicMock for my test

db.query(ModelA).distinct(ModelA.attr_1).group_by(ModelA.attr_2, ModelA.attr_1).filter(ModelA.attr_2 == 10).all()

I can not use distinct in my query

I would like to understand how these objects could be extended to allow recursive operations. Let me explain a little of what this might look like. For example, it is often the case where

def setUp():
session = UnifiedAlchemyMagicMock( ...)

def my_test(self):
first_item = session.query(Blah).filter(Blah.foo == 'bar').first()
all_items = first_item.my_relationship.filter(Foo.bar == 'blah blah').all()

I can successfully replicate using examples the first item; however I would like all objects that are subsequently passed back from the UnifiedAlchemyMagicMock() class instance to have have embedded the same mixin as the session that I originally setup. This would allow "first_item.my_relationship.filter(" type calls on those objects via the same single setup.

My ultimate goal is also to boostrap the UnifiedAlchemyMagicMock.data from a json file; that I have first streamed out of a real set of SQLAlchemy objects. This

SQL DB ---> json file --> UnifiedAlchemyMagicMock.data --> tests

This way I can decouple the SQL DB but re-use a complex objects graph that spans my data base directly in mocked tests. The code in this repo is very close to allowing the first step; but the subsequent calls to alternative relationships and association_proxy, hybrid_properties and methods needs to be ideally supported.

Hay,
I found a bug.
The filter_by does not affect the output of the query. It is no issue if not using the mock. sqlalchemy/sqlalchemy#6873

from sqlalchemy.orm import registry, Session
from sqlalchemy import Column, Integer
Base = registry().generate_base()
class Hardware(Base):
    """
    parent Hardwaretable
    """
    __tablename__   = 'hardware'
    hardware_id     = Column(Integer, primary_key=True)
session = UnifiedAlchemyMagicMock()
session.add(Hardware(hardware_id=0))
session.add(Hardware(hardware_id=1))
session.add(Hardware(hardware_id=2))
hardware = session.query(Hardware).filter_by(hardware_id=1).all()
print(hardware)

prints all 3 hardware objects and behaves like

hardware = session.query(Hardware).all()

Hello,

I have this query in my function to be tested:

query = session.query(
            whitelist.whitelistEmailId,
            whitelist.areaId,
            whitelist.businessNameId,
            whitelist.email,
            whitelist.identificationNumber,
            whitelist.isRegistered,
            whitelist.serviceTypeId,
            area.name.label('area'),
            business.name.label('businessName'),
            service.name.label('serviceType')
            ).\
        join(area, whitelist.areaId == area.areaId).\
        join(business, whitelist.businessNameId == business.businessNameId).\
        join(service, whitelist.serviceTypeId == service.serviceTypeId)

I am trying to mock it out like this:

return UnifiedAlchemyMagicMock(data=[
        (
            [mock.call.query(
                WhitelistedEmail.whitelistEmailId,
                WhitelistedEmail.areaId,
                WhitelistedEmail.businessNameId,
                WhitelistedEmail.email,
                WhitelistedEmail.identificationNumber,
                WhitelistedEmail.isRegistered,
                WhitelistedEmail.serviceTypeId,
                Area.name.label('area'),
                BusinessName.name.label('businessName'),
                ServiceType.name.label('serviceType')
            )], Factory.get_list()
        )
    ])

After Ive added these three guys, that make use of .label() does not work anymore.

    Area.name.label('area'),
    BusinessName.name.label('businessName'),
    ServiceType.name.label('serviceType')

Is this a known issue? or might be am I doing something wrong. any clue is more than welcome.

It appears that mocking does not work with the session.execute(text(<sql>), params).fetchall() call.

I have a multi-step mock where I am, in a large transaction, performing multiple lookups.

One of these example lookups is below, the one that doesn't work.

def get_data_source_url(type_id, issuer_id, session):
    sql_query = """
    SELECT e.config_key 
      FROM data_sources c
        INNER JOIN external_systems e on c.external_system_id = e.id 
     WHERE c.ctype_id = :type_id
       AND c.issuer_id = :issuer_id
    """
    result = session.execute(
        text(sql_query),
        {'type_id': type_id,
         'issuer_id': issuer_id}) \
        .fetchall()
    return result

And the test mock:

def mytest():
    ...
    mock_session = UnifiedAlchemyMagicMock(data=[
    ...# other mocks
        (
            [mock.call.execute(
                ANY,  # Match any text query
                {
                    'type_id': mock_type.id,
                    'issuer_id': issuer_id),
                }
            )],
           [  # Mock response
            (external_system_id,)  # Return a list with a single tuple containing the external system ID
           ]
    )
    # call test function with session, etc.

The result of the .fetchall() mock is always a mock instance rather than the specified data.

I end up having to use unittest.mock to work with my test, though that means I can't use alchemy-mock for this specific mock. I'd like to be able to use alchemy-mock.

Thank you for building this library. It's hard work to do things like this.

I have this code early on in the setup where it basically checks that a certain piece of data exists:

 group_name_query = _session.query(Group).filter(Group.name == group_name)
            if not _session.query(group_name_query.exists()).scalar():
                group = Group(name=group_name)
                _session.add(group)

And this works just fine but later on I go over it again this time retrieving the values:

for group_name in groups_to_add:
            value = _session.query(Group).filter(Group.name == group_name).first()
            i = value.id
            ids[group_name] = i
            ids_to_group[i] = group_name

No matter what I do this second time the mocked result will always be empty
Is there a way to add something so that it is returned on a second match?
I know in your example docs it does not return again and is empty.

I love this library! It's useful!

I want to contribute this project.

but I'm not good at understand sqlalchemy and this library

I have below problem.

could you help me?

"""
    in db session
    [A(id = 1, user_id="tester")]
    [B(id = 1, a_id = 1)]
"""
s = UnifiedAlchemyMagicMock()
for a in A_list:
    s.add(a)
for b in B_list:
    s.add(b)

filter = (A.id == B.a_id) & (A.user_id == 'tester')
result = session.query(B).filter(filter).all() 
(pdb) result 
[B(id=1, a_id=1)] # good!

filter = (A.id == B.a_id) & (A.user_id == 'whoareyou')
result = session.query(B).filter(filter).all() 
(pdb) result
[B(id=1, a_id=1)] # ?? filter is not working

Hello,

First of all, a lot of thanks for sharing this library! It's very useful.

If I create a session, later I try to delete all the entries of a session mocked with some conditions not accomplished, it returns a UnifiedAlchemyMagicMock instead of returning None.

Do you know if I am doing something wrong or this function is not implemented?

Example:

session = UnifiedAlchemyMagicMock(data=[
... (
... [mock.call.query(Model),
... mock.call.filter(Model.name == 'bob',
... Model.age == 29)],
... [Model(name='bob,
... age=29)]
... ),
... (
... [mock.call.query(Model),
... mock.call.filter(Model.name == 'alice',
... Model.age == 30)],
... [Model(name='alice',
... age=30)]
... ),
... ])

from sqlalchemy import Column, Integer, String, Text, DateTime, Float, Boolean, PickleType
from sqlalchemy.ext.declarative import declarative_base
from alchemy_mock.mocking import UnifiedAlchemyMagicMock
Base = declarative_base()
class Post(Base):
tablename = 'posts'
id = Column(Integer, primary_key=True)
title = Column(String(100), nullable=False)

session = UnifiedAlchemyMagicMock()
user = Post(id=2, title="new")
session.add(user)
abc = session.query(Post).get(2)
print(abc)

As per the documentation, this should work. But session.query(Post).get(2) returns None

Python version : 3.8
alchemy-mock : 0.4.3

Importing ABC directly from collections will be removed in Python 3.10 . Use collections.abc for Python 3 and collections for Python 2.

alchemy_mock/comparison.py
150:            elif isinstance(self.expr, collections.Mapping):

Hi,
I am using pytest and assert_has_calls to determine if my session makes certain queries, but I am running into the issue of having multiple tests on the same session. When this happens, the session still has the calls from the previous tests and so tests fail.

what is the best way to get around this? creating a new session every time? wiping the current session's call history?

the other aspect is that i am loading my mocked DB with data via the UnifiedAlchemyMagicMock(data=[ ...]) functionality but i don't want to have to do this for every test as it seems expensive and kind of clunky.

thanks in advance!

I have been unable to figure out the usefulness of this module.
Admittedly, I am not very experienced with sqlalchemy, but the examples in the readme - while helpful - seem too rudimentary.
It reads a bit like showing the usefulness of assert with examples like assert True == True.
It is effective to show what methods the module offers, but does not demonstrate usefulness.

I have done a bit of searching and have not yet found more realistic examples/articles using this library.

I believe this module would benefit greatly from the addition of more realistic example tests in a directory here and/or a section in the readme with links to additional reading and/or examples.

New Fork called mock-alchemy

Since this library appears to no longer be supported, I created a fork of this project with the improvements that I suggested in PR #39. This new library is called mock-alchemy and is available on GitHub and PyPI.

My motivation stems from working on a project in which the alchemy-mock library was very helpful, but I wanted some more functionality such as delete. I decided to add this functionality in a separate library (due to the alchemy-mock library no longer being supported). I also added documentation with some more detailed examples. There are two versions of this project: one for Python 2.7 users and one for Python 3.7 users and up. The releases under 0.2.0 are designed to support Python 2.7 and Python 3.6+ simultaneously.

I hope that this new active version of this project will allow others to contribute to this project and make changes to mock-alchemy to improve this package for the developer community.

For a full list of the changes I have made, check out the changelog. As of right now, not much has been greatly altered, but I hope, with active community contribution, we can improve this neat library Miroslav Shubernetskiy has created.

Credit

Full credit has been given to the original creators (Miroslav Shubernetskiy and Serkan Hoscai) for starting and building this project.

Selecting single columns does not work for me.
eg. the following is not working

db = UnifiedAlchemyMagicMock()
db.session.query(MyTable.col).all()

Hi,

I am getting the following error when trying to execute in 3.6.8 :

 def __eq__(self, other):
        _other = list(other)
        _other[-2] = UnorderedTuple(other[-2])
>       other = Call(tuple(_other), **vars(other))
E       TypeError: __new__() got an unexpected keyword argument '_mock_name'

../../.local/share/virtualenvs/jobsite-api-sGWrxjA3/lib/python3.6/site-packages/alchemy_mock/mocking.py:57: TypeError 

Using session, it is possible to execute plain SQL calls with execute method:

session.execute("select count(*) as count from my_table").fetchone[0]

I did not find out how I could mock such scenarios with UnifiedAlchemyMagicMock.
Did I oversee something?

I use a method to paginate my queries but in the library I don't have this implementation whenever I use session.query(Model).paginate(page=1, per_page=10, max_per_page=100) returns an empty list. If this method is implemented I do not know how it should be applied!

In sqlalchemy we have here how the implementation is usually done in the library https://flask-sqlalchemy.palletsprojects.com/en/2.x/api/#flask_sqlalchemy.Pagination

https://stackoverflow.com/q/69546839/6429049?sem=2

Can mock-alchemy or alchemy-mock be used to mock the following:

def fetch() -> List[MyEntity]:
  selectObject = sqlalchemy.select(MyEntity)
  with sqlalchemy.orm.Session(engine) as session:
    results = session.execute(selectObject)
    instances = results.scalars().fetchall()
    return instances

Hello,

Is there a way of mocking this kind of queries?

I am making use of .subquery() and .select_from() without success

salesParamsModel = Base.classes.SalesParams
lastSalesParams = aliased(salesParamsModel)
last_sales_params_derived = session.query(
        func.max(lastSalesParams.salesParamsId).label('salesParamsId')
    ).filter(lastSalesParams.recordId == recordId).subquery()

salesParams = aliased(salesParamsModel)
priceCalculation = Base.classes.PriceCalculation
main_query = session.query(
        salesParams.salesParamsId,
        salesParams.recordId,
        salesParams.ebcGuide,
        salesParams.kms,
        salesParams.color,
        salesParams.licencePlate,
        salesParams.licencePlateStateId,
        salesParams.debt,
        salesParams.invoiceDate,
        salesParams.isLcv,
        salesParams.uploadedByUserId,
        salesParams.uploadedAt,
        salesParams.updatedByUserId,
        salesParams.updatedAt,
        salesParams.isForSale,
        priceCalculation.value,
        priceCalculation.calculatedAt
    ).select_from(last_sales_params_derived).\
    join(salesParams, salesParams.salesParamsId == last_sales_params_derived.c.salesParamsId).\
    join(priceCalculation, priceCalculation.salesParamsId == salesParams.salesParamsId).\
    filter(priceCalculation.priceStructureId == 1)

Hello! I love this library!

I want to mock session.add()...

[
    mock.call.add(Model_instance),
    mock.call.commit()
]

help me plz..

The mocked get() function only accepts tuples, but the real version of the function can take in dictionaries, and more importantly scalar values.

I had to change the code to unnecessarily use tuples for this mock, so hopefully this is a simple fix.

It appears that mocking always returns an empty list with the session.execute(text(<sql>), params).mappings().all() call.

I have a multi-step mock where I am, in a large transaction, performing multiple lookups.

One of these example lookups is below, the one that doesn't work.

def get_pending_requests(credential_type_said, holder_id, issuer_id, session):
    sql_query = pending_requests_query(latest_cred_req_state_subquery) # returns a valid SQL query
    terminal_states = get_terminal_states() # tuple of strings
    completed_states_str = tuple(terminal_states)  # Convert list to tuple for the SQL query
    result = session.execute(
        text(sql_query),
        {
            'credential_type_said': credential_type_said,
            'holder_id': holder_id,
            'issuer_id': issuer_id,
            'completed_states': completed_states_str
        }
    ).mappings().all()

    # Process the results: separate CredentialRequest fields and status
    requests_with_status = []
    for row in result:
        row_dict = dict(row)  # Convert proxy row to dict
        status = row_dict.pop('status')  # Extract status, removing it from the row_dict
        cred_request = CredentialRequest(**row_dict)
        requests_with_status.append((cred_request, status))
    return requests_with_status

And the test mock:

def mytest():
    ...
    mock_session = UnifiedAlchemyMagicMock(data=[
    ...# other mocks
        (
        [mock.call.execute(
            ANY,  # This will match any query text
            {
                "credential_type_said": mock_credential_type.said,
                "issuer_id": issuer_id,
                "holder_id": holder_id,
                "completed_states": credentials.terminal_states_tuple()
            }
        )],
        [mocks.make_pending_request_list_mock()]  # returns a list of CredentialRequest objects
    ),
    # call test function with session, etc.

The result of the .mappings().all() mock is always an empty list rather than the specified data.

Like in the other issue, I end up having to use unittest.mock to work with my test, though that means I can't use alchemy-mock for this specific mock.

Thanks again for building this library.