test_routes.py

import datetime
import pytest

from funkwhale_network import db
from funkwhale_network import serializers


async def test_domains_get(db_conn, client, factories):

    await factories["Check"].c(private=True)
    checks = sorted(
        [
            await factories["Check"].c(private=False),
            await factories["Check"].c(private=False),
        ],
        key=lambda o: o["domain"],
    )
    for check in checks:
        domain = await db.get_domain(db_conn, check["domain"])
        check["first_seen"] = domain["first_seen"]
        check["node_name"] = domain["node_name"]
    resp = await client.get("/api/domains")
    assert resp.status == 200
    assert await resp.json() == {
        "count": len(checks),
        "next": None,
        "previous": None,
        "results": [serializers.serialize_domain_from_check(check) for check in checks],
    }


async def test_domains_get_page_size(db_conn, client, factories):

    checks = sorted(
        [
            await factories["Check"].c(private=False),
            await factories["Check"].c(private=False),
        ],
        key=lambda o: o["domain"],
    )
    for check in checks:
        domain = await db.get_domain(db_conn, check["domain"])
        check["first_seen"] = domain["first_seen"]
        check["node_name"] = domain["node_name"]
    resp = await client.get("/api/domains", params={"limit": 1})
    assert resp.status == 200
    assert await resp.json() == {
        "count": 2,
        "next": None,
        "previous": None,
        "results": [serializers.serialize_domain_from_check(checks[0])],
    }


@pytest.mark.parametrize(
    "field", ["up", "open_registrations", "anonymous_can_listen", "federation_enabled"]
)
@pytest.mark.parametrize("value", [True, False])
async def test_domains_get_filter_up(db_conn, client, factories, field, value):

    await factories["Check"].c(private=True)
    await factories["Check"].c(private=False, **{field: not value}),
    check = await factories["Check"].c(private=False, **{field: value})
    domain = await db.get_domain(db_conn, check["domain"])
    check["first_seen"] = domain["first_seen"]
    check["node_name"] = domain["node_name"]
    resp = await client.get("/api/domains", params={field: str(value)})
    assert resp.status == 200
    assert await resp.json() == {
        "count": 1,
        "next": None,
        "previous": None,
        "results": [serializers.serialize_domain_from_check(check)],
    }


async def test_domains_exclude_blocked(db_conn, client, factories):

    blocked = await factories["Domain"].c(blocked=True)
    await factories["Check"].c(private=False, domain=blocked["name"])
    check = await factories["Check"].c(private=False)
    domain = await db.get_domain(db_conn, check["domain"])
    check["first_seen"] = domain["first_seen"]
    check["node_name"] = domain["node_name"]
    resp = await client.get("/api/domains")
    assert resp.status == 200
    assert await resp.json() == {
        "count": 1,
        "next": None,
        "previous": None,
        "results": [serializers.serialize_domain_from_check(check)],
    }


async def test_domains_create(client, coroutine_mock, mocker):
    payload = {"name": "test.domain"}
    mocker.patch("funkwhale_network.crawler.fetch_nodeinfo", coroutine_mock())

    resp = await client.post("/api/domains", json=payload)
    assert resp.status == 201
    assert await resp.json() == {"name": payload["name"]}


async def test_domains_stats(client, mocker, coroutine_mock):

    payload = {"hello": "world"}
    mocker.patch("funkwhale_network.db.get_stats", coroutine_mock(return_value=payload))
    resp = await client.get("/api/domains/stats")
    assert resp.status == 200
    assert await resp.json() == payload