test_db.py

import aiopg

from funkwhale_network import settings
from funkwhale_network.db import DB


async def test_create_domain_ignore_duplicate(populated_db):
    async with DB() as db:
        r1 = await db.create_domain({"name": "test.domain"})
        r2 = await db.create_domain({"name": "test.domain"})

        assert r1 != r2
        assert r2 is None


async def test_db_create():
    async with DB() as db:
        try:
            await db.create()
            tables = ["domains", "checks"]
            conn = await aiopg.connect(settings.DB_DSN)
            async with conn.cursor() as cursor:
                for t in tables:
                    await cursor.execute(f"SELECT * from {t}")
                    await cursor.fetchall()
        finally:
            await db.clear()


# async def test_get_latest_checks_by_domain(factories):
#    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]
#    async with DB() as db:
#        for check in expected:
#            domain = await db.get_domain(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()
#        assert len(result) == 2
#        for i, row in enumerate(result):
#            assert dict(row) == dict(expected[i])
#
#
# async def test_get_stats(factories):
#    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},
#    }
#    async with DB() as db:
#        assert await db.get_stats() == expected