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]
    for check in expected:
        domain = await db.get_domain(db_conn, check["domain"])
        check["first_seen"] = domain["first_seen"]
        check["node_name"] = domain["node_name"]
        check["blocked"] = domain["blocked"]
        check["name"] = domain["name"]
    result = await db.get_latest_check_by_domain(db_conn)
    assert len(result) == 2
    for i, row in enumerate(result):
        assert dict(row) == dict(expected[i])


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,
        usage_downloads_total=30,
        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,
        usage_downloads_total=33,
        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},
        "downloads": {"total": 63},
    }
    assert await db.get_stats(db_conn) == expected