from pathlib import Path from fastapi.testclient import TestClient from sqlmodel import Session, select from app.core.config import get_settings from app.db.session import init_db, reset_db_for_tests from app.main import app from app.models.core import ActivityEvent, ExternalPost, PostDraft, ProductionQueueItem, Provider, UploadRequest, VideoAsset, WebsiteCompanion, YouTubeUploadAudit from app.services.importers.prepared_package import PreparedPackageImporter from app.services.reconciliation import detect_pending_website_duplicates, import_existing_audit from app.services.upload_requests import create_existing_upload_request AUDIT_025 = Path("/home/agent/jarvis_runtime/autoshortsbot/AutoShortsBot/data/youtube_upload_audits/erf-025-mfa-prompt-without-login_chatterbox_gianna_clone_premium_v1_readable_silent_2line_clean_ai_youtube_upload_audit.json") def setup_test_db(tmp_path): settings = get_settings() settings.database_url = "sqlite://" settings.storage_root = tmp_path / "storage" engine = init_db(settings) reset_db_for_tests(engine) return engine, TestClient(app) def test_import_existing_youtube_audit_creates_links(tmp_path): engine, _ = setup_test_db(tmp_path) with Session(engine) as session: result = import_existing_audit(session, AUDIT_025, dry_run=False) assert result["youtube_video_id"] == "J6ErKh55-KQ" audit = session.get(YouTubeUploadAudit, result["audit_id"]) video = session.get(VideoAsset, result["video_asset_id"]) draft = session.get(PostDraft, result["post_draft_id"]) external = session.exec(select(ExternalPost).where(ExternalPost.external_post_id == "J6ErKh55-KQ")).first() companion = session.exec(select(WebsiteCompanion).where(WebsiteCompanion.video_asset_id == video.id)).first() assert audit.upload_status == "external_existing_upload" assert video.external_youtube_id == "J6ErKh55-KQ" assert draft.external_post_id == "J6ErKh55-KQ" assert external is not None assert companion.status == "pushed" assert companion.slug == "mfa-prompt-without-login" def test_real_youtube_id_overwrites_dryrun_for_same_candidate(tmp_path): engine, _ = setup_test_db(tmp_path) with Session(engine) as session: video = VideoAsset(candidate_id="erf-025-mfa-prompt-without-login", package_id="erf-025-mfa-prompt-without-login", source="dry_run", file_path="", checksum="dryrunchecksum", working_title="MFA", external_youtube_id="dryrun_old") session.add(video); session.commit(); session.refresh(video) draft = PostDraft(video_asset_id=video.id, provider=Provider.youtube, title="MFA", description="desc", privacy_status="private", external_post_id="dryrun_old") session.add(draft); session.commit() result = import_existing_audit(session, AUDIT_025, dry_run=False) video = session.get(VideoAsset, video.id) draft = session.get(PostDraft, draft.id) assert result["youtube_video_id"] == "J6ErKh55-KQ" assert video.external_youtube_id == "J6ErKh55-KQ" assert draft.external_post_id == "J6ErKh55-KQ" def test_upload_request_outbox_and_agent_pending_contains_no_secrets(tmp_path): engine, client = setup_test_db(tmp_path) with Session(engine) as session: video = VideoAsset(candidate_id="erf-test", package_id="erf-test", source="manual", file_path="", checksum="abcdef1234567890", working_title="Upload Request Test") session.add(video); session.commit(); session.refresh(video) draft = PostDraft(video_asset_id=video.id, provider=Provider.youtube, title="Upload Request Test", description="desc", privacy_status="private") session.add(draft); session.commit(); session.refresh(draft) req = create_existing_upload_request(session, draft.id) assert req["status"] == "approved" outbox = Path(req["outbox_path"]) assert outbox.exists() text = outbox.read_text() assert "APPROVE_UPLOAD erf-test abcdef123456" in text assert "token" not in text.lower() pending = client.get("/api/agent/upload-requests/pending").json()["items"] assert len(pending) == 1 assert pending[0]["expected_candidate_id"] == "erf-test" def test_upload_request_agent_status_transitions_without_upload(tmp_path): engine, client = setup_test_db(tmp_path) with Session(engine) as session: video = VideoAsset(candidate_id="transition-test", package_id="transition-test", source="manual", file_path="", checksum="transitionchecksum", working_title="Transition Test") session.add(video); session.commit(); session.refresh(video) draft = PostDraft(video_asset_id=video.id, provider=Provider.youtube, title="Transition Test", description="desc", privacy_status="private") session.add(draft); session.commit(); session.refresh(draft) req = create_existing_upload_request(session, draft.id) sent = client.post(f"/api/agent/upload-requests/{req['id']}/mark-sent").json() assert sent["status"] == "sent_to_executor" completed = client.post(f"/api/agent/upload-requests/{req['id']}/complete", json={"youtube_video_id": "manual_completion_test"}).json() assert completed["status"] == "completed" assert completed["youtube_video_id"] == "manual_completion_test" def test_production_queue_ordering_locked_wins_and_approved_fallback(tmp_path): _, client = setup_test_db(tmp_path) low = client.post("/api/production-queue", json={"title": "Approved low", "brief": "B", "status": "approved_next", "position": 2, "priority": 1, "source": "manual"}).json() locked = client.post("/api/production-queue", json={"title": "Locked wins", "brief": "A", "status": "locked_next", "position": 9, "priority": 0, "source": "manual"}).json() nxt = client.get("/api/agent/production/next").json() assert nxt["queue_item_id"] == locked["id"] client.post(f"/api/production-queue/{locked['id']}", json={"status": "archived"}) fallback = client.get("/api/agent/production/next").json() assert fallback["queue_item_id"] == low["id"] def test_production_agent_lifecycle_and_ready_for_review(tmp_path): _, client = setup_test_db(tmp_path) created = client.post("/api/production-queue", json={"title": "Next locked item", "brief": "Make it short", "reason": "Manual test", "status": "suggested", "source": "manual"}).json() locked = client.post(f"/api/production-queue/{created['id']}/lock-next").json() assert locked["status"] == "locked_next" nxt = client.get("/api/agent/production/next").json() assert nxt["queue_item_id"] == created["id"] assert nxt["production_rules"]["no_final_wan_render_without_approval"] is True assert nxt["production_rules"]["ready_for_review_requires_quality_passed"] is True assert nxt["quality_profile"]["version"] == "autoshorts-quality-v1.1" assert "no_large_top_title" in nxt["quality_profile"]["visual_checks_required"] assert nxt["director_skill"]["required"] is True blocked = client.post(f"/api/agent/production/{created['id']}/mark-started") assert blocked.status_code == 409 concept = client.post(f"/api/agent/production/{created['id']}/concept-proposed").json() assert concept["concept"]["confirm_command"] == f"APPROVE_PRODUCTION {created['id']}" generate_blocked = client.post(f"/api/agent/production/{created['id']}/generate-review-package") assert generate_blocked.status_code == 409 approved = client.post(f"/api/agent/production/{created['id']}/approve-generation").json() assert approved["status"] == "concept_approved" generated = client.post(f"/api/agent/production/{created['id']}/generate-review-package").json() assert generated["ok"] is True assert generated["mode"] == "workflow_preview" assert generated["video_asset_id"] assert Path(generated["video_path"]).exists() assert Path(generated["manifest_path"]).exists() ready_item = client.get("/api/production-queue").json()["items"][0] assert ready_item["status"] == "in_review" assert ready_item["video_asset_id"] == generated["video_asset_id"] def test_send_to_production_legacy_route_cannot_start_without_concept_approval(tmp_path): _, client = setup_test_db(tmp_path) created = client.post("/api/production-queue", json={"title": "Safe legacy route", "brief": "B", "status": "suggested", "source": "manual"}).json() sent = client.post(f"/api/production-queue/{created['id']}/send-to-production").json() assert sent["status"] == "concept_proposed" blocked = client.post(f"/api/agent/production/{created['id']}/mark-started") assert blocked.status_code == 409 approved = client.post(f"/api/agent/production/{created['id']}/approve-generation").json() assert approved["status"] == "concept_approved" started = client.post(f"/api/agent/production/{created['id']}/mark-started").json() assert started["status"] == "producing" def test_prepared_package_import_links_manifest_to_queue_item(tmp_path): engine, _ = setup_test_db(tmp_path) settings = get_settings() package_id = "package_queue_test" source_video = tmp_path / "input.mp4" source_video.write_bytes(b"fake video bytes") with Session(engine) as session: item = ProductionQueueItem(title="Queue linked package", status="locked_next", source="manual", expected_package_id=package_id, expected_package_manifest=f"storage/incoming/{package_id}/manifest.json") session.add(item); session.commit(); session.refresh(item) manifest = { "package_id": package_id, "source": "prepared_package", "test_package": True, "queue_item_id": item.id, "working_title": "Queue linked package", "language": "en", "series": "Everyday Red Flags", "suggested_youtube_title": "Queue linked package", "suggested_youtube_description": "Test package", "suggested_hashtags": ["#test"], "script": "Script text", "disclosure": {"synthetic_media": True, "commercial_content": False}, "quality": { "visual_gate": "passed", "voice_gate": "passed", "hook_gate": "passed", "metadata_gate": "passed", "safety_gate": "passed", "package_integrity_gate": "passed", "overall": "passed", }, "status": "ready_for_review", } importer = PreparedPackageImporter(settings=settings, engine=engine) video_id = importer.import_manifest_payload(manifest, source_video) with Session(engine) as session: item = session.get(ProductionQueueItem, item.id) video = session.get(VideoAsset, video_id) event = session.exec(select(ActivityEvent).where(ActivityEvent.event_type == "production_package_imported")).first() assert item.status == "package_ready" assert item.video_asset_id == video_id assert item.actual_package_id == package_id assert video.package_id == package_id assert video.script == "Script text" assert event is not None def test_website_duplicate_detection_catches_pending_login_prompt(tmp_path): engine, _ = setup_test_db(tmp_path) with Session(engine) as session: result = detect_pending_website_duplicates(session) dupes = result["potential_duplicates"] assert dupes assert dupes[0]["existing_slug"] == "mfa-prompt-without-login" assert dupes[0]["pending_slug"] == "login-prompt-you-didn-t-start-don-t-approve" assert result["push_status"] == "waiting for approval" def test_production_queue_seeds_many_ranked_suggestions_and_duplicate_warning(tmp_path): _, client = setup_test_db(tmp_path) client.post("/api/production-queue", json={"title": "If This Message Says Your Account Will Be Locked, Stop", "status": "locked_next", "source": "manual"}) seeded = client.post("/api/production-queue/seed-recommendations", json={"minimum": 12}).json() assert len(seeded["items"]) >= 12 queue = client.get("/api/production-queue").json()["items"] suggestions = [item for item in queue if item["source"] == "recommendation" and item["status"] == "suggested"] assert len(suggestions) >= 12 assert all(item["hook"] or item["brief"] for item in suggestions) assert all("duplicate_risk" in item for item in suggestions) account_items = [item for item in suggestions if "account" in item["title"].lower() or "lock" in item["title"].lower()] if account_items: assert account_items[0]["duplicate_risk"] in {"medium", "high"} assert account_items[0]["similar_content"] def test_unlock_replace_and_agent_next_follow_queue_order(tmp_path): _, client = setup_test_db(tmp_path) first = client.post("/api/production-queue", json={"title": "First", "status": "locked_next", "source": "manual", "position": 0}).json() second = client.post("/api/production-queue", json={"title": "Second", "status": "suggested", "source": "recommendation", "position": 2}).json() unlocked = client.post(f"/api/production-queue/{first['id']}/unlock").json() assert unlocked["unlocked"] == [first["id"]] assert client.get("/api/agent/production/next").json()["queue_item_id"] == first["id"] replaced = client.post(f"/api/production-queue/{second['id']}/replace-locked").json() assert replaced["status"] == "locked_next" assert client.get("/api/agent/production/next").json()["queue_item_id"] == second["id"] def test_reset_active_workspace_archives_review_without_deleting_library(tmp_path): engine, client = setup_test_db(tmp_path) with Session(engine) as session: video = VideoAsset(candidate_id="test-old", package_id="package_real_003", source="prepared_package", file_path="", checksum="x", working_title="Old test preview", status="ready_for_review") session.add(video); session.commit(); session.refresh(video) draft = PostDraft(video_asset_id=video.id, provider=Provider.youtube, title="Old", description="desc", privacy_status="private", status="draft") session.add(draft); session.commit(); session.refresh(draft) create_existing_upload_request(session, draft.id) session.add(ProductionQueueItem(title="Old locked", status="locked_next", source="manual")); session.commit() result = client.post("/api/dev/reset-active-workspace", json={"archive_active_queue": True}).json() assert result["active_review_count"] == 0 assert result["videos_archived"] >= 1 assert result["upload_requests_archived"] >= 1 assert "analytics" in result["kept"] def test_existing_jarvis_uploader_is_ready_primary_executor(tmp_path): _, client = setup_test_db(tmp_path) executors = client.get("/api/upload-requests/executors").json()["items"] jarvis = next(e for e in executors if e["type"] == "existing_jarvis_uploader") assert jarvis["name"] == "Existing Jarvis YouTube uploader" assert jarvis["status"] == "ready" assert jarvis["credentials"] == "available" assert jarvis["mode"] == "private_upload"