"""Advertisement video file upload tests. Covers the video-upload workflow that reuses the image upload infrastructure: successful MP4/WebM uploads, format/MIME/size validation, replace & delete cleanup, switching media type, playback from the uploaded file, legacy external-URL backward compatibility, and image ads remaining unaffected. """ import io import os import shutil import tempfile import unittest from decimal import Decimal from pathlib import Path os.environ["TEST_DATABASE_URL"] = "sqlite:///:memory:" from werkzeug.security import generate_password_hash from werkzeug.datastructures import FileStorage from app import create_app from app.admin.advertisements.services import duplicate_advertisement, save_ad_video from app.advertisements.rate_limit import reset_rate_limits from app.advertisements.reward_token import issue_reward_token, reset_replay_guard from app.chat.services import get_messaging_state from app.extensions import db from app.models import ( Advertisement, AdvertisementMediaType, AdvertisementPlacement, AdvertisementType, Category, Conversation, ConversationStatus, Listing, ListingCondition, ListingStatus, ListingVisibility, ReviewStatus, User, UserRole, ) class AdVideoUploadTests(unittest.TestCase): def setUp(self): self.app = create_app("testing") self.ctx = self.app.app_context() self.ctx.push() db.create_all() reset_rate_limits() reset_replay_guard() self._upload_dir = Path(tempfile.mkdtemp()) self.app.config["ADVERTISEMENT_UPLOAD_FOLDER"] = self._upload_dir def tearDown(self): reset_rate_limits() shutil.rmtree(self._upload_dir, ignore_errors=True) db.session.remove() db.drop_all() self.ctx.pop() # --- file fixtures (valid magic bytes) ------------------------------------ def _mp4(self, name="ad.mp4", pad=16): return (io.BytesIO(b"\x00\x00\x00\x20ftypisom" + b"\x00" * pad), name) def _webm(self, name="ad.webm"): return (io.BytesIO(b"\x1a\x45\xdf\xa3" + b"\x00" * 16), name) def _fake_mp4(self, name="malware.mp4"): return (io.BytesIO(b"MZ\x90\x00" + b"\x00" * 40), name) # PE header, not video def _stored_file(self, media_url): return self._upload_dir / os.path.basename(media_url) # --- factories ------------------------------------------------------------- def _user(self, email, role=UserRole.STUDENT): u = User(email=email, password_hash=generate_password_hash("Passw0rd!"), is_verified=True, role=role) db.session.add(u) db.session.commit() return u def _admin_client(self): self._user("admin@c.edu", UserRole.SUPER_ADMIN) c = self.app.test_client() c.post("/login", data={"email": "admin@c.edu", "password": "Passw0rd!"}) return c def _base_form(self, **overrides): data = { "title": "Ad", "advertisement_type": "rewarded", "placement": "landing", "media_type": "image", "priority": "5", "is_active": "y", "submit": "Save advertisement", } data.update(overrides) return data def _create_video_ad(self, client, title="Video Ad", video=None): client.post("/admin/advertisements/new", data=self._base_form( title=title, media_type="video", video_duration="15", video=video or self._mp4(), ), content_type="multipart/form-data", follow_redirects=True) return db.session.scalar(db.select(Advertisement).filter_by(title=title)) def _ad_model(self, *, media_url, title="Legacy"): """A video ad created directly (external URL or a service-saved file path).""" ad = Advertisement(title=title, advertisement_type=AdvertisementType.REWARDED, placement=AdvertisementPlacement.CHAT, media_type=AdvertisementMediaType.VIDEO, media_url=media_url, video_duration=15, is_active=True, priority=10) db.session.add(ad) db.session.commit() return ad def _uploaded_video_ad(self): """A video ad whose file was saved via the real upload helper (no admin client needed — avoids two authenticated clients in one test context).""" f = FileStorage(stream=io.BytesIO(b"\x00\x00\x00\x20ftypisom" + b"\x00" * 16), filename="ad.mp4") return self._ad_model(media_url=save_ad_video(f), title="Uploaded Video") def _convo_client(self, ad): buyer = self._user("b@c.edu") 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="SKU-1", seller_id=seller.id, category_id=cat.id, title="Book", slug="book-1", 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, status=ConversationStatus.ACTIVE) db.session.add(convo) db.session.commit() c = self.app.test_client() c.post("/login", data={"email": "b@c.edu", "password": "Passw0rd!"}) return c, convo, buyer # --- successful uploads ---------------------------------------------------- def test_upload_mp4_stores_file(self): ad = self._create_video_ad(self._admin_client()) self.assertIsNotNone(ad) self.assertTrue(ad.media_url.endswith(".mp4")) self.assertIn("/static/uploads/advertisements/", ad.media_url) self.assertTrue(self._stored_file(ad.media_url).is_file()) def test_upload_webm_stores_file(self): client = self._admin_client() client.post("/admin/advertisements/new", data=self._base_form( title="WebM Ad", media_type="video", video=self._webm(), ), content_type="multipart/form-data", follow_redirects=True) ad = db.session.scalar(db.select(Advertisement).filter_by(title="WebM Ad")) self.assertTrue(ad.media_url.endswith(".webm")) self.assertTrue(self._stored_file(ad.media_url).is_file()) # --- validation ------------------------------------------------------------ def test_unsupported_extension_rejected(self): client = self._admin_client() resp = client.post("/admin/advertisements/new", data=self._base_form( title="Exe Ad", media_type="video", video=(io.BytesIO(b"MZ..."), "x.exe"), ), content_type="multipart/form-data", follow_redirects=True) self.assertIsNone(db.session.scalar(db.select(Advertisement).filter_by(title="Exe Ad"))) self.assertIn("videos only", resp.get_data(as_text=True).lower()) def test_mime_sniff_rejects_fake_video(self): client = self._admin_client() resp = client.post("/admin/advertisements/new", data=self._base_form( title="Fake Ad", media_type="video", video=self._fake_mp4(), ), content_type="multipart/form-data", follow_redirects=True) self.assertIsNone(db.session.scalar(db.select(Advertisement).filter_by(title="Fake Ad"))) self.assertIn("does not look like a valid", resp.get_data(as_text=True).lower()) def test_oversized_rejected(self): self.app.config["MAX_AD_VIDEO_SIZE"] = 100 # bytes client = self._admin_client() resp = client.post("/admin/advertisements/new", data=self._base_form( title="Big Ad", media_type="video", video=self._mp4(pad=300), ), content_type="multipart/form-data", follow_redirects=True) self.assertIsNone(db.session.scalar(db.select(Advertisement).filter_by(title="Big Ad"))) self.assertIn("too large", resp.get_data(as_text=True).lower()) def test_video_required_when_type_video(self): client = self._admin_client() resp = client.post("/admin/advertisements/new", data=self._base_form( title="No File", media_type="video", ), content_type="multipart/form-data", follow_redirects=True) self.assertIsNone(db.session.scalar(db.select(Advertisement).filter_by(title="No File"))) self.assertIn("video file is required", resp.get_data(as_text=True).lower()) # --- replace / delete cleanup --------------------------------------------- def test_replace_video_deletes_old_file(self): client = self._admin_client() ad = self._create_video_ad(client) old_file = self._stored_file(ad.media_url) self.assertTrue(old_file.is_file()) client.post(f"/admin/advertisements/{ad.public_id}/edit", data=self._base_form( title="Video Ad", media_type="video", video_duration="15", video=self._mp4("new.mp4"), ), content_type="multipart/form-data", follow_redirects=True) db.session.refresh(ad) self.assertTrue(self._stored_file(ad.media_url).is_file()) # new file present self.assertFalse(old_file.is_file()) # old file cleaned up def test_delete_ad_removes_video_file(self): client = self._admin_client() ad = self._create_video_ad(client) stored = self._stored_file(ad.media_url) self.assertTrue(stored.is_file()) client.post(f"/admin/advertisements/{ad.public_id}/delete", follow_redirects=True) self.assertFalse(stored.is_file()) def test_switch_to_image_removes_video_file(self): client = self._admin_client() ad = self._create_video_ad(client) stored = self._stored_file(ad.media_url) client.post(f"/admin/advertisements/{ad.public_id}/edit", data=self._base_form( title="Video Ad", media_type="image", icon="bi-stars", ), content_type="multipart/form-data", follow_redirects=True) db.session.refresh(ad) self.assertEqual(ad.media_type, AdvertisementMediaType.IMAGE) self.assertIsNone(ad.media_url) self.assertFalse(stored.is_file()) def test_duplicate_shares_reference_and_file_kept(self): client = self._admin_client() ad = self._create_video_ad(client) stored = self._stored_file(ad.media_url) copy = duplicate_advertisement(ad) self.assertEqual(copy.media_url, ad.media_url) # Deleting the original must NOT remove a file still referenced by the copy. client.post(f"/admin/advertisements/{ad.public_id}/delete", follow_redirects=True) self.assertTrue(stored.is_file()) # --- playback -------------------------------------------------------------- def test_playback_uses_uploaded_file(self): ad = self._uploaded_video_ad() self.assertTrue(self._stored_file(ad.media_url).is_file()) chat, convo, _ = self._convo_client(ad) html = chat.get(f"/app/chat/{convo.public_id}").get_data(as_text=True) self.assertIn("