"""Admin listing-management tests (Phase 6.1).""" import os import unittest from datetime import timedelta from decimal import Decimal os.environ["TEST_DATABASE_URL"] = "sqlite:///:memory:" from sqlalchemy import event from werkzeug.security import generate_password_hash from app import create_app from app.admin.listing_management import ( bulk_update_listings, get_listing_summary, get_listings, ) from app.extensions import db from app.models import ( Category, Conversation, Favorite, Listing, ListingCondition, ListingStatus, ListingView, ListingVisibility, Offer, ReviewStatus, User, UserProfile, ) from app.models.base import utcnow from app.models.enums import UserRole from app.marketplace.services import is_listing_currently_featured class _QC: def __init__(self): self.n = 0 def __enter__(self): self._e = db.engine event.listen(self._e, "after_cursor_execute", self._cb) return self def __exit__(self, *a): event.remove(self._e, "after_cursor_execute", self._cb) def _cb(self, *a): self.n += 1 class AdminListingManagementTests(unittest.TestCase): def setUp(self): self.app = create_app("testing") self.ctx = self.app.app_context() self.ctx.push() db.create_all() def tearDown(self): db.session.remove() db.drop_all() self.ctx.pop() # --- factories --- def _user(self, email, role=UserRole.STUDENT, name=None): u = User(is_verified=True, email=email, password_hash=generate_password_hash("Passw0rd!"), role=role) db.session.add(u) db.session.commit() if name: db.session.add(UserProfile(user_id=u.id, full_name=name)) db.session.commit() return u def _category(self, name="Books"): cat = db.session.scalar(db.select(Category).filter_by(name=name)) if cat is None: cat = Category(name=name, description="d", is_active=True, display_order=1) db.session.add(cat) db.session.commit() return cat def _listing(self, seller, idx=1, title=None, views=0, favorites=0, status=ListingStatus.AVAILABLE, category="Books", featured=False, deleted=False): cat = self._category(category) listing = Listing( sku=f"SKU-{seller.id}-{idx}", seller_id=seller.id, category_id=cat.id, title=title or f"Item {idx}", slug=f"item-{seller.id}-{idx}", description="d", price=Decimal("50.00"), condition=ListingCondition.GOOD, status=status, visibility=ListingVisibility.PUBLIC, review_status=ReviewStatus.APPROVED, view_count=views, favorite_count=favorites, ) if featured: listing.is_featured = True listing.featured_at = utcnow() listing.featured_until = utcnow() + timedelta(days=7) if deleted: listing.deleted_at = utcnow() listing.status = ListingStatus.REMOVED db.session.add(listing) db.session.commit() return listing def _offer(self, listing, buyer, amount=40): # One conversation per (buyer, seller, listing); reuse it for extra offers. convo = db.session.scalar(db.select(Conversation).filter_by( listing_id=listing.id, buyer_id=buyer.id, seller_id=listing.seller_id)) if convo is None: convo = Conversation(listing_id=listing.id, buyer_id=buyer.id, seller_id=listing.seller_id) db.session.add(convo) db.session.commit() offer = Offer(listing_id=listing.id, conversation_id=convo.id, buyer_id=buyer.id, seller_id=listing.seller_id, amount=Decimal(amount)) db.session.add(offer) db.session.commit() return offer def _admin_client(self): self._user("admin@c.edu", UserRole.SUPER_ADMIN) client = self.app.test_client() client.post("/login", data={"email": "admin@c.edu", "password": "Passw0rd!"}) return client # --- access --- def test_admin_access(self): client = self._admin_client() self.assertEqual(client.get("/admin/listings").status_code, 200) def test_non_admin_denied(self): self._user("s@c.edu") client = self.app.test_client() client.post("/login", data={"email": "s@c.edu", "password": "Passw0rd!"}) self.assertEqual(client.get("/admin/listings").status_code, 403) # --- search --- def test_search_by_title_and_seller(self): alice = self._user("alice@c.edu", name="Alice Adams") bob = self._user("bob@c.edu", name="Bob Brown") self._listing(alice, 1, title="Red Bicycle") self._listing(bob, 1, title="Blue Guitar") _, rows = get_listings(search="bicycle") self.assertEqual([r["title"] for r in rows], ["Red Bicycle"]) _, rows = get_listings(search="Brown") # seller name self.assertEqual([r["title"] for r in rows], ["Blue Guitar"]) _, rows = get_listings(search="alice@c.edu") # seller email self.assertEqual([r["title"] for r in rows], ["Red Bicycle"]) # --- filters --- def test_status_filter(self): u = self._user("u@c.edu") self._listing(u, 1, status=ListingStatus.AVAILABLE) self._listing(u, 2, status=ListingStatus.SOLD) self._listing(u, 3, deleted=True) _, rows = get_listings(status="active") self.assertEqual([r["status"] for r in rows], ["available"]) _, rows = get_listings(status="sold") self.assertEqual([r["status"] for r in rows], ["sold"]) _, rows = get_listings(status="deleted") self.assertEqual(len(rows), 1) self.assertTrue(rows[0]["is_deleted"]) # Default (all) hides soft-deleted listings. _, rows = get_listings(status="all") self.assertEqual(len(rows), 2) def test_featured_filter(self): u = self._user("u@c.edu") self._listing(u, 1, featured=True) self._listing(u, 2, featured=False) _, rows = get_listings(featured="featured") self.assertEqual(len(rows), 1) self.assertTrue(rows[0]["is_featured"]) _, rows = get_listings(featured="not_featured") self.assertEqual(len(rows), 1) self.assertFalse(rows[0]["is_featured"]) def test_category_filter(self): u = self._user("u@c.edu") self._listing(u, 1, category="Books") self._listing(u, 2, category="Electronics") electronics = self._category("Electronics") _, rows = get_listings(category=electronics.public_id) self.assertEqual([r["category"] for r in rows], ["Electronics"]) def test_seller_filter(self): a = self._user("a@c.edu") b = self._user("b@c.edu") self._listing(a, 1) self._listing(b, 1) _, rows = get_listings(seller=a.public_id) self.assertEqual([r["seller_email"] for r in rows], ["a@c.edu"]) # --- sort --- def test_sort_most_views(self): u = self._user("u@c.edu") self._listing(u, 1, title="Low", views=2) self._listing(u, 2, title="High", views=99) _, rows = get_listings(sort="most_views") self.assertEqual(rows[0]["title"], "High") def test_sort_most_offers(self): seller = self._user("s@c.edu") buyer = self._user("b@c.edu") popular = self._listing(seller, 1, title="Popular") self._listing(seller, 2, title="Quiet") self._offer(popular, buyer, 40) self._offer(popular, buyer, 45) _, rows = get_listings(sort="most_offers") self.assertEqual(rows[0]["title"], "Popular") self.assertEqual(rows[0]["offers"], 2) def test_sort_newest_oldest(self): u = self._user("u@c.edu") self._listing(u, 1, title="First") self._listing(u, 2, title="Second") _, rows = get_listings(sort="oldest") self.assertEqual(rows[0]["title"], "First") _, rows = get_listings(sort="newest") self.assertEqual(rows[0]["title"], "Second") # --- pagination --- def test_pagination(self): u = self._user("u@c.edu") for i in range(25): self._listing(u, i) pagination, rows = get_listings(page=1) self.assertEqual(pagination.per_page, 20) self.assertEqual(len(rows), 20) self.assertEqual(pagination.pages, 2) _, rows2 = get_listings(page=2) self.assertEqual(len(rows2), 5) # --- summary --- def test_summary_totals(self): u = self._user("u@c.edu") self._listing(u, 1, status=ListingStatus.AVAILABLE, views=10, favorites=2) self._listing(u, 2, status=ListingStatus.SOLD, views=20, favorites=4) self._listing(u, 3, status=ListingStatus.RESERVED, views=0, favorites=0) self._listing(u, 4, featured=True, views=0, favorites=0) self._listing(u, 5, deleted=True) s = get_listing_summary() self.assertEqual(s["total_listings"], 4) # excludes deleted self.assertEqual(s["active_listings"], 2) # available + featured(available) self.assertEqual(s["sold_listings"], 1) self.assertEqual(s["reserved_listings"], 1) self.assertEqual(s["featured_listings"], 1) self.assertEqual(s["deleted_listings"], 1) self.assertEqual(s["avg_views"], round(30 / 4, 1)) # --- action visibility --- def test_action_visibility_available(self): client = self._admin_client() u = self._user("u@c.edu") listing = self._listing(u, 1, status=ListingStatus.AVAILABLE) html = client.get(f"/admin/listings/{listing.public_id}").get_data(as_text=True) self.assertIn("Feature Listing", html) self.assertIn("Mark Reserved", html) self.assertIn("Mark Sold", html) self.assertIn("Soft Delete", html) self.assertNotIn("Restore", html) def test_action_visibility_deleted(self): client = self._admin_client() u = self._user("u@c.edu") listing = self._listing(u, 1, deleted=True) html = client.get(f"/admin/listings/{listing.public_id}").get_data(as_text=True) self.assertIn("Restore", html) self.assertNotIn("Soft Delete", html) self.assertNotIn("Feature Listing", html) self.assertNotIn("Mark Sold", html) # --- single actions perform --- def test_feature_and_unfeature(self): client = self._admin_client() u = self._user("u@c.edu") listing = self._listing(u, 1, status=ListingStatus.AVAILABLE) client.post(f"/admin/listings/{listing.public_id}/action", data={"action": "feature"}) db.session.expire_all() self.assertTrue(is_listing_currently_featured(db.session.get(Listing, listing.id))) client.post(f"/admin/listings/{listing.public_id}/action", data={"action": "unfeature"}) db.session.expire_all() self.assertFalse(db.session.get(Listing, listing.id).is_featured) def test_change_status_and_delete_restore(self): client = self._admin_client() u = self._user("u@c.edu") listing = self._listing(u, 1, status=ListingStatus.AVAILABLE) client.post(f"/admin/listings/{listing.public_id}/action", data={"action": "reserved"}) db.session.expire_all() self.assertEqual(db.session.get(Listing, listing.id).status, ListingStatus.RESERVED) client.post(f"/admin/listings/{listing.public_id}/action", data={"action": "delete"}) db.session.expire_all() self.assertIsNotNone(db.session.get(Listing, listing.id).deleted_at) client.post(f"/admin/listings/{listing.public_id}/action", data={"action": "restore"}) db.session.expire_all() restored = db.session.get(Listing, listing.id) self.assertIsNone(restored.deleted_at) self.assertEqual(restored.status, ListingStatus.AVAILABLE) def test_feature_rejected_for_sold_listing(self): u = self._user("u@c.edu") listing = self._listing(u, 1, status=ListingStatus.SOLD) from app.admin.listing_management import feature_listing_admin self.assertFalse(feature_listing_admin(listing)) self.assertFalse(db.session.get(Listing, listing.id).is_featured) # --- bulk actions --- def test_bulk_mark_sold(self): u = self._user("u@c.edu") a = self._listing(u, 1, status=ListingStatus.AVAILABLE) b = self._listing(u, 2, status=ListingStatus.AVAILABLE) changed = bulk_update_listings("sold", [a.public_id, b.public_id]) self.assertEqual(changed, 2) db.session.expire_all() self.assertEqual(db.session.get(Listing, a.id).status, ListingStatus.SOLD) self.assertEqual(db.session.get(Listing, b.id).status, ListingStatus.SOLD) def test_bulk_delete_via_route(self): client = self._admin_client() u = self._user("u@c.edu") a = self._listing(u, 1) b = self._listing(u, 2) client.post("/admin/listings/bulk", data={"action": "delete", "selected": [a.public_id, b.public_id]}) db.session.expire_all() self.assertIsNotNone(db.session.get(Listing, a.id).deleted_at) self.assertIsNotNone(db.session.get(Listing, b.id).deleted_at) def test_bulk_unknown_action_noop(self): u = self._user("u@c.edu") a = self._listing(u, 1) self.assertEqual(bulk_update_listings("explode", [a.public_id]), 0) # --- performance --- def test_no_n_plus_1(self): seller = self._user("s@c.edu") buyer = self._user("b@c.edu") for i in range(4): listing = self._listing(seller, i) self._offer(listing, buyer, 40) with _QC() as c1: get_listings(page=1) small = c1.n for i in range(4, 40): listing = self._listing(seller, i) self._offer(listing, buyer, 40) with _QC() as c2: get_listings(page=1) self.assertEqual(small, c2.n) # constant regardless of listing count if __name__ == "__main__": unittest.main()