"""Activity Feed tests.""" 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 ( ActivityEvent, Category, Conversation, Listing, ListingCondition, ListingStatus, ListingVisibility, Offer, OfferStatus, RelatedEntityType, ReviewStatus, User, ) from app.activity import services class ActivityTests(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() 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 _feed(self, user, **kwargs): with self.app.test_request_context(): _, items = services.get_user_activity(user, **kwargs) return items # --- creation + ordering --- def test_publish_creates_event(self): user = self._user("u@c.edu") ev = services.publish_activity(user, "welcome", "Welcome") self.assertIsNotNone(ev) self.assertEqual(ev.icon, "bi-emoji-smile") # icon defaulted from registry self.assertFalse(ev.is_read) def test_feed_is_newest_first(self): user = self._user("u@c.edu") services.publish_activity(user, "welcome", "First") services.publish_activity(user, "listing_created", "Second") titles = [i["title"] for i in self._feed(user)] self.assertEqual(titles[0], "Second") # --- grouping --- def test_similar_events_are_grouped(self): user = self._user("u@c.edu") for _ in range(3): services.publish_activity(user, "favorite_received", "Listing saved") items = self._feed(user) self.assertEqual(len(items), 1) self.assertEqual(items[0]["count"], 3) self.assertIn("3 people saved", items[0]["message"]) # --- deep links --- def test_deep_link_resolves_to_listing(self): seller = self._user("s@c.edu") cat = Category(name="Books", description="d", is_active=True, display_order=1) db.session.add(cat); db.session.commit() listing = Listing(sku="S1", seller_id=seller.id, category_id=cat.id, title="Item", slug="s1", description="d", price=Decimal("50"), condition=ListingCondition.GOOD, status=ListingStatus.AVAILABLE, visibility=ListingVisibility.PUBLIC, review_status=ReviewStatus.APPROVED) db.session.add(listing); db.session.commit() services.publish_activity(seller, "listing_created", "Listing created", related_entity_type=RelatedEntityType.LISTING, related_entity_id=listing.id) item = self._feed(seller)[0] self.assertIn(listing.public_id, item["action_url"]) self.assertEqual(item["action_label"], "View Listing") # --- read / unread --- def test_read_unread_and_filters(self): user = self._user("u@c.edu") a = services.publish_activity(user, "welcome", "A") services.publish_activity(user, "listing_created", "B") self.assertEqual(services.unread_count(user), 2) self.assertTrue(services.mark_activity_read(a.id, user)) self.assertEqual(services.unread_count(user), 1) self.assertEqual(len(self._feed(user, unread_only=True)), 1) def test_mark_all_read_and_clear_read(self): user = self._user("u@c.edu") services.publish_activity(user, "welcome", "A") services.publish_activity(user, "listing_created", "B") self.assertEqual(services.mark_all_read(user), 2) self.assertEqual(services.unread_count(user), 0) self.assertEqual(services.clear_read(user), 2) # both now read → cleared self.assertEqual(len(self._feed(user)), 0) def test_type_filter(self): user = self._user("u@c.edu") services.publish_activity(user, "welcome", "A") services.publish_activity(user, "listing_created", "B") items = self._feed(user, event_type="welcome") self.assertEqual([i["event_type"] for i in items], ["welcome"]) # --- dashboard preview --- def test_dashboard_preview_limited(self): user = self._user("u@c.edu") for i in range(8): services.publish_activity(user, "listing_created", f"L{i}") with self.app.test_request_context(): recent = services.get_recent_activity(user, limit=5) self.assertEqual(len(recent), 5) # --- source hooks (activities auto-published from existing services) --- def test_welcome_activity_on_registration(self): from app.auth.services import register_user user = register_user("New User", "new@c.edu", "Passw0rd!") types = [e.event_type for e in db.session.scalars( db.select(ActivityEvent).where(ActivityEvent.user_id == user.id)).all()] self.assertIn("welcome", types) def test_offer_accepted_publishes_activity(self): seller, buyer = self._user("s@c.edu"), self._user("b@c.edu") cat = Category(name="Books", description="d", is_active=True, display_order=1) db.session.add(cat); db.session.commit() listing = Listing(sku="S1", seller_id=seller.id, category_id=cat.id, title="Item", slug="s1", description="d", price=Decimal("50"), condition=ListingCondition.GOOD, status=ListingStatus.AVAILABLE, visibility=ListingVisibility.PUBLIC, review_status=ReviewStatus.APPROVED) db.session.add(listing); db.session.commit() convo = Conversation(listing_id=listing.id, buyer_id=buyer.id, seller_id=seller.id) db.session.add(convo); db.session.commit() from app.negotiation.services import accept_offer, create_offer offer = create_offer(convo, buyer, Decimal("40")) # publishes new_negotiation → seller accept_offer(offer, seller) # publishes offer_accepted → buyer # seller got "new negotiation", buyer got "offer accepted" seller_types = {e.event_type for e in db.session.scalars( db.select(ActivityEvent).where(ActivityEvent.user_id == seller.id)).all()} buyer_types = {e.event_type for e in db.session.scalars( db.select(ActivityEvent).where(ActivityEvent.user_id == buyer.id)).all()} self.assertIn("new_negotiation", seller_types) self.assertIn("offer_accepted", buyer_types) # --- route --- def test_activity_page_and_mark_all_read_route(self): user = self._user("u@c.edu") services.publish_activity(user, "welcome", "Hello there") client = self.app.test_client() client.post("/login", data={"email": "u@c.edu", "password": "Passw0rd!"}) html = client.get("/app/activity/").get_data(as_text=True) self.assertEqual(200, client.get("/app/activity/").status_code) self.assertIn("Hello there", html) client.post("/app/activity/mark-all-read", data={}) self.assertEqual(services.unread_count(user), 0) # --- redesign: category filter, date sections, card metadata --- def test_category_filter(self): user = self._user("u@c.edu") services.publish_activity(user, "offer_accepted", "Accepted") services.publish_activity(user, "listing_created", "Listed") with self.app.test_request_context(): _, items = services.get_user_activity(user, category="offers") self.assertEqual([i["event_type"] for i in items], ["offer_accepted"]) def test_group_by_date_buckets_today(self): user = self._user("u@c.edu") services.publish_activity(user, "welcome", "Hi") items = self._feed(user) sections = services.group_by_date(items) self.assertEqual(sections[0][0], "Today") def test_display_items_carry_presentation_metadata(self): user = self._user("u@c.edu") services.publish_activity(user, "offer_accepted", "Accepted") items = self._feed(user) self.assertEqual(items[0]["tone"], "success") self.assertEqual(items[0]["category"], "offers") self.assertEqual(items[0]["status_label"], "Accepted") if __name__ == "__main__": unittest.main()