"""Review & rating tests (Trust System — Phase 2).""" import os import unittest from decimal import Decimal os.environ["TEST_DATABASE_URL"] = "sqlite:///:memory:" from werkzeug.security import generate_password_hash from app import create_app from app.extensions import db from app.models import ( Category, CompletedTransaction, Conversation, Listing, ListingCondition, ListingStatus, ListingVisibility, Notification, Offer, OfferStatus, Review, ReviewStatus, User, ) from app.marketplace.services import complete_transaction from app.negotiation.services import accept_offer from app.reviews.services import ( DuplicateReviewError, InvalidReviewError, ReviewPermissionError, can_review, get_review_context, has_reviewed, submit_review, ) class ReviewTests(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): u = User(is_verified=True, email=email, password_hash=generate_password_hash("Passw0rd!")) db.session.add(u) db.session.commit() return u def _listing(self, seller, idx=1): cat = db.session.scalar(db.select(Category).filter_by(name="Books")) if cat is None: cat = Category(name="Books", description="d", is_active=True, display_order=1) db.session.add(cat) db.session.commit() listing = Listing( sku=f"SKU-{seller.id}-{idx}", seller_id=seller.id, category_id=cat.id, title=f"Item {idx}", slug=f"item-{seller.id}-{idx}", description="d", price=Decimal("50.00"), condition=ListingCondition.GOOD, status=ListingStatus.AVAILABLE, visibility=ListingVisibility.PUBLIC, review_status=ReviewStatus.APPROVED, ) db.session.add(listing) db.session.commit() return listing def _reserved(self): buyer, seller = self._user("b@c.edu"), self._user("s@c.edu") listing = self._listing(seller) convo = Conversation(listing_id=listing.id, buyer_id=buyer.id, seller_id=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=seller.id, amount=Decimal("40.00"), status=OfferStatus.PENDING) db.session.add(offer) db.session.commit() accept_offer(offer, seller) # → RESERVED, offer ACCEPTED return buyer, seller, listing def _completed(self): buyer, seller, listing = self._reserved() tx = complete_transaction(listing, seller) return buyer, seller, listing, tx # --- happy paths (independent buyer/seller reviews) --- def test_buyer_can_review_seller(self): buyer, seller, listing, tx = self._completed() review = submit_review(tx, buyer, 5, "Great seller") self.assertEqual(review.reviewer_id, buyer.id) self.assertEqual(review.reviewee_id, seller.id) self.assertEqual(review.rating, 5) self.assertEqual(review.comment, "Great seller") def test_seller_can_review_buyer(self): buyer, seller, listing, tx = self._completed() review = submit_review(tx, seller, 4) self.assertEqual(review.reviewer_id, seller.id) self.assertEqual(review.reviewee_id, buyer.id) self.assertIsNone(review.comment) def test_both_reviews_are_independent(self): buyer, seller, listing, tx = self._completed() submit_review(tx, buyer, 5) submit_review(tx, seller, 3) rows = db.session.scalars(db.select(Review).where(Review.transaction_id == tx.id)).all() self.assertEqual(len(rows), 2) # --- guards --- def test_duplicate_review_rejected(self): buyer, seller, listing, tx = self._completed() submit_review(tx, buyer, 5) with self.assertRaises(DuplicateReviewError): submit_review(tx, buyer, 3) rows = db.session.scalars( db.select(Review).where(Review.transaction_id == tx.id, Review.reviewer_id == buyer.id) ).all() self.assertEqual(len(rows), 1) # only one persisted def test_non_participant_rejected(self): buyer, seller, listing, tx = self._completed() stranger = self._user("x@c.edu") with self.assertRaises(ReviewPermissionError): submit_review(tx, stranger, 5) def test_review_before_completion_not_possible(self): buyer, seller, listing = self._reserved() # accepted offer, NOT completed # No transaction exists, so no review can be offered to either party. self.assertIsNone(get_review_context(listing, buyer)) self.assertIsNone(get_review_context(listing, seller)) def test_invalid_rating_rejected(self): buyer, seller, listing, tx = self._completed() for bad in (0, 6, -1, "abc", None): with self.assertRaises(InvalidReviewError): submit_review(tx, buyer, bad) self.assertEqual(len(db.session.scalars(db.select(Review)).all()), 0) def test_self_review_rejected_by_db_constraint(self): buyer, seller, listing, tx = self._completed() # A review can never target its own author (service always resolves the # counterparty); the DB check is the last line of defence. db.session.add(Review(transaction_id=tx.id, reviewer_id=buyer.id, reviewee_id=buyer.id, rating=5)) with self.assertRaises(Exception): db.session.commit() db.session.rollback() # --- notification --- def test_notification_sent_without_review_text(self): buyer, seller, listing, tx = self._completed() submit_review(tx, buyer, 5, "secret private feedback") notif = db.session.scalar( db.select(Notification).where(Notification.user_id == seller.id, Notification.title == "New Review") ) self.assertIsNotNone(notif) self.assertIn("recent transaction", notif.message) self.assertNotIn("secret private feedback", notif.message) # review text never leaked # --- eligibility helpers + context --- def test_can_review_and_context(self): buyer, seller, listing, tx = self._completed() self.assertTrue(can_review(tx, buyer)) ctx = get_review_context(listing, buyer) self.assertEqual(ctx["review_for"], "seller") self.assertFalse(ctx["already_reviewed"]) submit_review(tx, buyer, 5) self.assertFalse(can_review(tx, buyer)) self.assertTrue(get_review_context(listing, buyer)["already_reviewed"]) # --- route end-to-end (single client, no admin) --- def test_review_page_and_submit_via_route(self): buyer, seller, listing, tx = self._completed() client = self.app.test_client() client.post("/login", data={"email": buyer.email, "password": "Passw0rd!"}) page = client.get(f"/app/reviews/{tx.public_id}") self.assertEqual(page.status_code, 200) self.assertIn("review-stars", page.get_data(as_text=True)) # star widget rendered resp = client.post(f"/app/reviews/{tx.public_id}", data={"rating": "5", "comment": "Nice"}) self.assertEqual(resp.status_code, 302) review = db.session.scalar(db.select(Review).where(Review.reviewer_id == buyer.id)) self.assertIsNotNone(review) self.assertEqual(review.rating, 5) def test_non_participant_gets_404(self): buyer, seller, listing, tx = self._completed() self._user("x@c.edu") client = self.app.test_client() client.post("/login", data={"email": "x@c.edu", "password": "Passw0rd!"}) resp = client.get(f"/app/reviews/{tx.public_id}") self.assertEqual(resp.status_code, 404) if __name__ == "__main__": unittest.main()