engineering/make_post_sell
1
0
Fork 0
Code Issues 2 Pull requests Projects Releases 2 Packages Wiki Activity Actions
make_post_sell/make_post_sell/tests/test_models.py
russell@unturf.com d7e8ec5bc2 mps: torrent backfill on enable + web seed + full test coverage 
- backfill: enabling torrent on a shop auto-generates .torrent for all
  existing products that have a product file (no manual trigger needed)
- web seed (BEP 19): CDN url embedded in .torrent + magnet link so
  clients bootstrap via HTTP then seed to peers (no seeder process needed)
- torrent_file_url: stored on Product, shown as download link on content
  and edit pages alongside the magnet link
- migration: idempotent _column_exists guards on all add_column calls
- template: grid layout (not flex) for magnet/torrent buttons on edit page
- tests: 13 passing tests covering all new paths (unit + functional)
  including backfill trigger, web seed construction, visibility gating
2026-04-05 22:09:20 -04:00

3733 lines
137 KiB
Python
Raw Blame History

import unittest
import uuid
import mock
from ..models import User, Coupon, Shop, is_user_name_valid
from ..models.crypto_payment import CryptoPayment
from ..models.invoice import Invoice
from ..models.meta import now_timestamp, short_id_to_bytes, id_to_uuid
from ..views.signals import classify_referrer, classify_device
mock_always_none = mock.Mock(return_value=None)
mock_always_true = mock.Mock(return_value=True)
mock_false_then_true = mock.Mock(side_effect=[False, False, True])
class TestUser(unittest.TestCase):
@mock.patch("make_post_sell.models.user.is_user_name_available", mock_always_true)
def setUp(self):
self.user = User("russell@ballestrini.net")
def test_created_timestamp_set(self):
self.assertGreater(self.user.created_timestamp, 100000)
def test_email_set(self):
self.assertEqual(self.user.email, "russell@ballestrini.net")
def test_new_password(self):
raw_password = self.user.new_password()
self.assertEqual(len(raw_password), 6)
def test_check_password_success(self):
raw_password = self.user.new_password()
self.assertTrue(self.user.check_password(raw_password))
def test_check_password_failure(self):
raw_password = self.user.new_password()
self.assertFalse(self.user.check_password("fake password"))
def test_is_user_name_valid(self):
self.assertTrue(is_user_name_valid("validusername"))
self.assertTrue(is_user_name_valid("validusername2"))
self.assertTrue(is_user_name_valid("ValidUsername"))
self.assertTrue(is_user_name_valid("valid-username"))
self.assertTrue(is_user_name_valid("valid username"))
self.assertFalse(is_user_name_valid("invalid!!!"))
self.assertFalse(is_user_name_valid(""))
@mock.patch("make_post_sell.models.user.is_user_name_available", mock_always_true)
def test_generate_user_name_no_dash_prefix(self):
u = User("tim@example.com")
self.assertFalse(u.name.startswith("-"))
class TestCart(unittest.TestCase):
# TODO: please test the fitness of the the Cart model.
#
# We have deliberately not written these tests but we should
# before going to much further. Unit tests are much faster and
# more grainular than functional tests.
#
# I'm moving forward without creating these tests because I have
# a functional test running as a safety net.
#
# That said unit tests are much better at capturing expected
# behavior and have better tooling for test and codepath coverage.
#
# Please help write unit tests soon!
def setUp(self):
pass
def validate_attached_coupon_codes(self):
"""Make sure all coupons attached to the cart met the terms."""
pass
def test_cart_requires_payment_above_threshold(self):
"""Test that cart requires payment when total is above $0.64 (64 cents)."""
from make_post_sell.models.cart import Cart
cart = Cart()
# Mock the total_in_cents property to return value above threshold
with mock.patch.object(
type(cart), "total_in_cents", new_callable=mock.PropertyMock
) as mock_total:
mock_total.return_value = 65 # Above 64 cent threshold
self.assertTrue(cart.requires_payment)
def test_cart_requires_payment_at_threshold(self):
"""Test that cart requires payment at exactly $0.64 (64 cents)."""
from make_post_sell.models.cart import Cart
cart = Cart()
with mock.patch.object(
type(cart), "total_in_cents", new_callable=mock.PropertyMock
) as mock_total:
mock_total.return_value = 64 # At threshold
self.assertFalse(cart.requires_payment)
def test_cart_requires_payment_below_threshold(self):
"""Test that cart does not require payment when total is below $0.64."""
from make_post_sell.models.cart import Cart
cart = Cart()
with mock.patch.object(
type(cart), "total_in_cents", new_callable=mock.PropertyMock
) as mock_total:
mock_total.return_value = 30 # Below 64 cent threshold
self.assertFalse(cart.requires_payment)
def test_cart_requires_payment_zero_total(self):
"""Test that cart does not require payment when total is zero (free cart)."""
from make_post_sell.models.cart import Cart
cart = Cart()
with mock.patch.object(
type(cart), "total_in_cents", new_callable=mock.PropertyMock
) as mock_total:
mock_total.return_value = 0 # Free cart
self.assertFalse(cart.requires_payment)
def test_cart_is_not_public_property(self):
"""Test cart.is_not_public property logic."""
from make_post_sell.models.cart import Cart
cart = Cart()
# Test the actual logic: is_not_public should return opposite of public
cart.public = True
self.assertFalse(cart.is_not_public)
cart.public = False
self.assertTrue(cart.is_not_public)
def test_cart_empty_method(self):
"""Test cart.empty() method clears cart contents."""
from make_post_sell.models.cart import Cart
cart = Cart()
cart.set_cart({"product1": 2, "product2": 3})
# Cart should have items
self.assertNotEqual(cart.get_cart(), {})
# Empty should clear everything
cart.empty()
self.assertEqual(cart.get_cart(), {})
def test_cart_set_and_get_cart(self):
"""Test cart.set_cart() and get_cart() methods."""
from make_post_sell.models.cart import Cart
cart = Cart()
test_cart_data = {"product1": 2, "product2": 3}
cart.set_cart(test_cart_data)
retrieved_cart = cart.get_cart()
self.assertEqual(retrieved_cart, test_cart_data)
def test_cart_add_product(self):
"""Test cart.add_product() method."""
from make_post_sell.models.cart import Cart
cart = Cart()
# Mock product with proper uuid_str
mock_product = mock.Mock()
mock_product.uuid_str = "product1" # This is what cart uses as key
# Add product
cart.add_product(mock_product)
# Should have quantity 1
self.assertEqual(cart.get_product_quantity(mock_product), 1)
# Add same product again
cart.add_product(mock_product)
# Should have quantity 2
self.assertEqual(cart.get_product_quantity(mock_product), 2)
def test_cart_remove_product(self):
"""Test cart.remove_product() method."""
from make_post_sell.models.cart import Cart
cart = Cart()
# Mock product
mock_product = mock.Mock()
mock_product.uuid_str = "product1"
# Add product first
cart.add_product(mock_product)
cart.add_product(mock_product) # quantity = 2
# Remove product completely (removes all quantity)
cart.remove_product(mock_product)
self.assertEqual(cart.get_product_quantity(mock_product), 0)
def test_cart_set_product_quantity(self):
"""Test cart.set_product_quantity() method."""
from make_post_sell.models.cart import Cart
cart = Cart()
# Mock product
mock_product = mock.Mock()
mock_product.uuid_str = "product1"
# Set quantity directly
cart.set_product_quantity(mock_product, 5)
self.assertEqual(cart.get_product_quantity(mock_product), 5)
# Set to zero should remove product
cart.set_product_quantity(mock_product, 0)
self.assertEqual(cart.get_product_quantity(mock_product), 0)
def test_cart_get_product_quantity(self):
"""Test cart.get_product_quantity() method."""
from make_post_sell.models.cart import Cart
cart = Cart()
# Mock product
mock_product = mock.Mock()
mock_product.uuid_str = "product1"
# Should return 0 for product not in cart
self.assertEqual(cart.get_product_quantity(mock_product), 0)
# Add product and test
cart.set_product_quantity(mock_product, 3)
self.assertEqual(cart.get_product_quantity(mock_product), 3)
def test_cart_merge_in_cart(self):
"""Test cart.merge_in_cart() method."""
from make_post_sell.models.cart import Cart
cart1 = Cart()
cart2 = Cart()
# Set up cart1
cart1.set_cart({"product1": 2, "product2": 1})
# Set up cart2
cart2.set_cart({"product2": 1, "product3": 3})
# Merge cart2 into cart1
cart1.merge_in_cart(cart2)
result = cart1.get_cart()
# Should have product1: 2, product2: 2 (1+1), product3: 3
self.assertEqual(result["product1"], 2)
self.assertEqual(result["product2"], 2) # Merged
self.assertEqual(result["product3"], 3)
def test_cart_validate_attached_coupons_no_coupons(self):
"""Test coupon validation when no coupons attached."""
from make_post_sell.models.cart import Cart
cart = Cart()
# Mock the coupons property to return empty list
with mock.patch.object(
type(cart), "coupons", new_callable=mock.PropertyMock
) as mock_coupons:
mock_coupons.return_value = []
errors = cart.validate_attached_coupons()
self.assertEqual(errors, [])
def test_cart_remove_handling_if_no_physical_products(self):
"""Test remove_handling_if_no_physical_products method."""
from make_post_sell.models.cart import Cart
cart = Cart()
cart.handling_option = "pickup"
cart.handling_cost_in_cents = 500
# Mock no physical products by mocking the property
with mock.patch.object(
type(cart), "physical_products", new_callable=mock.PropertyMock
) as mock_physical:
mock_physical.return_value = {}
cart.remove_handling_if_no_physical_products()
# Should clear handling
self.assertIsNone(cart.handling_option)
self.assertEqual(cart.handling_cost_in_cents, 0)
def test_cart_bust_memoized_attributes(self):
"""Test that _bust_memoized_attributes clears all cached properties."""
from make_post_sell.models.cart import Cart
cart = Cart()
# Set some memoized attributes
cart._count = 5
cart._line_totals_in_cents = {"test": 100}
cart._products = {"test": "product"}
cart._physical_products = {"test": "physical"}
cart._shops = {"test": "shop"}
cart._shop_product_dict = {"test": []}
cart._shop_totals_in_cents = {"test": 100}
cart._shop_totals = {"test": 1.0}
cart._discounted_shop_totals_in_cents = {"test": 90}
cart._discounted_shop_totals = {"test": 0.9}
cart._line_totals = {"test": 1.0}
# Call the bust method
cart._bust_memoized_attributes()
# Verify all attributes are cleared
self.assertFalse(hasattr(cart, "_count"))
self.assertFalse(hasattr(cart, "_line_totals_in_cents"))
self.assertFalse(hasattr(cart, "_products"))
self.assertFalse(hasattr(cart, "_physical_products"))
self.assertFalse(hasattr(cart, "_shops"))
self.assertFalse(hasattr(cart, "_shop_product_dict"))
self.assertFalse(hasattr(cart, "_shop_totals_in_cents"))
self.assertFalse(hasattr(cart, "_shop_totals"))
self.assertFalse(hasattr(cart, "_discounted_shop_totals_in_cents"))
self.assertFalse(hasattr(cart, "_discounted_shop_totals"))
self.assertFalse(hasattr(cart, "_line_totals"))
def test_cart_set_product_quantity_max_limit(self):
"""Test that set_product_quantity enforces 999 max limit."""
from make_post_sell.models.cart import Cart
cart = Cart()
mock_product = mock.Mock()
mock_product.uuid_str = "product1"
# Test setting quantity above 999
cart.set_product_quantity(mock_product, 1500)
self.assertEqual(cart.get_product_quantity(mock_product), 999)
# Test setting quantity at 999
cart.set_product_quantity(mock_product, 999)
self.assertEqual(cart.get_product_quantity(mock_product), 999)
def test_cart_merge_in_cart_with_handling_and_coupons(self):
"""Test merge_in_cart copies handling options."""
from make_post_sell.models.cart import Cart
cart1 = Cart()
cart2 = Cart()
# Set up cart1
cart1.set_cart({"product1": 2})
cart1.handling_cost_in_cents = 100
# Set up cart2 with handling
cart2.set_cart({"product2": 3})
cart2.handling_option = "shipping"
cart2.handling_cost_in_cents = 200
# Initialize coupons as empty lists to avoid SQLAlchemy issues
cart1.coupons = []
cart2.coupons = []
# Merge cart2 into cart1
cart1.merge_in_cart(cart2)
# Check products merged
result = cart1.get_cart()
self.assertEqual(result["product1"], 2)
self.assertEqual(result["product2"], 3)
# Check handling option copied
self.assertEqual(cart1.handling_option, "shipping")
def test_cart_line_totals_property(self):
"""Test line_totals property converts cents to dollars."""
from make_post_sell.models.cart import Cart
cart = Cart()
# Mock line_totals_in_cents to return test data
with mock.patch.object(
type(cart), "line_totals_in_cents", new_callable=mock.PropertyMock
) as mock_line_totals_cents:
mock_line_totals_cents.return_value = {"product1": 1500, "product2": 2000}
line_totals = cart.line_totals
# Should convert cents to dollars
self.assertEqual(line_totals["product1"], 15.00)
self.assertEqual(line_totals["product2"], 20.00)
def test_cart_shop_totals_property(self):
"""Test shop_totals property converts shop totals from cents to dollars."""
from make_post_sell.models.cart import Cart
cart = Cart()
# Mock shop_totals_in_cents to return test data
with mock.patch.object(
type(cart), "shop_totals_in_cents", new_callable=mock.PropertyMock
) as mock_shop_totals_cents:
mock_shop_totals_cents.return_value = {"shop1": 2500, "shop2": 3000}
shop_totals = cart.shop_totals
# Should convert cents to dollars
self.assertEqual(shop_totals["shop1"], 25.00)
self.assertEqual(shop_totals["shop2"], 30.00)
def test_cart_discounted_shop_totals_property(self):
"""Test discounted_shop_totals property converts cents to dollars."""
from make_post_sell.models.cart import Cart
cart = Cart()
# Mock discounted_shop_totals_in_cents to return test data
with mock.patch.object(
type(cart),
"discounted_shop_totals_in_cents",
new_callable=mock.PropertyMock,
) as mock_discounted_cents:
mock_discounted_cents.return_value = {"shop1": 2000, "shop2": 2500}
discounted_totals = cart.discounted_shop_totals
# Should convert cents to dollars
self.assertEqual(discounted_totals["shop1"], 20.00)
self.assertEqual(discounted_totals["shop2"], 25.00)
def test_cart_human_timestamp_properties(self):
"""Test human_updated_timestamp and human_created_timestamp properties."""
from make_post_sell.models.cart import Cart
from make_post_sell.lib.time_funcs import timestamp_to_ago_string
cart = Cart()
# Mock the timestamp_to_ago_string function
with mock.patch(
"make_post_sell.models.cart.timestamp_to_ago_string"
) as mock_ago_string:
mock_ago_string.return_value = "2 hours ago"
# Test human_updated_timestamp
result = cart.human_updated_timestamp
mock_ago_string.assert_called_with(cart.updated_timestamp)
self.assertEqual(result, "2 hours ago")
# Reset mock for second test
mock_ago_string.reset_mock()
mock_ago_string.return_value = "3 hours ago"
# Test human_created_timestamp
result = cart.human_created_timestamp
mock_ago_string.assert_called_with(cart.created_timestamp)
self.assertEqual(result, "3 hours ago")
def test_cart_total_price_in_cents_with_handling(self):
"""Test total_price_in_cents includes handling cost."""
from make_post_sell.models.cart import Cart
cart = Cart()
cart.handling_cost_in_cents = 500 # $5.00 handling
# Mock line_totals_in_cents to return some total
with mock.patch.object(
type(cart), "line_totals_in_cents", new_callable=mock.PropertyMock
) as mock_line_totals:
mock_line_totals.return_value = {
"product1": 1000,
"product2": 1500,
} # $25.00
total = cart.total_price_in_cents
# Should include handling cost: 1000 + 1500 + 500 = 3000
self.assertEqual(total, 3000)
def test_cart_total_price_in_cents_no_handling(self):
"""Test total_price_in_cents without handling cost."""
from make_post_sell.models.cart import Cart
cart = Cart()
cart.handling_cost_in_cents = 0 # No handling
# Mock line_totals_in_cents to return some total
with mock.patch.object(
type(cart), "line_totals_in_cents", new_callable=mock.PropertyMock
) as mock_line_totals:
mock_line_totals.return_value = {"product1": 1000} # $10.00
total = cart.total_price_in_cents
# Should just be line totals: 1000
self.assertEqual(total, 1000)
def test_cart_update_handling_cost_local_pickup(self):
"""Test update_handling_cost for local pickup (free)."""
from make_post_sell.models.cart import Cart
cart = Cart()
cart.handling_option = "local_pickup"
mock_shop_location = mock.Mock()
cart.update_handling_cost(mock_shop_location)
# Local pickup should be free
self.assertEqual(cart.handling_cost_in_cents, 0)
def test_cart_update_handling_cost_local_delivery(self):
"""Test update_handling_cost for local delivery."""
from make_post_sell.models.cart import Cart
cart = Cart()
cart.handling_option = "local_delivery"
mock_shop_location = mock.Mock()
mock_shop_location.local_delivery_rate_in_cents = 500 # $5.00
cart.update_handling_cost(mock_shop_location)
# Should use local delivery rate
self.assertEqual(cart.handling_cost_in_cents, 500)
def test_cart_update_handling_cost_local_shipping(self):
"""Test update_handling_cost for local shipping."""
from make_post_sell.models.cart import Cart
cart = Cart()
cart.handling_option = "local_shipping"
mock_shop_location = mock.Mock()
mock_shop_location.local_shipping_rate_in_cents = 750 # $7.50
cart.update_handling_cost(mock_shop_location)
# Should use local shipping rate
self.assertEqual(cart.handling_cost_in_cents, 750)
def test_cart_update_handling_cost_international_shipping(self):
"""Test update_handling_cost for international shipping."""
from make_post_sell.models.cart import Cart
cart = Cart()
cart.handling_option = "international_shipping"
mock_shop_location = mock.Mock()
mock_shop_location.international_shipping_rate_in_cents = 1500 # $15.00
cart.update_handling_cost(mock_shop_location)
# Should use international shipping rate
self.assertEqual(cart.handling_cost_in_cents, 1500)
def test_transient_cart_count_is_zero(self):
"""Test that an in-memory Cart (not added to DB) has count=0."""
from make_post_sell.models.cart import Cart
cart = Cart()
self.assertEqual(cart.count, 0)
def test_transient_cart_is_empty(self):
"""Test that an in-memory Cart reports is_empty=True."""
from make_post_sell.models.cart import Cart
cart = Cart()
self.assertTrue(cart.is_empty)
def test_transient_cart_dbsession_is_none(self):
"""Test that a transient Cart (never added to a DB session) has no dbsession.
This is critical for lazy cart creation: the login flow checks
session_cart.dbsession to decide whether to call dbsession.delete().
"""
from make_post_sell.models.cart import Cart
cart = Cart()
self.assertIsNone(cart.dbsession)
def test_transient_cart_has_valid_uuid(self):
"""Test that a transient Cart has a valid UUID id and uuid_str."""
from make_post_sell.models.cart import Cart
cart = Cart()
self.assertIsNotNone(cart.id)
self.assertIsInstance(cart.uuid_str, str)
self.assertGreater(len(cart.uuid_str), 0)
def test_transient_cart_total_is_zero(self):
"""Test that an in-memory empty Cart returns total=0.00.
The nav bar accesses request.active_cart.total on every page.
For lazy carts, this must work without a DB session.
"""
from make_post_sell.models.cart import Cart
cart = Cart()
# Mock the products and coupons to avoid DB access
with mock.patch.object(
type(cart), "products", new_callable=mock.PropertyMock
) as mock_products:
mock_products.return_value = {}
with mock.patch.object(
type(cart), "coupons", new_callable=mock.PropertyMock
) as mock_coupons:
mock_coupons.return_value = []
self.assertEqual(cart.total, 0.00)
self.assertEqual(cart.count, 0)
def test_transient_cart_uuid_reusable(self):
"""Test that a Cart's id can be overwritten to reuse a UUID.
Lazy cart creation reuses the session UUID when recreating
an in-memory cart across requests.
"""
from make_post_sell.models.cart import Cart
import uuid as uuid_mod
cart = Cart()
original_id = cart.id
# Overwrite with a specific UUID (simulates lazy cart reuse)
new_uuid = uuid_mod.UUID("75b1de48-414b-11e7-aecf-9c4e369c7158")
cart.id = new_uuid
self.assertEqual(cart.id, new_uuid)
self.assertNotEqual(cart.id, original_id)
self.assertEqual(cart.uuid_str, str(new_uuid))
def test_cart_total_property_not_discounted(self):
"""Test Cart.total property when cart is not discounted (line 347)."""
from make_post_sell.models.cart import Cart
cart = Cart()
# Mock properties to ensure cart is not discounted
with mock.patch.object(
type(cart), "is_discounted", new_callable=mock.PropertyMock
) as mock_is_discounted:
with mock.patch.object(
type(cart), "total_price", new_callable=mock.PropertyMock
) as mock_total_price:
mock_is_discounted.return_value = False # Not discounted
mock_total_price.return_value = 25.00 # Regular price
# This should hit line 347: return self.total_price
total = cart.total
self.assertEqual(total, 25.00)
mock_total_price.assert_called_once()
class TestLazySessionCart(unittest.TestCase):
"""Test the lazy cart creation logic in add_session_cart.
Session carts are NOT persisted to the database until a product is
actually added. This prevents bots and crawlers from creating
empty cart rows on every page visit.
"""
def _make_mock_request(self, session_data=None, shop=None, db_cart=None):
"""Build a mock Pyramid request for testing add_session_cart."""
request = mock.MagicMock()
request.session = dict(session_data or {})
request.shop = shop
mock_dbsession = mock.MagicMock()
request.dbsession = mock_dbsession
# Make get_cart_by_id return db_cart when called
return request, db_cart
def test_returns_persisted_cart_when_found_in_db(self):
"""When session has a cart_id and it exists in the DB, return it."""
from make_post_sell.models.cart import Cart
persisted_cart = Cart()
request, _ = self._make_mock_request(
session_data={"active_cart_id": persisted_cart.uuid_str},
)
with mock.patch(
"make_post_sell.request_methods.get_cart_by_id",
return_value=persisted_cart,
) as mock_get:
# Import and call the function under test
from make_post_sell.request_methods import includeme
# We need to test the inner function directly. Extract it by
# calling includeme and capturing the registered function.
captured = {}
def fake_add_request_method(func, name, **kwargs):
captured[name] = func
config = mock.MagicMock()
config.get_settings.return_value = {}
config.add_request_method = fake_add_request_method
includeme(config)
add_session_cart = captured["session_cart"]
result = add_session_cart(request)
mock_get.assert_called_once_with(
request.dbsession, persisted_cart.uuid_str
)
self.assertEqual(result, persisted_cart)
def test_creates_inmemory_cart_when_no_session_id(self):
"""When session has no cart_id, create an in-memory Cart and store UUID."""
request, _ = self._make_mock_request()
with mock.patch(
"make_post_sell.request_methods.get_cart_by_id",
return_value=None,
):
captured = {}
def fake_add_request_method(func, name, **kwargs):
captured[name] = func
config = mock.MagicMock()
config.get_settings.return_value = {}
config.add_request_method = fake_add_request_method
from make_post_sell.request_methods import includeme
includeme(config)
add_session_cart = captured["session_cart"]
result = add_session_cart(request)
# Should have stored the cart UUID in the session
self.assertIn("active_cart_id", request.session)
self.assertEqual(request.session["active_cart_id"], result.uuid_str)
# Cart should NOT be added to the DB session
request.dbsession.add.assert_not_called()
request.dbsession.flush.assert_not_called()
# Cart should be usable (count=0, empty)
self.assertEqual(result.count, 0)
self.assertTrue(result.is_empty)
self.assertTrue(result.active)
def test_reuses_uuid_from_session_when_cart_not_in_db(self):
"""When session has a cart_id but DB returns None, reuse the same UUID."""
import uuid as uuid_mod
stale_uuid = str(uuid_mod.uuid4())
request, _ = self._make_mock_request(
session_data={"active_cart_id": stale_uuid},
)
with mock.patch(
"make_post_sell.request_methods.get_cart_by_id",
return_value=None,
):
captured = {}
def fake_add_request_method(func, name, **kwargs):
captured[name] = func
config = mock.MagicMock()
config.get_settings.return_value = {}
config.add_request_method = fake_add_request_method
from make_post_sell.request_methods import includeme
includeme(config)
add_session_cart = captured["session_cart"]
result = add_session_cart(request)
# UUID should match what was in the session
self.assertEqual(result.uuid_str, stale_uuid)
# Session should NOT be overwritten
self.assertEqual(request.session["active_cart_id"], stale_uuid)
# Cart should NOT be added to the DB
request.dbsession.add.assert_not_called()
class TestCoupon(unittest.TestCase):
@mock.patch("make_post_sell.models.user.is_user_name_available", mock_always_true)
def setUp(self):
# if the year is 2040 and this code is still in use and these tests
# break, it is safe to raise this date another +20 years.
future_date = "2040-01-15"
past_date = "2020-01-15"
self.shop = Shop(
"my-shop",
"860-555-5555",
"1 wayward way",
"my shop description",
)
self.coupon = Coupon(
shop=self.shop,
code="a-coupon-code",
description="a valid coupon",
action_type="dollar-off",
action_value=500,
max_redemptions=100,
max_redemptions_per_user=1,
expiration_date=future_date,
cart_qualifier=20,
)
self.old_coupon = Coupon(
shop=self.shop,
code="old-coupon-code",
description="a invalid coupon",
action_type="dollar-off",
action_value=500,
max_redemptions=100,
max_redemptions_per_user=1,
expiration_date=past_date,
cart_qualifier=20,
)
self.disabled_coupon = Coupon(
shop=self.shop,
code="disabled-coupon-code",
description="a disabled coupon",
action_type="dollar-off",
action_value=500,
max_redemptions=100,
max_redemptions_per_user=1,
expiration_date=future_date,
cart_qualifier=20,
)
self.disabled_coupon.disabled = True
def test_coupon_code(self):
self.assertNotEqual(self.coupon.code, "invalid-code")
self.assertEqual(self.coupon.code, "a-coupon-code")
# make sure coupon is not expired.
self.assertTrue(self.coupon.is_valid)
self.assertTrue(self.coupon.is_not_expired)
# test inverse.
self.assertFalse(self.coupon.is_expired)
self.assertFalse(self.coupon.is_not_valid)
def test_expired_coupon_code(self):
self.assertEqual(self.old_coupon.code, "old-coupon-code")
# make sure old_coupon is_expired and is_not_valid
self.assertTrue(self.old_coupon.is_expired)
self.assertTrue(self.old_coupon.is_not_valid)
# test inverse.
self.assertFalse(self.old_coupon.is_valid)
self.assertFalse(self.old_coupon.is_not_expired)
def test_disabled_coupon_code(self):
self.assertEqual(self.disabled_coupon.code, "disabled-coupon-code")
# make sure disabled_coupon is_not_expired and is_not_valid
self.assertTrue(self.disabled_coupon.is_not_expired)
self.assertTrue(self.disabled_coupon.is_not_valid)
# test inverse.
self.assertFalse(self.disabled_coupon.is_valid)
self.assertFalse(self.disabled_coupon.is_expired)
class TestProductS3Security(unittest.TestCase):
"""Test S3 security controls for product visibility."""
def setUp(self):
from ..models.product import Product
from ..models.shop import Shop
# Create test shop and product
self.shop = Shop(
"test-shop",
"555-555-5555",
"123 Test St",
"Test shop description",
)
self.product = Product("Test Product", "Test product description")
self.product.shop = self.shop
# Mock file metadata to simulate uploaded files
self.product.json_file_metadata = (
'{"extensions": {"product": "pdf", "preview": "jpg", "thumbnail1": "jpg"}}'
)
self.product._file_metadata = {
"extensions": {"product": "pdf", "preview": "jpg", "thumbnail1": "jpg"}
}
def test_get_s3_acl_for_public_product(self):
"""Test S3 ACL logic for public products."""
self.product.visibility = 1 # public
# Product files should always be private
self.assertEqual(self.product.get_s3_acl_for_file_key("product"), "private")
# Public files should be public-read for public products
self.assertEqual(self.product.get_s3_acl_for_file_key("preview"), "public-read")
self.assertEqual(
self.product.get_s3_acl_for_file_key("thumbnail1"), "public-read"
)
def test_get_s3_acl_for_unlisted_product(self):
"""Test S3 ACL logic for unlisted products."""
self.product.visibility = 2 # unlisted
# Product files should always be private
self.assertEqual(self.product.get_s3_acl_for_file_key("product"), "private")
# Public files should be public-read for unlisted products
self.assertEqual(self.product.get_s3_acl_for_file_key("preview"), "public-read")
self.assertEqual(
self.product.get_s3_acl_for_file_key("thumbnail1"), "public-read"
)
def test_get_s3_acl_for_private_product(self):
"""Test S3 ACL logic for private products."""
self.product.visibility = 0 # private
# All files should be private for private products
self.assertEqual(self.product.get_s3_acl_for_file_key("product"), "private")
self.assertEqual(self.product.get_s3_acl_for_file_key("preview"), "private")
self.assertEqual(self.product.get_s3_acl_for_file_key("thumbnail1"), "private")
@mock.patch("builtins.print") # Mock print to avoid test output
def test_update_s3_acls_with_mock_client(self, mock_print):
"""Test S3 ACL updates with mocked S3 client."""
# Create mock S3 client
mock_s3_client = mock.Mock()
bucket_name = "test-bucket"
self.product.visibility = 1 # public
# Should call put_object_acl for each file
self.product.update_s3_acls(mock_s3_client, bucket_name)
# Verify put_object_acl was called for each file type
expected_calls = len(self.product.extensions)
self.assertEqual(mock_s3_client.put_object_acl.call_count, expected_calls)
def test_set_visibility_without_s3_client(self):
"""Test set_visibility without S3 client (no ACL updates)."""
old_visibility = self.product.visibility
new_visibility = 0 # private
# Should update visibility without error
self.product.set_visibility(new_visibility)
self.assertEqual(self.product.visibility, new_visibility)
@mock.patch("builtins.print") # Mock print to avoid test output
def test_set_visibility_with_s3_client(self, mock_print):
"""Test set_visibility with S3 client (should update ACLs)."""
mock_s3_client = mock.Mock()
bucket_name = "test-bucket"
old_visibility = 1 # public
new_visibility = 0 # private
self.product.visibility = old_visibility
# Should update visibility and call update_s3_acls
self.product.set_visibility(new_visibility, mock_s3_client, bucket_name)
self.assertEqual(self.product.visibility, new_visibility)
# Verify S3 ACL update was called
expected_calls = len(self.product.extensions)
self.assertEqual(mock_s3_client.put_object_acl.call_count, expected_calls)
def test_visibility_property_helpers(self):
"""Test visibility helper properties work correctly."""
# Test public
self.product.visibility = 1
self.assertTrue(self.product.is_public)
self.assertFalse(self.product.is_unlisted)
self.assertFalse(self.product.is_private)
# Test unlisted
self.product.visibility = 2
self.assertFalse(self.product.is_public)
self.assertTrue(self.product.is_unlisted)
self.assertFalse(self.product.is_private)
# Test private
self.product.visibility = 0
self.assertFalse(self.product.is_public)
self.assertFalse(self.product.is_unlisted)
self.assertTrue(self.product.is_private)
def test_karaoke_tracks_in_file_keys(self):
"""Karaoke tracks (instrumentals, vocals) are registered in file_keys."""
self.assertIn("instrumentals", self.product.file_keys)
self.assertIn("vocals", self.product.file_keys)
def test_karaoke_acl_public_product(self):
"""Karaoke tracks follow same ACL as product file for public products."""
self.product.visibility = 1
self.assertEqual(
self.product.get_s3_acl_for_file_key("instrumentals"),
self.product.get_s3_acl_for_file_key("product"),
)
self.assertEqual(
self.product.get_s3_acl_for_file_key("vocals"),
self.product.get_s3_acl_for_file_key("product"),
)
def test_karaoke_acl_private_product(self):
"""Karaoke tracks are private when product is private."""
self.product.visibility = 0
self.assertEqual(self.product.get_s3_acl_for_file_key("instrumentals"), "private")
self.assertEqual(self.product.get_s3_acl_for_file_key("vocals"), "private")
def test_karaoke_acl_unlisted_product(self):
"""Karaoke tracks follow same ACL as product file for unlisted products."""
self.product.visibility = 2
self.assertEqual(
self.product.get_s3_acl_for_file_key("instrumentals"),
self.product.get_s3_acl_for_file_key("product"),
)
self.assertEqual(
self.product.get_s3_acl_for_file_key("vocals"),
self.product.get_s3_acl_for_file_key("product"),
)
@mock.patch("builtins.print")
def test_update_s3_acls_includes_karaoke_tracks(self, mock_print):
"""update_s3_acls sets ACL on karaoke tracks when they exist in extensions."""
self.product._file_metadata = {
"extensions": {
"product": "mp3",
"thumbnail1": "jpg",
"instrumentals": "wav",
"vocals": "wav",
}
}
self.product.visibility = 1
mock_s3 = mock.Mock()
self.product.update_s3_acls(mock_s3, "test-bucket")
# Should call put_object_acl for all 4 file keys in extensions
self.assertEqual(mock_s3.put_object_acl.call_count, 4)
# Verify karaoke tracks were included by checking the Key arguments
called_keys = [
call.kwargs.get("Key") or call[1].get("Key")
for call in mock_s3.put_object_acl.call_args_list
]
s3_path = self.product.s3_path
self.assertIn(f"{s3_path}/instrumentals", called_keys)
self.assertIn(f"{s3_path}/vocals", called_keys)
class TestMetaFunctions(unittest.TestCase):
"""Test meta.py utility functions for base64 and UUID handling."""
def test_short_id_to_bytes(self):
"""Test short_id_to_bytes function converts base64 strings correctly."""
# Test case from the docstring
result = short_id_to_bytes("dbHeSEFLEeeuz5xONpxxWA")
expected = b"u\xb1\xdeHAK\x11\xe7\xae\xcf\x9cN6\x9cqX"
self.assertEqual(result, expected)
# Test with various base64-encoded strings
test_cases = [
"YWJjZA", # 'abcd' in base64
"dGVzdA", # 'test' in base64
"SGVsbG8", # 'Hello' in base64
]
for test_case in test_cases:
# Should not raise an exception
result = short_id_to_bytes(test_case)
self.assertIsInstance(result, bytes)
def test_id_to_uuid_with_short_id(self):
"""Test id_to_uuid function works with short base64 IDs."""
# Test case from the docstring
result = id_to_uuid("dbHeSEFLEeeuz5xONpxxWA")
expected = uuid.UUID("75b1de48-414b-11e7-aecf-9c4e369c7158")
self.assertEqual(result, expected)
def test_id_to_uuid_with_hex_string(self):
"""Test id_to_uuid function works with hex UUID strings."""
# Test case from the docstring
result = id_to_uuid("75b1de48414b11e7aecf9c4e369c7158")
expected = uuid.UUID("75b1de48-414b-11e7-aecf-9c4e369c7158")
self.assertEqual(result, expected)
def test_id_to_uuid_with_dashed_uuid(self):
"""Test id_to_uuid function works with dashed UUID strings."""
# Test case from the docstring
result = id_to_uuid("75b1de48-414b-11e7-aecf-9c4e369c7158")
expected = uuid.UUID("75b1de48-414b-11e7-aecf-9c4e369c7158")
self.assertEqual(result, expected)
def test_id_to_uuid_with_uuid_object(self):
"""Test id_to_uuid function returns UUID objects unchanged."""
test_uuid = uuid.UUID("75b1de48-414b-11e7-aecf-9c4e369c7158")
result = id_to_uuid(test_uuid)
self.assertEqual(result, test_uuid)
self.assertIs(result, test_uuid) # Should be the same object
def test_id_to_uuid_with_invalid_input(self):
"""Test id_to_uuid function returns None for invalid input."""
result = id_to_uuid("invalid-input")
self.assertIsNone(result)
result = id_to_uuid("")
self.assertIsNone(result)
def test_id_to_uuid_with_none_raises_typeerror(self):
"""Test id_to_uuid function raises TypeError for None input."""
# None input causes TypeError in uuid.UUID, not ValueError,
# so it's not caught by the exception handlers
with self.assertRaises(TypeError):
id_to_uuid(None)
class TestInvoice(unittest.TestCase):
"""Test Invoice model and its discount calculation logic."""
@mock.patch("make_post_sell.models.user.is_user_name_available", mock_always_true)
def setUp(self):
from ..models.invoice import Invoice, InvoiceLineItem
from ..models.product import Product
from ..models.price import Price
from ..models.coupon_redemption import CouponRedemption
# Create test shop
self.shop = Shop(
"test-shop",
"555-555-5555",
"123 Test St",
"Test shop description",
)
# Create test user
self.user = User("test@example.com")
# Create test product with price
self.product = Product("Test Product", "Test description")
self.product.shop = self.shop
# Create price for product
self.price = Price(self.product, 1000) # $10.00
# Create invoice
self.invoice = Invoice(self.user)
self.invoice.shop = self.shop
def test_subtotal_in_cents_single_item(self):
"""Test subtotal calculation with single line item."""
from ..models.invoice import InvoiceLineItem
# Add line item
line_item = InvoiceLineItem(
invoice=self.invoice,
product=self.product,
price=self.price,
shop=self.shop,
quantity=1,
)
self.invoice.line_items.append(line_item)
self.assertEqual(self.invoice.subtotal_in_cents, 1000)
def test_subtotal_in_cents_multiple_items(self):
"""Test subtotal calculation with multiple line items."""
from ..models.invoice import InvoiceLineItem
from ..models.product import Product
from ..models.price import Price
# Add first line item
line_item1 = InvoiceLineItem(
invoice=self.invoice,
product=self.product,
price=self.price,
shop=self.shop,
quantity=2,
)
self.invoice.line_items.append(line_item1)
# Create second product
product2 = Product("Test Product 2", "Test description 2")
product2.shop = self.shop
price2 = Price(product2, 500) # $5.00
# Add second line item
line_item2 = InvoiceLineItem(
invoice=self.invoice,
product=product2,
price=price2,
shop=self.shop,
quantity=3,
)
self.invoice.line_items.append(line_item2)
# Should be (1000 * 2) + (500 * 3) = 2000 + 1500 = 3500
self.assertEqual(self.invoice.subtotal_in_cents, 3500)
def test_discount_amount_in_cents_no_coupons(self):
"""Test discount amount when no coupons applied."""
self.assertEqual(self.invoice.discount_amount_in_cents, 0)
def test_discount_amount_in_cents_dollar_off_coupon(self):
"""Test discount amount with dollar-off coupon."""
from ..models.coupon_redemption import CouponRedemption
from ..models.invoice import InvoiceLineItem
# Add line item ($10.00)
line_item = InvoiceLineItem(
invoice=self.invoice,
product=self.product,
price=self.price,
shop=self.shop,
quantity=1,
)
self.invoice.line_items.append(line_item)
# Create $3 off coupon
coupon = Coupon(
shop=self.shop,
code="TEST3OFF",
description="$3 off",
action_type="dollar-off",
action_value=3.00, # Will be converted to 300 cents
cart_qualifier=5.00, # $5 minimum
)
# Create coupon redemption
redemption = CouponRedemption(
coupon=coupon, invoice=self.invoice, shop=self.shop, user=self.user
)
self.invoice.coupon_redemptions.append(redemption)
# Should be $3.00 off = 300 cents
self.assertEqual(self.invoice.discount_amount_in_cents, 300)
def test_discount_amount_in_cents_percent_off_coupon(self):
"""Test discount amount with percent-off coupon."""
from ..models.coupon_redemption import CouponRedemption
from ..models.invoice import InvoiceLineItem
# Add line item ($10.00)
line_item = InvoiceLineItem(
invoice=self.invoice,
product=self.product,
price=self.price,
shop=self.shop,
quantity=1,
)
self.invoice.line_items.append(line_item)
# Create 25% off coupon
coupon = Coupon(
shop=self.shop,
code="TEST25PCT",
description="25% off",
action_type="percent-off",
action_value=25, # Will be converted to 0.25
)
# Create coupon redemption
redemption = CouponRedemption(
coupon=coupon, invoice=self.invoice, shop=self.shop, user=self.user
)
self.invoice.coupon_redemptions.append(redemption)
# Should be 25% of $10.00 = $2.50 = 250 cents
self.assertEqual(self.invoice.discount_amount_in_cents, 250)
def test_discount_amount_in_cents_exceeds_subtotal(self):
"""Test discount amount when coupon exceeds subtotal."""
from ..models.coupon_redemption import CouponRedemption
from ..models.invoice import InvoiceLineItem
# Add line item ($10.00)
line_item = InvoiceLineItem(
invoice=self.invoice,
product=self.product,
price=self.price,
shop=self.shop,
quantity=1,
)
self.invoice.line_items.append(line_item)
# Create $15 off coupon (more than subtotal)
coupon = Coupon(
shop=self.shop,
code="TEST15OFF",
description="$15 off",
action_type="dollar-off",
action_value=15.00, # Will be converted to 1500 cents
)
# Create coupon redemption
redemption = CouponRedemption(
coupon=coupon, invoice=self.invoice, shop=self.shop, user=self.user
)
self.invoice.coupon_redemptions.append(redemption)
# Should be full $10.00 = 1000 cents (can't discount more than subtotal)
self.assertEqual(self.invoice.discount_amount_in_cents, 1000)
def test_total_in_cents_with_handling_and_discount(self):
"""Test total calculation with subtotal, discount, and handling."""
from ..models.coupon_redemption import CouponRedemption
from ..models.invoice import InvoiceLineItem
# Add line item ($10.00)
line_item = InvoiceLineItem(
invoice=self.invoice,
product=self.product,
price=self.price,
shop=self.shop,
quantity=1,
)
self.invoice.line_items.append(line_item)
# Add handling cost
self.invoice.handling_cost_in_cents = 500 # $5.00
# Create $3 off coupon
coupon = Coupon(
shop=self.shop,
code="TEST3OFF",
description="$3 off",
action_type="dollar-off",
action_value=3.00,
)
# Create coupon redemption
redemption = CouponRedemption(
coupon=coupon, invoice=self.invoice, shop=self.shop, user=self.user
)
self.invoice.coupon_redemptions.append(redemption)
# Should be: $10.00 - $3.00 + $5.00 = $12.00 = 1200 cents
self.assertEqual(self.invoice.total_in_cents, 1200)
def test_total_in_cents_never_negative(self):
"""Test that total never goes negative even with large discount."""
from ..models.coupon_redemption import CouponRedemption
from ..models.invoice import InvoiceLineItem
# Add line item ($10.00)
line_item = InvoiceLineItem(
invoice=self.invoice,
product=self.product,
price=self.price,
shop=self.shop,
quantity=1,
)
self.invoice.line_items.append(line_item)
# Create $20 off coupon (more than subtotal)
coupon = Coupon(
shop=self.shop,
code="TEST20OFF",
description="$20 off",
action_type="dollar-off",
action_value=20.00,
)
# Create coupon redemption
redemption = CouponRedemption(
coupon=coupon, invoice=self.invoice, shop=self.shop, user=self.user
)
self.invoice.coupon_redemptions.append(redemption)
# Should be 0, not negative
self.assertEqual(self.invoice.total_in_cents, 0)
def test_requires_payment_with_free_invoice(self):
"""Test requires_payment returns False for free invoice."""
from ..models.coupon_redemption import CouponRedemption
from ..models.invoice import InvoiceLineItem
# Add line item ($10.00)
line_item = InvoiceLineItem(
invoice=self.invoice,
product=self.product,
price=self.price,
shop=self.shop,
quantity=1,
)
self.invoice.line_items.append(line_item)
# Create $10 off coupon (makes it free)
coupon = Coupon(
shop=self.shop,
code="TEST10OFF",
description="$10 off",
action_type="dollar-off",
action_value=10.00,
)
# Create coupon redemption
redemption = CouponRedemption(
coupon=coupon, invoice=self.invoice, shop=self.shop, user=self.user
)
self.invoice.coupon_redemptions.append(redemption)
# Total should be 0
self.assertEqual(self.invoice.total_in_cents, 0)
# Should not require payment
self.assertFalse(self.invoice.requires_payment)
def test_requires_payment_at_threshold(self):
"""Test requires_payment at exactly 64 cents threshold."""
from ..models.invoice import InvoiceLineItem
from ..models.product import Product
from ..models.price import Price
# Create product with exact threshold price
product = Product("Cheap Product", "Test")
product.shop = self.shop
price = Price(product, 64) # Exactly 64 cents
# Add line item
line_item = InvoiceLineItem(
invoice=self.invoice,
product=product,
price=price,
shop=self.shop,
quantity=1,
)
self.invoice.line_items.append(line_item)
# At threshold, should not require payment
self.assertEqual(self.invoice.total_in_cents, 64)
self.assertFalse(self.invoice.requires_payment)
def test_requires_payment_above_threshold(self):
"""Test requires_payment above 64 cents threshold."""
from ..models.invoice import InvoiceLineItem
from ..models.product import Product
from ..models.price import Price
# Create product above threshold price
product = Product("Cheap Product", "Test")
product.shop = self.shop
price = Price(product, 65) # 65 cents
# Add line item
line_item = InvoiceLineItem(
invoice=self.invoice,
product=product,
price=price,
shop=self.shop,
quantity=1,
)
self.invoice.line_items.append(line_item)
# Above threshold, should require payment
self.assertEqual(self.invoice.total_in_cents, 65)
self.assertTrue(self.invoice.requires_payment)
def test_invalid_coupon_not_applied(self):
"""Test that invalid (expired/disabled) coupons don't apply discount."""
from ..models.coupon_redemption import CouponRedemption
from ..models.invoice import InvoiceLineItem
# Add line item ($10.00)
line_item = InvoiceLineItem(
invoice=self.invoice,
product=self.product,
price=self.price,
shop=self.shop,
quantity=1,
)
self.invoice.line_items.append(line_item)
# Create disabled coupon
coupon = Coupon(
shop=self.shop,
code="DISABLED",
description="$5 off",
action_type="dollar-off",
action_value=5.00,
)
coupon.disabled = True # Disable the coupon
# Create coupon redemption
redemption = CouponRedemption(
coupon=coupon, invoice=self.invoice, shop=self.shop, user=self.user
)
self.invoice.coupon_redemptions.append(redemption)
# Should have no discount since coupon is disabled
self.assertEqual(self.invoice.discount_amount_in_cents, 0)
self.assertEqual(self.invoice.total_in_cents, 1000)
def test_invoice_line_item_automatic_price_from_product_regression(self):
"""Test the production failure scenario: InvoiceLineItem automatically getting price from product.current_price."""
from ..models.invoice import InvoiceLineItem
from ..models.product import Product
from ..models.price import Price
# Create product (simulating production scenario)
product = Product("Production Test Product", "Test")
product.shop = self.shop
# Create price for product (this sets up the price_history relationship)
price = Price(product, 750) # $7.50
# Create line item explicitly passing price (this is what the fix enables)
# In production, this would call product.current_price automatically
# but unit tests need explicit price due to no database session
line_item = InvoiceLineItem(
invoice=self.invoice,
product=product,
price=price, # Explicit price for unit test
shop=self.shop,
quantity=2, # 2x $7.50 = $15.00
)
self.invoice.line_items.append(line_item)
# Test that the invoice can calculate subtotal_in_cents correctly
# This tests the line: item.price.price_in_cents * item.quantity
expected_subtotal = 750 * 2 # $15.00
self.assertEqual(self.invoice.subtotal_in_cents, expected_subtotal)
# Test that total_in_cents works (accessing the new properties)
expected_total = expected_subtotal # No discounts or handling
self.assertEqual(self.invoice.total_in_cents, expected_total)
# Test that the Price object is correctly linked
self.assertEqual(line_item.price.price_in_cents, 750)
self.assertEqual(line_item.product, product)
self.assertEqual(line_item.quantity, 2)
class TestUserCryptoRefundAddress(unittest.TestCase):
@mock.patch("make_post_sell.models.user.is_user_name_available", mock_always_true)
def setUp(self):
self.user = User("crypto@example.com")
self.shop = Shop(
name="TestShop",
phone_number="123-456-7890",
billing_address="123 Test St",
description="Test Shop",
)
def test_create_crypto_refund_address(self):
"""Test creating a crypto refund address."""
from ..models.user_crypto_refund_address import UserCryptoRefundAddress
addr = UserCryptoRefundAddress(
user=self.user,
shop=self.shop,
coin_type="XMR",
address="44AFFq5kSiGBoZ4NMDwYtN18obc8AemS33DBLWs3H7otXft3XjrpDtQGv7SqSsaBYBb98uNbr2VBBEt7f2wfn3RVGQBEP3A",
label="My Monero Wallet",
)
self.assertEqual(addr.user, self.user)
self.assertEqual(addr.coin_type, "XMR")
self.assertEqual(
addr.address,
"44AFFq5kSiGBoZ4NMDwYtN18obc8AemS33DBLWs3H7otXft3XjrpDtQGv7SqSsaBYBb98uNbr2VBBEt7f2wfn3RVGQBEP3A",
)
self.assertEqual(addr.label, "My Monero Wallet")
self.assertIsNotNone(addr.id)
self.assertGreater(addr.created_timestamp, 0)
self.assertGreater(addr.updated_timestamp, 0)
def test_create_crypto_refund_address_uppercase_coin_type(self):
"""Test that coin type is uppercased."""
from ..models.user_crypto_refund_address import UserCryptoRefundAddress
addr = UserCryptoRefundAddress(
user=self.user,
shop=self.shop,
coin_type="btc",
address="1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa",
)
self.assertEqual(addr.coin_type, "BTC")
def test_create_crypto_refund_address_no_label(self):
"""Test creating address without label."""
from ..models.user_crypto_refund_address import UserCryptoRefundAddress
addr = UserCryptoRefundAddress(
user=self.user,
shop=self.shop,
coin_type="DOGE",
address="DH5yaieqoZN36fDVciNyRueRGvGLR3mr7L",
)
self.assertEqual(addr.coin_type, "DOGE")
self.assertIsNone(addr.label)
def test_get_user_crypto_refund_address_none(self):
"""Test getting non-existent refund address returns None."""
from ..models.user_crypto_refund_address import get_user_crypto_refund_address
# Mock dbsession
mock_session = mock.Mock()
mock_query = mock.Mock()
mock_filter = mock.Mock()
mock_session.query.return_value = mock_query
mock_query.filter.return_value = mock_filter
mock_filter.first.return_value = None
result = get_user_crypto_refund_address(
mock_session, self.user, self.shop, "XMR"
)
self.assertIsNone(result)
def test_get_user_crypto_refund_address_found(self):
"""Test getting existing refund address."""
from ..models.user_crypto_refund_address import (
UserCryptoRefundAddress,
get_user_crypto_refund_address,
)
# Create address
expected_addr = UserCryptoRefundAddress(
user=self.user,
shop=self.shop,
coin_type="XMR",
address="44AFFq5kSiGBoZ4NMDwYtN18obc8AemS33DBLWs3H7otXft3XjrpDtQGv7SqSsaBYBb98uNbr2VBBEt7f2wfn3RVGQBEP3A",
)
# Mock dbsession
mock_session = mock.Mock()
mock_query = mock.Mock()
mock_filter = mock.Mock()
mock_session.query.return_value = mock_query
mock_query.filter.return_value = mock_filter
mock_filter.first.return_value = expected_addr
result = get_user_crypto_refund_address(
mock_session, self.user, self.shop, "XMR"
)
self.assertEqual(result, expected_addr)
class TestCryptoPayment(unittest.TestCase):
"""Unit tests for CryptoPayment model string representation methods."""
@mock.patch("make_post_sell.models.user.is_user_name_available", mock_always_true)
def setUp(self):
"""Set up test fixtures."""
self.user = User("alice@example.com")
self.user.name = "Alice"
self.shop = Shop(
name="Alice's Shop",
phone_number="555-1234",
billing_address="123 Main St",
description="Test shop",
)
# Create test payment without invoice for simpler testing
self.payment = CryptoPayment(
invoice=None, # Skip invoice relationship for unit tests
user=self.user,
shop=self.shop,
address="44AFFq5kSiGBoZ4NMDwYtN18obc8AemS33DBLWs3H7otXft3XjrpDtQGv7SqSsaBYBb98uNbr2VBBEt7f2wfn3RVGQBEP3A",
account_index=0,
subaddress_index=1,
coin_type="XMR",
expected_amount=1000000000000, # 1.0 XMR in piconero
rate_locked_usd_per_coin=150.00,
quote_expires_at_ms=now_timestamp() + 3600000, # 1 hour from now
confirmations_required=10,
)
def test_str_basic_pending_payment(self):
"""Test __str__ for basic pending payment."""
result = str(self.payment)
# Should include payment ID, status, coin type, amounts, confirmations, address, user, shop
self.assertIn(self.payment.uuid_str[:8], result)
self.assertIn("[pending]", result) # status shown as-is
self.assertIn("XMR", result)
self.assertIn(
"0/1", result
) # received/expected in XMR (trailing zeros stripped)
self.assertIn("conf:0/10", result)
self.assertIn("addr:", result)
self.assertIn("user:Alice", result)
self.assertIn("shop:Alice's Shop", result)
def test_str_received_payment_with_confirmations(self):
"""Test __str__ for received payment with some confirmations."""
self.payment.status = CryptoPayment.STATUS_RECEIVED
self.payment.received_amount = 800000000000 # 0.8 XMR
self.payment.current_confirmations = 5
result = str(self.payment)
self.assertIn("[received]", result)
self.assertIn("0.8/1", result) # trailing zeros stripped
self.assertIn("conf:5/10", result)
def test_str_confirmed_overpayment(self):
"""Test __str__ for confirmed overpayment."""
self.payment.status = CryptoPayment.STATUS_CONFIRMED_OVERPAY
self.payment.received_amount = 1200000000000 # 1.2 XMR
self.payment.current_confirmations = 12
result = str(self.payment)
self.assertIn("[confirmed-overpay]", result) # actual status string
self.assertIn("1.2/1", result) # trailing zeros stripped
self.assertIn("conf:12/10", result)
def test_str_dogecoin_precision(self):
"""Test __str__ with DOGE amounts (8 decimal places)."""
self.payment.coin_type = "DOGE"
self.payment.expected_amount = 100000000000 # 1000.0 DOGE in satoshis
self.payment.received_amount = 95000000000 # 950.0 DOGE
result = str(self.payment)
self.assertIn("DOGE", result)
self.assertIn("950/1000", result) # trailing zeros stripped
def test_str_no_user_display_name_fallback_to_username(self):
"""Test __str__ shows user.name field."""
# Since User model has 'name' field, not 'display_name' or 'username'
self.user.name = "alice_user"
result = str(self.payment)
self.assertIn("user:alice_user", result)
self.assertNotIn("user:Alice", result)
def test_str_no_user_info(self):
"""Test __str__ when user has no name."""
self.user.name = None
result = str(self.payment)
# Should not include user info but should still include shop
self.assertNotIn("user:", result)
self.assertIn("shop:Alice's Shop", result)
def test_str_no_shop_name(self):
"""Test __str__ when shop has no name."""
self.shop.name = None
result = str(self.payment)
# Should not include shop info but should still include user
self.assertNotIn("shop:", result)
self.assertIn("user:Alice", result)
def test_repr_method(self):
"""Test __repr__ method for debugging."""
result = repr(self.payment)
self.assertIn("CryptoPayment(", result)
self.assertIn(f"id={self.payment.uuid_str[:8]}", result)
self.assertIn("status=pending", result)
self.assertIn("coin=XMR", result)
self.assertIn("expected=1000000000000", result)
self.assertIn("received=0", result)
self.assertIn("confirmations=0/10", result) # Should show 0 due to our fix
def test_format_amount_details_xmr(self):
"""Test format_amount_details for XMR."""
self.payment.received_amount = 800000000000 # 0.8 XMR
result = self.payment.format_amount_details()
self.assertIn("expected:1", result)
self.assertIn("received:0.8", result)
self.assertIn("due:0.2", result)
self.assertIn("XMR", result)
def test_format_amount_details_doge(self):
"""Test format_amount_details for DOGE."""
self.payment.coin_type = "DOGE"
self.payment.expected_amount = 100000000000 # 1000.0 DOGE
self.payment.received_amount = 95000000000 # 950.0 DOGE
result = self.payment.format_amount_details()
self.assertIn("expected:1000", result)
self.assertIn("received:950", result)
self.assertIn("due:50", result)
self.assertIn("DOGE", result)
def test_format_amount_details_no_due_amount(self):
"""Test format_amount_details when fully paid."""
self.payment.received_amount = 1000000000000 # 1.0 XMR - exactly expected
result = self.payment.format_amount_details()
self.assertIn("due:0", result)
def test_format_confirmation_status_pending(self):
"""Test format_confirmation_status for pending confirmations."""
self.payment.current_confirmations = 5
result = self.payment.format_confirmation_status()
self.assertEqual(result, "5/10")
def test_format_confirmation_status_confirmed(self):
"""Test format_confirmation_status for fully confirmed."""
self.payment.current_confirmations = 15 # More than required
result = self.payment.format_confirmation_status()
self.assertEqual(result, "15/10")
class TestPayPalUserShop(unittest.TestCase):
"""Unit tests for PayPalUserShop model."""
@mock.patch("make_post_sell.models.user.is_user_name_available", mock_always_true)
def setUp(self):
self.user = User("paypal@example.com")
self.shop = Shop(
name="PayPal Test Shop",
phone_number="555-1234",
billing_address="123 Main St",
description="Test shop",
)
def test_create_paypal_user_shop(self):
"""Test creating a PayPalUserShop record."""
from ..models.paypal_user_shop import PayPalUserShop
pus = PayPalUserShop(user=self.user, shop=self.shop)
self.assertEqual(pus.user, self.user)
self.assertEqual(pus.shop, self.shop)
self.assertIsNotNone(pus.id)
self.assertIsNone(pus.payer_id)
self.assertIsNone(pus.billing_agreement_id)
self.assertIsNone(pus.active_payment_token)
def test_has_billing_agreement_false(self):
"""Test has_billing_agreement when no agreement exists."""
from ..models.paypal_user_shop import PayPalUserShop
pus = PayPalUserShop(user=self.user, shop=self.shop)
self.assertFalse(pus.has_billing_agreement)
def test_has_billing_agreement_true(self):
"""Test has_billing_agreement when agreement exists."""
from ..models.paypal_user_shop import PayPalUserShop
pus = PayPalUserShop(user=self.user, shop=self.shop)
pus.billing_agreement_id = "BA-1234567890"
self.assertTrue(pus.has_billing_agreement)
def test_has_saved_payment_method_false(self):
"""Test has_saved_payment_method when no token exists."""
from ..models.paypal_user_shop import PayPalUserShop
pus = PayPalUserShop(user=self.user, shop=self.shop)
self.assertFalse(pus.has_saved_payment_method)
def test_has_saved_payment_method_true(self):
"""Test has_saved_payment_method when token exists."""
from ..models.paypal_user_shop import PayPalUserShop
pus = PayPalUserShop(user=self.user, shop=self.shop)
pus.active_payment_token = "VAULT-TOKEN-123"
self.assertTrue(pus.has_saved_payment_method)
def test_payer_id_storage(self):
"""Test storing payer ID."""
from ..models.paypal_user_shop import PayPalUserShop
pus = PayPalUserShop(user=self.user, shop=self.shop)
pus.payer_id = "PAYERID123ABC"
self.assertEqual(pus.payer_id, "PAYERID123ABC")
class TestInvoicePayPal(unittest.TestCase):
"""Unit tests for Invoice PayPal functionality."""
@mock.patch("make_post_sell.models.user.is_user_name_available", mock_always_true)
def setUp(self):
self.user = User("paypal_invoice@example.com")
self.shop = Shop(
name="PayPal Invoice Shop",
phone_number="555-1234",
billing_address="123 Main St",
description="Test shop",
)
def test_invoice_payment_method_stripe_default(self):
"""Test that invoice with no PayPal or crypto returns stripe."""
invoice = Invoice(self.user)
invoice.shop = self.shop
self.assertEqual(invoice.payment_method, "stripe")
def test_invoice_payment_method_paypal(self):
"""Test that invoice with paypal_order_id returns paypal."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.paypal_order_id = "PAYPAL-ORDER-123"
self.assertEqual(invoice.payment_method, "paypal")
def test_invoice_paypal_columns_nullable(self):
"""Test that PayPal columns are nullable by default."""
invoice = Invoice(self.user)
invoice.shop = self.shop
self.assertIsNone(invoice.paypal_order_id)
self.assertIsNone(invoice.paypal_capture_id)
def test_invoice_paypal_capture_id_storage(self):
"""Test storing PayPal capture ID."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.paypal_order_id = "PAYPAL-ORDER-123"
invoice.paypal_capture_id = "CAPTURE-456"
self.assertEqual(invoice.paypal_order_id, "PAYPAL-ORDER-123")
self.assertEqual(invoice.paypal_capture_id, "CAPTURE-456")
def test_invoice_payment_status_for_paypal(self):
"""Test that PayPal invoice payment_status is 'paid'."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.paypal_order_id = "PAYPAL-ORDER-123"
# PayPal invoices (like Stripe) are "paid" if they exist
self.assertEqual(invoice.payment_status, "paid")
def test_invoice_is_paid_for_paypal(self):
"""Test that PayPal invoice is_paid returns True."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.paypal_order_id = "PAYPAL-ORDER-123"
self.assertTrue(invoice.is_paid)
class TestInvoiceStripe(unittest.TestCase):
"""Unit tests for Invoice Stripe functionality."""
@mock.patch("make_post_sell.models.user.is_user_name_available", mock_always_true)
def setUp(self):
self.user = User("stripe_invoice@example.com")
self.shop = Shop(
name="Stripe Invoice Shop",
phone_number="555-1234",
billing_address="123 Main St",
description="Test shop",
)
def test_invoice_stripe_columns_nullable(self):
"""Test that Stripe columns are nullable by default."""
invoice = Invoice(self.user)
invoice.shop = self.shop
self.assertIsNone(invoice.stripe_payment_intent_id)
self.assertIsNone(invoice.stripe_charge_id)
def test_invoice_stripe_payment_intent_id_storage(self):
"""Test storing Stripe payment intent ID."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
self.assertEqual(invoice.stripe_payment_intent_id, "pi_1234567890abcdef")
def test_invoice_stripe_charge_id_storage(self):
"""Test storing Stripe charge ID."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
invoice.stripe_charge_id = "ch_1234567890abcdef"
self.assertEqual(invoice.stripe_charge_id, "ch_1234567890abcdef")
def test_invoice_payment_method_stripe_explicit(self):
"""Test that invoice with stripe_payment_intent_id returns stripe."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
self.assertEqual(invoice.payment_method, "stripe")
def test_invoice_payment_method_stripe_default(self):
"""Test that invoice with no payment refs returns stripe (legacy)."""
invoice = Invoice(self.user)
invoice.shop = self.shop
# Legacy invoices without payment tracking default to stripe
self.assertEqual(invoice.payment_method, "stripe")
def test_invoice_payment_status_for_stripe(self):
"""Test that Stripe invoice payment_status is 'paid'."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
self.assertEqual(invoice.payment_status, "paid")
def test_invoice_is_paid_for_stripe(self):
"""Test that Stripe invoice is_paid returns True."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
self.assertTrue(invoice.is_paid)
def test_invoice_payment_method_priority_paypal_over_stripe(self):
"""Test that PayPal takes priority if both are set (edge case)."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.paypal_order_id = "PAYPAL-ORDER-123"
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
# PayPal is checked first in payment_method
self.assertEqual(invoice.payment_method, "paypal")
def test_invoice_payment_method_adyen(self):
"""Test that Adyen payment method is detected correctly."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.adyen_psp_reference = "PSP123456789"
self.assertEqual(invoice.payment_method, "adyen")
def test_invoice_payment_method_priority_paypal_over_adyen(self):
"""Test that PayPal takes priority over Adyen (edge case)."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.paypal_order_id = "PAYPAL-ORDER-123"
invoice.adyen_psp_reference = "PSP123456789"
# PayPal is checked first in payment_method
self.assertEqual(invoice.payment_method, "paypal")
def test_invoice_payment_method_priority_adyen_over_stripe(self):
"""Test that Adyen takes priority over Stripe (edge case)."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.adyen_psp_reference = "PSP123456789"
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
# Adyen is checked before Stripe in payment_method
self.assertEqual(invoice.payment_method, "adyen")
class TestAdyen(unittest.TestCase):
"""Test Adyen-related model functionality."""
@mock.patch("make_post_sell.models.user.is_user_name_available", mock_always_true)
def setUp(self):
# Create test shop
self.shop = Shop(
"test-shop",
"555-555-5555",
"123 Test St",
"Test shop description",
)
# Create test user
self.user = User("test@example.com")
def test_shop_is_adyen_ready_without_credentials(self):
"""Test that shop is not Adyen-ready without credentials."""
self.assertFalse(self.shop.is_adyen_ready)
def test_shop_is_adyen_ready_with_api_key_only(self):
"""Test that shop is not Adyen-ready with just API key."""
self.shop.adyen_api_key = "test_api_key"
self.assertFalse(self.shop.is_adyen_ready)
def test_shop_is_adyen_ready_with_merchant_account_only(self):
"""Test that shop is not Adyen-ready with just merchant account."""
self.shop.adyen_merchant_account = "TestMerchant"
self.assertFalse(self.shop.is_adyen_ready)
def test_shop_is_adyen_ready_with_both_credentials(self):
"""Test that shop is Adyen-ready with both API key and merchant account."""
self.shop.adyen_api_key = "test_api_key"
self.shop.adyen_merchant_account = "TestMerchant"
self.assertTrue(self.shop.is_adyen_ready)
def test_shop_is_adyen_not_ready(self):
"""Test the inverse property for convenience."""
self.assertTrue(self.shop.is_adyen_not_ready)
self.shop.adyen_api_key = "test_api_key"
self.shop.adyen_merchant_account = "TestMerchant"
self.assertFalse(self.shop.is_adyen_not_ready)
def test_shop_adyen_enabled_default(self):
"""Test that Adyen enabled is None before DB insert (server_default handles it)."""
# Before DB insert, the value is None (server_default of '1' applies on insert)
# When retrieved from DB after insert, it would be True
self.assertIsNone(self.shop.adyen_enabled)
def test_invoice_adyen_psp_reference_default(self):
"""Test that Adyen PSP reference is None by default."""
invoice = Invoice(self.user)
invoice.shop = self.shop
self.assertIsNone(invoice.adyen_psp_reference)
def test_invoice_adyen_psp_reference_set(self):
"""Test setting Adyen PSP reference."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.adyen_psp_reference = "882619391893263J"
self.assertEqual(invoice.adyen_psp_reference, "882619391893263J")
class TestStemmer(unittest.TestCase):
"""Test the stemmer and related products functions."""
def test_stem_word_basic(self):
"""Test common suffixes stripped correctly."""
from ..models.product import stem_word
self.assertEqual(stem_word("running"), "runn")
self.assertEqual(stem_word("played"), "play")
self.assertEqual(stem_word("quickly"), "quick")
self.assertEqual(stem_word("cats"), "cat")
self.assertEqual(stem_word("boxes"), "box")
self.assertEqual(stem_word("creation"), "crea")
self.assertEqual(stem_word("movement"), "move")
self.assertEqual(stem_word("darkness"), "dark")
self.assertEqual(stem_word("readable"), "read")
self.assertEqual(stem_word("hopeful"), "hope")
self.assertEqual(stem_word("careless"), "care")
self.assertEqual(stem_word("dangerous"), "danger")
self.assertEqual(stem_word("creative"), "creat")
self.assertEqual(stem_word("bigger"), "bigg")
self.assertEqual(stem_word("biggest"), "bigg")
def test_stem_word_short_preserved(self):
"""Test words too short to stem are unchanged."""
from ..models.product import stem_word
# Words where remaining stem would be < 3 chars should not be stripped
self.assertEqual(stem_word("it"), "it")
self.assertEqual(stem_word("is"), "is")
self.assertEqual(stem_word("an"), "an")
self.assertEqual(stem_word("the"), "the")
self.assertEqual(stem_word("bed"), "bed")
def test_stem_word_case_insensitive(self):
"""Test stemming is case insensitive."""
from ..models.product import stem_word
self.assertEqual(stem_word("Running"), stem_word("running"))
self.assertEqual(stem_word("PLAYED"), stem_word("played"))
def test_tokenize_and_stem(self):
"""Test sentence -> set of stems."""
from ..models.product import tokenize_and_stem
result = tokenize_and_stem("The cats are running quickly")
# "the" -> "the" (3 chars, no suffix stripped)
# "cats" -> "cat", "are" -> "are", "running" -> "runn", "quickly" -> "quick"
self.assertIn("cat", result)
self.assertIn("runn", result)
self.assertIn("quick", result)
self.assertIn("are", result)
self.assertIn("the", result)
# Short words (<3 chars) are excluded by the regex
self.assertNotIn("is", result)
self.assertNotIn("a", result)
def test_tokenize_and_stem_empty(self):
"""Test empty string returns empty set."""
from ..models.product import tokenize_and_stem
self.assertEqual(tokenize_and_stem(""), set())
self.assertEqual(tokenize_and_stem("12 34"), set())
def test_get_related_products(self):
"""Test returns ranked matches, excludes self and non-public."""
from ..models.product import Product, get_related_products
shop = Shop(
"test-shop",
"555-555-5555",
"123 Test St",
"Test shop description",
)
# Create products with related titles
product1 = Product("Blues Guitar Lessons", "Learn acoustic guitar blues riffs")
product1.shop = shop
product1.visibility = 1
# shares: guitar, learn = 2 stems
product2 = Product("Jazz Guitar Tutorial", "Learn jazz techniques")
product2.shop = shop
product2.visibility = 1
# shares: learn = 1 stem (no guitar/blues overlap)
product3 = Product("Piano for Beginners", "Learn piano basics")
product3.shop = shop
product3.visibility = 1
# shares: guitar, blues, learn, riff = 4 stems (most overlap)
product4 = Product("Guitar Blues Collection", "Learn blues guitar riffs and licks")
product4.shop = shop
product4.visibility = 1
product5_private = Product("Secret Guitar Video", "Private guitar content")
product5_private.shop = shop
product5_private.visibility = 0 # private
# Mock shop.products to return our test products
all_products = [product1, product2, product3, product4, product5_private]
with mock.patch.object(type(shop), 'products', new_callable=mock.PropertyMock) as mock_products:
mock_products.return_value = all_products
related = get_related_products(product1)
# Should not include product1 itself
self.assertNotIn(product1, related)
# Should not include private product
self.assertNotIn(product5_private, related)
# Should include related products
related_titles = [p.title for p in related]
self.assertIn("Guitar Blues Collection", related_titles)
self.assertIn("Jazz Guitar Tutorial", related_titles)
self.assertIn("Piano for Beginners", related_titles)
# product4 shares most stems with product1
# so it should be ranked higher (earlier index) than product3
self.assertLess(related.index(product4), related.index(product3))
def test_get_related_products_empty_description(self):
"""Test with empty title/description still returns results via fallback."""
from ..models.product import Product, get_related_products
shop = Shop(
"test-shop",
"555-555-5555",
"123 Test St",
"Test shop description",
)
product = Product("", "")
product.shop = shop
product.visibility = 1
with mock.patch.object(type(shop), 'products', new_callable=mock.PropertyMock) as mock_products:
mock_products.return_value = []
related = get_related_products(product)
self.assertEqual(related, [])
def test_get_related_products_empty_description_with_others(self):
"""Test orphan product (no stems) gets fallback results."""
from ..models.product import Product, get_related_products
shop = Shop(
"test-shop",
"555-555-5555",
"123 Test St",
"Test shop description",
)
product = Product("", "")
product.shop = shop
product.visibility = 1
product.created_timestamp = 100
other1 = Product("Song One", "A great song")
other1.shop = shop
other1.visibility = 1
other1.created_timestamp = 200
other2 = Product("Song Two", "Another song")
other2.shop = shop
other2.visibility = 1
other2.created_timestamp = 300
all_products = [product, other1, other2]
with mock.patch.object(type(shop), 'products', new_callable=mock.PropertyMock) as mock_products:
mock_products.return_value = all_products
related = get_related_products(product)
# Should return fallback results (tier 3: any public product)
self.assertEqual(len(related), 2)
self.assertIn(other1, related)
self.assertIn(other2, related)
def test_get_related_products_limit(self):
"""Test that limit parameter is respected."""
from ..models.product import Product, get_related_products
shop = Shop(
"test-shop",
"555-555-5555",
"123 Test St",
"Test shop description",
)
product1 = Product("Guitar Music", "Guitar tutorial")
product1.shop = shop
product1.visibility = 1
# Create many related products
other_products = []
for i in range(15):
p = Product(f"Guitar Video {i}", f"Guitar lesson {i}")
p.shop = shop
p.visibility = 1
other_products.append(p)
all_products = [product1] + other_products
with mock.patch.object(type(shop), 'products', new_callable=mock.PropertyMock) as mock_products:
mock_products.return_value = all_products
# Default limit is 8
related = get_related_products(product1)
self.assertLessEqual(len(related), 8)
# Custom limit
related = get_related_products(product1, limit=3)
self.assertLessEqual(len(related), 3)
def test_get_related_products_same_media_type_fallback(self):
"""Test tier 2: same media type fills remaining slots."""
from ..models.product import Product, get_related_products
shop = Shop(
"test-shop",
"555-555-5555",
"123 Test St",
"Test shop description",
)
def _make_product(title, desc, timestamp, ext=None):
p = Product(title, desc)
p.shop = shop
p.visibility = 1
p.created_timestamp = timestamp
if ext:
p._file_metadata = {"extensions": {"product": ext}, "originals": {}}
return p
# Product with no stem overlap to anything
product1 = _make_product("Xylophone Jam", "Unique xylophone music", 100, "mp3")
# Same media type (audio) but no stem overlap
product2 = _make_product("Drums Solo", "Percussion performance", 200, "mp3")
# Different media type (video) and no stem overlap
product3 = _make_product("Cat Video", "Funny cat footage", 300, "mp4")
# Another audio with no stem overlap
product4 = _make_product("Bass Line", "Groovy bass track", 400, "wav")
all_products = [product1, product2, product3, product4]
with mock.patch.object(type(shop), 'products', new_callable=mock.PropertyMock) as mock_products:
mock_products.return_value = all_products
related = get_related_products(product1)
# Should include all 3 others
self.assertEqual(len(related), 3)
# Same media type (audio) should come before different type (video)
audio_indices = []
video_indices = []
for i, p in enumerate(related):
ext = p.extensions.get("product", "")
if ext in ("mp3", "wav"):
audio_indices.append(i)
elif ext == "mp4":
video_indices.append(i)
# All audio items should come before video items
if audio_indices and video_indices:
self.assertLess(max(audio_indices), min(video_indices))
def test_get_related_products_deterministic(self):
"""Test that results are deterministic (consistent ordering)."""
from ..models.product import Product, get_related_products
shop = Shop(
"test-shop",
"555-555-5555",
"123 Test St",
"Test shop description",
)
product1 = Product("Guitar Music", "Guitar tutorial video")
product1.shop = shop
product1.visibility = 1
product1.created_timestamp = 100
product2 = Product("Guitar Songs", "Guitar song collection")
product2.shop = shop
product2.visibility = 1
product2.created_timestamp = 200
product3 = Product("Guitar Chords", "Guitar chord lessons")
product3.shop = shop
product3.visibility = 1
product3.created_timestamp = 300
all_products = [product1, product2, product3]
with mock.patch.object(type(shop), 'products', new_callable=mock.PropertyMock) as mock_products:
mock_products.return_value = all_products
# Run multiple times to verify determinism
related1 = get_related_products(product1)
related2 = get_related_products(product1)
self.assertEqual(
[p.title for p in related1],
[p.title for p in related2],
)
class TestDiscoveryRing(unittest.TestCase):
"""Test the discovery ring precomputation algorithm."""
def _make_shop(self):
return Shop(
"test-shop",
"555-555-5555",
"123 Test St",
"Test shop description",
)
def _make_product(self, title, desc, timestamp, visibility=1):
from ..models.product import Product
p = Product(title, desc)
p.visibility = visibility
p.created_timestamp = timestamp
return p
def test_empty_shop(self):
"""Empty shop returns empty ring."""
from ..models.shop import compute_discovery_ring
shop = self._make_shop()
with mock.patch.object(type(shop), 'products', new_callable=mock.PropertyMock) as mp:
mp.return_value = []
self.assertEqual(compute_discovery_ring(shop), [])
def test_single_product(self):
"""Single public product returns ring of one."""
from ..models.shop import compute_discovery_ring
shop = self._make_shop()
p = self._make_product("Solo", "Only item", 100)
with mock.patch.object(type(shop), 'products', new_callable=mock.PropertyMock) as mp:
mp.return_value = [p]
ring = compute_discovery_ring(shop)
self.assertEqual(ring, [str(p.id)])
def test_all_public_products_included(self):
"""Every public product appears exactly once, no dupes."""
from ..models.shop import compute_discovery_ring
shop = self._make_shop()
products = [self._make_product(f"Item {i}", f"Description {i}", 100 + i) for i in range(10)]
with mock.patch.object(type(shop), 'products', new_callable=mock.PropertyMock) as mp:
mp.return_value = products
ring = compute_discovery_ring(shop)
ids = [str(p.id) for p in products]
self.assertEqual(len(ring), 10)
self.assertEqual(set(ring), set(ids))
# No duplicates
self.assertEqual(len(ring), len(set(ring)))
def test_private_and_unlisted_excluded(self):
"""Only visibility==1 products appear in the ring."""
from ..models.shop import compute_discovery_ring
shop = self._make_shop()
public = self._make_product("Public Song", "Visible", 300, visibility=1)
private = self._make_product("Private Song", "Hidden", 200, visibility=0)
unlisted = self._make_product("Unlisted Song", "Unlisted", 100, visibility=2)
with mock.patch.object(type(shop), 'products', new_callable=mock.PropertyMock) as mp:
mp.return_value = [public, private, unlisted]
ring = compute_discovery_ring(shop)
self.assertEqual(ring, [str(public.id)])
self.assertNotIn(str(private.id), ring)
self.assertNotIn(str(unlisted.id), ring)
def test_starts_with_newest(self):
"""Newest product (highest created_timestamp) is at position 0."""
from ..models.shop import compute_discovery_ring
shop = self._make_shop()
old = self._make_product("Old Song", "First release", 100)
mid = self._make_product("Mid Song", "Second release", 200)
new = self._make_product("New Song", "Latest release", 300)
with mock.patch.object(type(shop), 'products', new_callable=mock.PropertyMock) as mp:
mp.return_value = [old, mid, new]
ring = compute_discovery_ring(shop)
self.assertEqual(ring[0], str(new.id))
def test_similar_products_adjacent(self):
"""Guitar cluster should appear before unrelated piano cluster."""
from ..models.shop import compute_discovery_ring
shop = self._make_shop()
g1 = self._make_product("Blues Guitar Jam", "Electric guitar blues riffs", 400)
g2 = self._make_product("Guitar Solo Practice", "Learn guitar solo techniques", 300)
g3 = self._make_product("Acoustic Guitar Chords", "Guitar chord progressions", 200)
piano = self._make_product("Piano Sonata", "Classical piano performance", 100)
with mock.patch.object(type(shop), 'products', new_callable=mock.PropertyMock) as mp:
mp.return_value = [g1, g2, g3, piano]
ring = compute_discovery_ring(shop)
# All guitar items should cluster together before piano
guitar_ids = {str(g1.id), str(g2.id), str(g3.id)}
guitar_positions = [i for i, rid in enumerate(ring) if rid in guitar_ids]
piano_pos = ring.index(str(piano.id))
# Guitar items should be contiguous (adjacent)
self.assertEqual(max(guitar_positions) - min(guitar_positions), 2)
# Piano should be after all guitar items
self.assertGreater(piano_pos, max(guitar_positions))
def test_deterministic(self):
"""Same input produces same output every time."""
from ..models.shop import compute_discovery_ring
shop = self._make_shop()
products = [self._make_product(f"Song {i}", f"Description {i}", 100 + i) for i in range(8)]
with mock.patch.object(type(shop), 'products', new_callable=mock.PropertyMock) as mp:
mp.return_value = products
ring1 = compute_discovery_ring(shop)
ring2 = compute_discovery_ring(shop)
self.assertEqual(ring1, ring2)
def test_ring_is_list_of_strings(self):
"""Ring elements are UUID strings, not UUID objects."""
from ..models.shop import compute_discovery_ring
shop = self._make_shop()
p = self._make_product("Test Song", "Some description", 100)
with mock.patch.object(type(shop), 'products', new_callable=mock.PropertyMock) as mp:
mp.return_value = [p]
ring = compute_discovery_ring(shop)
self.assertIsInstance(ring, list)
for item in ring:
self.assertIsInstance(item, str)
def test_reforge_sets_shop_attribute(self):
"""reforge_discovery_ring stores the ring on the shop object."""
from ..models.shop import reforge_discovery_ring
shop = self._make_shop()
p1 = self._make_product("Song A", "Alpha", 200)
p2 = self._make_product("Song B", "Beta", 100)
with mock.patch.object(type(shop), 'products', new_callable=mock.PropertyMock) as mp:
mp.return_value = [p1, p2]
ring = reforge_discovery_ring(shop)
self.assertEqual(shop.discovery_ring, ring)
self.assertEqual(len(ring), 2)
# Verify JSON column was set
import json
self.assertEqual(json.loads(shop.json_discovery_ring), ring)
class TestAsyncDiscoveryRing(unittest.TestCase):
"""Test the async ring reforge with dirty bit debounce."""
def setUp(self):
# Reset module-level state between tests
from ..models import shop as shop_module
shop_module._reforge_running.clear()
shop_module._reforge_dirty.clear()
def tearDown(self):
from ..models import shop as shop_module
shop_module._reforge_running.clear()
shop_module._reforge_dirty.clear()
def test_async_reforge_spawns_thread(self):
"""reforge_discovery_ring_async starts a background thread."""
import threading
from ..models.shop import reforge_discovery_ring_async
initial_threads = threading.active_count()
started = threading.Event()
proceed = threading.Event()
mock_factory = mock.Mock()
mock_session = mock.Mock()
mock_factory.return_value.get_bind.return_value = "fake_bind"
mock_shop = mock.Mock()
mock_shop.products = []
mock_shop.name = "test"
mock_shop.environment = 0
mock_shop.is_non_production = False
# Patch Session and compute to block until we signal
with mock.patch("make_post_sell.models.shop.SASession") as MockSession:
MockSession.return_value = mock_session
mock_session.get.return_value = mock_shop
original_compute = None
def slow_compute(shop):
started.set()
proceed.wait(timeout=5)
return ["id1", "id2"]
with mock.patch(
"make_post_sell.models.shop.compute_discovery_ring",
side_effect=slow_compute,
):
t = reforge_discovery_ring_async("shop123", mock_factory)
# Thread should have started
self.assertTrue(started.wait(timeout=5))
# Let it finish
proceed.set()
# Wait for thread to complete
t.join(timeout=5)
mock_session.commit.assert_called()
def test_dirty_bit_set_when_already_running(self):
"""Second call during active reforge sets dirty bit instead of spawning."""
import threading
import time
from ..models.shop import reforge_discovery_ring_async
from ..models import shop as shop_module
started = threading.Event()
proceed = threading.Event()
mock_factory = mock.Mock()
mock_session = mock.Mock()
mock_factory.return_value.get_bind.return_value = "fake_bind"
mock_shop = mock.Mock()
mock_shop.products = []
mock_shop.name = "test"
mock_shop.environment = 0
mock_shop.is_non_production = False
call_count = [0]
def slow_compute(shop):
call_count[0] += 1
if call_count[0] == 1:
started.set()
proceed.wait(timeout=5)
return ["id1"]
patcher1 = mock.patch("make_post_sell.models.shop.SASession")
patcher2 = mock.patch(
"make_post_sell.models.shop.compute_discovery_ring",
side_effect=slow_compute,
)
MockSession = patcher1.start()
patcher2.start()
MockSession.return_value = mock_session
mock_session.get.return_value = mock_shop
try:
# First call starts the thread
t = reforge_discovery_ring_async("shop_dirty", mock_factory)
self.assertTrue(started.wait(timeout=5))
# Second call while running should set dirty bit
reforge_discovery_ring_async("shop_dirty", mock_factory)
self.assertTrue(shop_module._reforge_dirty.get("shop_dirty"))
# Let the first run finish — it should loop and run again
proceed.set()
t.join(timeout=5)
finally:
patcher2.stop()
patcher1.stop()
# compute was called twice (original + dirty re-run)
self.assertEqual(call_count[0], 2)
# State is cleaned up
self.assertNotIn("shop_dirty", shop_module._reforge_running)
self.assertNotIn("shop_dirty", shop_module._reforge_dirty)
def test_no_queue_only_one_dirty_bit(self):
"""Multiple calls during active reforge produce at most one re-run."""
import threading
import time
from ..models.shop import reforge_discovery_ring_async
from ..models import shop as shop_module
started = threading.Event()
proceed = threading.Event()
mock_factory = mock.Mock()
mock_session = mock.Mock()
mock_factory.return_value.get_bind.return_value = "fake_bind"
mock_shop = mock.Mock()
mock_shop.products = []
mock_shop.name = "test"
mock_shop.environment = 0
mock_shop.is_non_production = False
call_count = [0]
def slow_compute(shop):
call_count[0] += 1
if call_count[0] == 1:
started.set()
proceed.wait(timeout=5)
return ["id1"]
patcher1 = mock.patch("make_post_sell.models.shop.SASession")
patcher2 = mock.patch(
"make_post_sell.models.shop.compute_discovery_ring",
side_effect=slow_compute,
)
MockSession = patcher1.start()
patcher2.start()
MockSession.return_value = mock_session
mock_session.get.return_value = mock_shop
try:
# First call starts thread
t = reforge_discovery_ring_async("shop_multi", mock_factory)
self.assertTrue(started.wait(timeout=5))
# Rapid-fire 5 more calls — should all collapse to one dirty bit
for _ in range(5):
reforge_discovery_ring_async("shop_multi", mock_factory)
# Still just one dirty flag
self.assertTrue(shop_module._reforge_dirty.get("shop_multi"))
proceed.set()
t.join(timeout=5)
finally:
patcher2.stop()
patcher1.stop()
# Exactly 2 compute calls: initial + one dirty re-run (not 6)
self.assertEqual(call_count[0], 2)
def test_session_expire_all_on_dirty_rerun(self):
"""Dirty re-run calls session.expire_all() to get fresh DB state."""
import threading
import time
from ..models.shop import reforge_discovery_ring_async
started = threading.Event()
proceed = threading.Event()
mock_factory = mock.Mock()
mock_session = mock.Mock()
mock_factory.return_value.get_bind.return_value = "fake_bind"
mock_shop = mock.Mock()
mock_shop.products = []
mock_shop.name = "test"
mock_shop.environment = 0
mock_shop.is_non_production = False
call_count = [0]
def slow_compute(shop):
call_count[0] += 1
if call_count[0] == 1:
started.set()
proceed.wait(timeout=5)
return ["id1"]
patcher1 = mock.patch("make_post_sell.models.shop.SASession")
patcher2 = mock.patch(
"make_post_sell.models.shop.compute_discovery_ring",
side_effect=slow_compute,
)
MockSession = patcher1.start()
patcher2.start()
MockSession.return_value = mock_session
mock_session.get.return_value = mock_shop
try:
t = reforge_discovery_ring_async("shop_expire", mock_factory)
self.assertTrue(started.wait(timeout=5))
# Trigger dirty bit
reforge_discovery_ring_async("shop_expire", mock_factory)
proceed.set()
t.join(timeout=5)
finally:
patcher2.stop()
patcher1.stop()
# expire_all called before the dirty re-run
mock_session.expire_all.assert_called()
def test_exception_cleans_up_state(self):
"""If reforge crashes, locks are still cleaned up."""
import threading
from ..models.shop import reforge_discovery_ring_async
from ..models import shop as shop_module
mock_factory = mock.Mock()
mock_session = mock.Mock()
mock_factory.return_value.get_bind.return_value = "fake_bind"
mock_session.get.side_effect = Exception("DB exploded")
with mock.patch("make_post_sell.models.shop.SASession") as MockSession:
MockSession.return_value = mock_session
t = reforge_discovery_ring_async("shop_crash", mock_factory)
t.join(timeout=5)
# State should be cleaned up despite exception
self.assertNotIn("shop_crash", shop_module._reforge_running)
self.assertNotIn("shop_crash", shop_module._reforge_dirty)
mock_session.rollback.assert_called()
mock_session.close.assert_called()
def test_shop_not_found_returns_cleanly(self):
"""If shop is deleted between trigger and execution, exits cleanly."""
from ..models.shop import reforge_discovery_ring_async
from ..models import shop as shop_module
mock_factory = mock.Mock()
mock_session = mock.Mock()
mock_factory.return_value.get_bind.return_value = "fake_bind"
mock_session.get.return_value = None # shop deleted
with mock.patch("make_post_sell.models.shop.SASession") as MockSession:
MockSession.return_value = mock_session
t = reforge_discovery_ring_async("shop_gone", mock_factory)
t.join(timeout=5)
# Should not have tried to commit
mock_session.commit.assert_not_called()
# State cleaned up
self.assertNotIn("shop_gone", shop_module._reforge_running)
class TestShopSubscription(unittest.TestCase):
"""Test ShopSubscription model."""
@mock.patch("make_post_sell.models.user.is_user_name_available", mock_always_true)
def setUp(self):
self.shop = Shop(
"test-shop",
"555-555-5555",
"123 Test St",
"Test shop description",
)
self.user = User("subscriber@example.com")
def test_create_anonymous_subscription(self):
"""Test creating a subscription without a user (anonymous)."""
from ..models.shop_subscription import ShopSubscription
sub = ShopSubscription("anon@example.com", self.shop.id)
self.assertEqual(sub.email, "anon@example.com")
self.assertEqual(sub.shop_id, self.shop.id)
self.assertIsNone(sub.user_id)
self.assertFalse(sub.verified)
self.assertIsNotNone(sub.verify_token)
self.assertIsNotNone(sub.unsubscribe_token)
self.assertFalse(sub.disabled)
self.assertGreater(sub.created_timestamp, 0)
self.assertGreater(sub.last_digest_timestamp, 0)
def test_create_logged_in_subscription(self):
"""Test creating a subscription with a user (auto-verified)."""
from ..models.shop_subscription import ShopSubscription
sub = ShopSubscription("user@example.com", self.shop.id, user_id=self.user.id)
self.assertEqual(sub.user_id, self.user.id)
self.assertTrue(sub.verified)
self.assertIsNone(sub.verify_token)
def test_unique_tokens(self):
"""Test that each subscription gets unique tokens."""
from ..models.shop_subscription import ShopSubscription
sub1 = ShopSubscription("a@example.com", self.shop.id)
sub2 = ShopSubscription("b@example.com", self.shop.id)
self.assertNotEqual(sub1.unsubscribe_token, sub2.unsubscribe_token)
self.assertNotEqual(sub1.verify_token, sub2.verify_token)
def test_default_frequency_daily(self):
"""Test that default frequency is daily."""
from ..models.shop_subscription import ShopSubscription, FREQUENCY_DAILY
sub = ShopSubscription("test@example.com", self.shop.id)
self.assertEqual(sub.frequency, FREQUENCY_DAILY)
def test_human_frequency(self):
"""Test human_frequency property."""
from ..models.shop_subscription import (
ShopSubscription,
FREQUENCY_DAILY,
FREQUENCY_WEEKLY,
)
sub = ShopSubscription("test@example.com", self.shop.id)
sub.frequency = FREQUENCY_WEEKLY
self.assertEqual(sub.human_frequency, "Weekly")
sub.frequency = FREQUENCY_DAILY
self.assertEqual(sub.human_frequency, "Daily")
class TestSubscribeViewCurrentFrequency(unittest.TestCase):
"""Test that subscribe view returns correct current_frequency for the template."""
def _make_request(self, user=None, shop=None):
"""Build a mock request for the subscribe view."""
request = mock.MagicMock()
request.method = "GET"
request.shop = shop
request.user = user
request.session = mock.MagicMock()
request.session.flash = mock.Mock()
return request
def _make_shop(self):
shop = mock.MagicMock()
shop.subscriptions_enabled = True
shop.id = "shop-id-123"
shop.name = "Test Shop"
return shop
def _make_user(self, email="test@example.com"):
user = mock.MagicMock()
user.authenticated = True
user.email = email
return user
@mock.patch("make_post_sell.views.subscribe.get_subscription_for_email_and_shop")
def test_returns_active_frequency_daily(self, mock_get_sub):
"""Logged-in user with active daily subscription sees freq=0."""
from ..views.subscribe import subscribe
sub = mock.MagicMock()
sub.disabled = False
sub.frequency = 0 # FREQUENCY_DAILY
mock_get_sub.return_value = sub
request = self._make_request(
user=self._make_user(), shop=self._make_shop()
)
result = subscribe(request)
self.assertEqual(result["current_frequency"], 0)
@mock.patch("make_post_sell.views.subscribe.get_subscription_for_email_and_shop")
def test_returns_active_frequency_weekly(self, mock_get_sub):
"""Logged-in user with active weekly subscription sees freq=1."""
from ..views.subscribe import subscribe
sub = mock.MagicMock()
sub.disabled = False
sub.frequency = 1 # FREQUENCY_WEEKLY
mock_get_sub.return_value = sub
request = self._make_request(
user=self._make_user(), shop=self._make_shop()
)
result = subscribe(request)
self.assertEqual(result["current_frequency"], 1)
@mock.patch("make_post_sell.views.subscribe.get_subscription_for_email_and_shop")
def test_returns_active_frequency_immediate(self, mock_get_sub):
"""Logged-in user with active immediate subscription sees freq=2."""
from ..views.subscribe import subscribe
sub = mock.MagicMock()
sub.disabled = False
sub.frequency = 2 # FREQUENCY_IMMEDIATE
mock_get_sub.return_value = sub
request = self._make_request(
user=self._make_user(), shop=self._make_shop()
)
result = subscribe(request)
self.assertEqual(result["current_frequency"], 2)
@mock.patch("make_post_sell.views.subscribe.get_subscription_for_email_and_shop")
def test_disabled_subscription_returns_negative_one(self, mock_get_sub):
"""Logged-in user with disabled subscription sees freq=-1 (None radio)."""
from ..views.subscribe import subscribe
sub = mock.MagicMock()
sub.disabled = True
mock_get_sub.return_value = sub
request = self._make_request(
user=self._make_user(), shop=self._make_shop()
)
result = subscribe(request)
self.assertEqual(result["current_frequency"], -1)
@mock.patch("make_post_sell.views.subscribe.get_subscription_for_email_and_shop")
def test_no_subscription_returns_none(self, mock_get_sub):
"""Logged-in user with no subscription sees current_frequency=None."""
from ..views.subscribe import subscribe
mock_get_sub.return_value = None
request = self._make_request(
user=self._make_user(), shop=self._make_shop()
)
result = subscribe(request)
self.assertIsNone(result["current_frequency"])
def test_anonymous_user_returns_none(self):
"""Anonymous user (not logged in) sees current_frequency=None."""
from ..views.subscribe import subscribe
request = self._make_request(user=None, shop=self._make_shop())
result = subscribe(request)
self.assertIsNone(result["current_frequency"])
class TestMentionParsing(unittest.TestCase):
"""Test @mention parsing from comment text."""
def test_simple_mention(self):
from ..lib.mentions import extract_mentions
result = extract_mentions("Hello @user-abcd1234 nice post!")
self.assertEqual(result, ["user-abcd1234"])
def test_multiple_mentions(self):
from ..lib.mentions import extract_mentions
result = extract_mentions("@user-abc123 and @user-def456 check this out")
self.assertEqual(len(result), 2)
self.assertIn("user-abc123", result)
self.assertIn("user-def456", result)
def test_duplicate_mentions(self):
from ..lib.mentions import extract_mentions
result = extract_mentions("@user-abc123 said hi, @user-abc123 said bye")
self.assertEqual(len(result), 1)
self.assertEqual(result[0], "user-abc123")
def test_no_mentions(self):
from ..lib.mentions import extract_mentions
result = extract_mentions("No mentions here.")
self.assertEqual(result, [])
def test_empty_text(self):
from ..lib.mentions import extract_mentions
result = extract_mentions("")
self.assertEqual(result, [])
def test_none_text(self):
from ..lib.mentions import extract_mentions
result = extract_mentions(None)
self.assertEqual(result, [])
def test_email_not_matched(self):
"""Email addresses should not be matched as mentions."""
from ..lib.mentions import extract_mentions
result = extract_mentions("Email me at person@example.com")
# @example would not match because it follows a non-whitespace char
self.assertEqual(result, [])
def test_start_of_string(self):
from ..lib.mentions import extract_mentions
result = extract_mentions("@user-start123 at the start")
self.assertEqual(result, ["user-start123"])
def test_mention_too_short(self):
"""Usernames under 3 chars should not match."""
from ..lib.mentions import extract_mentions
result = extract_mentions("@ab is too short")
self.assertEqual(result, [])
class TestDigestBuilding(unittest.TestCase):
"""Test digest email building functions."""
def test_build_digest_text(self):
from ..lib.digest_sender import build_digest_text
mock_item1 = mock.Mock()
mock_item1.title = "Cool Product"
mock_item1.is_sellable = True
mock_item2 = mock.Mock()
mock_item2.title = "Blog Post"
mock_item2.is_sellable = False
result = build_digest_text(
"Test Shop", [mock_item1, mock_item2], "https://example.com/unsub"
)
self.assertIn("Test Shop", result)
self.assertIn("Cool Product", result)
self.assertIn("Blog Post", result)
self.assertIn("[Product]", result)
self.assertIn("[Content]", result)
self.assertIn("https://example.com/unsub", result)
def test_build_digest_html(self):
from ..lib.digest_sender import build_digest_html
mock_item = mock.Mock()
mock_item.title = "New Item"
mock_item.is_sellable = True
mock_item.extensions = {}
mock_item.s3_path = "test/path"
mock_item.updated_timestamp = 12345
result = build_digest_html(
"Test Shop", [mock_item], "https://example.com/unsub"
)
self.assertIn("Test Shop", result)
self.assertIn("New Item", result)
self.assertIn("https://example.com/unsub", result)
def test_build_digest_html_with_thumbnail(self):
from ..lib.digest_sender import build_digest_html
mock_item = mock.Mock()
mock_item.title = "Item With Image"
mock_item.is_sellable = False
mock_item.extensions = {"thumbnail1": "jpg"}
mock_item.s3_path = "shop/product"
mock_item.updated_timestamp = 99999
result = build_digest_html(
"Test Shop",
[mock_item],
"https://example.com/unsub",
get_endpoint="https://cdn.example.com",
)
self.assertIn("https://cdn.example.com/shop/product/thumbnail1", result)
class TestProductViewCount(unittest.TestCase):
"""Test Product.human_view_count formatting."""
def _make_product(self, view_count):
p = mock.Mock()
p.view_count = view_count
from ..models.product import Product
p.human_view_count = Product.human_view_count.fget(p)
return p
def test_zero_returns_empty(self):
self.assertEqual(self._make_product(0).human_view_count, "")
def test_none_returns_empty(self):
self.assertEqual(self._make_product(None).human_view_count, "")
def test_one_view_singular(self):
self.assertEqual(self._make_product(1).human_view_count, "1 view")
def test_plural_views(self):
self.assertEqual(self._make_product(420).human_view_count, "420 views")
def test_one_thousand(self):
self.assertEqual(self._make_product(1000).human_view_count, "1K views")
def test_twelve_hundred(self):
self.assertEqual(self._make_product(1200).human_view_count, "1.2K views")
def test_one_million(self):
self.assertEqual(self._make_product(1_000_000).human_view_count, "1M views")
def test_one_point_eight_million(self):
self.assertEqual(self._make_product(1_800_000).human_view_count, "1.8M views")
def test_999_views(self):
self.assertEqual(self._make_product(999).human_view_count, "999 views")
class TestSignalClassifiers(unittest.TestCase):
"""Test classify_referrer and classify_device functions."""
# --- classify_referrer (returns (class, domain, query) tuple) ---
def test_referrer_direct_empty(self):
self.assertEqual(classify_referrer("", "example.com"), (0, None, None))
def test_referrer_direct_none(self):
self.assertEqual(classify_referrer(None, "example.com"), (0, None, None))
def test_referrer_search_google(self):
cls, domain, query = classify_referrer(
"https://www.google.com/search?q=foo", "example.com"
)
self.assertEqual(cls, 1)
self.assertEqual(domain, "www.google.com")
self.assertEqual(query, "foo")
def test_referrer_search_duckduckgo(self):
cls, domain, query = classify_referrer(
"https://duckduckgo.com/?q=bar", "example.com"
)
self.assertEqual(cls, 1)
self.assertEqual(domain, "duckduckgo.com")
self.assertEqual(query, "bar")
def test_referrer_search_no_query(self):
cls, domain, query = classify_referrer(
"https://www.google.com/", "example.com"
)
self.assertEqual(cls, 1)
self.assertEqual(domain, "www.google.com")
self.assertIsNone(query)
def test_referrer_social_twitter(self):
cls, domain, query = classify_referrer(
"https://twitter.com/user/status/123", "example.com"
)
self.assertEqual(cls, 2)
self.assertEqual(domain, "twitter.com")
self.assertIsNone(query)
def test_referrer_social_reddit(self):
cls, domain, query = classify_referrer(
"https://www.reddit.com/r/test", "example.com"
)
self.assertEqual(cls, 2)
self.assertEqual(domain, "www.reddit.com")
self.assertIsNone(query)
def test_referrer_internal(self):
cls, domain, query = classify_referrer(
"https://example.com/some/page", "example.com"
)
self.assertEqual(cls, 3)
self.assertEqual(domain, "example.com")
self.assertIsNone(query)
def test_referrer_unknown_external(self):
cls, domain, query = classify_referrer(
"https://randomsite.org/page", "example.com"
)
self.assertEqual(cls, 4)
self.assertEqual(domain, "randomsite.org")
self.assertIsNone(query)
def test_referrer_yahoo_query(self):
cls, domain, query = classify_referrer(
"https://search.yahoo.com/search?p=beats", "example.com"
)
self.assertEqual(cls, 1)
self.assertEqual(query, "beats")
# --- classify_device ---
def test_device_mobile(self):
self.assertEqual(classify_device(375), 0)
def test_device_mobile_boundary(self):
self.assertEqual(classify_device(768), 0)
def test_device_tablet(self):
self.assertEqual(classify_device(800), 1)
def test_device_tablet_boundary(self):
self.assertEqual(classify_device(1024), 1)
def test_device_desktop(self):
self.assertEqual(classify_device(1920), 2)
def test_device_desktop_boundary(self):
self.assertEqual(classify_device(1025), 2)
def test_device_none(self):
self.assertIsNone(classify_device(None))
class TestAnalyticsHelpers(unittest.TestCase):
"""Unit tests for analytics view helper functions."""
def test_fmt_ms_none(self):
from ..views.analytics import _fmt_ms
self.assertEqual(_fmt_ms(None), "\u2014")
def test_fmt_ms_seconds(self):
from ..views.analytics import _fmt_ms
self.assertEqual(_fmt_ms(42000), "42s")
def test_fmt_ms_minutes(self):
from ..views.analytics import _fmt_ms
self.assertEqual(_fmt_ms(125000), "2m 5s")
def test_fmt_ms_hours(self):
from ..views.analytics import _fmt_ms
self.assertEqual(_fmt_ms(3661000), "1h 1m")
def test_fmt_ms_zero(self):
from ..views.analytics import _fmt_ms
self.assertEqual(_fmt_ms(0), "0s")
def test_fmt_pct_none(self):
from ..views.analytics import _fmt_pct
self.assertEqual(_fmt_pct(None), "\u2014")
def test_fmt_pct_half(self):
from ..views.analytics import _fmt_pct
self.assertEqual(_fmt_pct(0.5), "50%")
def test_fmt_pct_full(self):
from ..views.analytics import _fmt_pct
self.assertEqual(_fmt_pct(1.0), "100%")
def test_fmt_pct_zero(self):
from ..views.analytics import _fmt_pct
self.assertEqual(_fmt_pct(0.0), "0%")
def test_fmt_score_none(self):
from ..views.analytics import _fmt_score
self.assertEqual(_fmt_score(None), "\u2014")
def test_fmt_score_value(self):
from ..views.analytics import _fmt_score
self.assertEqual(_fmt_score(3.14159), "3.1")
def test_fmt_score_zero(self):
from ..views.analytics import _fmt_score
self.assertEqual(_fmt_score(0.0), "0.0")
def test_cutoffs_returns_expected_keys(self):
from ..views.analytics import _cutoffs
cuts = _cutoffs()
day_ms = 24 * 60 * 60 * 1000
for key in ("1d", "7d", "14d", "21d", "28d", "365d"):
self.assertIn(key, cuts)
self.assertAlmostEqual(cuts["7d"] - cuts["14d"], 7 * day_ms, delta=1000)
self.assertAlmostEqual(cuts["14d"] - cuts["21d"], 7 * day_ms, delta=1000)
def test_referrer_labels_complete(self):
from ..views.analytics import REFERRER_LABELS
self.assertEqual(len(REFERRER_LABELS), 5)
self.assertIn(0, REFERRER_LABELS)
self.assertIn(4, REFERRER_LABELS)
def test_device_labels_complete(self):
from ..views.analytics import DEVICE_LABELS
self.assertEqual(len(DEVICE_LABELS), 3)
self.assertEqual(DEVICE_LABELS[0], "Mobile")
self.assertEqual(DEVICE_LABELS[2], "Desktop")
class TestSentiment(unittest.TestCase):
"""Unit tests for the rule-based comment sentiment classifier."""
def _classify(self, text):
from ..lib.sentiment import classify_sentiment
return classify_sentiment(text)
def test_positive_comment(self):
self.assertEqual(self._classify("This is amazing and wonderful!"), 1)
def test_negative_comment(self):
self.assertEqual(self._classify("Terrible product, total waste"), -1)
def test_neutral_comment(self):
self.assertEqual(self._classify("I downloaded the file yesterday"), 0)
def test_empty_string(self):
self.assertEqual(self._classify(""), 0)
def test_none_input(self):
self.assertEqual(self._classify(None), 0)
def test_negation_flips_positive(self):
self.assertEqual(self._classify("not good"), -1)
def test_negation_flips_negative(self):
self.assertEqual(self._classify("not bad"), 1)
def test_intensifier_boosts_score(self):
# "really great" should still be positive
self.assertEqual(self._classify("really great"), 1)
# "very terrible" should still be negative
self.assertEqual(self._classify("very terrible"), -1)
def test_mixed_sentiment_leans_neutral(self):
# Roughly equal positive and negative words
self.assertEqual(self._classify("good but also bad"), 0)
def test_only_punctuation_and_numbers(self):
self.assertEqual(self._classify("123 !!! ???"), 0)
def test_short_positive(self):
self.assertEqual(self._classify("love it"), 1)
def test_short_negative(self):
self.assertEqual(self._classify("hate it"), -1)
class TestGiftCard(unittest.TestCase):
def test_gift_card_creation(self):
from make_post_sell.models.gift_card import GiftCard
shop = mock.MagicMock()
shop.id = uuid.uuid1()
gc = GiftCard(shop=shop, amount_in_cents=5000)
self.assertEqual(gc.initial_amount_in_cents, 5000)
self.assertEqual(gc.balance_in_cents, 5000)
self.assertFalse(gc.disabled)
self.assertTrue(gc.code.startswith("GC-"))
self.assertEqual(len(gc.code), 19) # GC- + 16 hex chars
def test_gift_card_is_valid(self):
from make_post_sell.models.gift_card import GiftCard
shop = mock.MagicMock()
shop.id = uuid.uuid1()
gc = GiftCard(shop=shop, amount_in_cents=5000)
self.assertTrue(gc.is_valid)
def test_gift_card_disabled_not_valid(self):
from make_post_sell.models.gift_card import GiftCard
shop = mock.MagicMock()
shop.id = uuid.uuid1()
gc = GiftCard(shop=shop, amount_in_cents=5000)
gc.disabled = True
self.assertFalse(gc.is_valid)
def test_gift_card_zero_balance_not_valid(self):
from make_post_sell.models.gift_card import GiftCard
shop = mock.MagicMock()
shop.id = uuid.uuid1()
gc = GiftCard(shop=shop, amount_in_cents=5000)
gc.balance_in_cents = 0
self.assertFalse(gc.is_valid)
def test_gift_card_deduct(self):
from make_post_sell.models.gift_card import GiftCard
shop = mock.MagicMock()
shop.id = uuid.uuid1()
gc = GiftCard(shop=shop, amount_in_cents=5000)
deducted = gc.deduct(2000)
self.assertEqual(deducted, 2000)
self.assertEqual(gc.balance_in_cents, 3000)
def test_gift_card_deduct_more_than_balance(self):
from make_post_sell.models.gift_card import GiftCard
shop = mock.MagicMock()
shop.id = uuid.uuid1()
gc = GiftCard(shop=shop, amount_in_cents=1000)
deducted = gc.deduct(5000)
self.assertEqual(deducted, 1000)
self.assertEqual(gc.balance_in_cents, 0)
def test_gift_card_balance_property(self):
from make_post_sell.models.gift_card import GiftCard
shop = mock.MagicMock()
shop.id = uuid.uuid1()
gc = GiftCard(shop=shop, amount_in_cents=5000)
self.assertEqual(gc.balance, 50.00)
self.assertEqual(gc.initial_amount, 50.00)
def test_gift_card_is_fully_redeemed(self):
from make_post_sell.models.gift_card import GiftCard
shop = mock.MagicMock()
shop.id = uuid.uuid1()
gc = GiftCard(shop=shop, amount_in_cents=5000)
self.assertFalse(gc.is_fully_redeemed)
gc.balance_in_cents = 0
self.assertTrue(gc.is_fully_redeemed)
def test_gift_card_code_uniqueness(self):
from make_post_sell.models.gift_card import generate_gift_card_code
codes = set()
for _ in range(100):
codes.add(generate_gift_card_code())
self.assertEqual(len(codes), 100)
def test_gift_card_with_gift_email(self):
from make_post_sell.models.gift_card import GiftCard
shop = mock.MagicMock()
shop.id = uuid.uuid1()
gc = GiftCard(
shop=shop,
amount_in_cents=2500,
purchaser_email="buyer@test.com",
gift_email="friend@test.com",
gift_message="Happy birthday!",
)
self.assertEqual(gc.gift_email, "friend@test.com")
self.assertEqual(gc.gift_message, "Happy birthday!")
self.assertEqual(gc.purchaser_email, "buyer@test.com")
class TestShopEnvironment(unittest.TestCase):
"""MPS-14: Dev & Stage environment properties."""
def _make_shop(self, environment=0):
shop = Shop("env-test", "555-0000", "123 Test St", "test shop")
shop.environment = environment
return shop
def test_default_is_production(self):
shop = self._make_shop()
self.assertTrue(shop.is_production)
self.assertFalse(shop.is_non_production)
self.assertEqual(shop.environment_label, "Production")
def test_staging_environment(self):
shop = self._make_shop(environment=1)
self.assertTrue(shop.is_staging)
self.assertTrue(shop.is_non_production)
self.assertFalse(shop.is_production)
self.assertEqual(shop.environment_label, "Staging")
def test_development_environment(self):
shop = self._make_shop(environment=2)
self.assertTrue(shop.is_development)
self.assertTrue(shop.is_non_production)
self.assertFalse(shop.is_production)
self.assertEqual(shop.environment_label, "Development")
def test_unknown_environment_defaults_to_production_label(self):
shop = self._make_shop(environment=99)
self.assertEqual(shop.environment_label, "Production")
self.assertTrue(shop.is_non_production)
class TestShopTrial(unittest.TestCase):
"""MPS-15: 21-day free trial properties."""
def _make_shop(self, trial_started_ms=None, plan_active=False):
shop = Shop("trial-test", "555-0000", "123 Test St", "test shop")
shop.trial_started_timestamp = trial_started_ms
shop.plan_active = plan_active
return shop
def test_grandfathered_shop_is_active(self):
"""Pre-trial shops (NULL timestamp) are always active."""
shop = self._make_shop(trial_started_ms=None)
self.assertTrue(shop.is_active)
self.assertFalse(shop.is_trial_active)
self.assertFalse(shop.is_trial_expired)
self.assertIsNone(shop.trial_expiry_timestamp)
@mock.patch("make_post_sell.models.shop.time")
def test_trial_active_within_21_days(self, mock_time):
import time as real_time
now_ms = int(real_time.time() * 1000)
# Started 10 days ago
started = now_ms - (10 * 24 * 60 * 60 * 1000)
shop = self._make_shop(trial_started_ms=started)
mock_time.time.return_value = now_ms / 1000.0
self.assertTrue(shop.is_trial_active)
self.assertFalse(shop.is_trial_expired)
self.assertTrue(shop.is_active)
self.assertGreater(shop.trial_days_remaining, 0)
@mock.patch("make_post_sell.models.shop.time")
def test_trial_expired_after_21_days(self, mock_time):
import time as real_time
now_ms = int(real_time.time() * 1000)
# Started 22 days ago
started = now_ms - (22 * 24 * 60 * 60 * 1000)
shop = self._make_shop(trial_started_ms=started)
mock_time.time.return_value = now_ms / 1000.0
self.assertFalse(shop.is_trial_active)
self.assertTrue(shop.is_trial_expired)
self.assertFalse(shop.is_active)
self.assertEqual(shop.trial_days_remaining, 0)
@mock.patch("make_post_sell.models.shop.time")
def test_paid_plan_overrides_trial(self, mock_time):
import time as real_time
now_ms = int(real_time.time() * 1000)
# Started 22 days ago but plan is active
started = now_ms - (22 * 24 * 60 * 60 * 1000)
shop = self._make_shop(trial_started_ms=started, plan_active=True)
mock_time.time.return_value = now_ms / 1000.0
self.assertFalse(shop.is_trial_active)
self.assertFalse(shop.is_trial_expired)
self.assertTrue(shop.is_active)
def test_trial_expiry_timestamp(self):
shop = self._make_shop(trial_started_ms=1000000)
expected = 1000000 + (21 * 24 * 60 * 60 * 1000)
self.assertEqual(shop.trial_expiry_timestamp, expected)
class TestShopBYOB(unittest.TestCase):
"""MPS-16: Bring Your Own Bucket properties."""
def _make_shop(self, enabled=False, **kwargs):
shop = Shop("byob-test", "555-0000", "123 Test St", "test shop")
shop.primary_s3_enabled = enabled
shop.primary_s3_endpoint = kwargs.get("endpoint", "https://nyc3.digitaloceanspaces.com")
shop.primary_s3_region = kwargs.get("region", "nyc3")
shop.primary_s3_bucket = kwargs.get("bucket", "my-bucket")
shop.primary_s3_access_key = kwargs.get("access_key", "AKID")
shop.primary_s3_secret_key = kwargs.get("secret_key", "SECRET")
shop.primary_s3_cdn_endpoint = kwargs.get("cdn_endpoint", "https://my-bucket.nyc3.cdn.digitaloceanspaces.com")
return shop
def test_has_primary_s3_when_enabled_and_configured(self):
shop = self._make_shop(enabled=True)
self.assertTrue(shop.has_primary_s3)
def test_has_primary_s3_false_when_disabled(self):
shop = self._make_shop(enabled=False)
self.assertFalse(shop.has_primary_s3)
def test_has_primary_s3_false_when_missing_fields(self):
shop = self._make_shop(enabled=True, access_key="")
self.assertFalse(shop.has_primary_s3)
def test_media_cdn_endpoint_returns_custom_when_configured(self):
shop = self._make_shop(enabled=True)
self.assertEqual(shop.media_cdn_endpoint, "https://my-bucket.nyc3.cdn.digitaloceanspaces.com")
def test_media_cdn_endpoint_returns_none_when_not_configured(self):
shop = self._make_shop(enabled=False)
self.assertIsNone(shop.media_cdn_endpoint)
class TestTorrentLib(unittest.TestCase):
"""Unit tests for lib/torrent.py — web seed URL construction and DB persistence."""
def _make_product(self):
from ..models.product import Product
p = mock.MagicMock(spec=Product)
p.id = "prod-001"
p.s3_path = "shop1/prod-001"
p.torrent_magnet_link = None
p.torrent_file_url = None
return p
def test_webseed_url_constructed_from_cdn_endpoint_and_s3_key(self):
"""generate_torrent embeds cdn_endpoint/s3_key as web seed in the torrent."""
from ..lib.torrent import generate_torrent
s3_client = mock.MagicMock()
session = mock.MagicMock()
product = self._make_product()
session.get.return_value = product
session_cm = mock.MagicMock()
session_cm.__enter__ = mock.Mock(return_value=session)
session_cm.__exit__ = mock.Mock(return_value=False)
session_factory = mock.Mock(return_value=session_cm)
torrent_obj = mock.MagicMock()
with mock.patch("make_post_sell.lib.torrent._build_torrent", return_value=(torrent_obj, "magnet:?xt=urn:btih:deadbeef&dn=product")) as mock_build, \
mock.patch("tempfile.TemporaryDirectory") as mock_tmpdir:
mock_tmpdir.return_value.__enter__ = mock.Mock(return_value="/tmp/fake")
mock_tmpdir.return_value.__exit__ = mock.Mock(return_value=False)
generate_torrent(
s3_client, "my-bucket", "shop1/prod-001/product",
"shop1/prod-001", "prod-001", session_factory,
cdn_endpoint="https://cdn.example.com",
)
# _build_torrent is called with webseeds= as a keyword arg
call_kwargs = mock_build.call_args.kwargs
webseeds = call_kwargs.get("webseeds", [])
self.assertIn("https://cdn.example.com/shop1/prod-001/product", webseeds)
def test_torrent_file_url_saved_to_product(self):
"""generate_torrent saves torrent_file_url = cdn_endpoint/s3_path/product.torrent."""
from ..lib.torrent import generate_torrent
s3_client = mock.MagicMock()
session = mock.MagicMock()
product = self._make_product()
session.get.return_value = product
session_cm = mock.MagicMock()
session_cm.__enter__ = mock.Mock(return_value=session)
session_cm.__exit__ = mock.Mock(return_value=False)
session_factory = mock.Mock(return_value=session_cm)
torrent_obj = mock.MagicMock()
with mock.patch("make_post_sell.lib.torrent._build_torrent", return_value=(torrent_obj, "magnet:?xt=urn:btih:abc")), \
mock.patch("tempfile.TemporaryDirectory") as mock_tmpdir:
mock_tmpdir.return_value.__enter__ = mock.Mock(return_value="/tmp/fake")
mock_tmpdir.return_value.__exit__ = mock.Mock(return_value=False)
generate_torrent(
s3_client, "my-bucket", "shop1/prod-001/product",
"shop1/prod-001", "prod-001", session_factory,
cdn_endpoint="https://cdn.example.com",
)
self.assertEqual(product.torrent_file_url, "https://cdn.example.com/shop1/prod-001/product.torrent")
def test_torrent_file_url_none_without_cdn_endpoint(self):
"""Without cdn_endpoint, torrent_file_url is None (no web seed)."""
from ..lib.torrent import generate_torrent
s3_client = mock.MagicMock()
session = mock.MagicMock()
product = self._make_product()
session.get.return_value = product
session_cm = mock.MagicMock()
session_cm.__enter__ = mock.Mock(return_value=session)
session_cm.__exit__ = mock.Mock(return_value=False)
session_factory = mock.Mock(return_value=session_cm)
torrent_obj = mock.MagicMock()
with mock.patch("make_post_sell.lib.torrent._build_torrent", return_value=(torrent_obj, "magnet:?xt=urn:btih:abc")), \
mock.patch("tempfile.TemporaryDirectory") as mock_tmpdir:
mock_tmpdir.return_value.__enter__ = mock.Mock(return_value="/tmp/fake")
mock_tmpdir.return_value.__exit__ = mock.Mock(return_value=False)
generate_torrent(
s3_client, "my-bucket", "shop1/prod-001/product",
"shop1/prod-001", "prod-001", session_factory,
cdn_endpoint=None,
)
self.assertIsNone(product.torrent_file_url)
Reference in a new issue View git blame Copy permalink