Newer
Older
from funkwhale_network.db import DB
from funkwhale_network import settings
import aiopg
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("SELECT * from {}".format(t))
await cursor.fetchall()
finally:
await db.clear()
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
#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