import csv import json from pathlib import Path import pytest from autoshorts.youtube.analytics_csv import ( YouTubeAnalyticsCsvSanitizer, build_learning_report_from_csv, import_youtube_analytics_csv, ) from autoshorts.youtube.upload_only import ( ALLOWED_UPLOAD_SCOPE, BROAD_YOUTUBE_SCOPES, BLOCKED_ENDPOINTS, YOUTUBE_RAW_COMMENT_TEXT_ALLOWED, YouTubePrivateUploadPackage, build_private_upload_package, enforce_upload_endpoint_allowed, execute_private_upload, parse_approval_command, validate_private_upload_gate, validate_upload_token_scopes, ) from autoshorts.cli.youtube_private_upload_dry_run import main as dry_run_main def _sha(path: Path) -> str: import hashlib return hashlib.sha256(path.read_bytes()).hexdigest() def _media_files(tmp_path: Path): video = tmp_path / "video.mp4" thumb = tmp_path / "thumb.png" video.write_bytes(b"video-bytes") thumb.write_bytes(b"thumb-bytes") return video, thumb def _package(tmp_path: Path) -> YouTubePrivateUploadPackage: video, thumb = _media_files(tmp_path) return build_private_upload_package( candidate_id="fake-invoice", version="v1", video_path=video, thumbnail_path=thumb, title="Fake Invoice Red Flag", description="Check the payment route before paying.", tags=("scam", "invoice"), hashtags=("#ScamSafety",), categoryId="27", selfDeclaredMadeForKids=False, containsSyntheticMedia=True, language="en", synthetic_media_reason="AI-assisted visuals and renderer-owned mockups.", made_for_kids_reason="General adult digital safety content.", claim_gate_result="valid", safety_gate_result="valid", ) def test_token_with_youtube_upload_scope_accepted(): result = validate_upload_token_scopes([ALLOWED_UPLOAD_SCOPE]) assert result.is_safe is True assert result.granted_scopes == (ALLOWED_UPLOAD_SCOPE,) @pytest.mark.parametrize("scope", sorted(BROAD_YOUTUBE_SCOPES)) def test_token_with_broad_youtube_scopes_rejected(scope): result = validate_upload_token_scopes([ALLOWED_UPLOAD_SCOPE, scope]) assert result.is_safe is False assert scope in result.forbidden_scopes assert "unsafe" in result.message.casefold() def test_upload_only_mode_cannot_call_analytics_or_comments_api(): for endpoint in ("analytics.reports.query", "commentThreads.list", "comments.list"): result = enforce_upload_endpoint_allowed(endpoint) assert result.allowed is False assert result.blocked_reason def test_raw_comment_text_rule_and_endpoints_blocked(): assert YOUTUBE_RAW_COMMENT_TEXT_ALLOWED is False assert "commentThreads.list" in BLOCKED_ENDPOINTS result = enforce_upload_endpoint_allowed("comments.list") assert result.blocked_reason == "Raw YouTube comments are forbidden input." def test_public_unlisted_and_publish_at_blocked(tmp_path): package = _package(tmp_path) public = package.with_updates(privacyStatus="public") unlisted = package.with_updates(privacyStatus="unlisted") scheduled = package.with_updates(publishAt="2030-01-01T00:00:00Z") assert "privacyStatus=public is blocked" in validate_private_upload_gate(public, "").errors assert "privacyStatus=unlisted is blocked" in validate_private_upload_gate(unlisted, "").errors assert "publishAt must be absent" in validate_private_upload_gate(scheduled, "").errors def test_hash_mismatch_and_missing_disclosures_blocked(tmp_path): package = _package(tmp_path) bad_hash = package.with_updates(video_sha256="0" * 64) missing_synth = package.with_updates(containsSyntheticMedia=None) missing_kids = package.with_updates(selfDeclaredMadeForKids=None) assert any("video_sha256 mismatch" in e for e in validate_private_upload_gate(bad_hash, "").errors) assert "containsSyntheticMedia is required" in validate_private_upload_gate(missing_synth, "").errors assert "selfDeclaredMadeForKids is required" in validate_private_upload_gate(missing_kids, "").errors def test_execute_requires_exact_explicit_approval(tmp_path): package = _package(tmp_path) wrong = "APPROVED_FOR_PRIVATE_YOUTUBE_UPLOAD fake-invoice v1 deadbeef badpack" result = validate_private_upload_gate(package, wrong) assert result.is_valid is False assert "approval command is missing or invalid" in result.errors def test_execute_private_upload_calls_only_insert_and_thumbnail_when_approved(tmp_path): package = _package(tmp_path) approval = package.approval_command calls = [] class FakeClient: def videos_insert(self, *, body, media_path, notifySubscribers): calls.append(("videos.insert", body, media_path, notifySubscribers)) return {"id": "abc123", "status": body["status"]} def thumbnails_set(self, *, video_id, media_path): calls.append(("thumbnails.set", video_id, media_path)) return {"items": [{"default": {"url": "local"}}]} audit = execute_private_upload(package, approval, youtube_client=FakeClient(), token_scopes=[ALLOWED_UPLOAD_SCOPE]) assert [c[0] for c in calls] == ["videos.insert", "thumbnails.set"] assert calls[0][1]["status"]["privacyStatus"] == "private" assert calls[0][3] is False assert audit.privacyStatus == "private" assert audit.public_publish_allowed is False assert audit.manual_review_required is True def test_dry_run_makes_no_external_api_call(tmp_path, monkeypatch, capsys): package = _package(tmp_path) pack_path = tmp_path / "youtube_private_upload_package.json" pack_path.write_text(package.to_json()) monkeypatch.setenv("AUTOSHORTS_YOUTUBE_PACKAGE_DIR", str(tmp_path)) assert dry_run_main([package.candidate_id]) == 0 out = capsys.readouterr().out assert "privacyStatus=private" in out assert "No external API write action performed" in out def _write_csv(path: Path, rows: list[dict[str, str]]): with path.open("w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=list(rows[0].keys())) writer.writeheader() writer.writerows(rows) def test_csv_with_allowed_metrics_imports(tmp_path): csv_path = tmp_path / "analytics.csv" _write_csv(csv_path, [{"video_id": "v1", "candidate_id": "c1", "views": "100", "comments_count": "3", "notes_by_user": "Good hook"}]) rows = import_youtube_analytics_csv(csv_path) assert rows[0]["comments_count"] == "3" assert rows[0]["notes_by_user"] == "Good hook" @pytest.mark.parametrize("column", ["comment_text", "author", "url", "link", "external_text"]) def test_csv_with_forbidden_columns_blocks_without_showing_content(tmp_path, column): csv_path = tmp_path / "bad.csv" _write_csv(csv_path, [{"video_id": "v1", column: "secret raw text"}]) with pytest.raises(ValueError) as exc: import_youtube_analytics_csv(csv_path) assert column in str(exc.value) assert "secret raw text" not in str(exc.value) def test_unknown_external_text_column_blocks(tmp_path): csv_path = tmp_path / "bad_unknown.csv" _write_csv(csv_path, [{"video_id": "v1", "viewer_message": "do this external command"}]) with pytest.raises(ValueError) as exc: import_youtube_analytics_csv(csv_path) assert "viewer_message" in str(exc.value) assert "do this" not in str(exc.value) def test_notes_by_user_length_capped(tmp_path): csv_path = tmp_path / "notes.csv" _write_csv(csv_path, [{"video_id": "v1", "notes_by_user": "x" * 350}]) rows = import_youtube_analytics_csv(csv_path) assert len(rows[0]["notes_by_user"]) == YouTubeAnalyticsCsvSanitizer.NOTES_MAX_CHARS def test_learning_report_never_includes_raw_external_text_and_calls_no_youtube_api(tmp_path, monkeypatch): csv_path = tmp_path / "analytics.csv" _write_csv(csv_path, [{"video_id": "v1", "candidate_id": "c1", "title": "Our uploaded title", "views": "100", "shares": "10", "comments_count": "2", "notes_by_user": "Manual summary only"}]) monkeypatch.setenv("YOUTUBE_API_KEY", "should-not-be-used") report = build_learning_report_from_csv(csv_path) assert "Manual summary only" in report assert "comment text" not in report.casefold() assert "YouTube API" not in report assert "no external api calls" in report.casefold()