""" R2-B: approval-token + per-voter vote round-trip tests. - Pure unit tests for `app.services.approval` (no DB). - DB-backed tests for `consume_token` single-use + the two vote routes. """ from __future__ import annotations import time import uuid from datetime import date, timedelta import pytest from fastapi import HTTPException from app.services import approval as approval_service # --------------------------------------------------------------------------- # Pure-unit token tests. # --------------------------------------------------------------------------- def test_issue_and_verify_token_roundtrip(): item_id = uuid.uuid4() voter_id = uuid.uuid4() token = approval_service.issue_token(item_id, voter_id) payload = approval_service.verify_token(token) assert payload["item"] == str(item_id) assert payload["voter"] == str(voter_id) def test_verify_rejects_tampered_token(): item_id = uuid.uuid4() voter_id = uuid.uuid4() token = approval_service.issue_token(item_id, voter_id) tampered = token[:-2] + ("AA" if not token.endswith("AA") else "BB") with pytest.raises(HTTPException) as excinfo: approval_service.verify_token(tampered) assert excinfo.value.status_code == 401 def test_verify_rejects_expired_token(monkeypatch): """itsdangerous reads `time.time()` — patch the time module's `time` attribute on `itsdangerous.timed` to simulate clock advance. """ import itsdangerous.timed as _timed_mod item_id = uuid.uuid4() voter_id = uuid.uuid4() # Issue at "now". token = approval_service.issue_token(item_id, voter_id) # Advance the clock 8 days past issue time (default TTL is 7 days). real_now = time.time() fake_now = real_now + (8 * 24 * 3600) class _FakeTime: @staticmethod def time(): return fake_now monkeypatch.setattr(_timed_mod, "time", _FakeTime) with pytest.raises(HTTPException) as excinfo: approval_service.verify_token(token) assert excinfo.value.status_code == 401 # --------------------------------------------------------------------------- # DB-backed fixtures: build a minimal scenario reused by several tests. # --------------------------------------------------------------------------- @pytest.fixture() def scenario(db): """Create profile + 2 voters + recipe + plan + item, all in one shot. Uses the per-test transactional session, so changes roll back on teardown (no cross-test pollution). """ from app.models import ( FamilyMember, FamilyMemberRole, FamilyProfile, MealPlan, MealPlanItem, MealPlanStatus, MealType, Recipe, ) suffix = uuid.uuid4().hex[:8] profile = FamilyProfile( name=f"Test Family {suffix}", household_size=2, adult_count=2, child_count=0, ) db.add(profile) db.flush() voter_a = FamilyMember( family_profile_id=profile.id, name="Alice", email=f"alice+{suffix}@example.com", role=FamilyMemberRole.ADULT, ) voter_b = FamilyMember( family_profile_id=profile.id, name="Bob", email=f"bob+{suffix}@example.com", role=FamilyMemberRole.ADULT, ) db.add_all([voter_a, voter_b]) db.flush() recipe = Recipe( family_profile_id=profile.id, name=f"Test Pasta {suffix}", servings=2, ingredients=[{"name": "pasta", "qty": "200g"}], instructions=["Boil", "Drain"], is_manually_added=True, ) db.add(recipe) db.flush() plan = MealPlan( family_profile_id=profile.id, week_start_date=date.today() + timedelta(days=14), status=MealPlanStatus.DRAFT, ) db.add(plan) db.flush() item = MealPlanItem( meal_plan_id=plan.id, recipe_id=recipe.id, day_of_week=1, meal_type=MealType.DINNER, ) db.add(item) db.flush() return { "profile": profile, "voter_a": voter_a, "voter_b": voter_b, "recipe": recipe, "plan": plan, "item": item, } # --------------------------------------------------------------------------- # DB-backed tests. # --------------------------------------------------------------------------- @pytest.mark.requires_postgres def test_consume_token_single_use(db, scenario): from app.models import MealPlanVote item = scenario["item"] voter = scenario["voter_a"] token = approval_service.issue_token(item.id, voter.id) # First call: succeeds, returns the voter. out = approval_service.consume_token(db, token, item.id) assert out.id == voter.id # Simulate the route writing the vote row (consume_token itself does NOT # write — single-use is enforced by the presence of the vote row). db.add(MealPlanVote( meal_plan_item_id=item.id, family_member_id=voter.id, vote=True, )) db.flush() # Second call: raises 409. with pytest.raises(HTTPException) as excinfo: approval_service.consume_token(db, token, item.id) assert excinfo.value.status_code == 409 @pytest.mark.requires_postgres def test_vote_get_renders_html(client, db, scenario): item = scenario["item"] voter = scenario["voter_a"] token = approval_service.issue_token(item.id, voter.id) r = client.get(f"/api/meals/vote/{item.id}", params={"token": token}) assert r.status_code == 200, r.text assert "text/html" in r.headers.get("content-type", "") # Voter name appears in the visible body. assert "Alice" in r.text # Token is allowed inside the form `action` attribute (the link's href) # but must not appear in the visible meal-card text. body = r.text card_start = body.find('
') card_end = body.find("
", card_start) assert card_start != -1 and card_end != -1 visible_card = body[card_start:card_end] assert token not in visible_card @pytest.mark.requires_postgres def test_vote_post_records_and_decides(client, db, scenario): """First voter approves -> still pending (Bob hasn't voted). Then Bob approves -> approved. """ item = scenario["item"] voter_a = scenario["voter_a"] voter_b = scenario["voter_b"] token_a = approval_service.issue_token(item.id, voter_a.id) r1 = client.post( f"/api/meals/vote/{item.id}", params={"token": token_a}, json={"vote": "approve"}, ) assert r1.status_code == 200, r1.text body1 = r1.json() assert body1["status"] == "recorded" assert body1["item_status"] in ("pending", "approved") # With 2 voters, after 1 approve we should still be pending. assert body1["item_status"] == "pending" token_b = approval_service.issue_token(item.id, voter_b.id) r2 = client.post( f"/api/meals/vote/{item.id}", params={"token": token_b}, json={"vote": "approve"}, ) assert r2.status_code == 200, r2.text body2 = r2.json() assert body2["item_status"] == "approved"