"""Listing view tracking tests (Phase 4.1). Covers de-duplication (logged-in + anonymous), the 30-minute window, per-user and per-session separation, the ListingView row contents, view_count synchronisation, IP hashing (never stored raw), and that analytics read the cached count. """ import os import unittest from datetime import timedelta from decimal import Decimal os.environ["TEST_DATABASE_URL"] = "sqlite:///:memory:" from flask import request as flask_request from flask import session as flask_session from flask_login import AnonymousUserMixin from sqlalchemy import func from werkzeug.security import generate_password_hash from app import create_app from app.extensions import db from app.marketplace.services import ( VIEW_DEDUP_WINDOW_MINUTES, _hash_value, listing_view_statistics, record_listing_view, ) from app.models import ( Category, Listing, ListingCondition, ListingStatus, ListingView, ListingVisibility, ReviewStatus, User, ) from app.models.base import utcnow from app.premium.analytics import get_premium_dashboard from app.premium.services import activate_premium class ListingViewTrackingTests(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 _listing(self, seller): cat = Category(name="Books", description="B", is_active=True, display_order=1) db.session.add(cat) db.session.commit() listing = Listing( sku="SKU-1", seller_id=seller.id, category_id=cat.id, title="Item", slug="item-1", 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 _record(self, listing_id, *, user_id=None, sid=None, ip="1.2.3.4", ua="UA-1"): """Record a view inside a fresh request context (controls session/IP/UA).""" with self.app.test_request_context( "/", headers={"User-Agent": ua}, environ_base={"REMOTE_ADDR": ip} ): listing = db.session.get(Listing, listing_id) if sid is not None: flask_session["view_sid"] = sid current = db.session.get(User, user_id) if user_id else AnonymousUserMixin() return record_listing_view(listing, current, flask_request) def _count(self, listing_id): db.session.expire_all() return db.session.get(Listing, listing_id).view_count def _rows(self, listing_id): return db.session.scalar( db.select(func.count()).select_from(ListingView).where(ListingView.listing_id == listing_id) ) or 0 def _backdate(self, listing_id, minutes): for v in db.session.scalars(db.select(ListingView).where(ListingView.listing_id == listing_id)): v.viewed_at = utcnow() - timedelta(minutes=minutes) db.session.commit() # --- logged-in --- def test_first_logged_in_view_increments(self): seller = self._user("s@c.edu") viewer = self._user("v@c.edu") listing = self._listing(seller) self.assertTrue(self._record(listing.id, user_id=viewer.id)) self.assertEqual(self._count(listing.id), 1) self.assertEqual(self._rows(listing.id), 1) def test_refresh_within_window_does_not_increment(self): seller = self._user("s@c.edu") viewer = self._user("v@c.edu") listing = self._listing(seller) self._record(listing.id, user_id=viewer.id) self.assertFalse(self._record(listing.id, user_id=viewer.id)) # refresh self.assertEqual(self._count(listing.id), 1) self.assertEqual(self._rows(listing.id), 1) def test_view_after_window_increments(self): seller = self._user("s@c.edu") viewer = self._user("v@c.edu") listing = self._listing(seller) self._record(listing.id, user_id=viewer.id) self._backdate(listing.id, VIEW_DEDUP_WINDOW_MINUTES + 1) # older than window self.assertTrue(self._record(listing.id, user_id=viewer.id)) self.assertEqual(self._count(listing.id), 2) def test_different_user_counts_separately(self): seller = self._user("s@c.edu") v1 = self._user("v1@c.edu") v2 = self._user("v2@c.edu") listing = self._listing(seller) self.assertTrue(self._record(listing.id, user_id=v1.id)) self.assertTrue(self._record(listing.id, user_id=v2.id)) self.assertEqual(self._count(listing.id), 2) def test_owner_view_not_counted(self): seller = self._user("s@c.edu") listing = self._listing(seller) self.assertFalse(self._record(listing.id, user_id=seller.id)) self.assertEqual(self._count(listing.id), 0) # --- anonymous --- def test_anonymous_view_increments(self): seller = self._user("s@c.edu") listing = self._listing(seller) self.assertTrue(self._record(listing.id, sid="sess-A", ua="UA-A")) self.assertEqual(self._count(listing.id), 1) row = db.session.scalar(db.select(ListingView).where(ListingView.listing_id == listing.id)) self.assertIsNone(row.viewer_id) # anonymous def test_anonymous_refresh_blocked(self): seller = self._user("s@c.edu") listing = self._listing(seller) self._record(listing.id, sid="sess-A", ua="UA-A") self.assertFalse(self._record(listing.id, sid="sess-A", ua="UA-A")) # same session + IP/UA self.assertEqual(self._count(listing.id), 1) def test_different_session_counts_separately(self): seller = self._user("s@c.edu") listing = self._listing(seller) # Different session AND different user-agent, so neither dedup clause matches. self.assertTrue(self._record(listing.id, sid="sess-A", ua="UA-A")) self.assertTrue(self._record(listing.id, sid="sess-B", ua="UA-B")) self.assertEqual(self._count(listing.id), 2) def test_same_ip_and_agent_blocks_even_new_session(self): seller = self._user("s@c.edu") listing = self._listing(seller) # New session but identical IP + user agent -> blocked by the IP/UA clause. self._record(listing.id, sid="sess-A", ua="UA-A", ip="9.9.9.9") self.assertFalse(self._record(listing.id, sid="sess-B", ua="UA-A", ip="9.9.9.9")) self.assertEqual(self._count(listing.id), 1) # --- row contents + security --- def test_listing_view_row_created_correctly(self): seller = self._user("s@c.edu") viewer = self._user("v@c.edu") listing = self._listing(seller) self._record(listing.id, user_id=viewer.id, ip="8.8.8.8", ua="Mozilla/5.0") row = db.session.scalar(db.select(ListingView).where(ListingView.listing_id == listing.id)) self.assertEqual(row.viewer_id, viewer.id) self.assertIsNotNone(row.viewed_at) # IP/UA stored only as 64-char SHA-256 hashes — never raw. self.assertEqual(len(row.ip_hash), 64) self.assertNotIn("8.8.8.8", row.ip_hash) self.assertEqual(row.ip_hash, _hash_value("8.8.8.8")) self.assertNotEqual(row.user_agent_hash, "Mozilla/5.0") def test_raw_ip_never_stored_anywhere(self): seller = self._user("s@c.edu") listing = self._listing(seller) self._record(listing.id, sid="s", ip="203.0.113.7", ua="UA-A") rows = db.session.scalars(db.select(ListingView)).all() for r in rows: self.assertNotEqual(r.ip_hash, "203.0.113.7") self.assertNotIn("203.0.113", r.ip_hash or "") # --- synchronisation + analytics --- def test_cached_count_stays_synchronized(self): seller = self._user("s@c.edu") listing = self._listing(seller) self._record(listing.id, user_id=self._user("a@c.edu").id) self._record(listing.id, user_id=self._user("b@c.edu").id) self._record(listing.id, sid="x", ua="UA-X") self.assertEqual(self._count(listing.id), self._rows(listing.id)) stats = listing_view_statistics(db.session.get(Listing, listing.id)) self.assertEqual(stats["cached_view_count"], stats["tracked_views"]) def test_analytics_display_correct_values(self): seller = self._user("s@c.edu") activate_premium(seller) listing = self._listing(seller) self._record(listing.id, user_id=self._user("a@c.edu").id) self._record(listing.id, user_id=self._user("b@c.edu").id) dash = get_premium_dashboard(db.session.get(User, seller.id)) self.assertEqual(dash["summary"]["total_views"], 2) self.assertEqual(dash["top_listings"][0]["views"], 2) def test_route_records_view(self): seller = self._user("s@c.edu") viewer = self._user("v@c.edu") listing = self._listing(seller) client = self.app.test_client() client.post("/login", data={"email": "v@c.edu", "password": "Passw0rd!"}) client.get(f"/app/listings/{listing.public_id}", headers={"User-Agent": "UA-A"}) self.assertEqual(self._count(listing.id), 1) # Refresh from the same client must not double-count. client.get(f"/app/listings/{listing.public_id}", headers={"User-Agent": "UA-A"}) self.assertEqual(self._count(listing.id), 1) if __name__ == "__main__": unittest.main()