feat: ritual + judgment + favor + cross-over (Workstream B)
Implements Workstream B of the character-depth-ghost-log spec: ritual_start/ritual_step/judgment WS handlers, the pure judgment.py logic module, and the User.favor / Entity.at_peace columns + migration. - app/judgment.py: pure ritual success roll (base 65%, floored at 30%, driven by an entity's power+deceptiveness difficulty), the "stuck spirit" cross_over rule (alignment >= 0.5 and volatility > 0.6, ~20% of entities), judgment correctness/favor-delta/essence-delta/ consequence resolution for all four verdicts, favor clamping, the favor-to-trait-roll bias applied at mint time, and tell-line generation (opaque behavioral flavor text, never a raw stat). - app/ws.py: wires ritual_start/ritual_step/judgment frames, emits ritual_complete/tell/judgment_result/item_drop per the spec's Contract; traits are added to serialize_entity for internal server-side use but stripped from the outbound `entity` frame via a new _public_entity helper so hidden ground truth never reaches the client outside ritual_complete; _summon excludes at-peace entities from signature re-contact and mints a fresh (salted-signature) entity instead; new entities' traits are nudged by the discovering user's favor before being persisted. - models/user.py, models/entity.py, main.py: User.favor and Entity.at_peace columns plus their idempotent ADD COLUMN IF NOT EXISTS migration lines in lifespan, alongside the existing ones. - tests/test_judgment.py, tests/test_ws_ritual_judgment.py: 56 new tests covering the ritual/judgment correctness matrix, favor clamping/bias, essence crediting, at_peace persistence + re-contact, and the entity-frame trait leak guard. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Indiana
614
backend/tests/test_ws_ritual_judgment.py
Normal file
614
backend/tests/test_ws_ritual_judgment.py
Normal file
@@ -0,0 +1,614 @@
|
||||
"""Workstream B WS integration tests: ritual_start/ritual_step/judgment
|
||||
handlers in app.ws, plus the favor/essence/at_peace side effects and the
|
||||
favor read-back bias on trait rolls at mint time."""
|
||||
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
|
||||
import app.judgment as judgment_module
|
||||
import app.ws
|
||||
from app.entities import fallback_profile
|
||||
from app.inventory import (
|
||||
CORRECT_JUDGMENT_ESSENCE,
|
||||
CROSS_OVER_ESSENCE,
|
||||
RITUAL_SUCCESS_ESSENCE,
|
||||
)
|
||||
from app.models.entity import Entity
|
||||
from app.models.user import User
|
||||
from app.rate_limit import RateLimiter
|
||||
|
||||
|
||||
class FakeSpiritService:
|
||||
async def mint_profile(self, signature, channel, anomalies, language="en"):
|
||||
return fallback_profile(signature)
|
||||
|
||||
async def fragment(self, source, anomaly, language="en"):
|
||||
return "listen"
|
||||
|
||||
async def wire_whisper(self, telemetry, language="en"):
|
||||
return "the wire hums"
|
||||
|
||||
def chat_stream(self, entity, question, history, language="en"):
|
||||
async def gen():
|
||||
for token in ["I ", "am ", "here."]:
|
||||
yield token
|
||||
|
||||
return gen()
|
||||
|
||||
def ambient_ready(self):
|
||||
return False
|
||||
|
||||
|
||||
async def _fake_synth(text, voice, profile, instability=0.0):
|
||||
return b"RIFFfake wav bytes"
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _fake_spirits(monkeypatch):
|
||||
monkeypatch.setattr(app.ws, "spirit_service", FakeSpiritService())
|
||||
monkeypatch.setattr(app.ws, "synthesize_spirit_voice", _fake_synth)
|
||||
# Generous, test-scoped limiters — the module-level ones are shared
|
||||
# singletons that accumulate real hit counts across the whole test
|
||||
# session (see backend/tests/test_ws_session.py's precedent), and this
|
||||
# file summons repeatedly.
|
||||
monkeypatch.setattr(app.ws, "summon_limiter", RateLimiter(max_requests=1000, window_seconds=60))
|
||||
monkeypatch.setattr(app.ws, "summon_ip_limiter", RateLimiter(max_requests=1000, window_seconds=60))
|
||||
monkeypatch.setattr(app.ws, "question_limiter", RateLimiter(max_requests=1000, window_seconds=60))
|
||||
monkeypatch.setattr(app.ws, "question_ip_limiter", RateLimiter(max_requests=1000, window_seconds=60))
|
||||
monkeypatch.setattr(app.ws, "fragment_limiter", RateLimiter(max_requests=1000, window_seconds=60))
|
||||
monkeypatch.setattr(app.ws, "fragment_ip_limiter", RateLimiter(max_requests=1000, window_seconds=60))
|
||||
|
||||
|
||||
def _read_until(ws, msg_type, max_frames=60, **match):
|
||||
for _ in range(max_frames):
|
||||
frame = ws.receive_json()
|
||||
if frame.get("type") != msg_type:
|
||||
continue
|
||||
if all(frame.get(key) == value for key, value in match.items()):
|
||||
return frame
|
||||
raise AssertionError(f"never saw frame of type {msg_type!r} matching {match!r}")
|
||||
|
||||
|
||||
def _login(sync_client, username):
|
||||
sync_client.post("/auth/register", json={"username": username, "password": "spookyspooky"})
|
||||
sync_client.post("/auth/login", json={"username": username, "password": "spookyspooky"})
|
||||
return sync_client.cookies.get("qm_session")
|
||||
|
||||
|
||||
def _ws_connect(sync_client, token):
|
||||
return sync_client.websocket_connect(
|
||||
"/ws/session", headers={"cookie": f"qm_session={token}"}
|
||||
)
|
||||
|
||||
|
||||
def _summon(ws):
|
||||
ws.send_json({"type": "summon"})
|
||||
entity_frame = _read_until(ws, "entity")
|
||||
_read_until(ws, "utterance", kind="greeting")
|
||||
return entity_frame
|
||||
|
||||
|
||||
def _run_ritual(ws, steps=4):
|
||||
ws.send_json({"type": "ritual_start"})
|
||||
for i in range(1, steps + 1):
|
||||
ws.send_json({"type": "ritual_step", "step": i})
|
||||
result = _read_until(ws, "ritual_complete")
|
||||
# `_handle_ritual_step`'s reward call (essence credit / item roll) runs
|
||||
# *after* the `ritual_complete` frame is queued for send, so receiving
|
||||
# that frame doesn't by itself guarantee the reward has landed yet (the
|
||||
# sender task drains the queue concurrently with the handler's own
|
||||
# in-flight awaits). A trailing ping/pong forces a full round trip
|
||||
# through the single connection's sequential message loop, which can't
|
||||
# read the next message until the ritual_step handler (reward included)
|
||||
# has fully returned — so seeing the pong is a hard guarantee, not a
|
||||
# poll-and-hope.
|
||||
ws.send_json({"type": "ping"})
|
||||
_read_until(ws, "pong")
|
||||
return result
|
||||
|
||||
|
||||
# --- entity frame never leaks traits ---------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_entity_frame_never_includes_traits(sync_client):
|
||||
_login(sync_client, "no-leak")
|
||||
with _ws_connect(sync_client, sync_client.cookies.get("qm_session")) as ws:
|
||||
_read_until(ws, "session")
|
||||
entity_frame = _summon(ws)
|
||||
assert "traits" not in entity_frame["entity"]
|
||||
|
||||
|
||||
# --- ritual ------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ritual_success_reveals_true_traits_and_credits_essence(
|
||||
sync_client, db_session, monkeypatch
|
||||
):
|
||||
monkeypatch.setattr(app.ws.judgment, "roll_ritual_success", lambda traits, rng=None: True)
|
||||
|
||||
token = _login(sync_client, "ritual-winner")
|
||||
user_id = uuid.UUID(
|
||||
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
|
||||
)
|
||||
|
||||
with _ws_connect(sync_client, token) as ws:
|
||||
_read_until(ws, "session")
|
||||
_summon(ws)
|
||||
result = _run_ritual(ws)
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["revealed"] is not None
|
||||
assert set(result["revealed"].keys()) == {
|
||||
"alignment",
|
||||
"power",
|
||||
"volatility",
|
||||
"deceptiveness",
|
||||
}
|
||||
for v in result["revealed"].values():
|
||||
assert 0.0 <= v <= 1.0
|
||||
|
||||
# trickle (summon) + ritual success milestone. `_run_ritual`'s trailing
|
||||
# ping/pong (see its docstring comment) guarantees the reward has fully
|
||||
# landed before we get here.
|
||||
from app.inventory import SUMMON_ESSENCE_TRICKLE
|
||||
|
||||
db_session.expire_all()
|
||||
user = await db_session.get(User, user_id)
|
||||
assert user.essence == SUMMON_ESSENCE_TRICKLE + RITUAL_SUCCESS_ESSENCE
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ritual_failure_reveals_nothing_and_grants_no_essence(
|
||||
sync_client, db_session, monkeypatch
|
||||
):
|
||||
monkeypatch.setattr(app.ws.judgment, "roll_ritual_success", lambda traits, rng=None: False)
|
||||
|
||||
token = _login(sync_client, "ritual-loser")
|
||||
user_id = uuid.UUID(
|
||||
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
|
||||
)
|
||||
|
||||
with _ws_connect(sync_client, token) as ws:
|
||||
_read_until(ws, "session")
|
||||
_summon(ws)
|
||||
result = _run_ritual(ws)
|
||||
|
||||
assert result["success"] is False
|
||||
assert result["revealed"] is None
|
||||
|
||||
db_session.expire_all()
|
||||
user = await db_session.get(User, user_id)
|
||||
from app.inventory import SUMMON_ESSENCE_TRICKLE
|
||||
|
||||
assert user.essence == SUMMON_ESSENCE_TRICKLE
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ritual_complete_only_fires_after_all_steps(sync_client, monkeypatch):
|
||||
monkeypatch.setattr(app.ws.judgment, "roll_ritual_success", lambda traits, rng=None: True)
|
||||
_login(sync_client, "ritual-partial")
|
||||
|
||||
with _ws_connect(sync_client, sync_client.cookies.get("qm_session")) as ws:
|
||||
_read_until(ws, "session")
|
||||
_summon(ws)
|
||||
ws.send_json({"type": "ritual_start"})
|
||||
ws.send_json({"type": "ritual_step", "step": 1})
|
||||
ws.send_json({"type": "ritual_step", "step": 2})
|
||||
ws.send_json({"type": "ping"})
|
||||
pong = _read_until(ws, "pong")
|
||||
assert pong == {"type": "pong"}
|
||||
# No ritual_complete should have arrived yet (only 2/4 steps done) —
|
||||
# if it had, it'd have been consumed as the "pong" read above
|
||||
# skipped it via _read_until's scan, so assert explicitly by
|
||||
# finishing the remaining steps and confirming exactly one
|
||||
# ritual_complete follows.
|
||||
ws.send_json({"type": "ritual_step", "step": 3})
|
||||
ws.send_json({"type": "ritual_step", "step": 4})
|
||||
result = _read_until(ws, "ritual_complete")
|
||||
assert result["success"] is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fresh_summon_resets_ritual_progress(sync_client, monkeypatch):
|
||||
monkeypatch.setattr(app.ws.judgment, "roll_ritual_success", lambda traits, rng=None: True)
|
||||
_login(sync_client, "ritual-reset")
|
||||
|
||||
with _ws_connect(sync_client, sync_client.cookies.get("qm_session")) as ws:
|
||||
_read_until(ws, "session")
|
||||
_summon(ws)
|
||||
ws.send_json({"type": "ritual_start"})
|
||||
ws.send_json({"type": "ritual_step", "step": 1})
|
||||
ws.send_json({"type": "ritual_step", "step": 2})
|
||||
# A brand-new summon should discard the in-progress ritual — the
|
||||
# next 4 steps on the new entity shouldn't complete after only 2
|
||||
# more (i.e. carry over the old count).
|
||||
_summon(ws)
|
||||
ws.send_json({"type": "ritual_start"})
|
||||
ws.send_json({"type": "ritual_step", "step": 1})
|
||||
ws.send_json({"type": "ritual_step", "step": 2})
|
||||
ws.send_json({"type": "ping"})
|
||||
pong = _read_until(ws, "pong")
|
||||
assert pong == {"type": "pong"}
|
||||
|
||||
|
||||
# --- judgment: trust / banish -----------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_judgment_trust_correct_on_benevolent_entity(sync_client, db_session, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
app.ws.judgment,
|
||||
"judge_verdict",
|
||||
lambda verdict, traits, **kw: judgment_module.JudgmentOutcome(
|
||||
True, 0.05, CORRECT_JUDGMENT_ESSENCE, False, "reward"
|
||||
),
|
||||
)
|
||||
token = _login(sync_client, "truster")
|
||||
user_id = uuid.UUID(
|
||||
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
|
||||
)
|
||||
|
||||
with _ws_connect(sync_client, token) as ws:
|
||||
_read_until(ws, "session")
|
||||
_summon(ws)
|
||||
ws.send_json({"type": "judgment", "verdict": "trust"})
|
||||
result = _read_until(ws, "judgment_result")
|
||||
|
||||
assert result == {
|
||||
"type": "judgment_result",
|
||||
"correct": True,
|
||||
"favor_delta": 0.05,
|
||||
"essence_delta": CORRECT_JUDGMENT_ESSENCE,
|
||||
"at_peace": False,
|
||||
"consequence": "reward",
|
||||
}
|
||||
|
||||
db_session.expire_all()
|
||||
user = await db_session.get(User, user_id)
|
||||
assert user.favor == pytest.approx(0.05)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_judgment_wrong_trust_emits_escalation_consequence(sync_client, db_session, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
app.ws.judgment,
|
||||
"judge_verdict",
|
||||
lambda verdict, traits, **kw: judgment_module.JudgmentOutcome(
|
||||
False, -0.10, 0, False, "escalation"
|
||||
),
|
||||
)
|
||||
token = _login(sync_client, "wrong-truster")
|
||||
user_id = uuid.UUID(
|
||||
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
|
||||
)
|
||||
|
||||
with _ws_connect(sync_client, token) as ws:
|
||||
_read_until(ws, "session")
|
||||
_summon(ws)
|
||||
ws.send_json({"type": "judgment", "verdict": "trust"})
|
||||
result = _read_until(ws, "judgment_result")
|
||||
|
||||
assert result["consequence"] == "escalation"
|
||||
assert result["correct"] is False
|
||||
assert result["favor_delta"] == -0.10
|
||||
|
||||
db_session.expire_all()
|
||||
user = await db_session.get(User, user_id)
|
||||
assert user.favor == pytest.approx(-0.10)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_judgment_favor_clamped_at_negative_one(sync_client, db_session, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
app.ws.judgment,
|
||||
"judge_verdict",
|
||||
lambda verdict, traits, **kw: judgment_module.JudgmentOutcome(
|
||||
False, -0.10, 0, False, "escalation"
|
||||
),
|
||||
)
|
||||
token = _login(sync_client, "favor-floor")
|
||||
user_id = uuid.UUID(
|
||||
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
|
||||
)
|
||||
|
||||
async with app.ws.session_maker() as db:
|
||||
user = await db.get(User, user_id)
|
||||
user.favor = -0.95
|
||||
await db.commit()
|
||||
|
||||
with _ws_connect(sync_client, token) as ws:
|
||||
_read_until(ws, "session")
|
||||
_summon(ws)
|
||||
ws.send_json({"type": "judgment", "verdict": "trust"})
|
||||
result = _read_until(ws, "judgment_result")
|
||||
|
||||
assert result["favor_delta"] == -0.10 # the raw per-event delta, unclamped
|
||||
db_session.expire_all()
|
||||
user = await db_session.get(User, user_id)
|
||||
assert user.favor == pytest.approx(-1.0) # but the stored balance is clamped
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_judgment_favor_clamped_at_positive_one(sync_client, db_session, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
app.ws.judgment,
|
||||
"judge_verdict",
|
||||
lambda verdict, traits, **kw: judgment_module.JudgmentOutcome(
|
||||
True, 0.08, CROSS_OVER_ESSENCE, True, "crossed_over"
|
||||
),
|
||||
)
|
||||
token = _login(sync_client, "favor-ceiling")
|
||||
user_id = uuid.UUID(
|
||||
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
|
||||
)
|
||||
|
||||
async with app.ws.session_maker() as db:
|
||||
user = await db.get(User, user_id)
|
||||
user.favor = 0.97
|
||||
await db.commit()
|
||||
|
||||
with _ws_connect(sync_client, token) as ws:
|
||||
_read_until(ws, "session")
|
||||
_summon(ws)
|
||||
ws.send_json({"type": "judgment", "verdict": "cross_over"})
|
||||
_read_until(ws, "judgment_result")
|
||||
|
||||
db_session.expire_all()
|
||||
user = await db_session.get(User, user_id)
|
||||
assert user.favor == pytest.approx(1.0)
|
||||
|
||||
|
||||
# --- judgment: cross_over / at_peace ----------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_judgment_cross_over_correct_sets_at_peace_and_pays_most_essence(
|
||||
sync_client, db_session, monkeypatch
|
||||
):
|
||||
monkeypatch.setattr(
|
||||
app.ws.judgment,
|
||||
"judge_verdict",
|
||||
lambda verdict, traits, **kw: judgment_module.JudgmentOutcome(
|
||||
True, 0.08, CROSS_OVER_ESSENCE, True, "crossed_over"
|
||||
),
|
||||
)
|
||||
token = _login(sync_client, "crosser")
|
||||
user_id = uuid.UUID(
|
||||
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
|
||||
)
|
||||
|
||||
with _ws_connect(sync_client, token) as ws:
|
||||
_read_until(ws, "session")
|
||||
entity_frame = _summon(ws)
|
||||
entity_id = uuid.UUID(entity_frame["entity"]["id"])
|
||||
ws.send_json({"type": "judgment", "verdict": "cross_over"})
|
||||
result = _read_until(ws, "judgment_result")
|
||||
|
||||
assert result["consequence"] == "crossed_over"
|
||||
assert result["at_peace"] is True
|
||||
assert result["essence_delta"] == CROSS_OVER_ESSENCE
|
||||
|
||||
db_session.expire_all()
|
||||
entity = await db_session.get(Entity, entity_id)
|
||||
assert entity.at_peace is True
|
||||
|
||||
from app.inventory import SUMMON_ESSENCE_TRICKLE
|
||||
|
||||
user = await db_session.get(User, user_id)
|
||||
assert user.essence == SUMMON_ESSENCE_TRICKLE + CROSS_OVER_ESSENCE
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_judgment_cross_over_resisted_on_demon_has_no_effect(sync_client, db_session, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
app.ws.judgment,
|
||||
"judge_verdict",
|
||||
lambda verdict, traits, **kw: judgment_module.JudgmentOutcome(
|
||||
False, 0.0, 0, False, "resisted"
|
||||
),
|
||||
)
|
||||
token = _login(sync_client, "resisted-demon")
|
||||
user_id = uuid.UUID(
|
||||
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
|
||||
)
|
||||
|
||||
with _ws_connect(sync_client, token) as ws:
|
||||
_read_until(ws, "session")
|
||||
entity_frame = _summon(ws)
|
||||
entity_id = uuid.UUID(entity_frame["entity"]["id"])
|
||||
ws.send_json({"type": "judgment", "verdict": "cross_over"})
|
||||
result = _read_until(ws, "judgment_result")
|
||||
|
||||
assert result == {
|
||||
"type": "judgment_result",
|
||||
"correct": False,
|
||||
"favor_delta": 0.0,
|
||||
"essence_delta": 0,
|
||||
"at_peace": False,
|
||||
"consequence": "resisted",
|
||||
}
|
||||
|
||||
db_session.expire_all()
|
||||
entity = await db_session.get(Entity, entity_id)
|
||||
assert entity.at_peace is False
|
||||
from app.inventory import SUMMON_ESSENCE_TRICKLE
|
||||
|
||||
user = await db_session.get(User, user_id)
|
||||
assert user.essence == SUMMON_ESSENCE_TRICKLE
|
||||
assert user.favor == 0.0
|
||||
|
||||
|
||||
# --- judgment: test ----------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_judgment_test_verdict_is_always_allowed_and_neutral(sync_client, db_session):
|
||||
token = _login(sync_client, "tester")
|
||||
user_id = uuid.UUID(
|
||||
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
|
||||
)
|
||||
|
||||
with _ws_connect(sync_client, token) as ws:
|
||||
_read_until(ws, "session")
|
||||
_summon(ws)
|
||||
# No ritual attempted at all — should not crash, just no effect.
|
||||
ws.send_json({"type": "judgment", "verdict": "test"})
|
||||
result = _read_until(ws, "judgment_result")
|
||||
|
||||
assert result == {
|
||||
"type": "judgment_result",
|
||||
"correct": False,
|
||||
"favor_delta": 0,
|
||||
"essence_delta": 0,
|
||||
"at_peace": False,
|
||||
"consequence": "neutral",
|
||||
}
|
||||
|
||||
db_session.expire_all()
|
||||
user = await db_session.get(User, user_id)
|
||||
assert user.favor == 0.0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_judgment_test_verdict_correct_after_successful_ritual(sync_client, monkeypatch):
|
||||
monkeypatch.setattr(app.ws.judgment, "roll_ritual_success", lambda traits, rng=None: True)
|
||||
_login(sync_client, "tester-after-ritual")
|
||||
|
||||
with _ws_connect(sync_client, sync_client.cookies.get("qm_session")) as ws:
|
||||
_read_until(ws, "session")
|
||||
_summon(ws)
|
||||
_run_ritual(ws)
|
||||
ws.send_json({"type": "judgment", "verdict": "test"})
|
||||
result = _read_until(ws, "judgment_result")
|
||||
|
||||
assert result["correct"] is True
|
||||
assert result["consequence"] == "neutral"
|
||||
assert result["favor_delta"] == 0
|
||||
assert result["essence_delta"] == 0
|
||||
|
||||
|
||||
# --- at_peace + re-contact ----------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_at_peace_entity_is_not_recontacted_a_fresh_one_mints_instead(
|
||||
sync_client, db_session
|
||||
):
|
||||
anomalies = [
|
||||
{"type": "anomaly", "source": "radio", "frequency": 101.0 + i, "magnitude": 5.0 + i}
|
||||
for i in range(4)
|
||||
]
|
||||
|
||||
token = _login(sync_client, "peace-seeker")
|
||||
with _ws_connect(sync_client, token) as ws:
|
||||
_read_until(ws, "session")
|
||||
for anomaly in anomalies:
|
||||
ws.send_json(anomaly)
|
||||
first_frame = _read_until(ws, "entity")
|
||||
first_id = uuid.UUID(first_frame["entity"]["id"])
|
||||
first_signature = None # not exposed to the client; fetched below
|
||||
|
||||
# Manually mark the entity at_peace, as a completed cross_over would.
|
||||
async with app.ws.session_maker() as db:
|
||||
entity = await db.get(Entity, first_id)
|
||||
entity.at_peace = True
|
||||
first_signature = entity.signature
|
||||
await db.commit()
|
||||
|
||||
with _ws_connect(sync_client, token) as ws:
|
||||
_read_until(ws, "session")
|
||||
for anomaly in anomalies:
|
||||
ws.send_json(anomaly)
|
||||
second_frame = _read_until(ws, "entity")
|
||||
|
||||
assert second_frame["is_new"] is True
|
||||
second_id = uuid.UUID(second_frame["entity"]["id"])
|
||||
assert second_id != first_id
|
||||
|
||||
db_session.expire_all()
|
||||
second_entity = await db_session.get(Entity, second_id)
|
||||
assert second_entity.at_peace is False
|
||||
# Salted variant of the same base signature, not a raw collision.
|
||||
assert second_entity.signature != first_signature
|
||||
assert second_entity.signature.startswith(first_signature + ":")
|
||||
|
||||
first_entity = await db_session.get(Entity, first_id)
|
||||
assert first_entity is not None # memorialized, never deleted
|
||||
assert first_entity.at_peace is True
|
||||
|
||||
|
||||
# --- favor read-back bias on trait rolls at mint time -----------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_high_favor_user_mints_less_volatile_deceptive_entities(sync_client, db_session):
|
||||
token = _login(sync_client, "high-favor-summoner")
|
||||
user_id = uuid.UUID(
|
||||
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
|
||||
)
|
||||
async with app.ws.session_maker() as db:
|
||||
user = await db.get(User, user_id)
|
||||
user.favor = 1.0
|
||||
await db.commit()
|
||||
|
||||
anomalies = [
|
||||
{"type": "anomaly", "source": "radio", "frequency": 201.0 + i, "magnitude": 5.0 + i}
|
||||
for i in range(4)
|
||||
]
|
||||
with _ws_connect(sync_client, token) as ws:
|
||||
_read_until(ws, "session")
|
||||
for anomaly in anomalies:
|
||||
ws.send_json(anomaly)
|
||||
entity_frame = _read_until(ws, "entity")
|
||||
|
||||
entity_id = uuid.UUID(entity_frame["entity"]["id"])
|
||||
db_session.expire_all()
|
||||
entity = await db_session.get(Entity, entity_id)
|
||||
|
||||
# Recompute what the unbiased roll would have been for this signature
|
||||
# and confirm the stored traits were nudged toward more legible
|
||||
# (lower volatility/deceptiveness), never the other direction.
|
||||
from app.entities import roll_traits
|
||||
|
||||
unbiased = roll_traits(entity.signature)
|
||||
assert entity.traits["volatility"] <= unbiased["volatility"]
|
||||
assert entity.traits["deceptiveness"] <= unbiased["deceptiveness"]
|
||||
# alignment/power are untouched by the bias.
|
||||
assert entity.traits["alignment"] == pytest.approx(unbiased["alignment"])
|
||||
assert entity.traits["power"] == pytest.approx(unbiased["power"])
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_low_favor_user_mints_more_volatile_deceptive_entities(sync_client, db_session):
|
||||
token = _login(sync_client, "low-favor-summoner")
|
||||
user_id = uuid.UUID(
|
||||
sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"]
|
||||
)
|
||||
async with app.ws.session_maker() as db:
|
||||
user = await db.get(User, user_id)
|
||||
user.favor = -1.0
|
||||
await db.commit()
|
||||
|
||||
anomalies = [
|
||||
{"type": "anomaly", "source": "radio", "frequency": 301.0 + i, "magnitude": 5.0 + i}
|
||||
for i in range(4)
|
||||
]
|
||||
with _ws_connect(sync_client, token) as ws:
|
||||
_read_until(ws, "session")
|
||||
for anomaly in anomalies:
|
||||
ws.send_json(anomaly)
|
||||
entity_frame = _read_until(ws, "entity")
|
||||
|
||||
entity_id = uuid.UUID(entity_frame["entity"]["id"])
|
||||
db_session.expire_all()
|
||||
entity = await db_session.get(Entity, entity_id)
|
||||
|
||||
from app.entities import roll_traits
|
||||
|
||||
unbiased = roll_traits(entity.signature)
|
||||
assert entity.traits["volatility"] >= unbiased["volatility"]
|
||||
assert entity.traits["deceptiveness"] >= unbiased["deceptiveness"]
|
||||
Reference in New Issue
Block a user