How would you unit test this SQLAlchemy Core query/function?

Question

I'm working on learning how to unit test properly. Given this function...

def get_user_details(req_user_id):
    users = sa.Table('users', db.metadata, autoload=True)

    s = sa.select([users.c.username,
                   users.c.favorite_color,
                   users.c.favorite_book
                   ]).select_from(users)

    s = s.where(users.c.user_id == req_user_id)

    result = db.connection.execute(s).fetchone()
    return dict(result)

...what is the proper way to unit test it?

Here's where I am now...

1. From what I've read, testing the "construct" of the query is unnecessary as that's part of the already well-tested SQLAlchemy library. So I don't need to test the raw SQL generated, right? But should I test the parameters passed and if so, how?
2. I've read about mocking the response that comes from the db.connection.execute but how is that really testing anything? Ultimately, I want to make sure the function is generating the proper SQL and getting the right database result?

Any advice/guidance is much appreciated. Thank you!

Answer 1

Following from this comment:

what you need to test is if the statements in your code produce expected results. – Shod

and your code, here's what I'm aiming to answer: How can I test a method that is dynamically generating SQLAlchemy queries ?

That is the issue I was having, as I wanted to make sure the queries generated were indeed correct - not because of SQL Alchemy, but because of the logic that puts them together.

the code we will test

def add_filters(query, target_table, filter_type: str, filter_value: str):
    if filter_type == "favourite_book":
        query = query.filter(target_table.c.favourite_book == filter_value)
    elif filter_type == "favourite_color":
        query = query.filter(target_table.c.favourite_color == filter_value)

    return query

and so I want to test that indeed the favourite_book filter is correctly added to the query.

In order to do this, we will create a temporary sqlite3 database, with a table containing data, and run the queries against it. Finally we test the result of the query. NOTE: you'll want both good and bad data in the data for comprehensive testing.

setting up the testing database

import pytest
from sqlalchemy.orm import Session
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
import pandas as pd

@pytest.fixture
def engine():
    my_engine = create_engine("sqlite:///file:test_db?mode=memory&cache=shared&uri=true", echo=True)
    return my_engine


@pytest.fixture
def target_table(engine):
    meta = MetaData()
    table_name = "users"

    users = Table(
        table_name,
        meta,
        Column("id", Integer, primary_key=True),
        Column("username", String),
        Column("favourite_color", String),
        Column("favourite_book", String),
    )

    meta.create_all(engine)

    # you can choose to skip the whole table declaration as df.to_sql will create
    # the table for you if it doesn't exist

    records = [
        {"username": "john", "favourite_color": "blue", "favourite_book": "Harry Potter"},
        {"username": "jane", "favourite_color": "red", "favourite_book": "The Power of Now"},
        {"username": "bob", "favourite_color": "green", "favourite_book": "Extreme Ownership"},
    ]

    df = pd.DataFrame(records)
    df.to_sql(table_name, engine, if_exists="append", index=False)
    return users

And finally the actual test

def test_query(engine, target_table):
    with Session(engine) as session:
        query = session.query(target_table)
        query = add_filters(query, target_table, "favourite_book", "Extreme Ownership")
        df = pd.read_sql_query(query.statement, session.bind)
        assert df["favourite_book"].unique().tolist() == ["Extreme Ownership"]

and you can see, the test is not very comprehensive, as it only tests one case. However, we can use pytest.mark.parametrize to extend that. (see last reference)

@pytest.mark.parametrize(
    "filter_type,filter_value,expected_result",
    [
        ("favourite_book", "Extreme Ownership", True),
        ("favourite_book", "Extreme", False),
        ("favourite_color", "blue", True),
        ("favourite_color", "purple", False),
    ],
)
def test_query(engine, target_table, filter_type, filter_value, expected_result):
    with Session(engine) as session:
        query = session.query(target_table)
        query = add_filters(query, target_table, filter_type, filter_value)
        df = pd.read_sql_query(query.statement, session.bind)
        assert (df[filter_type].unique().tolist() == [filter_value]) == expected_result

some references:

• https://www.tutorialspoint.com/pytest/pytest_fixtures.htm
• https://smirnov-am.github.io/pytest-testing_database/
• Python - How to connect SQLAlchemy to existing database in memory
• Mocking database calls in python using pytest-mock