git / moonbite-network / tests/test_forum_moderation.py · raw
"""Forum moderation: auto-mod guards, moderator login/actions (key + CSRF),reports, pinned/locked/removed behaviour, and the weekly scheduled threads."""import ioimport timeimport pytestimport forumimport forum_botimport web_appSEED = "abandon ability able about above absent absorb abstract absurd abuse access accident"KEY = "test-mod-key-123"@pytest.fixture(autouse=True)def board(tmp_path, monkeypatch): monkeypatch.setattr(forum, "_DB_PATH", str(tmp_path / "forum.db")) monkeypatch.setenv("MOONBITE_FORUM_MOD_KEY", KEY)# ---------------------------------------------------------------- auto-moddef test_seed_phrases_and_keys_are_refused(): with pytest.raises(ValueError, match="seed phrase"): forum.create_thread("help", "x", "my words are " + SEED) with pytest.raises(ValueError, match="seed phrase"): forum.create_thread("help", "x", "key 5HueCGU8rMjxEXxiPuD5BDku4MkFqeZyd4dZ1jvhTVqvbTLvyTJ") # ordinary prose full of common words is fine forum.create_thread("ok", "x", "I think the miner is able to find a block about once an hour on my laptop.")def test_link_spam_and_duplicates_are_refused(): with pytest.raises(ValueError, match="Too many links"): forum.create_thread("x", "x", " ".join(f"https://e{i}.com" for i in range(6))) forum.create_thread("Same title", "x", "same words") with pytest.raises(ValueError, match="already posted"): forum.create_thread("same TITLE", "y", "Same words") t = forum.create_thread("other", "x", "body") forum.add_reply(t["id"], "a", "+1") forum.add_reply(t["id"], "b", "+1") # short and from someone else: fine with pytest.raises(ValueError): forum.add_reply(t["id"], "a", "+1") # same person, same text spam = "Buy cheap followers now at the best price on the internet" forum.add_reply(t["id"], "c", spam) with pytest.raises(ValueError): forum.add_reply(t["id"], "d", spam) # long copy-paste from anyone# ---------------------------------------------------------------- flagsdef test_pinned_locked_removed(): t = forum.create_thread("rules", "x", "body") other = forum.create_thread("other", "x", "other body") forum.set_flag("t", t["id"], "pinned", True) assert [p["id"] for p in forum.pinned_threads()] == [t["id"]] assert t["id"] not in [x["id"] for x in forum.list_threads()["threads"]] forum.set_flag("t", other["id"], "locked", True) with pytest.raises(ValueError, match="locked"): forum.add_reply(other["id"], "y", "hello") forum.add_reply(other["id"], None, "moderator note", official=True) # mods can still reply r = forum.add_reply(t["id"], "y", "spam reply") forum.set_flag("r", r["id"], "removed", True) assert forum.get_thread(t["id"])["replies"][0]["removed"] forum.set_flag("t", other["id"], "removed", True) assert forum.get_thread(other["id"]) is None assert forum.get_thread(other["id"], include_removed=True) is not None with pytest.raises(ValueError): forum.set_flag("r", r["id"], "pinned", True)def test_reports_group_and_resolve(): t = forum.create_thread("x", "x", "body") assert forum.report("t", t["id"], "spam", "net1") assert not forum.report("t", t["id"], "scam", "net1") # one per network assert forum.report("t", t["id"], "scam", "net2") rep = forum.open_reports() assert rep[0]["count"] == 2 and rep[0]["reasons"] == ["scam", "spam"] forum.resolve_reports("t", t["id"]) assert forum.open_reports() == []def test_metrics_count_people_not_the_official_account(): t = forum.create_thread("q", "alice", "question") forum.add_reply(t["id"], "bob", "answer") forum.create_thread("report", None, "numbers", "network", official=True, automated=True) m = forum.metrics() assert m["this_week"] == {"threads": 1, "replies": 1, "posters": 2}# ---------------------------------------------------------------- webdef _client(): web_app.app.config["SESSION_COOKIE_SECURE"] = False return web_app.app.test_client()def _csrf(c): with c.session_transaction() as s: return s.get("forum_csrf")def test_mod_login_and_actions_need_key_and_csrf(): c = _client() t = forum.create_thread("x", "x", "body") # not logged in: action refused c.post("/discussions/mod/action", data={"kind": "t", "id": t["id"], "action": "pin", "csrf": "x"}) assert not forum.pinned_threads() assert b"Moderator login" in c.get("/discussions/mod").data c.post("/discussions/mod/login", data={"key": "wrong"}) assert b"Moderator login" in c.get("/discussions/mod").data c.post("/discussions/mod/login", data={"key": KEY}) assert b"Moderator desk" in c.get("/discussions/mod").data # logged in but no / wrong CSRF token: refused c.post("/discussions/mod/action", data={"kind": "t", "id": t["id"], "action": "pin", "csrf": "bad"}) assert not forum.pinned_threads() c.post("/discussions/mod/action", data={"kind": "t", "id": t["id"], "action": "pin", "csrf": _csrf(c), "next": "//evil.example"}) assert forum.pinned_threads()def test_only_moderators_can_post_as_verified(): c = _client() t = forum.create_thread("x", "x", "body") c.post(f"/discussions/{t['id']}/reply", data={"body": "trying to fake it", "as_official": "1"}) assert not forum.get_thread(t["id"])["replies"][0]["is_official"] c.post("/discussions/mod/login", data={"key": KEY}) c.post(f"/discussions/{t['id']}/reply", data={"body": "real mod reply", "as_official": "1", "csrf": _csrf(c)}) rep = forum.get_thread(t["id"])["replies"][1] assert rep["is_official"] and rep["author"] == forum.OFFICIAL_NAMEdef test_report_api_is_same_origin_json(): c = _client() t = forum.create_thread("x", "x", "body") assert c.post("/api/discussions/report", json={"kind": "t", "id": t["id"], "reason": "spam"}, headers={"Origin": "https://evil.example"}).status_code == 403 assert c.post("/api/discussions/report", json={"kind": "t", "id": t["id"], "reason": "spam"}).status_code == 200 assert forum.open_reports()[0]["id"] == t["id"]def test_mod_desk_off_without_key(monkeypatch): monkeypatch.delenv("MOONBITE_FORUM_MOD_KEY") assert _client().get("/discussions/mod").status_code == 404# ---------------------------------------------------------------- weekly threadsclass Node: def getblockcount(self): return 100 def getblockhash(self, h): return f"h{h}" def call(self, method, *params): if method == "getblockheader": return {"time": int(time.time()) - (100 - int(params[0][1:])) * 120} if method == "getnetworkhashps": return 1600.0 raise AssertionError(method)def _day(weekday: int) -> int: """A UTC timestamp at 23:00 (after every posting slot) on the given weekday (Monday = 0).""" base = 1_800_000_000 - 1_800_000_000 % 86400 + 82_800 return base + ((weekday - time.gmtime(base).tm_wday) % 7) * 86400def test_weekly_threads_post_on_their_day_once(): for wd, prefix in ((0, "Show your rig"), (3, "Hot take"), (5, "Predictions")): posted = forum_bot._weekly(Node(), _day(wd), None, False, io.StringIO()) assert len(posted) == 1 assert forum_bot._weekly(Node(), _day(wd), None, False, io.StringIO()) == [] assert any(t["title"].startswith(prefix) for t in forum.list_threads(limit=50)["threads"]) assert forum_bot._weekly(Node(), _day(1), None, False, io.StringIO()) == [] # Tuesday: nothingdef test_predictions_get_real_numbers_next_week(): forum_bot._weekly(Node(), _day(5), None, False, io.StringIO()) first = forum_bot._thread_for(f"weekly:predict:{forum_bot._week(_day(5))}") forum_bot._weekly(Node(), _day(5) + 7 * 86400, None, False, io.StringIO()) replies = forum.get_thread(first)["replies"] assert len(replies) == 1 and replies[0]["is_official"] and "Chain height: 100" in replies[0]["body"]# ---------------------------------------------------------------- pacingdef test_posts_wait_for_their_human_hours_slot(): day = _day(0) - 82_800 # 00:00 UTC on a Monday key = f"weekly:rig:{forum_bot._week(day)}" s = forum_bot.slot(key, *forum_bot.WEEKLY_WINDOW) assert 9 * 3600 <= s < 18 * 3600 assert forum_bot._weekly(Node(), day + s - 60, None, False, io.StringIO()) == [] assert len(forum_bot._weekly(Node(), day + s, None, False, io.StringIO())) == 1 # different keys land at different times slots = {forum_bot.slot(f"daily:2026-10-{d:02d}", *forum_bot.DAILY_WINDOW) for d in range(1, 15)} assert len(slots) > 10 and all(6 * 3600 <= x < 9 * 3600 for x in slots)def test_queue_publishes_founder_posts_on_time_and_holds_unresolved(tmp_path): import json q = tmp_path / "queue.json" q.write_text(json.dumps([ {"key": "ama", "publish_at": "2026-10-25T10:00:00Z", "category": "general", "title": "Ask us anything", "body": "Opening post", "pin": True}, {"key": "ama-2", "publish_at": "2026-10-26T15:00:00Z", "reply_to": "ama", "body": "Second batch"}, {"key": "draft", "publish_at": "2026-10-25T10:00:00Z", "title": "x", "body": "[NEEDS FOUNDER INPUT: something]"}, ]), encoding="utf-8") t0 = forum_bot._parse_time("2026-10-25T09:59:00Z") assert forum_bot.publish_queue(t0, str(q), out=io.StringIO()) == [] assert forum_bot.publish_queue(t0 + 120, str(q), out=io.StringIO()) == ["queue:ama"] thread = forum.pinned_threads()[0] assert thread["is_official"] and not thread["automated"] # written by a person assert forum_bot.publish_queue(t0 + 2 * 86400, str(q), out=io.StringIO()) == ["queue:ama-2"] assert forum_bot.publish_queue(t0 + 3 * 86400, str(q), out=io.StringIO()) == [] # once only assert forum.get_thread(thread["id"])["replies"][0]["body"] == "Second batch" assert forum.list_threads()["total"] == 0 # the held draft never posted