MPS-21: offer views — open, page, counter, accept/decline, withdraw

Routes:
- POST /p/{product_id}/offer  open new offer (login required)
- GET  /o/{offer_id}          offer detail page (buyer + seller only)
- POST /o/{offer_id}/counter  counter the current amount
- POST /o/{offer_id}/accept   accept current amount (terminal)
- POST /o/{offer_id}/decline  decline current amount (terminal)
- POST /o/{offer_id}/withdraw buyer-only terminal pull

offer_open registered before product_slug catch-all so /p/{id}/offer is
not shadowed (same shadowing rule as auction routes).

views/offer.py:
- _user_party resolves request.user → OFFER_PARTY_BUYER or _SELLER (or
  None for third party), gates offer detail page accordingly
- _serialize_offer: dict shape used by both template ctx and JSON;
  events array carries human-readable event_human
- offer_open enforces product.offers_allowed and self-offer block
  (buyer is shop owner)
- _offer_action wraps counter/accept/decline/withdraw, parses amount
  for counter, blocks non-buyer withdraw
- All write actions return JSON {ok, offer} or {error}

templates/offer.j2:
- Live page with state badge, current amount, both messages, action
  forms (accept / counter / decline; withdraw for buyer only) when it's
  the actor's turn
- Event timeline rendered from offer.events

Functional tests (12): anon redirected to login, seller cannot offer
on own product (403), buyer queues mid-range offer, auto-accept high
offer, auto-decline low offer, 404 unknown offer id, third-party 404,
full negotiation (open → counter → accept), wrong-party counter
rejected (400), buyer withdraw, seller cannot withdraw (403), offers
disabled on fixed-price product (403).

Cart integration deferred to commit 7 (bundling auction + offer).

Total: 904 tests pass (was 892 + 12).
This commit is contained in:
russell@unturf.com 2026-05-09 19:43:42 -04:00
parent f34e7e9ec5
commit 9b612182b8
No known key found for this signature in database
4 changed files with 555 additions and 0 deletions

View file

@ -183,6 +183,8 @@ def includeme(config):
config.add_route("product_edit", "/p/{product_id}/edit")
config.add_route("product_edit2", "/p/{product_id}/{slug:.*}/edit")
# Offer entry point — must register before product_slug catch-all (MPS-21).
config.add_route("offer_open", "/p/{product_id}/offer")
config.add_route("product_slug", "/p/{product_id}/{slug:.*}")
# media player routes
@ -244,3 +246,11 @@ def includeme(config):
config.add_route("auction_buy_now", "/a/{auction_id}/buy-now")
config.add_route("auction_watch", "/a/{auction_id}/watch")
config.add_route("auction_page", "/a/{auction_id}")
# Offers (MPS-21). offer_open is registered earlier near product routes
# to avoid shadow by product_slug. /o/{id}/* sub-paths first, bare last.
config.add_route("offer_counter", "/o/{offer_id}/counter")
config.add_route("offer_accept", "/o/{offer_id}/accept")
config.add_route("offer_decline", "/o/{offer_id}/decline")
config.add_route("offer_withdraw", "/o/{offer_id}/withdraw")
config.add_route("offer_page", "/o/{offer_id}")

View file

@ -0,0 +1,76 @@
{% extends "base.j2" -%}
{%- block append_to_head_tag_section %}
<title>Offer — {{ product_title }}</title>
{%- endblock %}
{% block content %}
<section class="one-column offer-page" data-offer-id="{{ id }}">
<div class="offer-header well">
<h1 class="type-title">Offer for "{{ product_title }}"</h1>
<div class="offer-state-badge offer-state-{{ state }}">{{ state_human }}</div>
<p>
<strong>Current amount:</strong> ${{ "%.2f"|format(current_amount) }}
&middot;
Round {{ round_count }}
</p>
{% if buyer_message %}
<p><em>Buyer:</em> {{ buyer_message }}</p>
{% endif %}
{% if seller_message %}
<p><em>Seller:</em> {{ seller_message }}</p>
{% endif %}
</div>
{% if can_act %}
<section class="offer-actions well">
<h3 class="type-title">Your turn</h3>
<form method="post" action="/o/{{ id }}/accept">
<input type="text" name="message" placeholder="Optional message" />
<input type="submit" class="mps-submit" value="Accept ${{ "%.2f"|format(current_amount) }}" />
</form>
<form method="post" action="/o/{{ id }}/counter" style="margin-top: var(--space-3);">
<label>Counter offer (USD)</label>
<input type="number" step="0.01" min="0.01" name="amount" required />
<input type="text" name="message" placeholder="Optional message" />
<input type="submit" class="mps-button" value="Counter" />
</form>
<form method="post" action="/o/{{ id }}/decline" style="margin-top: var(--space-3);">
<input type="text" name="message" placeholder="Reason (optional)" />
<input type="submit" class="mps-button" value="Decline" />
</form>
{% if actor_party == 0 %}
<form method="post" action="/o/{{ id }}/withdraw" style="margin-top: var(--space-3);">
<input type="submit" class="mps-button" value="Withdraw offer" />
</form>
{% endif %}
</section>
{% elif is_open %}
<section class="well">
<p><em>Waiting on the other party.</em></p>
</section>
{% endif %}
<section class="offer-timeline well">
<h3 class="type-title">History</h3>
<ol class="offer-events">
{% for e in events %}
<li class="offer-event">
<strong>{{ e.event_human }}</strong>
{% if e.actor_email %} by {{ e.actor_email }}{% else %} (system){% endif %}
{% if e.amount is not none %} &middot; ${{ "%.2f"|format(e.amount) }}{% endif %}
{% if e.message %} &mdash; <em>{{ e.message }}</em>{% endif %}
</li>
{% endfor %}
</ol>
</section>
</section>
{% endblock %}

View file

@ -5703,3 +5703,242 @@ class TestAuctionRoutes(_AuthenticatedBase):
u2 = get_or_create_user_by_email(self.dbsession, "test2@example.com")
self.assertEqual(auction.state, AUCTION_STATE_ENDED)
self.assertEqual(auction.winner, u2)
class TestOfferRoutes(_AuthenticatedBase):
"""MPS-21: HTTP-level coverage for offer views."""
def _make_offer_product(self, list_price=10000, offer_enabled=True,
auto_accept_pct=95, auto_decline_pct=50,
owner_creds=None):
"""Shop owner = user1 by default; product accepts offers."""
from ..models.product import Product
if owner_creds is None:
owner_creds = self.user1_creds
shop = self._create_shop_helper(user_creds=owner_creds)
# Configure shop's offer settings via direct ORM (faster than UI flow).
shop.offer_enabled = offer_enabled
shop.offer_auto_accept_threshold_pct = auto_accept_pct
shop.offer_auto_decline_threshold_pct = auto_decline_pct
self.dbsession.add(shop)
product = Product(title="Negotiable thing", description="...")
product.shop = shop
product.price_in_cents = list_price
product.is_physical = False
product.is_sellable = True
product.pricing_mode = 3 # offer mode
self.dbsession.add(product)
self.dbsession.flush()
product_id = product.uuid_str
transaction.commit()
return product_id
def test_open_offer_anon_redirected(self):
product_id = self._make_offer_product()
self.testapp.get("/log-out")
# user_required redirects with flash; should not reach 200.
res = self.testapp.post(
f"/p/{product_id}/offer",
{"amount": "70.00"},
expect_errors=True,
)
self.assertIn(res.status_int, (302, 303, 401, 403))
def test_seller_cannot_offer_on_own_product(self):
product_id = self._make_offer_product()
# user1 (shop owner) is logged in.
res = self.testapp.post(
f"/p/{product_id}/offer",
{"amount": "70.00"},
expect_errors=True,
)
self.assertEqual(res.status_int, 403)
def test_buyer_opens_queued_offer(self):
product_id = self._make_offer_product(list_price=10000)
self.testapp.get("/log-out")
self.log_in_user(self.user2_creds)
# 70% of list = $70 — between auto-decline (50%) and auto-accept (95%).
res = self.testapp.post(
f"/p/{product_id}/offer",
{"amount": "70.00", "message": "any room?"},
status=200,
)
self.assertTrue(res.json["ok"])
offer = res.json["offer"]
self.assertEqual(offer["state"], 0) # PENDING
self.assertEqual(offer["state_human"], "Pending")
self.assertEqual(offer["current_amount_in_cents"], 7000)
self.assertEqual(offer["round_count"], 0)
def test_auto_accept_high_offer(self):
product_id = self._make_offer_product(list_price=10000)
self.testapp.get("/log-out")
self.log_in_user(self.user2_creds)
# 96% of list → above 95% auto-accept threshold.
res = self.testapp.post(
f"/p/{product_id}/offer",
{"amount": "96.00"},
status=200,
)
self.assertEqual(res.json["offer"]["state"], 1) # ACCEPTED
def test_auto_decline_low_offer(self):
product_id = self._make_offer_product(list_price=10000)
self.testapp.get("/log-out")
self.log_in_user(self.user2_creds)
# 40% of list → below 50% auto-decline threshold.
res = self.testapp.post(
f"/p/{product_id}/offer",
{"amount": "40.00"},
status=200,
)
self.assertEqual(res.json["offer"]["state"], 3) # DECLINED
def test_offer_page_404_unknown(self):
self.testapp.get(
"/o/00000000000000000000000000000000",
status=404,
)
def test_offer_page_third_party_blocked(self):
product_id = self._make_offer_product()
self.testapp.get("/log-out")
self.log_in_user(self.user2_creds)
# Open as user2 (buyer).
res = self.testapp.post(
f"/p/{product_id}/offer",
{"amount": "70.00"},
status=200,
)
offer_id = res.json["offer_id"]
# Create a third user not involved.
from ..models.user import get_or_create_user_by_email
third = get_or_create_user_by_email(self.dbsession, "third@example.com")
third_password = third.new_password()
self.dbsession.add(third)
self.dbsession.flush()
transaction.commit()
third_creds = ("third@example.com", third_password)
self.testapp.get("/log-out")
self.log_in_user(third_creds)
self.testapp.get(f"/o/{offer_id}", status=404)
def test_full_negotiation(self):
product_id = self._make_offer_product(list_price=10000)
# Buyer opens offer.
self.testapp.get("/log-out")
self.log_in_user(self.user2_creds)
res = self.testapp.post(
f"/p/{product_id}/offer",
{"amount": "70.00"},
status=200,
)
offer_id = res.json["offer_id"]
# Seller (user1) counters.
self.testapp.get("/log-out")
self.log_in_user(self.user1_creds)
res2 = self.testapp.post(
f"/o/{offer_id}/counter",
{"amount": "85.00", "message": "how about this?"},
status=200,
)
self.assertTrue(res2.json["ok"])
self.assertEqual(res2.json["offer"]["current_amount_in_cents"], 8500)
self.assertEqual(res2.json["offer"]["round_count"], 1)
self.assertEqual(res2.json["offer"]["state"], 2) # COUNTERED
# Buyer accepts.
self.testapp.get("/log-out")
self.log_in_user(self.user2_creds)
res3 = self.testapp.post(
f"/o/{offer_id}/accept",
status=200,
)
self.assertEqual(res3.json["offer"]["state"], 1) # ACCEPTED
def test_wrong_party_counter_rejected(self):
product_id = self._make_offer_product()
self.testapp.get("/log-out")
self.log_in_user(self.user2_creds)
res = self.testapp.post(
f"/p/{product_id}/offer",
{"amount": "70.00"},
status=200,
)
offer_id = res.json["offer_id"]
# Buyer tries to counter their own offer when it's seller's turn.
res2 = self.testapp.post(
f"/o/{offer_id}/counter",
{"amount": "75.00"},
expect_errors=True,
)
self.assertEqual(res2.status_int, 400)
def test_buyer_withdraw(self):
product_id = self._make_offer_product()
self.testapp.get("/log-out")
self.log_in_user(self.user2_creds)
res = self.testapp.post(
f"/p/{product_id}/offer",
{"amount": "70.00"},
status=200,
)
offer_id = res.json["offer_id"]
res2 = self.testapp.post(
f"/o/{offer_id}/withdraw",
status=200,
)
self.assertEqual(res2.json["offer"]["state"], 5) # WITHDRAWN
def test_seller_cannot_withdraw(self):
product_id = self._make_offer_product()
self.testapp.get("/log-out")
self.log_in_user(self.user2_creds)
res = self.testapp.post(
f"/p/{product_id}/offer",
{"amount": "70.00"},
status=200,
)
offer_id = res.json["offer_id"]
self.testapp.get("/log-out")
self.log_in_user(self.user1_creds)
res2 = self.testapp.post(
f"/o/{offer_id}/withdraw",
expect_errors=True,
)
self.assertEqual(res2.status_int, 403)
def test_offers_disabled_on_fixed_price_product(self):
# Create product with pricing_mode=0 (fixed). Offer endpoint should 403.
from ..models.product import Product
shop = self._create_shop_helper()
product = Product(title="Fixed price", description="...")
product.shop = shop
product.price_in_cents = 5000
product.is_physical = False
product.is_sellable = True
product.pricing_mode = 0 # fixed, not offer
self.dbsession.add(product)
self.dbsession.flush()
product_id = product.uuid_str
transaction.commit()
self.testapp.get("/log-out")
self.log_in_user(self.user2_creds)
res = self.testapp.post(
f"/p/{product_id}/offer",
{"amount": "30.00"},
expect_errors=True,
)
self.assertEqual(res.status_int, 403)

View file

@ -0,0 +1,230 @@
"""Offer views — MPS-21.
Routes:
POST /p/{product_id}/offer open a new offer (login required)
GET /o/{offer_id} offer detail page (buyer + seller + admin)
POST /o/{offer_id}/counter counter the current amount
POST /o/{offer_id}/accept accept current amount (terminal)
POST /o/{offer_id}/decline decline current amount (terminal)
POST /o/{offer_id}/withdraw buyer pulls offer (terminal)
Identity resolution lives in this module: the pure logic in lib/offer.py
takes actor_party (BUYER vs SELLER) and does not know who owns the shop.
View handlers translate request.user into a party and reject self-offers
or third-party access.
"""
from pyramid.httpexceptions import HTTPFound, HTTPNotFound
from pyramid.view import view_config
from ..lib.offer import (
OfferRejected,
open_offer,
counter_offer,
accept_offer,
decline_offer,
withdraw_offer,
)
from ..lib.currency import cents_to_dollars
from ..models.offer import (
OFFER_PARTY_BUYER,
OFFER_PARTY_SELLER,
OFFER_EVENT_INT_TO_HUMAN,
get_offer_by_id,
)
from ..models.product import get_product_by_id
from ..views import user_required
def _user_party(request, offer):
"""Return (party_int, can_view).
party_int: OFFER_PARTY_BUYER if user is the buyer,
OFFER_PARTY_SELLER if user is in shop.owners,
None if neither.
can_view: True for buyer or seller; False otherwise.
"""
user = request.user
if user is None:
return (None, False)
if user == offer.buyer:
return (OFFER_PARTY_BUYER, True)
if user in offer.shop.owners:
return (OFFER_PARTY_SELLER, True)
return (None, False)
def _serialize_offer(offer):
return {
"id": offer.uuid_str,
"product_id": offer.product.uuid_str,
"product_title": offer.product.title,
"shop_id": offer.shop.uuid_str,
"buyer_id": offer.buyer.uuid_str,
"buyer_email": offer.buyer.email,
"state": offer.state,
"state_human": offer.state_human,
"is_open": offer.is_open,
"is_terminal": offer.is_terminal,
"is_paid": offer.is_paid,
"current_amount_in_cents": offer.current_amount_in_cents,
"current_amount": offer.current_amount,
"current_party": offer.current_party,
"round_count": offer.round_count,
"expires_timestamp": offer.expires_timestamp,
"time_remaining_ms": offer.time_remaining_ms,
"buyer_message": offer.buyer_message,
"seller_message": offer.seller_message,
"events": [
{
"event_type": e.event_type,
"event_human": OFFER_EVENT_INT_TO_HUMAN.get(
e.event_type, "Unknown"
),
"actor_email": e.actor.email if e.actor else None,
"amount_in_cents": e.amount_in_cents,
"amount": (
cents_to_dollars(e.amount_in_cents)
if e.amount_in_cents is not None
else None
),
"message": e.message,
"created_timestamp": e.created_timestamp,
}
for e in offer.events
],
}
@view_config(route_name="offer_open", request_method="POST", renderer="json")
@user_required(flash_msg="Please log in to make an offer.")
def offer_open(request):
product = get_product_by_id(
request.dbsession, request.matchdict["product_id"]
)
if product is None:
raise HTTPNotFound()
if not product.offers_allowed:
request.response.status_int = 403
return {"error": "this product is not accepting offers"}
# Block self-offer.
if request.user in product.shop.owners:
request.response.status_int = 403
return {"error": "you cannot offer on your own product"}
try:
amount_in_cents = int(round(float(request.params.get("amount", "")) * 100))
except (TypeError, ValueError):
request.response.status_int = 400
return {"error": "invalid amount"}
buyer_message = (request.params.get("message") or "").strip() or None
try:
offer = open_offer(
request.dbsession,
product=product,
shop=product.shop,
buyer=request.user,
amount_in_cents=amount_in_cents,
buyer_message=buyer_message,
)
except OfferRejected as e:
request.response.status_int = 400
return {"error": str(e)}
return {
"ok": True,
"offer_id": offer.uuid_str,
"offer": _serialize_offer(offer),
}
@view_config(route_name="offer_page", renderer="offer.j2")
def offer_page(request):
offer = get_offer_by_id(request.dbsession, request.matchdict["offer_id"])
if offer is None:
raise HTTPNotFound()
actor_party, can_view = _user_party(request, offer)
if not can_view:
if request.user is None:
return HTTPFound(
f"/join-or-log-in?next=/o/{offer.uuid_str}"
)
raise HTTPNotFound()
ctx = _serialize_offer(offer)
ctx["actor_party"] = actor_party
ctx["can_act"] = (
actor_party is not None
and offer.is_open
and actor_party == offer.current_party
)
return ctx
def _offer_action(request, action_fn, message_required=False):
"""Common handling for counter / accept / decline / withdraw."""
offer = get_offer_by_id(request.dbsession, request.matchdict["offer_id"])
if offer is None:
raise HTTPNotFound()
actor_party, can_view = _user_party(request, offer)
if not can_view or actor_party is None:
request.response.status_int = 403
return {"error": "not authorized"}
message = (request.params.get("message") or "").strip() or None
if message_required and not message:
# Optional — no validation here; counter has its own amount check.
pass
try:
if action_fn is counter_offer:
try:
new_amount = int(round(float(
request.params.get("amount", "")) * 100))
except (TypeError, ValueError):
request.response.status_int = 400
return {"error": "invalid amount"}
action_fn(
offer, request.user, actor_party, new_amount, message=message,
)
elif action_fn is withdraw_offer:
# Buyer-only.
if actor_party != OFFER_PARTY_BUYER:
request.response.status_int = 403
return {"error": "only the buyer can withdraw"}
action_fn(offer, request.user, message=message)
else:
action_fn(offer, request.user, actor_party, message=message)
except OfferRejected as e:
request.response.status_int = 400
return {"error": str(e)}
return {"ok": True, "offer": _serialize_offer(offer)}
@view_config(route_name="offer_counter", request_method="POST", renderer="json")
@user_required(flash_msg="Please log in to act on this offer.")
def offer_counter(request):
return _offer_action(request, counter_offer)
@view_config(route_name="offer_accept", request_method="POST", renderer="json")
@user_required(flash_msg="Please log in to act on this offer.")
def offer_accept(request):
return _offer_action(request, accept_offer)
@view_config(route_name="offer_decline", request_method="POST", renderer="json")
@user_required(flash_msg="Please log in to act on this offer.")
def offer_decline(request):
return _offer_action(request, decline_offer)
@view_config(route_name="offer_withdraw", request_method="POST", renderer="json")
@user_required(flash_msg="Please log in to act on this offer.")
def offer_withdraw(request):
return _offer_action(request, withdraw_offer)