"""Advertisement media-management UI tests. Covers the rich-media list enhancements: media-type filter, per-ad rewarded metrics (batched, no N+1), storage statistics (+ caching), video thumbnails and the media preview modal (which must record NO analytics), the per-ad analytics page, and the drag-and-drop uploader markup. """ import os import shutil import tempfile import unittest from pathlib import Path 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.advertisements import services as ad_admin from app.extensions import db from app.models import ( Advertisement, AdvertisementMediaType, AdvertisementPlacement, AdvertisementType, RewardedAdEvent, User, UserRole, ) class _QueryCounter: def __enter__(self): self.n = 0 self._engine = db.engine event.listen(self._engine, "after_cursor_execute", self._cb) return self def __exit__(self, *a): event.remove(self._engine, "after_cursor_execute", self._cb) def _cb(self, *a): self.n += 1 class AdMediaManagementTests(unittest.TestCase): def setUp(self): self.app = create_app("testing") self.ctx = self.app.app_context() self.ctx.push() db.create_all() self._upload_dir = Path(tempfile.mkdtemp()) self.app.config["ADVERTISEMENT_UPLOAD_FOLDER"] = self._upload_dir ad_admin.invalidate_storage_stats() def tearDown(self): ad_admin.invalidate_storage_stats() shutil.rmtree(self._upload_dir, ignore_errors=True) db.session.remove() db.drop_all() self.ctx.pop() # --- factories ------------------------------------------------------------- def _admin_client(self): u = User(email="a@c.edu", password_hash=generate_password_hash("Passw0rd!"), is_verified=True, role=UserRole.SUPER_ADMIN) db.session.add(u) db.session.commit() c = self.app.test_client() c.post("/login", data={"email": "a@c.edu", "password": "Passw0rd!"}) return c def _write_asset(self, name, size): (self._upload_dir / name).write_bytes(b"\x00" * size) return f"/static/uploads/advertisements/{name}" def _video_ad(self, title="Video Ad", *, media_url=None, duration=15, active=True): ad = Advertisement(title=title, advertisement_type=AdvertisementType.REWARDED, placement=AdvertisementPlacement.CHAT, media_type=AdvertisementMediaType.VIDEO, media_url=media_url or "/static/uploads/advertisements/v.mp4", video_duration=duration, is_active=active, priority=5) db.session.add(ad) db.session.commit() return ad def _image_ad(self, title="Image Ad", *, image_path="/static/uploads/advertisements/i.png"): ad = Advertisement(title=title, advertisement_type=AdvertisementType.BANNER, placement=AdvertisementPlacement.MARKETPLACE, image_path=image_path, is_active=True, priority=5) db.session.add(ad) db.session.commit() return ad def _event(self, ad, event_type, reason=None): db.session.add(RewardedAdEvent(advertisement_id=ad.id, event_type=event_type, reason=reason)) db.session.commit() # --- media type filter ----------------------------------------------------- def test_media_type_filter_service(self): self._image_ad("Img") self._video_ad("Vid") vids = [a.title for a in ad_admin.list_advertisements(1, media_type="video").items] imgs = [a.title for a in ad_admin.list_advertisements(1, media_type="image").items] self.assertEqual(vids, ["Vid"]) self.assertEqual(imgs, ["Img"]) def test_media_type_filter_route(self): self._image_ad("ImgOnly") self._video_ad("VidOnly") html = self._admin_client().get("/admin/advertisements?media_type=video").get_data(as_text=True) # Assert on the table rows (data-title) — the global charts embed all titles. self.assertIn('data-title="VidOnly"', html) self.assertNotIn('data-title="ImgOnly"', html) self.assertIn('name="media_type"', html) # filter control present # --- per-ad rewarded metrics ---------------------------------------------- def test_rewarded_stats_map_values(self): ad = self._video_ad() for et in ("impression", "impression", "reward_granted", "reward_rejected", "video_error"): self._event(ad, et) stats = ad_admin.rewarded_stats_map([ad.id])[ad.id] self.assertEqual(stats["impressions"], 2) self.assertEqual(stats["rewards_granted"], 1) self.assertEqual(stats["rejections"], 1) self.assertEqual(stats["video_failures"], 1) self.assertEqual(stats["completion_rate"], 50.0) self.assertTrue(stats["has_activity"]) def test_rewarded_stats_map_no_n_plus_1(self): ads = [self._video_ad(f"V{i}") for i in range(20)] for ad in ads: self._event(ad, "impression") ids = [a.id for a in ads] # resolve ids before counting with _QueryCounter() as qc: result = ad_admin.rewarded_stats_map(ids) self.assertEqual(len(result), 20) self.assertEqual(qc.n, 1) # single grouped query regardless of ad count def test_rewarded_metrics_render_in_list(self): ad = self._video_ad("Rewarded Vid") for et in ("impression", "impression", "reward_granted"): self._event(ad, et) html = self._admin_client().get("/admin/advertisements").get_data(as_text=True) self.assertIn("ad-rw-chip", html) # completion/rewards chips on the rewarded row # --- storage stats --------------------------------------------------------- def test_storage_stats_values(self): img = self._write_asset("pic.png", 1000) vid = self._write_asset("clip.mp4", 5000) self._image_ad("I", image_path=img) self._video_ad("V", media_url=vid) self._write_asset("orphan.mp4", 200) # on disk, unreferenced s = ad_admin.advertisement_storage_stats(force=True) self.assertEqual(s["total_bytes"], 6000) self.assertEqual(s["media_file_count"], 2) self.assertEqual(s["largest_video_bytes"], 5000) self.assertEqual(s["average_bytes"], 3000) self.assertEqual(s["unused_files"], 1) self.assertEqual(s["ad_count"], 2) def test_storage_stats_cached_and_invalidated(self): self._image_ad("I", image_path=self._write_asset("pic.png", 1000)) first = ad_admin.advertisement_storage_stats(force=True) # Add another file+ad WITHOUT force → cached value unchanged. self._video_ad("V", media_url=self._write_asset("clip.mp4", 5000)) cached = ad_admin.advertisement_storage_stats() self.assertEqual(cached["total_bytes"], first["total_bytes"]) # Invalidate → recomputes. ad_admin.invalidate_storage_stats() fresh = ad_admin.advertisement_storage_stats() self.assertEqual(fresh["total_bytes"], 6000) def test_storage_human_format(self): self._image_ad("I", image_path=self._write_asset("big.png", 2 * 1024 * 1024)) s = ad_admin.advertisement_storage_stats(force=True) self.assertEqual(s["total_human"], "2.0 MB") # --- video thumbnail + preview modal -------------------------------------- def test_list_renders_video_thumbnail(self): self._video_ad("Vid", duration=22) html = self._admin_client().get("/admin/advertisements").get_data(as_text=True) self.assertIn("ad-thumb-play", html) # play overlay self.assertIn(">22s<", html) # duration badge self.assertIn('data-media-type="video"', html) self.assertIn("ad-media-badge--video", html) def test_preview_modal_present_with_controls(self): self._image_ad("Img") html = self._admin_client().get("/admin/advertisements").get_data(as_text=True) self.assertIn('id="mediaPreviewModal"', html) self.assertIn("ad-media-video", html) self.assertIn("controls", html) self.assertIn("not counted as an advertisement impression", html) def test_preview_page_records_no_analytics(self): self._video_ad("Vid") before = db.session.scalar(db.select(db.func.count(RewardedAdEvent.id))) self._admin_client().get("/admin/advertisements") after = db.session.scalar(db.select(db.func.count(RewardedAdEvent.id))) self.assertEqual(before, after) # rendering the list records no impressions self.assertEqual(after, 0) # --- per-ad analytics page ------------------------------------------------- def test_per_ad_analytics_page(self): ad = self._video_ad("Vid") for et in ("impression", "reward_granted"): self._event(ad, et) html = self._admin_client().get(f"/admin/advertisements/{ad.public_id}/stats").get_data(as_text=True) self.assertIn("Rewarded engagement", html) self.assertIn("Completion rate", html) self.assertIn("chartAdFunnel", html) # chart rendered when there is activity def test_list_links_to_per_ad_analytics(self): ad = self._video_ad("Vid") html = self._admin_client().get("/admin/advertisements").get_data(as_text=True) self.assertIn(f"/admin/advertisements/{ad.public_id}/stats", html) # --- drag & drop uploader -------------------------------------------------- def test_create_form_has_dropzone(self): html = self._admin_client().get("/admin/advertisements/new").get_data(as_text=True) self.assertIn("uploader-dropzone", html) self.assertIn("data-uploader", html) self.assertIn("Drag & drop", html) self.assertIn("Recommended 720p", html) # video upload guidance if __name__ == "__main__": unittest.main()