engineering/make_post_sell
1
0
Fork 0
Code Issues 2 Pull requests Projects Releases 2 Packages Wiki Activity Actions

http-verbs-get-to-post-csrf

Browse source
This commit is contained in:
Russell Ballestrini 2025-05-07 10:55:51 +00:00
parent 590d13d77b
commit 07722bb962
10 changed files with 574 additions and 253 deletions
Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -12,6 +12,7 @@ vars.sh
coverage.xml
build/
dist/
data/
src/
.tox/
nosetests.xml

4
development.ini
View file

@ -31,7 +31,9 @@ session.secret = test-secret
session.timeout = 31104000
session.max_age = 31104000
session.reissue_time = 15552000
session.samesite = None
session.secure = false
session.domain = localhost.localhost
session.samesite = Lax
###
# app "business logic" settings

17
make_post_sell/request_methods.py
View file

@ -44,21 +44,22 @@ def includeme(config):
return get_product_by_id(request.dbsession, product_id)
def add_shop(request):
"""return Shop object or None."""
# get shop_id first from query, then route, or None.
shop_id = request.params.get("shop_id", request.matchdict.get("shop_id"))
# print(f"add_shop: shop_id={shop_id}, params={request.params}, matchdict={request.matchdict}")
shop = None
if shop_id:
return get_shop_by_id(request.dbsession, shop_id)
shop = get_shop_by_id(request.dbsession, shop_id)
# print(f"get_shop_by_id returned: {shop}")
elif request.product:
return request.product.shop
shop = request.product.shop
else:
if request.is_saas_domain:
if request.user and request.user.active_shop:
return request.user.active_shop
shop = request.user.active_shop
else:
return get_shop_by_domain_name(request.dbsession, request.domain)
shop = get_shop_by_domain_name(request.dbsession, request.domain)
# print(f"add_shop returning: {shop}")
return shop
def add_active_cart(request):
"""

93
make_post_sell/static/css/common.css
View file

@ -283,11 +283,13 @@ img.product-main {
cursor: pointer;
}
a.mps-button-blue {
a.mps-button-blue,
button.mps-button-blue {
background-color: #98b6fa;
}
a.mps-button-green {
a.mps-button-green,
button.mps-button-green {
background-color: #a3c765;
}
@ -319,35 +321,6 @@ a.log-out-button {
background-color: #5871ad;
}
a.cart-checkout-button {
background-color: #a3c765;
}
a.cart-save-button {
background-color: #98b6fa;
}
a.cart-activate-button {
background-color: #5871ad;
}
a.cart-public-button span.lock {
font-size: 24px;
}
a.cart-public-button {
color: #666666;
border-color: #666666;
border-style: solid;
border-width: 1px;
padding-top: 10px;
padding-bottom: 10px;
}
a.cart-delete-button {
background-color: #CC6958;
}
a.shop-switch-button {
color: #666666;
border-color: #666666;
@ -368,6 +341,64 @@ a.cart-and-count {
}
/* -------------------------------------------------------------------
Form-buttons: copy your <a>-class styles to <button> elements
------------------------------------------------------------------- */
/* “Remove” link: no button chrome, just a link color + hover underline */
button.cart-remove-link {
background: none;
border: none;
padding: 0;
font: inherit;
color: #5f6368;
text-decoration: none;
cursor: pointer;
}
button.cart-remove-link:hover {
text-decoration: underline;
}
/* “Add To Cart”, “Download”, “Preview” buttons */
button.product-download-button,
button.product-preview-button {
/* inherits .mps-button defaults */
}
button.product-download-button {
background-color: #a3c765;
}
button.product-preview-button {
background-color: #98b6fa;
}
/* Cart action buttons */
button.cart-checkout-button {
background-color: #a3c765;
}
button.cart-save-button {
background-color: #98b6fa;
}
button.cart-activate-button {
background-color: #5871ad;
}
button.cart-delete-button {
background-color: #CC6958;
color: white;
}
/* “Make Public” / “Make Private” */
button.cart-public-button {
background: none;
color: #666666;
border: 1px solid #666666;
padding-top: 10px;
padding-bottom: 10px;
cursor: pointer;
}
button.cart-public-button span.lock {
font-size: 24px;
}
.coupon-apply-button {
background-color: #a3c765;
}

40
make_post_sell/templates/cart.j2
View file

@ -118,7 +118,10 @@
<br/>
<!-- change this to a form to prevent CSRF -->
<a href="/cart/{{ cart.id }}/remove?product_id={{ product.id }}">remove</a>
<form method="POST" action="{{ request.route_url('cart_remove_product', cart_id=cart.id, product_id=product.id) }}" style="display:inline;">
{% include "snippets/csrf.j2" %}
<button type="submit" class="cart-remove-link">remove</button>
</form>
</div>
<div style="text-align: right;">
@ -203,7 +206,10 @@
{% if not cart.is_empty and request.user == cart.user %}
<a href="/u/cart/{{ cart.id }}/checkout" class="cart-checkout-button mps-button">Checkout</a>
<form method="POST" action="{{ request.route_url('user_cart_checkout', cart_id=cart.id) }}">
{% include "snippets/csrf.j2" %}
<button type="submit" class="cart-checkout-button mps-button">Checkout</button>
</form>
<br/>
<br/>
@ -218,16 +224,25 @@
{% if request.active_cart == cart and cart.count > 0 %}
<a href="/u/cart/save" class="cart-save-button mps-button">Save Cart</a>
<form method="POST" action="{{ request.route_url('user_cart_save', cart_id=cart.id) }}">
{% include "snippets/csrf.j2" %}
<button type="submit" class="cart-save-button mps-button">Save Cart</button>
</form>
<br/>
{% elif request.active_cart != cart %}
{% if request.user and request.user == cart.user %}
<a href="/u/cart/{{ cart.id }}/activate" class="cart-activate-button mps-button">Make Cart Active</a>
<form method="POST" action="{{ request.route_url('user_cart_activate', cart_id=cart.id) }}">
{% include "snippets/csrf.j2" %}
<button type="submit" class="cart-activate-button mps-button">Make Cart Active</button>
</form>
<br/>
{% else %}
<a href="/u/cart/{{ cart.id }}/activate" class="cart-activate-button mps-button">Copy & Make Cart Active</a>
<form method="POST" action="{{ request.route_url('user_cart_activate', cart_id=cart.id) }}">
{% include "snippets/csrf.j2" %}
<button type="submit" class="cart-activate-button mps-button">Copy & Make Cart Active</button>
</form>
<br/>
{% endif %}
@ -236,10 +251,16 @@
{% if request.user and request.user.owns_cart(cart) %}
{% if cart.is_not_public and cart.count > 0 %}
<a href="/u/cart/{{ cart.id }}/public" class="cart-public-button mps-button"><span class="lock">&#128275;</span> Make Cart Public</a>
<form method="POST" action="{{ request.route_url('user_cart_public', cart_id=cart.id) }}">
{% include "snippets/csrf.j2" %}
<button type="submit" class="cart-public-button mps-button"><span class="lock">&#128275;</span> Make Cart Public</button>
</form>
<br/>
{% elif cart.public %}
<a href="/u/cart/{{ cart.id }}/unpublic" class="cart-public-button mps-button"><span class="lock">&#128274;</span> Make Cart Private</a>
<form method="POST" action="{{ request.route_url('user_cart_unpublic', cart_id=cart.id) }}">
{% include "snippets/csrf.j2" %}
<button type="submit" class="cart-public-button mps-button"><span class="lock">&#128274;</span> Make Cart Private</button>
</form>
<br/>
<br/>
@ -249,7 +270,10 @@
{% endif %}
{% if request.active_cart != cart %}
<a href="/u/cart/{{ cart.id }}/delete" class="cart-delete-button mps-button">Delete Cart</a>
<form method="POST" action="{{ request.route_url('user_cart_delete', cart_id=cart.id) }}">
{% include "snippets/csrf.j2" %}
<button type="submit" class="cart-delete-button mps-button">Delete Cart</button>
</form>
<br/>
{% endif %}

5
make_post_sell/templates/cart_checkout.j2
View file

@ -38,7 +38,10 @@
<br/>
<br/>
<a href="/u/cart/{{ cart.id }}/complete/checkout" class="cart-checkout-button mps-button">Yes, Complete Checkout</a>
<form method="POST" action="{{ request.route_url('user_cart_complete_checkout', cart_id=cart.id) }}">
{% include "snippets/csrf.j2" %}
<button type="submit" class="cart-checkout-button mps-button">Yes, Complete Checkout</button>
</form>
</section>

12
make_post_sell/templates/product.j2
View file

@ -80,7 +80,11 @@
{% if product.is_physical %}
{% set inventory = product.inventories | selectattr('shop_location_id', 'equalto', request.shop_location.id) | first %}
{% if inventory and inventory.quantity > 0 %}
<a href="/cart/add?product_id={{ product.id }}" class="product-download-button mps-button">Add To Cart</a>
<form method="POST" action="{{ request.route_url('cart_add_product') }}">
{% include "snippets/csrf.j2" %}
<input type="hidden" name="product_id" value="{{ product.id }}">
<button type="submit" class="product-download-button mps-button">Add To Cart</button>
</form>
{% else %}
<button class="product-download-button mps-button" style="background-color: red; cursor: not-allowed;" disabled>Sold Out</button>
{% if request.shop.shop_locations.all()|length > 1 %}
@ -89,7 +93,11 @@
{% endif %}
{% endif %}
{% else %}
<a href="/cart/add?product_id={{ product.id }}" class="product-download-button mps-button">Add To Cart</a>
<form method="POST" action="{{ request.route_url('cart_add_product') }}">
{% include "snippets/csrf.j2" %}
<input type="hidden" name="product_id" value="{{ product.id }}">
<button type="submit" class="product-download-button mps-button">Add To Cart</button>
</form>
{% endif %}
<br/>

6
make_post_sell/templates/snippets/csrf.j2
View file

@ -1,3 +1,3 @@
{% if request.csrf_token %}
<input type="hidden" name="csrf_token" value="{{ request.csrf_token }}">
{% endif %}
{# Always include the session’s CSRF token #}
<input type="hidden" name="csrf_token" value="{{ request.session.get_csrf_token() }}" />

457
make_post_sell/tests/test_functional.py
View file

@ -7,6 +7,8 @@ import unittest
import webtest
import stripe
import re
from ..models import get_tm_session
from ..models.meta import Base
@ -17,7 +19,7 @@ from ..models.user import get_or_create_user_by_email
from ..models.cart import get_cart_by_id, get_all_carts
from ..models.stripe_user_shop import get_all_stripe_customer_objects
from ..models.stripe_user_shop import get_all_stripe_customer_objects, StripeUserShop
from ..models.product import get_product_by_id, get_all_products
@ -28,7 +30,7 @@ from ..lib.currency import dollars_to_cents, cents_to_dollars
from pyramid.paster import get_appsettings
import mock
from mock import patch
from mock import patch, MagicMock
# todo we should pick a new file to put test helpers.
@ -36,7 +38,7 @@ mock_always_true = mock.Mock(return_value=True)
# todo we should pick a new file to put test helpers.
class FunctionalTests(unittest.TestCase, object):
class FunctionalTests(unittest.TestCase):
def setUp(self):
from make_post_sell import main
@ -59,6 +61,27 @@ class FunctionalTests(unittest.TestCase, object):
transaction.abort()
Base.metadata.drop_all(bind=self.engine)
def get_csrf_token(self, shop_id, keywords="test"):
"""
GET the search page scoped to the given shop_id (and a dummy keywords param
so the form actually renders) and extract the CSRF hidden input via regex.
"""
url = f"/search?shop_id={shop_id}&keywords={keywords}"
res = self.testapp.get(url, status=200)
html = res.body.decode("utf-8")
m = re.search(
r'name=["\']csrf_token["\'].*?value=["\']([^"\']+)["\']',
html,
flags=re.IGNORECASE | re.DOTALL,
)
if not m:
self.fail(
f"CSRF token not found on {url} page.\n\n"
f"HTML snippet:\n{html[:500]}…"
)
return m.group(1)
class UnauthenticatedFunctionalTests(FunctionalTests):
def test_root_home_page(self):
@ -419,15 +442,10 @@ class AuthenticatedFunctionalTests(FunctionalTests):
self.assertEqual(product.price_history.count(), 1)
self.assertEqual(product.price_in_cents, 350)
# this patch requires an argument added to test method `mock_smtp`.
@patch("smtplib.SMTP")
@patch("make_post_sell.models.Product.is_ready", mock_always_true)
def test_cart_checkout_for_shop(self, mock_smtp):
# 1. log in as user1.
# 2. create new shop.
# 3. make new shop ready for checkout.
# 4. create new product.
# 5. log out user1.
# 1. Log in as user1, create shop, make shop ready, create product, log out.
self.test_new_product(
user_creds=self.user1_creds,
shop_params=self.shop1_params,
@ -435,20 +453,35 @@ class AuthenticatedFunctionalTests(FunctionalTests):
log_out_user=True,
)
# get the Product object from database.
# Get the Product and Shop objects from the database.
all_products = get_all_products(self.dbsession)
product = all_products.one()
shop = get_shop_by_name(self.dbsession, self.shop1_params["name"])
# print("{},{},{}".format(product.id, product.title, product.price))
# log in user2.
# Log in user2.
self.log_in_user(self.user2_creds)
# add product to cart.
redirect_res1 = self.testapp.get(
f"/cart/add?product_id={product.id}&shop_id={shop.id}"
# Step 1: Get the product page and follow the redirect to the slug version to get raw HTML.
res_csrf_redirect = self.testapp.get(f"/p/{product.uuid_str}")
res_csrf = (
res_csrf_redirect.follow()
) # Follow the single redirect to the slug version.
# Step 2: Extract the CSRF token from the raw HTML.
csrf_token = self.get_csrf_token(shop.uuid_str)
if not csrf_token:
self.fail(
"CSRF token not found in product page HTML or session cookie (_csrft_)"
)
# Step 3: Add product to cart using a POST request with CSRF token.
redirect_res1 = self.testapp.post(
"/cart/add",
{
"product_id": product.id,
"shop_id": shop.id,
"csrf_token": csrf_token, # Include CSRF token as a form value.
},
)
redirect_res2 = redirect_res1.follow()
res = redirect_res2.follow()
@ -458,7 +491,7 @@ class AuthenticatedFunctionalTests(FunctionalTests):
carts = get_all_carts(self.dbsession)
# make sure we have 4 carts. request.active_cart & request.session_cart for each user.
# Make sure we have 4 carts (request.active_cart & request.session_cart for each user).
self.assertEqual(4, carts.count())
self.dbsession.refresh(shop)
@ -468,64 +501,25 @@ class AuthenticatedFunctionalTests(FunctionalTests):
user_1_cart = shop.get_active_cart_for_user(self.user1)
user_2_cart = shop.get_active_cart_for_user(self.user2)
# make sure user1 has 0 items in active Cart.
# Make sure user1 has 0 items in active Cart.
self.assertEqual(0, user_1_cart.count)
# make sure user2 has 1 item in active Cart.
# Make sure user2 has 1 item in active Cart.
self.assertEqual(1, user_2_cart.count)
redirect_res = self.testapp.get(
f"/u/cart/{user_2_cart.id}/checkout?shop_id={shop.id}"
)
redirect_res2 = redirect_res1.follow()
res = redirect_res2.follow()
self.assertIn("Please enter your payment information.", res.body.decode())
# the stripe migration to SetupIntents broke this end-to-end test routine.
"""
res = self.testapp.post(
"/billing/add-card?shop_id={}".format(shop.id),
# Proceed to checkout (extract CSRF token again if needed).
csrf_token = self.get_csrf_token(shop.uuid_str)
res_csrf_checkout = self.testapp.post(
f"/u/cart/{user_2_cart.uuid_str}/checkout",
{
"email": self.user2.email,
"stripeToken": "tok_visa",
"shop_id": shop.id,
"csrf_token": csrf_token, # Include CSRF token as a form value.
},
)
res = self.testapp.get(
"/u/cart/{}/checkout?shop_id={}".format(
user_2_cart.id,
shop.id
)
res_csrf_checkout = res_csrf_checkout.follow().follow()
self.assertIn(
"Please enter your payment information.", res_csrf_checkout.body.decode()
)
res_body = res.body.decode()
self.assertIn("Please confirm your order.", res_body)
self.assertIn("Visa", res_body)
self.assertIn("Are you sure you want to charge", res_body)
redirect_res1 = self.testapp.get(
"/u/cart/{}/complete/checkout?shop_id={}".format(
user_2_cart.id,
shop.id
)
)
redirect_res2 = redirect_res1.follow()
res = redirect_res2.follow()
res_body = res.body.decode()
self.assertIn("Success, you have completed the purchase!", res_body)
# refresh shop attributes from database.
self.dbsession.refresh(shop)
stripe_customer = shop.stripe_customer(self.user2)
stripe_charge = shop.list_stripe_charges(stripe_customer).data[0]
self.assertEqual(
stripe_charge.amount,
dollars_to_cents(self.product1_params["price"]),
)
"""
@patch("make_post_sell.models.Product.is_ready", mock_always_true)
def make_new_coupon_for_shop(self, coupon_params=None):
@ -555,6 +549,10 @@ class AuthenticatedFunctionalTests(FunctionalTests):
f"/s/{shop.id}/coupon/new",
coupon_params,
)
# Commit the transaction to persist shop, product, and coupon
transaction.manager.commit()
return res
def test_new_coupon_for_shop(self):
@ -569,78 +567,307 @@ class AuthenticatedFunctionalTests(FunctionalTests):
self.assertIn("Please submit all required fields.", res.body.decode())
# this patch requires an argument added to test method `mock_smtp`.
@patch("make_post_sell.models.Product.is_ready")
@patch("smtplib.SMTP")
@patch("make_post_sell.models.Product.is_ready", mock_always_true)
def test_checkout_with_coupon_code(self, mock_smtp):
@patch("stripe.SetupIntent.retrieve")
@patch("stripe.Customer.list_payment_methods")
def test_checkout_with_coupon_code(
self,
mock_list_payment_methods,
mock_setup_intent,
mock_smtp,
mock_product_ready,
):
# Debug: Confirm mock objects
print(f"Mock SetupIntent: {mock_setup_intent}")
print(f"Mock SMTP: {mock_smtp}")
print(f"Mock Product.is_ready: {mock_product_ready}")
print(f"Mock list_payment_methods: {mock_list_payment_methods}")
# Configure mocks
mock_product_ready.return_value = True
mock_setup_intent.return_value = {
"id": "seti_123",
"status": "succeeded",
"payment_method": "pm_card_visa",
"customer": "cus_123",
"client_secret": "seti_123_secret",
"payment_method_types": ["card"],
"payment_method_options": {"card": {}},
"metadata": {"card_id": "pm_card_visa"},
}
# Mock PaymentMethod object
payment_method = MagicMock()
payment_method.id = "pm_card_visa"
payment_method.type = "card"
payment_method.card.brand = "visa"
payment_method.card.last4 = "4242"
mock_list_payment_methods.return_value = {"data": [payment_method]}
# Step 1: Create a coupon as user1
res = self.make_new_coupon_for_shop()
res = res.follow()
self.assertIn("You created a new coupon!", res.body.decode())
# log out user1.
self.testapp.get("/log-out")
# log in user2.
self.log_in_user(self.user2_creds)
# query the new shop from database.
# Re-query users and shop
self.user1 = get_or_create_user_by_email(self.dbsession, self.user1_creds[0])
self.user2 = get_or_create_user_by_email(self.dbsession, self.user2_creds[0])
shop = get_shop_by_name(self.dbsession, self.shop1_params["name"])
# Log out user1
self.testapp.get("/log-out")
# Log in user2
self.log_in_user(self.user2_creds)
# Refresh the shop to ensure it's committed
self.dbsession.refresh(shop)
print(
f"Shop UUID: {shop.uuid_str}, Ready: {shop.is_ready}, "
f"Public key: {shop.stripe_public_api_key}, "
f"Secret key: {shop.stripe_secret_api_key}"
)
# Verify shop is ready
self.assertTrue(
shop.is_ready,
f"Shop {shop.name} is not ready. "
f"stripe_public_api_key: {shop.stripe_public_api_key}, "
f"stripe_secret_api_key: {shop.stripe_secret_api_key}",
)
# Step 2: Visit /billing to create StripeUserShop record for user2
shop_id = str(shop.uuid_str)
res_billing_init = self.testapp.get(f"/billing?shop_id={shop_id}", status=200)
print(f"Billing init response: {res_billing_init.body.decode()}") # Debug
self.assertIn(
"The active card will be charged during checkout.",
res_billing_init.body.decode(),
) # Verify billing page
# Commit transaction to persist StripeUserShop
transaction.manager.commit()
# Re-query users and shop
self.user1 = get_or_create_user_by_email(self.dbsession, self.user1_creds[0])
self.user2 = get_or_create_user_by_email(self.dbsession, self.user2_creds[0])
shop = get_shop_by_name(self.dbsession, self.shop1_params["name"])
# Step 3: Add billing details for user2
csrf_token = self.get_csrf_token(shop.uuid_str)
if not csrf_token:
self.fail("CSRF token not found in session cookie (_csrft_)")
print(f"Using shop_id for billing: {shop_id}")
res_billing_redirect = self.testapp.post(
f"/billing/add-card?shop_id={shop_id}",
{
"email": self.user2.email,
"setup_intent": "seti_123",
"setup_intent_client_secret": "seti_123_secret",
"csrf_token": csrf_token,
},
status=302,
)
res_billing = res_billing_redirect.follow()
res_billing = res_billing.follow()
print(f"Billing response: {res_billing.body.decode()}") # Debug
self.assertIn(
"You saved a new card.", res_billing.body.decode()
) # Match view's flash message
# Commit transaction to persist card
transaction.manager.commit()
# Re-query users and shop
self.user1 = get_or_create_user_by_email(self.dbsession, self.user1_creds[0])
self.user2 = get_or_create_user_by_email(self.dbsession, self.user2_creds[0])
shop = get_shop_by_name(self.dbsession, self.shop1_params["name"])
# Verify StripeUserShop and set active card
stripe_user_shop = (
self.dbsession.query(StripeUserShop)
.filter_by(user_id=self.user2.id, shop_id=shop.id)
.first()
)
print(f"StripeUserShop for {self.user2.email}: {stripe_user_shop}")
print(
f"Card IDs before: {stripe_user_shop.stripe_card_ids if stripe_user_shop else None}"
)
if stripe_user_shop and not stripe_user_shop.active_card_id:
stripe_user_shop.active_card_id = "pm_card_visa"
self.dbsession.add(stripe_user_shop)
self.dbsession.flush()
print(f"Manually set active card ID: {stripe_user_shop.active_card_id}")
print(
f"Active card ID: {stripe_user_shop.active_card_id if stripe_user_shop else None}"
)
# Commit transaction to persist StripeUserShop changes
transaction.manager.commit()
# Re-query users and shop
self.user1 = get_or_create_user_by_email(self.dbsession, self.user1_creds[0])
self.user2 = get_or_create_user_by_email(self.dbsession, self.user2_creds[0])
shop = get_shop_by_name(self.dbsession, self.shop1_params["name"])
# Re-query StripeUserShop to verify persistence
stripe_user_shop = (
self.dbsession.query(StripeUserShop)
.filter_by(user_id=self.user2.id, shop_id=shop.id)
.first()
)
print(f"StripeUserShop after commit: {stripe_user_shop}")
print(
f"Active card ID after commit: {stripe_user_shop.active_card_id if stripe_user_shop else None}"
)
print(
f"Card IDs after commit: {stripe_user_shop.stripe_card_ids if stripe_user_shop else None}"
)
# Query the coupon from the database
coupons = get_coupons_by_code(self.dbsession, self.coupon1_params["code"])
coupon = coupons[0]
# apply coupon to user2's active cart.
self.testapp.post(f"/coupon/apply?&coupon_id={coupon.id}&shop_id={shop.id}")
# Step 4: Extract CSRF token for applying the coupon
csrf_token = self.get_csrf_token(shop.uuid_str)
if not csrf_token:
self.fail("CSRF token not found in session cookie (_csrft_)")
# verify coupon was applied.
res_redirect1 = self.testapp.get(f"/cart?shop_id={shop.id}")
res = res_redirect1.follow()
res_body = res.body.decode()
# Step 5: Apply coupon to user2's active cart
res_coupon_apply_redirect = self.testapp.post(
f"/coupon/apply?shop_id={shop_id}",
{
"coupon_id": coupon.id,
"shop_id": shop_id,
"csrf_token": csrf_token,
},
status=302,
)
res_coupon_apply = res_coupon_apply_redirect.follow()
res_body = res_coupon_apply.body.decode()
# Verify coupon was applied
self.assertIn("Slippery Halloween Party", res_body)
self.assertIn("3.50", res_body)
self.assertIn("camp31", res_body)
# get the Product object from database.
# Commit transaction to persist coupon application
transaction.manager.commit()
# Re-query users and shop
self.user1 = get_or_create_user_by_email(self.dbsession, self.user1_creds[0])
self.user2 = get_or_create_user_by_email(self.dbsession, self.user2_creds[0])
shop = get_shop_by_name(self.dbsession, self.shop1_params["name"])
# Step 6: Query the product from the database
all_products = get_all_products(self.dbsession)
product = all_products.one()
product_id = product.uuid_str
# add product to user2's active cart.
self.testapp.get(f"/cart/add?product_id={product.uuid_str}&shop_id={shop.id}")
# Step 7: Extract CSRF token for adding product to cart
csrf_token = self.get_csrf_token(shop.uuid_str)
if not csrf_token:
self.fail("CSRF token not found in session cookie (_csrft_)")
# the stripe migration to SetupIntents broke this end-to-end test routine.
"""
# add billing details for user2 for this shop.
res = self.testapp.post(
"/billing/add-card?shop_id={}".format(shop.id),
# Step 8: Add product to user2's active cart
res_add_product_redirect = self.testapp.post(
f"/cart/add?shop_id={shop_id}",
{
"email": self.user2.email,
"stripeToken": "tok_visa",
"product_id": product_id,
"shop_id": shop_id,
"csrf_token": csrf_token,
},
status=302,
)
res_add_product = res_add_product_redirect.follow()
res_add_product = res_add_product.follow()
res_body = res_add_product.body.decode()
self.assertIn('You added "russell\'s product" to your cart.', res_body)
# why is this None?
# Commit transaction to persist cart
transaction.manager.commit()
# Re-query users, shop, and cart
self.user1 = get_or_create_user_by_email(self.dbsession, self.user1_creds[0])
self.user2 = get_or_create_user_by_email(self.dbsession, self.user2_creds[0])
shop = get_shop_by_name(self.dbsession, self.shop1_params["name"])
user_2_cart = shop.get_active_cart_for_user(self.user2)
# attempt to checkout before coupon is properly qualified.
# shop total must be over cart_qualifier (test defaults to $10).
res_redirect1 = self.testapp.get(
"/u/cart/{}/checkout?shop_id={}".format(
user_2_cart.id,
shop.id
)
)
res = res_redirect1.follow()
self.assertIn("Please review coupon terms: shop total not met.", res.body.decode())
# Step 9: Get user2's active cart
self.assertIsNotNone(user_2_cart, "User2's active cart should not be None")
# add more product to user2's active cart, to pass cart_qualifier.
self.testapp.get("/cart/add?product_id={}&shop_id={}".format(product.uuid_str, shop.id))
self.testapp.get("/cart/add?product_id={}&shop_id={}".format(product.uuid_str, shop.id))
# Step 10: Attempt checkout with insufficient cart total
csrf_token = self.get_csrf_token(shop.uuid_str)
if not csrf_token:
self.fail("CSRF token not found in session cookie (_csrft_)")
# checkout coupon confirmation unblocked.
res = self.testapp.get(
"/u/cart/{}/checkout?shop_id={}".format(
user_2_cart.id,
shop.id
)
res_checkout = self.testapp.post(
f"/u/cart/{user_2_cart.uuid_str}/checkout?shop_id={shop_id}",
{
"shop_id": shop_id,
"csrf_token": csrf_token,
},
status=302,
)
self.assertIn("Please confirm your order.", res.body.decode())
"""
res_body = res_checkout.follow().body.decode()
print(f"First checkout response: {res_body}") # Debug
self.assertIn("Please review the terms for coupon", res_body)
self.assertIn("shop total not met.", res_body)
# Commit transaction to persist checkout attempt
transaction.manager.commit()
# Re-query users, shop, and cart
self.user1 = get_or_create_user_by_email(self.dbsession, self.user1_creds[0])
self.user2 = get_or_create_user_by_email(self.dbsession, self.user2_creds[0])
shop = get_shop_by_name(self.dbsession, self.shop1_params["name"])
user_2_cart = shop.get_active_cart_for_user(self.user2)
# Step 11: Add more products to meet coupon qualifier
for _ in range(2):
csrf_token = self.get_csrf_token(shop.uuid_str)
if not csrf_token:
self.fail("CSRF token not found in session cookie (_csrft_)")
res_add_product_redirect = self.testapp.post(
f"/cart/add?shop_id={shop_id}",
{
"product_id": product_id,
"shop_id": shop_id,
"csrf_token": csrf_token,
},
status=302,
)
res_add_product = res_add_product_redirect.follow()
res_add_product = res_add_product.follow()
print(f"Add product response: {res_add_product.body.decode()}") # Debug
transaction.manager.commit()
# Re-query users, shop, and cart
self.user1 = get_or_create_user_by_email(
self.dbsession, self.user1_creds[0]
)
self.user2 = get_or_create_user_by_email(
self.dbsession, self.user2_creds[0]
)
shop = get_shop_by_name(self.dbsession, self.shop1_params["name"])
user_2_cart = shop.get_active_cart_for_user(self.user2)
# Step 12: Attempt checkout again
csrf_token = self.get_csrf_token(shop.uuid_str)
if not csrf_token:
self.fail("CSRF token not found in session cookie (_csrft_)")
print(f"Shop before checkout: UUID: {shop.uuid_str}, Ready: {shop.is_ready}")
res_checkout_final = self.testapp.post(
f"/u/cart/{user_2_cart.uuid_str}/checkout?shop_id={shop_id}",
{
"shop_id": shop_id,
"csrf_token": csrf_token,
},
status=200,
)
res_body = res_checkout_final.body.decode()
print(f"Checkout final response: {res_body}") # Debug
self.assertIn("Please confirm your order.", res_body)

192
make_post_sell/views/cart.py
View file

@ -1,7 +1,5 @@
from pyramid.view import view_config
from transaction import TransactionManager
from . import (
user_required,
get_referer_or_home,
@ -19,6 +17,8 @@ from ..lib.mail import (
send_sale_email,
)
import stripe
def get_cart_from_matchdict(request):
"""
@ -212,7 +212,7 @@ def cart_by_id(request):
return HTTPFound(get_referer_or_home(request))
@view_config(route_name="cart_add_product")
@view_config(route_name="cart_add_product", request_method="POST", require_csrf=True)
def cart_add_product(request):
product_id = request.params.get("product_id", None)
@ -247,7 +247,7 @@ def cart_add_product(request):
return HTTPFound(get_referer_or_home(request))
@view_config(route_name="cart_remove_product")
@view_config(route_name="cart_remove_product", request_method="POST", require_csrf=True)
def cart_remove_product(request):
product_id = request.params.get("product_id", None)
@ -285,7 +285,9 @@ def cart_remove_product(request):
return HTTPFound(get_referer_or_home(request))
@view_config(route_name="cart_quantity_product")
@view_config(
route_name="cart_quantity_product", request_method="POST", require_csrf=True
)
def cart_quantity_product(request):
product_id = request.params.get("product_id", None)
quantity = request.params.get("quantity", None)
@ -380,8 +382,18 @@ def cart_handling_option(request):
return HTTPFound(f"/cart/{cart.id}")
@view_config(route_name="cart_checkout", renderer="cart_checkout.j2")
@view_config(route_name="user_cart_checkout", renderer="cart_checkout.j2")
@view_config(
route_name="cart_checkout",
renderer="cart_checkout.j2",
request_method="POST",
require_csrf=True,
)
@view_config(
route_name="user_cart_checkout",
renderer="cart_checkout.j2",
request_method="POST",
require_csrf=True,
)
@user_required(
flash_msg="To <b>checkout</b> your cart, please verify your email address below.",
flash_level="info",
@ -459,7 +471,10 @@ def cart_checkout(request):
request.session.flash(msg)
return HTTPFound(get_referer_or_home(request))
@view_config(route_name="user_cart_complete_checkout")
@view_config(
route_name="user_cart_complete_checkout", request_method="POST", require_csrf=True
)
@user_required()
@shop_is_ready_required()
def cart_complete_checkout(request):
@ -467,102 +482,111 @@ def cart_complete_checkout(request):
if cart is None:
msg = ("That cart_id does not exist.", "error")
request.session.flash(msg)
return HTTPFound(get_referer_or_home(request))
elif cart.is_not_public and request.user.does_not_own_cart(cart):
msg = ("That cart is not public and you do not own that cart.", "error")
request.session.flash(msg)
return HTTPFound(get_referer_or_home(request))
elif cart.is_empty:
msg = ("That cart is empty, you cannot checkout.", "error")
request.session.flash(msg)
return HTTPFound(get_referer_or_home(request))
elif cart.requires_payment and request.shop.stripe_customer(request.user) is None:
msg = ("Please enter your payment information.", "info")
request.session.flash(msg)
return HTTPFound("/billing")
else:
# make sure all coupon terms are met before proceeding.
error_messages = cart.validate_attached_coupons()
if error_messages:
for error_message in error_messages:
request.session.flash((error_message, "error"))
return HTTPFound(get_referer_or_home(request))
error_messages = cart.validate_attached_coupons()
if error_messages:
for error_message in error_messages:
request.session.flash((error_message, "error"))
return HTTPFound(get_referer_or_home(request))
tm = TransactionManager()
tm = request.tm
try:
tm.begin()
invoices = []
successful_checkout = True
try:
tm.begin()
for shop_id, product_quantity_tuple in cart.shop_product_dict.items():
shop = cart.shops[shop_id]
invoice = Invoice(request.user)
invoice.shop = shop
invoice.shop_id = shop.id
invoice.handling_option = cart.handling_option
invoice.handling_cost_in_cents = cart.handling_cost_in_cents
invoices = []
if cart.physical_products:
invoice.delivery_address = request.user.active_address.data
# First, prepare invoices without persisting
for shop_id, product_quantity_tuple in cart.shop_product_dict.items():
shop = cart.shops[shop_id]
invoice = Invoice(request.user)
invoice.shop = shop
invoice.shop_id = shop.id
invoice.handling_option = cart.handling_option
invoice.handling_cost_in_cents = cart.handling_cost_in_cents
for product, quantity in product_quantity_tuple:
product.unlock_for_user(request.user)
request.dbsession.add(product)
invoice.new_line_item(
product=product,
quantity=quantity,
)
if cart.physical_products:
invoice.delivery_address = request.user.active_address.data
for coupon in cart.coupons:
invoice.new_coupon_redemption(coupon)
for product, quantity in product_quantity_tuple:
invoice.new_line_item(product=product, quantity=quantity)
request.dbsession.add(invoice)
invoices.append(invoice)
for coupon in cart.coupons:
invoice.new_coupon_redemption(coupon)
for invoice in invoices:
shop = invoice.shop
invoices.append(invoice)
if invoice.requires_payment:
stripe_user_shop = shop.stripe_user_shop(request.user)
try:
shop.stripe.PaymentIntent.create(
amount=invoice.total_in_cents,
currency="usd",
customer=stripe_user_shop.cus_id,
payment_method=stripe_user_shop.active_card_id,
off_session=True,
confirm=True,
)
except shop.stripe.error.CardError as e:
successful_checkout = False
raise
# Attempt payment BEFORE adding invoices to session
for invoice in invoices:
shop = invoice.shop
if successful_checkout:
tm.commit()
cart.update_inventory(request.shop_location)
msg = ("Success, you have completed the purchase!", "success")
request.session.flash(msg)
for invoice in invoices:
send_purchase_email(
request,
request.user.email,
[item.product for item in invoice.line_items],
invoice.total,
)
send_sale_email(
request,
invoice.shop,
[item.product for item in invoice.line_items],
invoice.total,
)
save_cart(request)
return HTTPFound("/u/purchases")
else:
tm.abort()
msg = ("Payment failed. Please check your card details.", "error")
request.session.flash(msg)
return HTTPFound("/billing")
if invoice.requires_payment:
stripe_user_shop = shop.stripe_user_shop(request.user)
shop.stripe.PaymentIntent.create(
amount=invoice.total_in_cents,
currency="usd",
customer=stripe_user_shop.cus_id,
payment_method=stripe_user_shop.active_card_id,
off_session=True,
confirm=True,
)
except Exception as e:
request.session.flash(("Payment failed: Please check your card details. {}".format(str(e)), "error"))
return HTTPFound("/billing")
# Only persist data after successful payment
for invoice in invoices:
for line_item in invoice.line_items:
line_item.product.unlock_for_user(request.user)
request.dbsession.add(line_item.product)
request.dbsession.add(invoice)
tm.commit()
cart.update_inventory(request.shop_location)
msg = ("Success, you have completed the purchase!", "success")
request.session.flash(msg)
for invoice in invoices:
send_purchase_email(
request,
request.user.email,
[item.product for item in invoice.line_items],
invoice.total,
)
send_sale_email(
request,
invoice.shop,
[item.product for item in invoice.line_items],
invoice.total,
)
save_cart(request)
return HTTPFound("/u/purchases")
except stripe.error.CardError as e:
tm.abort()
msg = ("Payment failed. Please check your card details.", "error")
request.session.flash(msg)
return HTTPFound("/billing")
except Exception as e:
tm.abort()
msg = (f"Payment failed: {str(e)}", "error")
request.session.flash(msg)
return HTTPFound("/billing")