forked from retoor/devplacepy
A new post attachment type beside polls: the composer gains a Start
Opinion War builder (same disabled-inputs opt-in as the poll builder)
that names exactly two factions; the battle runs for exactly 7 days
from post creation. Members join a side, may defect at any time
(damage already dealt stays with the faction it was dealt to), and
fight once per 24 hours per battle. A fight spends 25 Code Farm coins
and deals deterministic level-weighted damage: 100 + 10 * min(level,
20) HP, so a newcomer deals 110 and a veteran caps at 300 - no
randomness anywhere.
The battle renders on the post card as a CSS pixel-art battlefield
(box-shadow sprites: castles, faction flags, marching soldiers, a
flickering campfire; steps() animation, disabled under reduced motion)
with live HP bars, a countdown, the viewer's faction strip, top
contributors and an event ticker. Live frames ride pub/sub on
public.battle.{uid} via a relay on the service-lock owner, with the
durable opinion_war_events trail (per-war atomic seq) as the source of
truth and a 15s incremental poller as fallback. /battles lists battles
with active/ended/mine filters, search and pagination.
Every mutation is a conditional UPDATE via conditional_update_row: the
fight sequence claims the cooldown first, then spends coins, then lands
the damage, compensating earlier steps on any later refusal so a crash
costs a turn, never coins. Resolution is lazy on read (no cron):
an exactly-once CAS computes the winner in the statement, awards XP
(participation, winner bonus, top damage dealer bonus; draws pay
participation only), emits the result event and notifies fighters. The
OpinionWarService backstop resolves unviewed wars and sends
fight-ready notifications, exactly-once via a marker CAS.
Fan-out: battle notification type, four badges, audit keys
(battle.create/join/switch/fight/resolve), Devii actions (join/fight
confirm-gated), API docs group, docs prose page, sitemap and topnav
entries, REPORTABLE_TARGETS registration, post-delete cascades,
README and nested CLAUDE.md documentation.
Verified with the four-layer procedure: property checks over the full
damage domain, 1200-step stateful fuzz (hp-sum invariant, coins never
negative, resolved totals frozen), and real 8-process races proving
exactly-once semantics for concurrent fights, double-spends across two
wars, resolution XP and double-joins. Persisted tests in
tests/unit/services/opinionwar, tests/api/battles, tests/e2e/battles
and tests/api/posts/create.py.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
95 lines
3.3 KiB
Python
95 lines
3.3 KiB
Python
# retoor <retoor@molodetz.nl>
|
|
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import requests
|
|
|
|
from devplacepy.database import get_table, refresh_snapshot
|
|
from devplacepy.utils.rewards import XP_BATTLE_PART, XP_BATTLE_TOP, XP_BATTLE_WIN
|
|
from tests.api.battles._helpers import (
|
|
_create_war_post,
|
|
_fight,
|
|
_join,
|
|
_session_battles,
|
|
_user_row,
|
|
_war_row,
|
|
)
|
|
from tests.conftest import BASE_URL
|
|
|
|
|
|
def _warp_ended(war_uid):
|
|
past = (datetime.now(timezone.utc) - timedelta(minutes=1)).isoformat()
|
|
get_table("opinion_wars").update({"uid": war_uid, "ends_at": past}, ["uid"])
|
|
|
|
|
|
def test_resolution_awards_xp_exactly_once(app_server):
|
|
s, winner_name = _session_battles()
|
|
other, loser_name = _session_battles()
|
|
_, war = _create_war_post(s, f"resolve-{int(time.time() * 1000)}")
|
|
_join(s, war["uid"], "a")
|
|
_join(other, war["uid"], "b")
|
|
assert _fight(s, war["uid"]).status_code == 200
|
|
winner_before = int(_user_row(winner_name)["xp"] or 0)
|
|
loser_before = int(_user_row(loser_name)["xp"] or 0)
|
|
|
|
_warp_ended(war["uid"])
|
|
r = requests.get(f"{BASE_URL}/battles/{war['uid']}")
|
|
assert r.status_code == 200
|
|
state = r.json()
|
|
assert state["status"] == "resolved"
|
|
assert state["winner"] == "a"
|
|
assert state["winner_label"] == "Tabs"
|
|
|
|
row = _war_row(war["uid"])
|
|
assert row["status"] == "resolved"
|
|
assert row["resolved_at"]
|
|
|
|
winner_after = int(_user_row(winner_name)["xp"] or 0)
|
|
loser_after = int(_user_row(loser_name)["xp"] or 0)
|
|
assert winner_after == winner_before + XP_BATTLE_PART + XP_BATTLE_WIN + XP_BATTLE_TOP
|
|
assert loser_after == loser_before
|
|
|
|
events = requests.get(f"{BASE_URL}/battles/{war['uid']}/events").json()["events"]
|
|
results = [event for event in events if event["kind"] == "result"]
|
|
assert len(results) == 1
|
|
|
|
refresh_snapshot()
|
|
notes = list(
|
|
get_table("notifications").find(
|
|
type="battle", user_uid=_user_row(loser_name)["uid"]
|
|
)
|
|
)
|
|
assert any("Battle over" in (n.get("message") or "") for n in notes)
|
|
|
|
r = requests.get(f"{BASE_URL}/battles/{war['uid']}")
|
|
assert r.json()["status"] == "resolved"
|
|
assert int(_user_row(winner_name)["xp"] or 0) == winner_after
|
|
events = requests.get(f"{BASE_URL}/battles/{war['uid']}/events").json()["events"]
|
|
assert len([e for e in events if e["kind"] == "result"]) == 1
|
|
|
|
|
|
def test_draw_awards_participation_only(app_server):
|
|
s, name = _session_battles()
|
|
_, war = _create_war_post(s, f"draw-{int(time.time() * 1000)}")
|
|
_join(s, war["uid"], "a")
|
|
before = int(_user_row(name)["xp"] or 0)
|
|
_warp_ended(war["uid"])
|
|
state = requests.get(f"{BASE_URL}/battles/{war['uid']}").json()
|
|
assert state["winner"] == "draw"
|
|
assert int(_user_row(name)["xp"] or 0) == before
|
|
|
|
|
|
def test_post_page_resolves_due_war(app_server):
|
|
s, _ = _session_battles()
|
|
post, war = _create_war_post(s, f"lazy-{int(time.time() * 1000)}")
|
|
_join(s, war["uid"], "b")
|
|
assert _fight(s, war["uid"]).status_code == 200
|
|
_warp_ended(war["uid"])
|
|
slug = post.get("slug") or post["uid"]
|
|
r = s.get(f"{BASE_URL}/posts/{slug}")
|
|
assert r.status_code == 200
|
|
assert "wins the war" in r.text
|
|
assert _war_row(war["uid"])["status"] == "resolved"
|
|
assert _war_row(war["uid"])["winner"] == "b"
|