gemini-antenna/tests/test_queuefeed.py

import os
from pathlib import Path
from typing import Optional

import pytest
from pytest import CaptureFixture

from gemini_antenna.URLHelper import URLHelper
from gemini_antenna.cgi import queuefeed
from gemini_antenna.db import AntennaDB

queuefeed_ask_input = "10 Feed URL:\r\n"
queuefeed_success = """20 text/gemini\r
Thank you for your submission!

This feed will be processed in a few minutes.
"""
queuefeed_invalid = """20 text/gemini\r
Error adding feed.

Feed URL is not valid.
"""
queuefeed_blocked = """20 text/gemini\r
Error adding feed.

Feed URL is blocked by instance rules.
"""


@pytest.mark.parametrize("query", [None, ""])
def test_ask_input(tmp_path: Path, capfd: CaptureFixture[str],
                   query: Optional[str]):
    if query is not None:
        os.environ['QUERY_STRING'] = query
    elif 'QUERY_STRING' in os.environ:
        del os.environ['QUERY_STRING']

    queuefeed.main()

    assert capfd.readouterr().out == queuefeed_ask_input


@pytest.mark.blocklist_rules("gemini://example.com/~antenna", "http://")
@pytest.mark.parametrize("query", [
    "https%3A%2F%2Fexample.com%2Ffeed.atom",
    "gemini.circumlunar.space/news/",
    "gemini://example.com/gemlog/about-http%3A%2F%2F-scheme",
])
def test_success(tmp_path: Path, db: AntennaDB, capfd: CaptureFixture[str],
                 query: str):
    os.environ['QUERY_STRING'] = query
    queuefeed.main()

    assert capfd.readouterr().out == queuefeed_success
    assert URLHelper.resolve(query) in db.getQueue()


@pytest.mark.blocklist_rules("gemini://example.com/~antenna", "http://",)
@pytest.mark.parametrize("query", [
    "/etc/passwd",
    "my cool feed",
    "42",
])
def test_invalid(tmp_path: Path, db: AntennaDB, capfd: CaptureFixture[str],
                 query: str):
    os.environ['QUERY_STRING'] = query
    queuefeed.main()

    assert capfd.readouterr().out == queuefeed_invalid
    assert not URLHelper.resolve(query) in db.getQueue()


@pytest.mark.blocklist_rules("gemini://example.com/~antenna", "http")
@pytest.mark.parametrize("query", [
    "http://example.org/feed",
    "https://example.org/feed",
    "example.com/~antenna",
    "gemini://example.com/./~antenna",
    "gemini://example.com/~user/../~antenna/feed.atom",
])
def test_blocked(tmp_path: Path, db: AntennaDB, capfd: CaptureFixture[str],
                 query: str):
    os.environ['QUERY_STRING'] = query
    queuefeed.main()

    assert capfd.readouterr().out == queuefeed_blocked
    assert not URLHelper.resolve(query) in db.getQueue()