# FILE: _St/Shared/test_shared_storage.py | ROLE: Shared SQLite 저장·조회 회귀 테스트 from __future__ import annotations import json import re import sqlite3 import sys from pathlib import Path import pytest SHARED_DIR = Path(__file__).resolve().parent if str(SHARED_DIR.parent) not in sys.path: sys.path.insert(0, str(SHARED_DIR.parent)) from Shared.storage.sort_storage import ( DEFAULT_STORAGE_CODE, SCHEMA_VERSION, SOURCE_MODES, build_instagram_media_path, find_instagram_media_file, connect_database, canonical_instagram_post_url, create_download_batch, create_download_job, default_database_path, initialize_database, list_database_files, open_database_file, create_database_file, save_database_as, record_media_saved, record_downloaded_file, upsert_observation, list_accounts, list_account_posts, get_post_stats, soft_delete_account, soft_delete_post, list_deleted_accounts, normalize_collection_username, list_collection_accounts, replace_collection_accounts, ) REQUIRED_TABLES = { "schema_migrations", "collector_profile", "device_profile", "storage_roots", "social_accounts", "collection_profiles", "account_collection_rules", "posts", "post_stat_latest", "post_stat_history", "media_assets", "transcripts", "hook_segments", "categories", "post_category_assignments", "tags", "post_tag_assignments", "post_notes", "review_templates", "review_items", "post_reviews", "review_scores", "post_comments", "aggregate_definitions", "aggregate_accounts", "p2p_inventory", "sync_queue", "point_ledger_cache", "download_batches", "download_jobs", "media_saved_events", "post_stat_10m", "deletion_events", "collection_account_queue", } def make_db(tmp_path: Path) -> tuple[Path, Path]: st_root = tmp_path / "bundle" / "_St" st_root.mkdir(parents=True) db_path = st_root / "Shared" / "data" / "sort_local.sqlite3" initialize_database(db_path, st_root) return st_root, db_path def insert_post(db_path: Path, suffix: str = "A") -> tuple[int, int]: conn = connect_database(db_path) try: account_id = conn.execute( "INSERT INTO social_accounts(username) VALUES (?)", (f"account_{suffix}",), ).lastrowid post_id = conn.execute( "INSERT INTO posts(account_id, shortcode, post_url) VALUES (?, ?, ?)", ( account_id, f"SHORT{suffix}", f"https://www.instagram.com/reel/SHORT{suffix}/", ), ).lastrowid conn.commit() return int(account_id), int(post_id) finally: conn.close() def test_sqlite_initialization_and_fixed_stdown(tmp_path: Path) -> None: st_root, db_path = make_db(tmp_path) result = initialize_database(db_path, st_root) expected = tmp_path / "bundle" / "_StDown" / "instagram" assert expected.is_dir() and result["storage_root"] == str(expected.resolve()) assert result["schema_version"] == SCHEMA_VERSION == 7 assert result["journal_mode"] == "wal" and result["foreign_keys"] == 1 conn = connect_database(db_path) try: tables = { row[0] for row in conn.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'" ) } assert REQUIRED_TABLES.issubset(tables) root = conn.execute( "SELECT root_code, root_path, is_default FROM storage_roots" ).fetchone() assert root["root_code"] == DEFAULT_STORAGE_CODE assert root["root_path"] == str(expected.resolve()) and root["is_default"] == 1 versions = { row[0] for row in conn.execute("SELECT version FROM schema_migrations") } assert {1, 2, 3, 4, 5, 6}.issubset(versions) finally: conn.close() def test_migrate_existing_v1_database_to_v7(tmp_path: Path) -> None: st_root = tmp_path / "bundle" / "_St" st_root.mkdir(parents=True) db_path = st_root / "Shared" / "data" / "sort_local.sqlite3" db_path.parent.mkdir(parents=True) conn = sqlite3.connect(db_path) try: schema_v1 = (SHARED_DIR / "storage" / "schema_sqlite_v1.sql").read_text(encoding="utf-8") conn.executescript(schema_v1) conn.commit() assert conn.execute("PRAGMA user_version").fetchone()[0] == 1 finally: conn.close() result = initialize_database(db_path, st_root) assert result["schema_version"] == 7 conn = connect_database(db_path) try: tables = { row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'") } assert {"download_batches", "download_jobs", "media_saved_events", "post_stat_10m", "deletion_events", "collection_account_queue"}.issubset(tables) for table_name in ("post_stat_latest", "post_stat_history", "post_stat_10m"): columns = {row["name"] for row in conn.execute(f"PRAGMA table_info({table_name})")} assert {"sort_by", "sort_value", "source_kind"}.issubset(columns) finally: conn.close() def test_sqlite_idempotent_and_post_unique(tmp_path: Path) -> None: st_root, db_path = make_db(tmp_path) initialize_database(db_path, st_root) conn = connect_database(db_path) try: assert conn.execute("SELECT COUNT(*) FROM storage_roots").fetchone()[0] == 1 conn.execute( "INSERT INTO posts(platform, shortcode, post_url) VALUES('instagram','ABC123','https://www.instagram.com/p/ABC123/')" ) with pytest.raises(sqlite3.IntegrityError): conn.execute( "INSERT INTO posts(platform, shortcode, post_url) VALUES('instagram','ABC123','https://www.instagram.com/reel/ABC123/')" ) finally: conn.close() @pytest.mark.parametrize("source_mode", sorted(SOURCE_MODES)) def test_unified_media_saved_flow_for_every_source_mode( tmp_path: Path, source_mode: str ) -> None: _, db_path = make_db(tmp_path) account_id, post_id = insert_post(db_path, source_mode[:3]) batch_id = create_download_batch( db_path, source_mode=source_mode, account_id=account_id, sort_by="views" if source_mode == "TOP_SORTED" else "", filter_days=7 if source_mode in {"TOP_SORTED", "BULK_FILTERED"} else None, requested_count=1, ) job_id = create_download_job( db_path, batch_id=batch_id, post_id=post_id, source_mode=source_mode, asset_kind="VIDEO", rank_no=1 if source_mode == "TOP_SORTED" else None, source_url="https://cdn.example/video.mp4", expected_file_name="video.mp4", ) event_uuid = f"event-{source_mode}" result = record_media_saved( db_path, post_id=post_id, source_mode=source_mode, asset_kind="VIDEO", relative_path=f"account_{source_mode[:3]}/video.mp4", file_name="video.mp4", file_size=12345, sha256="a" * 64, source_url="https://cdn.example/video.mp4", download_job_id=job_id, event_uuid=event_uuid, rank_no=1 if source_mode == "TOP_SORTED" else None, ) duplicate = record_media_saved( db_path, post_id=post_id, source_mode=source_mode, asset_kind="VIDEO", relative_path=f"account_{source_mode[:3]}/video.mp4", file_name="video.mp4", file_size=12345, sha256="a" * 64, event_uuid=event_uuid, ) assert result["idempotent"] is False assert duplicate["idempotent"] is True and duplicate["event_id"] == result["event_id"] conn = connect_database(db_path) try: event = conn.execute( "SELECT * FROM media_saved_events WHERE id = ?", (result["event_id"],) ).fetchone() asset = conn.execute( "SELECT * FROM media_assets WHERE id = ?", (result["asset_id"],) ).fetchone() job = conn.execute("SELECT * FROM download_jobs WHERE id = ?", (job_id,)).fetchone() post = conn.execute("SELECT media_state FROM posts WHERE id = ?", (post_id,)).fetchone() sync = conn.execute( "SELECT * FROM sync_queue WHERE entity_type = 'MEDIA_SAVED'" ).fetchall() assert event["source_mode"] == source_mode assert event["folder_name"] == f"account_{source_mode[:3]}" assert asset["local_status"] == "AVAILABLE" and asset["sha256"] == "a" * 64 assert asset["folder_name"] == f"account_{source_mode[:3]}" assert asset["file_name"] == "video.mp4" batch = conn.execute("SELECT * FROM download_batches WHERE id = ?", (batch_id,)).fetchone() assert job["job_status"] == "COMPLETE" assert batch["batch_status"] == "COMPLETE" assert asset["storage_root_id"] is not None assert post["media_state"] == "MEDIA_AVAILABLE" assert len(sync) == 1 sync_payload = json.loads(sync[0]["payload_json"]) assert sync_payload["source_mode"] == source_mode assert "relative_path" not in sync_payload and "local_path" not in sync_payload assert "folder_name" not in sync_payload finally: conn.close() def test_media_saved_rejects_path_escape(tmp_path: Path) -> None: _, db_path = make_db(tmp_path) _, post_id = insert_post(db_path, "PATH") with pytest.raises(ValueError, match="storage root"): record_media_saved( db_path, post_id=post_id, source_mode="MANUAL_SINGLE", asset_kind="IMAGE", relative_path="../outside.jpg", file_name="outside.jpg", file_size=1, ) def test_mysql_additive_schema() -> None: v45 = ( SHARED_DIR / "storage" / "mysql" / "migration_v45_from_yellow_v44.sql" ).read_text(encoding="utf-8") v46 = ( SHARED_DIR / "storage" / "mysql" / "migration_v46_media_saved_flow.sql" ).read_text(encoding="utf-8") combined = v45 + "\n" + v46 upper = combined.upper() assert "DROP TABLE" not in upper and "TRUNCATE TABLE" not in upper assert "ALTER TABLE `IG_MEDIA`" not in upper assert "ALTER TABLE `IG_MEDIA_USER_DOWNLOAD`" not in upper created = { match.group(1) for match in re.finditer( r"CREATE\s+TABLE\s+IF\s+NOT\s+EXISTS\s+`([^`]+)`", combined, re.I ) } assert { "collector_members", "collector_devices", "collector_groups", "collector_posts", "collector_post_stat_latest", "collector_asset_inventory", "collector_transcripts", "collector_hook_segments", "collector_categories", "collector_reviews", "collector_comments", "collector_aggregates", "collector_tasks", "collector_sync_events", "collector_point_ledger", "collector_download_batches", "collector_download_jobs", "collector_media_saved_events", }.issubset(created) assert "`collector_code` varchar(8) NOT NULL" in v45 assert "`token_hash` char(64)" in v45 assert "local_path" not in v46.lower() assert "relative_path" not in v46.lower() assert "source_mode" in v46 and "client_event_id" in v46 def test_record_downloaded_file_creates_context_for_actual_download_root(tmp_path: Path) -> None: st_root = tmp_path / "_St" st_root.mkdir() media_dir = tmp_path / "BrowserDownloads" / "campaign" / "account_a" media_dir.mkdir(parents=True) media_file = media_dir / "account_a_upload_unknown_No없음_CODE123.mp4" media_file.write_bytes(b"video-data") result = record_downloaded_file( st_root=st_root, file_path=media_file, source_mode="BULK_ALL", shortcode="CODE123", owner_username="account_a", asset_kind="VIDEO", source_url="https://cdn.example/video.mp4", post_url="https://www.instagram.com/reel/CODE123/", md5="m" * 32, sha256="s" * 64, file_size=media_file.stat().st_size, account_path="account_a", batch_key="batch-all-1", requested_count=1, event_uuid="event-all-1", stats={"views": 100, "likes": 10, "comments": 2, "reposts": 1, "captured_at": "2026-07-19T00:00:00"}, ) db_path = default_database_path(st_root) with connect_database(db_path) as conn: event = conn.execute("SELECT * FROM media_saved_events WHERE id = ?", (result["event_id"],)).fetchone() post = conn.execute("SELECT * FROM posts WHERE id = ?", (result["post_id"],)).fetchone() batch = conn.execute("SELECT * FROM download_batches WHERE id = ?", (result["batch_id"],)).fetchone() job = conn.execute("SELECT * FROM download_jobs WHERE id = ?", (result["job_id"],)).fetchone() stat = conn.execute("SELECT * FROM post_stat_latest WHERE post_id = ?", (result["post_id"],)).fetchone() root = conn.execute("SELECT * FROM storage_roots WHERE id = ?", (result["storage_root_id"],)).fetchone() assert event["relative_path"] == "account_a/" + media_file.name assert Path(root["root_path"]) == media_dir.parent assert post["media_state"] == "MEDIA_AVAILABLE" assert batch["source_mode"] == "BULK_ALL" assert batch["batch_status"] == "COMPLETE" assert job["job_status"] == "COMPLETE" assert stat["views"] == 100 payload = json.loads(conn.execute("SELECT payload_json FROM sync_queue WHERE entity_type = 'MEDIA_SAVED'").fetchone()[0]) assert "relative_path" not in payload assert "storage_root_path" not in payload def test_record_downloaded_file_is_idempotent_across_ws_retry(tmp_path: Path) -> None: st_root = tmp_path / "_St" st_root.mkdir() media_dir = tmp_path / "downloads" / "top" / "account_b" media_dir.mkdir(parents=True) media_file = media_dir / "account_b_upload_unknown_No001_CODE999.jpg" media_file.write_bytes(b"image-data") kwargs = dict( st_root=st_root, file_path=media_file, source_mode="TOP_SORTED", shortcode="CODE999", owner_username="account_b", asset_kind="IMAGE", rank_no=1, account_path="account_b", batch_key="top-batch-1", event_uuid="top-event-1", file_size=media_file.stat().st_size, ) first = record_downloaded_file(**kwargs) second = record_downloaded_file(**kwargs) assert first["event_id"] == second["event_id"] assert second["idempotent"] is True with connect_database(default_database_path(st_root)) as conn: assert conn.execute("SELECT COUNT(*) FROM media_saved_events").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM sync_queue WHERE entity_type = 'MEDIA_SAVED'").fetchone()[0] == 1 def test_download_batch_completes_only_after_requested_files(tmp_path: Path) -> None: st_root = tmp_path / "_St" st_root.mkdir() media_dir = tmp_path / "downloads" / "bulk" / "account_c" media_dir.mkdir(parents=True) first_file = media_dir / "account_c_upload_unknown_No없음_AAA.jpg" second_file = media_dir / "account_c_upload_unknown_No없음_BBB.jpg" first_file.write_bytes(b"one") second_file.write_bytes(b"two") first = record_downloaded_file( st_root=st_root, file_path=first_file, source_mode="BULK_FILTERED", shortcode="AAA", owner_username="account_c", asset_kind="IMAGE", account_path="account_c", batch_key="filtered-batch-two", requested_count=2, event_uuid="filtered-event-one", file_size=3, ) db_path = default_database_path(st_root) with connect_database(db_path) as conn: status = conn.execute("SELECT batch_status FROM download_batches WHERE id = ?", (first["batch_id"],)).fetchone()[0] assert status == "RUNNING" second = record_downloaded_file( st_root=st_root, file_path=second_file, source_mode="BULK_FILTERED", shortcode="BBB", owner_username="account_c", asset_kind="IMAGE", account_path="account_c", batch_key="filtered-batch-two", requested_count=2, event_uuid="filtered-event-two", file_size=3, ) assert second["batch_id"] == first["batch_id"] with connect_database(db_path) as conn: status = conn.execute("SELECT batch_status FROM download_batches WHERE id = ?", (first["batch_id"],)).fetchone()[0] assert status == "COMPLETE" def test_observation_10m_repository_and_soft_delete(tmp_path: Path) -> None: _, db_path = make_db(tmp_path) first = upsert_observation( db_path, username="account_a", shortcode="POST1", views=100, likes=10, comments=2, observed_at="2026-07-19T14:02:37Z", ) same_bucket = upsert_observation( db_path, username="account_a", shortcode="POST1", views=130, likes=12, comments=3, observed_at="2026-07-19T14:09:59Z", ) next_bucket = upsert_observation( db_path, username="account_a", shortcode="POST1", views=190, likes=18, comments=5, observed_at="2026-07-19T14:12:01Z", ) assert first["bucket_at"] == same_bucket["bucket_at"] == "2026-07-19T14:00:00Z" assert next_bucket["bucket_at"] == "2026-07-19T14:10:00Z" stats = get_post_stats(db_path, first["post_id"]) assert len(stats) == 2 assert stats[0]["views"] == 190 and stats[1]["views"] == 130 assert stats[1]["sample_count"] == 2 accounts = list_accounts(db_path) assert accounts["total"] == 1 and accounts["items"][0]["username"] == "account_a" posts = list_account_posts(db_path, account_id=first["account_id"], sort_by="views_delta") assert posts["total"] == 1 assert posts["items"][0]["views_delta"] == 60 assert posts["items"][0]["views_per_hour"] == 360 deleted_post = soft_delete_post(db_path, first["post_id"]) assert deleted_post["deleted"] is True assert list_account_posts(db_path, account_id=first["account_id"])["total"] == 0 upsert_observation( db_path, username="account_b", shortcode="POST2", views=20, observed_at="2026-07-19T15:01:00Z", ) account_b = next(item for item in list_accounts(db_path)["items"] if item["username"] == "account_b") deleted_account = soft_delete_account(db_path, account_b["id"]) assert deleted_account["username"] == "account_b" ignored = upsert_observation( db_path, username="account_b", shortcode="POST3", views=999, observed_at="2026-07-19T15:12:00Z", ) assert ignored["ignored_deleted"] is True and ignored["deleted_entity"] == "ACCOUNT" assert all(item["username"] != "account_b" for item in list_accounts(db_path)["items"]) assert any(item["username"] == "account_b" for item in list_deleted_accounts(db_path)) conn = connect_database(db_path) try: rule = conn.execute("SELECT active, deleted_at FROM social_accounts WHERE id=?", (account_b["id"],)).fetchone() assert rule["active"] == 0 and rule["deleted_at"] sync_ops = conn.execute("SELECT entity_type, operation FROM sync_queue WHERE operation='DELETE'").fetchall() assert {(row["entity_type"], row["operation"]) for row in sync_ops} >= {("POST", "DELETE"), ("ACCOUNT", "DELETE")} finally: conn.close() def test_observation_skips_unchanged_metrics_across_new_bucket(tmp_path: Path) -> None: _, db_path = make_db(tmp_path) first = upsert_observation( db_path, username="unchanged_account", shortcode="UNCHANGED1", views=500, views_available=1, likes=50, comments=5, reposts=2, sort_by="views", sort_value=500, source_kind="LOCAL", observed_at="2026-07-25T12:02:00Z", ) duplicate = upsert_observation( db_path, username="unchanged_account", shortcode="UNCHANGED1", views=500, views_available=1, likes=50, comments=5, reposts=2, sort_by="likes", sort_value=50, source_kind="LOCAL", observed_at="2026-07-25T12:12:00Z", ) assert first["saved"] is True assert duplicate["saved"] is False and duplicate["skipped_unchanged"] is True assert duplicate["skipped_all_writes"] is True with connect_database(db_path) as conn: history_count = conn.execute("SELECT COUNT(*) FROM post_stat_history WHERE post_id=?", (first["post_id"],)).fetchone()[0] bucket_count = conn.execute("SELECT COUNT(*) FROM post_stat_10m WHERE post_id=?", (first["post_id"],)).fetchone()[0] latest = conn.execute("SELECT views, views_available, likes, comments, reposts FROM post_stat_latest WHERE post_id=?", (first["post_id"],)).fetchone() post_observed = conn.execute("SELECT last_observed_at FROM posts WHERE id=?", (first["post_id"],)).fetchone()[0] assert history_count == 1 assert bucket_count == 1 assert tuple(latest) == (500, 1, 50, 5, 2) assert post_observed == "2026-07-25T12:02:00Z" missing_views = upsert_observation( db_path, username="unchanged_account", shortcode="UNCHANGED1", views=0, views_available=0, likes=51, comments=5, reposts=2, sort_by="likes", sort_value=51, source_kind="LOCAL", observed_at="2026-07-25T12:22:00Z", ) assert missing_views["saved"] is True with connect_database(db_path) as conn: latest = conn.execute("SELECT views, views_available, likes FROM post_stat_latest WHERE post_id=?", (first["post_id"],)).fetchone() assert tuple(latest) == (500, 1, 51) def test_observation_v7_preserves_sort_mode_source_and_views_availability(tmp_path: Path) -> None: _, db_path = make_db(tmp_path) first = upsert_observation( db_path, username="stats_account", shortcode="STATS1", views=100, views_available=1, likes=10, comments=2, reposts=1, sort_by="likes", sort_value=10, source_kind="LOCAL", observed_at="2026-07-25T11:02:00Z", ) second = upsert_observation( db_path, username="stats_account", shortcode="STATS1", views=150, views_available=1, likes=18, comments=4, reposts=2, sort_by="likes", sort_value=18, source_kind="SERVER", observed_at="2026-07-25T11:08:00Z", ) assert first["bucket_at"] == second["bucket_at"] == "2026-07-25T11:00:00Z" with connect_database(db_path) as conn: latest = conn.execute("SELECT * FROM post_stat_latest WHERE post_id=?", (first["post_id"],)).fetchone() tenm = conn.execute("SELECT * FROM post_stat_10m WHERE post_id=?", (first["post_id"],)).fetchone() assert latest["views_available"] == 1 assert latest["sort_by"] == "likes" and latest["sort_value"] == 18 and latest["source_kind"] == "SERVER" assert tenm["sample_count"] == 2 assert tenm["views_available"] == 1 assert tenm["sort_by"] == "likes" and tenm["sort_value"] == 18 and tenm["source_kind"] == "SERVER" def test_canonical_instagram_post_url_rejects_scope_and_profile_urls() -> None: code = "Da0WTq-OosH" assert canonical_instagram_post_url(code, "https://www.instagram.com/reel/Da0WTq-OosH/?utm_source=x") == "https://www.instagram.com/reel/Da0WTq-OosH/" assert canonical_instagram_post_url(code, "https://www.instagram.com/reels/Da0WTq-OosH/") == "https://www.instagram.com/reel/Da0WTq-OosH/" assert canonical_instagram_post_url(code, "https://www.instagram.com/explore/") == "https://www.instagram.com/p/Da0WTq-OosH/" assert canonical_instagram_post_url(code, "https://www.instagram.com/account_name/") == "https://www.instagram.com/p/Da0WTq-OosH/" assert canonical_instagram_post_url(code, "https://cdn.example/video.mp4") == "https://www.instagram.com/p/Da0WTq-OosH/" assert canonical_instagram_post_url(code, "https://www.instagram.com/explore/", fallback_kind="reel") == "https://www.instagram.com/reel/Da0WTq-OosH/" def test_download_and_analyze_reads_never_expose_scope_url_as_post_url(tmp_path: Path) -> None: st_root = tmp_path / "_St" st_root.mkdir() media_dir = tmp_path / "downloads" / "account_scope" media_dir.mkdir(parents=True) media_file = media_dir / "account_scope_upload_unknown_No없음_Da0WTq-OosH.jpg" media_file.write_bytes(b"image") result = record_downloaded_file( st_root=st_root, file_path=media_file, source_mode="MANUAL_SINGLE", shortcode="Da0WTq-OosH", owner_username="account_scope", asset_kind="IMAGE", post_url="https://www.instagram.com/explore/", account_path="account_scope", event_uuid="scope-url-event", file_size=media_file.stat().st_size, ) db_path = default_database_path(st_root) with connect_database(db_path) as conn: stored = conn.execute("SELECT post_url FROM posts WHERE id=?", (result["post_id"],)).fetchone()[0] account_id = conn.execute("SELECT account_id FROM posts WHERE id=?", (result["post_id"],)).fetchone()[0] assert stored == "https://www.instagram.com/p/Da0WTq-OosH/" conn.execute("UPDATE posts SET post_url='https://www.instagram.com/explore/' WHERE id=?", (result["post_id"],)) conn.commit() listed = list_account_posts(db_path, account_id=account_id)["items"][0] assert listed["post_url"] == "https://www.instagram.com/p/Da0WTq-OosH/" def test_collection_account_queue_username_only_and_scope_rejection(tmp_path: Path) -> None: _, db_path = make_db(tmp_path) assert normalize_collection_username("@Double_King0.1") == "double_king0.1" assert normalize_collection_username("https://www.instagram.com/Alpha.User/") == "alpha.user" assert normalize_collection_username("https://www.instagram.com/Alpha.User/reels/") == "alpha.user" for bad in ( "https://www.instagram.com/explore/", "https://www.instagram.com/reels/", "https://www.instagram.com/reel/CODE1/", "https://www.instagram.com/p/CODE1/", ): assert normalize_collection_username(bad) == "" saved = replace_collection_accounts( db_path, [ {"username": "Double_King0.1", "position": 99}, {"account_url": "https://www.instagram.com/alpha.user/"}, {"username": "explore"}, {"username": "double_king0.1"}, ], ) assert saved["total"] == 2 items = list_collection_accounts(db_path) assert [item["username"] for item in items] == ["double_king0.1", "alpha.user"] assert [item["position"] for item in items] == [1, 2] assert items[0]["account_url"] == "https://www.instagram.com/double_king0.1/" with connect_database(db_path) as conn: sync = conn.execute( "SELECT operation, payload_json FROM sync_queue WHERE entity_type='COLLECTION_ACCOUNT_QUEUE' ORDER BY id DESC LIMIT 1" ).fetchone() assert sync["operation"] == "REPLACE" payload = json.loads(sync["payload_json"]) assert [item["username"] for item in payload["items"]] == ["double_king0.1", "alpha.user"] def test_duplicate_deleted_collection_entry_always_wins(tmp_path: Path) -> None: _, db_path = make_db(tmp_path) result = replace_collection_accounts( db_path, [ {"username": "@beta", "enabled": True}, {"account_url": "https://www.instagram.com/alpha/"}, {"username": "beta", "is_deleted": True}, {"username": "ALPHA"}, ], ) assert [item["username"] for item in result["items"]] == ["beta", "alpha"] beta = next(item for item in result["items"] if item["username"] == "beta") assert beta["is_deleted"] is True and beta["enabled"] is False def test_deleted_account_stays_deleted_in_collection_queue(tmp_path: Path) -> None: _, db_path = make_db(tmp_path) result = upsert_observation( db_path, username="alpha", shortcode="DELETE1", views=1, observed_at="2026-07-19T09:00:00Z", ) replace_collection_accounts(db_path, [{"username": "alpha"}, {"username": "beta"}]) soft_delete_account(db_path, result["account_id"]) items = {item["username"]: item for item in list_collection_accounts(db_path)} assert items["alpha"]["is_deleted"] is True and items["alpha"]["enabled"] is False replaced = replace_collection_accounts( db_path, [ {"account_url": "https://www.instagram.com/alpha/", "enabled": True}, {"username": "@beta"}, {"username": "alpha"}, ], ) assert [item["username"] for item in replaced["items"]] == ["alpha", "beta"] alpha = next(item for item in replaced["items"] if item["username"] == "alpha") assert alpha["is_deleted"] is True and alpha["enabled"] is False def test_stdown_original_filename_folder_is_stored_and_listed(tmp_path: Path) -> None: st_root = tmp_path / "bundle" / "_St" st_root.mkdir(parents=True) media_dir = tmp_path / "bundle" / "_StDown" / "instagram" / "account_original" media_dir.mkdir(parents=True) original_name = "17912345678901234_n.mp4" media_file = media_dir / original_name media_file.write_bytes(b"original-media") result = record_downloaded_file( st_root=st_root, file_path=media_file, source_mode="MANUAL_SINGLE", shortcode="ORIGINAL1", owner_username="account_original", asset_kind="VIDEO", account_path="account_original", event_uuid="original-file-event", file_size=media_file.stat().st_size, ) assert result["relative_path"] == f"account_original/{original_name}" assert result["folder_name"] == "account_original" assert result["file_name"] == original_name listed = list_account_posts( default_database_path(st_root), account_id=result["account_id"] )["items"][0] assert listed["folder_name"] == "account_original" assert listed["file_name"] == original_name assert listed["relative_path"] == f"account_original/{original_name}" def test_mysql_v47_hourly_statistics_migration_is_non_destructive() -> None: sql = (SHARED_DIR / "storage" / "mysql" / "migration_v47_media_stat_hour.sql").read_text(encoding="utf-8") upper = sql.upper() for forbidden in ("DROP TABLE", "TRUNCATE TABLE", "DELETE FROM"): assert forbidden not in upper for token in ( "ADD COLUMN `views_available`", "ADD COLUMN `stat_hour_bucket_utc`", "CREATE TABLE IF NOT EXISTS `ig_media_stat_hour`", "UNIQUE KEY `uq_collector_shortcode_hour`", "`hour_bucket_utc` datetime NOT NULL", "`sample_count` int(10) unsigned NOT NULL DEFAULT 1", ): assert token in sql, token def test_sqlite_v7_preserves_views_availability(tmp_path: Path) -> None: st_root, db_path = make_db(tmp_path) result = upsert_observation( db_path, username="private_views", shortcode="PV1", views=0, views_available=0, likes=120, comments=30, reposts=7, observed_at="2026-07-26T01:07:00Z", ) upsert_observation( db_path, username="public_views", shortcode="PUB1", views=4321, views_available=1, likes=120, comments=30, reposts=7, observed_at="2026-07-26T01:08:00Z", ) with connect_database(db_path) as conn: assert int(conn.execute("PRAGMA user_version").fetchone()[0]) == 7 for table in ("post_stat_latest", "post_stat_history", "post_stat_10m"): assert "views_available" in {row["name"] for row in conn.execute(f"PRAGMA table_info({table})")} private = conn.execute("SELECT views, views_available FROM post_stat_latest WHERE post_id=?", (result["post_id"],)).fetchone() public = conn.execute("""SELECT s.views, s.views_available FROM post_stat_latest s JOIN posts p ON p.id=s.post_id WHERE p.shortcode='PUB1'""").fetchone() assert tuple(private) == (0, 0) assert tuple(public) == (4321, 1) listed = list_account_posts(db_path, account_id=result["account_id"])["items"][0] assert listed["views_available"] == 0 assert get_post_stats(db_path, result["post_id"])[0]["views_available"] == 0 def test_database_filename_selection_new_open_and_save_as(tmp_path: Path) -> None: st_root = tmp_path / "bundle" / "_St" st_root.mkdir(parents=True) default_path = default_database_path(st_root) initialize_database(default_path, st_root) replace_collection_accounts(default_path, [{"username": "alpha"}, {"username": "beta"}]) copied = save_database_as(st_root, default_path, "archive") assert copied["filename"] == "archive.sqlite3" assert default_database_path(st_root).name == "archive.sqlite3" assert [item["username"] for item in list_collection_accounts(default_database_path(st_root))] == ["alpha", "beta"] created = create_database_file(st_root, "fresh.sqlite3") assert created["filename"] == "fresh.sqlite3" assert default_database_path(st_root).name == "fresh.sqlite3" assert list_collection_accounts(default_database_path(st_root)) == [] opened = open_database_file(st_root, "archive.sqlite3") assert opened["filename"] == "archive.sqlite3" assert [item["username"] for item in list_collection_accounts(default_database_path(st_root))] == ["alpha", "beta"] info = list_database_files(st_root) assert info["active_filename"] == "archive.sqlite3" assert {"sort_local.sqlite3", "archive.sqlite3", "fresh.sqlite3"}.issubset(set(info["files"])) with pytest.raises(ValueError): create_database_file(st_root, "archive.sqlite3") with pytest.raises(ValueError): open_database_file(st_root, "../outside.sqlite3") with pytest.raises(ValueError): create_database_file(st_root, "../new-outside.sqlite3") def test_shortcode_media_path_contract_uses_only_shortcode_actual_extension(tmp_path: Path) -> None: st_root, _ = make_db(tmp_path) target = build_instagram_media_path(st_root, "ameliauusa", "ABC123", ".webp") assert target == tmp_path / "bundle" / "_StDown" / "instagram" / "ameliauusa" / "ABC123.webp" target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(b"webp") assert find_instagram_media_file(st_root, "ameliauusa", "ABC123") == target target.unlink() legacy = target.with_name("ABC123_thumbnail.jpg") legacy.write_bytes(b"jpg") assert find_instagram_media_file(st_root, "ameliauusa", "ABC123") is None def test_cache_and_auto_seen_source_modes_are_supported() -> None: assert {"CACHE8701_CDP", "AUTO_SEEN_JPG"}.issubset(SOURCE_MODES)