test_db.py

import pytest

from funkwhale_network import db


async def test_db_create(db_pool):
    try:
        async with db_pool.acquire() as conn:
            await db.create(conn)
            tables = ["domains", "checks"]
            async with conn.cursor() as cursor:
                for t in tables:
                    await cursor.execute("SELECT * from {}".format(t))
                    await cursor.fetchall()
    finally:
        async with db_pool.acquire() as conn:
            await db.clear(conn)


async def test_get_latest_checks_by_domain(factories, db_conn):
    await factories["Check"].c(domain="test1.domain", private=False)
    check2 = await factories["Check"].c(domain="test1.domain", private=False)
    check3 = await factories["Check"].c(domain="test2.domain", private=False)

    expected = [check2, check3]
    assert await db.get_latest_check_by_domain(db_conn) == expected


async def test_get_stats(factories, db_conn):
    await factories["Check"].c(domain="test1.domain", private=False)
    await factories["Check"].c(
        domain="test1.domain",
        private=False,
        open_registrations=False,
        anonymous_can_listen=False,
        usage_users_total=2,
        usage_users_active_half_year=1,
        usage_users_active_month=2,
        usage_listenings_total=20,
        library_tracks_total=6,
        library_albums_total=30,
        library_artists_total=36,
    )
    await factories["Check"].c(
        domain="test2.domain",
        private=False,
        open_registrations=True,
        anonymous_can_listen=True,
        usage_users_total=3,
        usage_users_active_half_year=3,
        usage_users_active_month=1,
        usage_listenings_total=22,
        library_tracks_total=15,
        library_albums_total=13,
        library_artists_total=40,
    )

    expected = {
        "users": {"total": 5, "activeMonth": 3, "activeHalfyear": 4},
        "instances": {"total": 2, "anonymousCanListen": 1, "openRegistrations": 1},
        "artists": {"total": 76},
        "albums": {"total": 43},
        "tracks": {"total": 21},
        "listenings": {"total": 42},
    }
    assert await db.get_stats(db_conn) == expected


@pytest.mark.parametrize(
    "kwargs, expected_query, expected_params",
    [
        (
            {},
            "SELECT DISTINCT on (domain) domain, * FROM checks ORDER BY domain, time DESC",
            [],
        ),
        (
            {"up": True},
            "SELECT DISTINCT on (domain) domain, * FROM checks WHERE up = %s ORDER BY domain, time DESC",
            [True],
        ),
        (
            {"up": True, "open_registrations": False},
            "SELECT DISTINCT on (domain) domain, * FROM checks WHERE open_registrations = %s AND up = %s ORDER BY domain, time DESC",
            [False, True],
        ),
        (
            {"up": True, "private": False},
            "SELECT DISTINCT on (domain) domain, * FROM checks WHERE private = %s AND up = %s ORDER BY domain, time DESC",
            [False, True],
        ),
    ],
)
def test_get_domain_query(kwargs, expected_query, expected_params):
    query, params = db.get_domain_query(**kwargs)
    assert query == expected_query
    assert params == expected_params