test_routes.py

import datetime
import pytest

from funkwhale_network import db
from funkwhale_network import routes
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):
    domain = factories["Domain"]()
    await factories["Check"].c(domain=domain["name"], private=True)
    await factories["Check"].c(
        domain=domain["name"], private=False, **{field: not value}
    ),
    check = await factories["Check"].c(
        domain=domain["name"], 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


async def test_domains_rss(db_conn, factories, client, mocker):

    blocked = await factories["Domain"].c(blocked=True)
    await factories["Check"].c(private=False, domain=blocked["name"])

    await factories["Domain"].c(blocked=True)
    await factories["Check"].c(),
    await factories["Check"].c(),
    expected = serializers.serialize_rss_feed_from_checks(
        await db.get_domains(db_conn, **{})
    )
    resp = await client.get("/api/domains", params={"format": "rss"})
    assert resp.status == 200
    assert await resp.text() == expected
    assert resp.headers["content-type"] == "application/rss+xml"


@pytest.mark.parametrize(
    "input,expected",
    [
        ("example.com", "example.com"),
        ("http://example.com", "example.com"),
        ("https://example.com", "example.com"),
    ],
)
def test_validate_domain(input, expected):
    assert routes.validate_domain(input) == expected