"""Strategic-log feedback service + token helpers. Covers the pure service path (set_vote, get_counts, clear) plus the token sign/verify round-trip. The web endpoint POST /api/log/{id}/feedback is covered in tests/test_api_feedback.py (separate file because it needs the full FastAPI + auth stack).""" from __future__ import annotations import asyncio def _build_db(tmp_path): from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app import db as db_mod from app.db import Base import app.models # noqa: F401 engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/fb.db") factory = async_sessionmaker(engine, expire_on_commit=False) db_mod._engine = engine db_mod._session_factory = factory async def _seed(): from app.models import StrategicLog, User from app.db import utcnow async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) async with factory() as s: s.add(User(id=1, email="alice@example.com", tier="free", settings_json={}, created_at=utcnow())) s.add(User(id=2, email="bob@example.com", tier="free", settings_json={}, created_at=utcnow())) s.add(StrategicLog( id=10, generated_at=utcnow(), model="m", anchor_date=None, prompt_version=1, tone="INTERMEDIATE", analysis="DRY", content="x", prompt_tokens=1, completion_tokens=1, cost_usd=0.0, )) await s.commit() asyncio.run(_seed()) return factory # --------------------------------------------------------------------------- # Service: set_vote / get_counts # --------------------------------------------------------------------------- def test_set_vote_up_writes_one_row(tmp_path): factory = _build_db(tmp_path) async def _go(): from app.services.log_feedback import set_vote async with factory() as s: counts = await set_vote(s, log_id=10, user_id=1, vote="up") return counts counts = asyncio.run(_go()) assert counts.up == 1 assert counts.down == 0 assert counts.user_vote == "up" def test_set_vote_flip_replaces_existing(tmp_path): """Voting again with a different value flips the existing row; no duplicate is written, the up/down counts swap.""" factory = _build_db(tmp_path) async def _go(): from app.services.log_feedback import set_vote async with factory() as s: await set_vote(s, log_id=10, user_id=1, vote="up") return await set_vote(s, log_id=10, user_id=1, vote="down") counts = asyncio.run(_go()) assert counts.up == 0 assert counts.down == 1 assert counts.user_vote == "down" def test_set_vote_clear_removes_row(tmp_path): factory = _build_db(tmp_path) async def _go(): from app.services.log_feedback import set_vote, get_counts async with factory() as s: await set_vote(s, log_id=10, user_id=1, vote="up") await set_vote(s, log_id=10, user_id=1, vote="clear") return await get_counts(s, log_id=10, user_id=1) counts = asyncio.run(_go()) assert counts.up == 0 assert counts.down == 0 assert counts.user_vote is None def test_aggregate_counts_across_users(tmp_path): """Two distinct users vote — counts aggregate; each user's own_vote field reflects only their own row.""" factory = _build_db(tmp_path) async def _go(): from app.services.log_feedback import set_vote, get_counts async with factory() as s: await set_vote(s, log_id=10, user_id=1, vote="up") await set_vote(s, log_id=10, user_id=2, vote="down") alice_view = await get_counts(s, log_id=10, user_id=1) bob_view = await get_counts(s, log_id=10, user_id=2) return alice_view, bob_view alice, bob = asyncio.run(_go()) assert alice.up == 1 and alice.down == 1 and alice.user_vote == "up" assert bob.up == 1 and bob.down == 1 and bob.user_vote == "down" def test_invalid_vote_rejected(tmp_path): factory = _build_db(tmp_path) async def _go(): from app.services.log_feedback import FeedbackError, set_vote async with factory() as s: try: await set_vote(s, log_id=10, user_id=1, vote="meh") except FeedbackError as e: return str(e) return "no error" msg = asyncio.run(_go()) assert "up" in msg.lower() and "down" in msg.lower() # --------------------------------------------------------------------------- # Token sign / verify # --------------------------------------------------------------------------- def test_feedback_token_round_trips(monkeypatch): """A signed token decodes back to the same (user, log, vote) tuple.""" monkeypatch.setenv("CASSANDRA_SESSION_SECRET", "test-secret-32-chars-long-okay") from app.config import get_settings get_settings.cache_clear() from app.services.log_feedback import sign_feedback_token, verify_feedback_token tok = sign_feedback_token(user_id=42, log_id=99, vote="up") payload = verify_feedback_token(tok) assert payload == {"user_id": 42, "log_id": 99, "vote": "up"} def test_feedback_token_tampered_returns_none(monkeypatch): monkeypatch.setenv("CASSANDRA_SESSION_SECRET", "test-secret-32-chars-long-okay") from app.config import get_settings get_settings.cache_clear() from app.services.log_feedback import sign_feedback_token, verify_feedback_token tok = sign_feedback_token(user_id=42, log_id=99, vote="up") tampered = tok[:-1] + ("a" if tok[-1] != "a" else "b") assert verify_feedback_token(tampered) is None def test_feedback_token_garbage_returns_none(monkeypatch): monkeypatch.setenv("CASSANDRA_SESSION_SECRET", "test-secret-32-chars-long-okay") from app.config import get_settings get_settings.cache_clear() from app.services.log_feedback import verify_feedback_token assert verify_feedback_token("not.a.real.token") is None assert verify_feedback_token("") is None def test_feedback_token_clear_is_not_a_valid_email_link_vote(monkeypatch): """The 'clear' sentinel is a web-only path; the email link can only apply a positive vote (up or down). Trying to sign 'clear' raises.""" monkeypatch.setenv("CASSANDRA_SESSION_SECRET", "test-secret-32-chars-long-okay") from app.config import get_settings get_settings.cache_clear() from app.services.log_feedback import FeedbackError, sign_feedback_token try: sign_feedback_token(user_id=1, log_id=10, vote="clear") except FeedbackError: return raise AssertionError("expected FeedbackError")