devplacepy/tests/unit/services/community/discoverer.py

# retoor <retoor@molodetz.nl>

from tests.conftest import run_async

from devplacepy.services.community.discoverer import CommunityDiscoverer
from devplacepy.services.community.spec import STATUS_FAILED, STATUS_LOGIN, STATUS_READY
from tests.unit.services.community.fakes import (
    FakeFetcher,
    club_community,
    html_community,
    json_community,
    rss_community,
    spa_community,
)


def discover(routes: dict, query: str, credentials: dict | None = None):
    fetcher = FakeFetcher(routes)
    return run_async(CommunityDiscoverer(fetcher, credentials=credentials).discover(query)), fetcher


def test_json_api_community_is_learned():
    result, _ = discover(json_community(), "https://forum.test/api/posts")
    assert result.status == STATUS_READY, result.error
    feed = result.schema["feeds"][0]
    assert feed["type"] == "json_api"
    assert feed["list_path"] == "$.items"
    assert feed["post"]["fields"]["id"] == ["$.id"]
    assert feed["post"]["fields"]["content"][0] == "$.content_html"
    assert feed["post"]["fields"]["created_at"] == ["$.created_at"]
    assert feed["author"]["fields"]["username"] == ["$.author.username"]
    assert feed["author"]["fields"]["id"] == ["$.author.id"]
    assert feed["pagination"] == {"style": "page", "param": "page", "start": 1}
    assert feed["comments"]["mode"] == "inline"
    assert feed["comments"]["fields"]["content"] == ["$.body"]
    assert feed["comments"]["author"]["fields"]["username"] == ["$.user.username"]
    records = result.sample["records"]
    assert len(records["posts"]) == 8
    assert len(records["comments"]) == 6
    assert {user["username"] for user in records["users"]} >= {"writer0", "writer1", "replier0"}


def test_rss_community_is_learned_from_advertised_feed():
    result, _ = discover(rss_community(), "https://blog.test/")
    assert result.status == STATUS_READY, result.error
    feed = result.schema["feeds"][0]
    assert feed["type"] == "rss"
    assert feed["post"]["fields"]["created_at"] == ["$.pubDate"]
    assert result.sample["records"]["users"][0]["username"] == "Alice"
    assert result.sample["records"]["posts"][0]["content"].startswith("Full body of entry")
    assert feed["comments"]["mode"] == "page"
    assert feed["comments"]["item_selector"] == "div.comment"
    assert feed["post"]["detail"]["content"] == ["article"]
    comments = result.sample["records"]["comments"]
    assert len(comments) == 9
    assert sum(1 for comment in comments if comment.get("parent_uid")) == 3


def test_html_list_community_is_learned():
    result, _ = discover(html_community(), "https://board.test/")
    assert result.status == STATUS_READY, result.error
    feed = result.schema["feeds"][0]
    assert feed["type"] == "html_list"
    assert feed["item_selector"] == "article.post"
    assert len(result.sample["records"]["posts"]) == 6
    assert result.sample["records"]["posts"][0]["title"].startswith("Thread 1")
    assert feed["comments"]["mode"] == "page"
    assert feed["comments"]["fields"]["id"] == ["::attr(id)"]
    assert feed["comments"]["author"]["fields"]["username"] == [".author"]
    assert len(result.sample["records"]["comments"]) == 9


def test_unreachable_community_fails_with_reason():
    result, _ = discover({}, "https://nothing.test/")
    assert result.status == STATUS_FAILED
    assert result.error


def test_javascript_only_site_is_rendered():
    result, _ = discover(spa_community(), "https://spa.test/")
    assert result.status == STATUS_READY, result.error
    feed = result.schema["feeds"][0]
    assert feed["render"] is True
    assert len(result.sample["records"]["posts"]) == 6


def test_login_wall_without_credentials_is_flagged_not_failed():
    state = {}
    result, _ = discover(club_community(state), "https://club.test/")
    assert result.status == STATUS_LOGIN
    assert result.schema["login"]["mode"] == "form"
    assert result.schema["login"]["username_field"] == "user"
    assert result.schema["login_required"] is True


def test_login_wall_with_credentials_is_learned_behind_the_wall():
    state = {}
    result, fetcher = discover(club_community(state), "https://club.test/", {"username": "bob", "password": "pw"})
    assert result.status == STATUS_READY, result.error
    assert result.schema["login_required"] is True
    assert result.schema["feeds"][0]["type"] == "rss"
    assert fetcher.posts[0] == ("https://club.test/login", {"csrf": "t1", "user": "bob", "pass": "pw"})


def test_wrong_credentials_fail_loudly():
    result, _ = discover(club_community({}), "https://club.test/", {"username": "bob", "password": "nope"})
    assert result.status == STATUS_FAILED
    assert "rejected the stored credentials" in result.error


def test_feeds_listing_the_same_posts_are_learned_once_and_comment_feeds_are_skipped():
    routes = rss_community()
    routes["https://blog.test/"] = (
        '<html><head><link rel="alternate" type="application/rss+xml" href="/feed.xml">'
        '<link rel="alternate" type="application/rss+xml" href="/comments.xml"></head>'
        '<body><a href="/latest">Latest posts</a></body></html>'
    )
    routes["https://blog.test/comments.xml"] = routes["https://blog.test/feed.xml"]
    items = "".join(
        f'<article class="post"><h2><a href="/p/{n}">Entry {n} on testing</a></h2><span class="author">Alice</span>'
        f"<p>Body of entry {n} with enough text to count as content.</p></article>"
        for n in range(1, 7)
    )
    routes["https://blog.test/latest"] = f"<html><body><main>{items}</main></body></html>"
    result, fetcher = discover(routes, "https://blog.test/")
    assert result.status == STATUS_READY, result.error
    assert [feed["type"] for feed in result.schema["feeds"]] == ["rss"]
    assert not any("comments.xml" in request for request in fetcher.requests)


def test_generic_probe_paths_are_skipped_when_a_feed_is_advertised():
    _, fetcher = discover(rss_community(), "https://blog.test/")
    assert not any(request.endswith("/api/posts") or request.endswith("/rss.xml") for request in fetcher.requests)


def test_a_captcha_on_the_login_wall_is_reported_and_never_attempted():
    routes = club_community({})
    captcha_wall = routes["https://club.test/"]().replace("<form", '<div class="g-recaptcha" data-sitekey="k"></div><form', 1)
    routes["https://club.test/"] = captcha_wall
    routes["https://club.test/login"] = captcha_wall
    result, fetcher = discover(routes, "https://club.test/")
    assert result.status == "login_required" and "captcha" in result.error
    assert result.schema["login"]["captcha"] is True and fetcher.posts == []
    result, fetcher = discover(routes, "https://club.test/", {"username": "bob", "password": "pw"})
    assert result.status == "login_required" and "captcha" in result.error and fetcher.posts == []


def test_a_second_factor_with_a_secret_logs_in_and_without_one_asks_a_human(monkeypatch):
    from devplacepy.services.community import runtime
    from tests.unit.services.community.login import OTP_PAGE

    monkeypatch.setattr(runtime, "totp", lambda secret: "123456")
    state = {}
    routes = club_community(state)
    password_step = routes["post:https://club.test/login"]

    def first_step(data):
        answer = password_step(data)
        if state.get("in"):
            state["in"] = False
            return (200, OTP_PAGE)
        return answer

    def second_step(data):
        state["in"] = data.get("otp_code") == "123456"
        return (200, "welcome" if state["in"] else OTP_PAGE)

    routes["post:https://club.test/login"] = first_step
    routes["post:https://club.test/2fa"] = second_step
    result, _ = discover(routes, "https://club.test/", {"username": "bob", "password": "pw", "totp": "JBSWY3DPEHPK3PXP"})
    assert result.status == "ready", result.error
    state.clear()
    result, _ = discover(routes, "https://club.test/", {"username": "bob", "password": "pw"})
    assert result.status == "login_required" and "authenticator secret" in result.error