import argparse import csv import importlib.util import json from pathlib import Path import pytest from analytics import AnalyticsRepository PROJECT_ROOT = Path(__file__).resolve().parents[1] MODULE_PATH = PROJECT_ROOT / "scripts" / "manage_app_catalog.py" MODULE_SPEC = importlib.util.spec_from_file_location("manage_app_catalog", MODULE_PATH) manage_app_catalog = importlib.util.module_from_spec(MODULE_SPEC) assert MODULE_SPEC.loader is not None MODULE_SPEC.loader.exec_module(manage_app_catalog) @pytest.fixture def db_path(tmp_path) -> str: path = tmp_path / "monitoring.db" AnalyticsRepository(db_path=str(path)) return str(path) def _normalize_tags(value): if value is None: return [] if isinstance(value, str): text = value.strip() if not text: return [] try: parsed = json.loads(text) if isinstance(parsed, list): return [str(item).strip() for item in parsed if str(item).strip()] except (TypeError, ValueError): pass return [text] if isinstance(value, (list, tuple, set)): return [str(item).strip() for item in value if str(item).strip()] return [str(value).strip()] if str(value).strip() else [] def _failure_parts(failure_type: str): category, _, code_text = str(failure_type or "").partition("/") try: code = int(code_text) except (TypeError, ValueError): code = None return category or None, code def seed_catalog_rows(db_path: str, rows) -> None: repo = AnalyticsRepository(db_path=db_path) with repo._write_lock, repo._connect() as connection: for index, row in enumerate(rows): package_name = row["package_name"] app_name = row.get("app_name", package_name) tags = _normalize_tags( row.get( "batch_tags", row.get("incremental_batch_tags", row.get("incremental_batch_tag", "")), ) ) last_updated = row.get("last_updated", "") country_code = row.get("country_code", "US") device_type = row.get("device_type", "emulator") app_magic_label = row.get("app_magic_label", "") payload = row.get("task_payload") or { "task_key": f"{app_name}_{package_name}", "app_name": app_name, "package_name": package_name, "app_magic_label": app_magic_label, "last_updated": last_updated, "country_code": country_code, "country_codes": [country_code] if country_code else [], "device_type": device_type, "available_sources": ["google_play", "local"], "original_row": { "app_name": app_name, "package_name": package_name, "app_magic_label": app_magic_label, "last_updated": last_updated, "country_code": country_code, "device_type": device_type, }, } created_at = f"2026-01-01 00:00:{index:02d}" connection.execute( """ INSERT INTO app_catalog ( package_name, app_name, batch_tags, app_magic_label, last_updated, country_code, device_type, task_payload_json, last_update_interval_days, source_order, is_active, downloads, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( package_name, app_name, json.dumps(tags, ensure_ascii=False), app_magic_label, last_updated, country_code, device_type, json.dumps(payload, ensure_ascii=False), int(row.get("last_update_interval_days", 0) or 0), row.get("source_order"), int(row.get("catalog_active", row.get("is_active", 1))), row.get("downloads", 0), created_at, created_at, ), ) if row.get("create_task", True) is False: continue failure_type = row.get("latest_failure_type", "") collection_status = row.get("collection_status") latest_status = row.get("latest_status", "") if collection_status is None: if latest_status == "success": collection_status = "qualified" elif failure_type: collection_status = "failed_terminal" else: collection_status = "pending" task_status = "pending" execution_status = None completed_at = None if latest_status or failure_type or collection_status in {"qualified", "failed_terminal"}: task_status = "completed" execution_status = latest_status or ("success" if collection_status == "qualified" else "failed") completed_at = created_at if execution_status == "success": total_traffic_bytes = int(row.get("total_traffic_bytes", 100) or 0) self_ratio = float(row.get("self_ratio", 10.0) or 0.0) self_traffic_bytes = int(row.get("self_traffic_bytes", total_traffic_bytes * self_ratio / 100) or 0) num_nodes = int(row.get("num_nodes", 12) or 0) else: total_traffic_bytes = int(row.get("total_traffic_bytes", 0) or 0) self_traffic_bytes = int(row.get("self_traffic_bytes", 0) or 0) num_nodes = int(row.get("num_nodes", 0) or 0) error_category, error_code = _failure_parts(failure_type) task_type = row.get("collection_task_type", "new_app") connection.execute( """ INSERT INTO collection_task ( package_name, batch_tag, run_kind, attempt, task_key, app_name, app_magic_label, is_new_app, task_status, execution_status, worker_id, created_at, completed_at, error_category, error_code, error_reason, error_details, num_nodes, total_traffic_bytes, self_traffic_bytes, server_traffic_bytes, unrecognized_traffic_bytes ) VALUES (?, ?, 'ranking', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( package_name, tags[-1] if tags else "seed", int(row.get("attempt", 1) or 1), f"{app_name}_{package_name}", app_name, app_magic_label, 1 if task_type == "new_app" else 0, task_status, execution_status, row.get("worker_id", "seed-worker"), created_at, completed_at, error_category, error_code, row.get("collection_status_reason", failure_type or collection_status or "seed"), row.get("latest_task_detail", failure_type or ""), num_nodes, total_traffic_bytes, self_traffic_bytes, int(row.get("server_traffic_bytes", 0) or 0), int(row.get("unrecognized_traffic_bytes", 0) or 0), ), ) def fetch_summary_row(db_path: str, package_name: str) -> dict: repo = AnalyticsRepository(db_path=db_path) return repo.get_collection_row(package_name) or {} def fetch_catalog_row(db_path: str, package_name: str) -> dict: repo = AnalyticsRepository(db_path=db_path) with repo._connect() as connection: row = connection.execute( "SELECT * FROM app_catalog WHERE package_name = ?", (package_name,), ).fetchone() return dict(row) if row else {} def fetch_collection_tasks(db_path: str, package_name: str): repo = AnalyticsRepository(db_path=db_path) with repo._connect() as connection: rows = connection.execute( """ SELECT * FROM collection_task WHERE package_name = ? ORDER BY attempt ASC """, (package_name,), ).fetchall() return [dict(row) for row in rows] def count_pending_task_rows(db_path: str, package_name: str) -> int: return sum(1 for row in fetch_collection_tasks(db_path, package_name) if row["task_status"] == "pending") def write_catalog_csv(path: Path, rows) -> None: normalized_rows = [] base_fields = ["app_name", "package_name", "last_updated", "country_code", "device_type"] fieldnames = list(base_fields) for item in rows: if isinstance(item, dict): row = dict(item) row.setdefault("app_name", row["package_name"]) row.setdefault("last_updated", "") row.setdefault("country_code", "US") row.setdefault("device_type", "emulator") else: row = { "app_name": item, "package_name": item, "last_updated": "", "country_code": "US", "device_type": "emulator", } normalized_rows.append(row) for key in row: if key not in fieldnames: fieldnames.append(key) with path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames) writer.writeheader() writer.writerows(normalized_rows) def test_sync_from_csv_persists_csv_payload_columns_and_creates_pending_task(db_path, tmp_path, capsys): csv_path = tmp_path / "catalog.csv" write_catalog_csv( csv_path, [ { "app_name": "New App", "package_name": "com.example.new", "last_updated": "2026-04-15", "country_code": "JP", "device_type": "physical", "app_magic_label": "finance", "downloads": "1M", "available_sources": "google_play,local", "custom_column": "kept", } ], ) manage_app_catalog.sync_from_csv( argparse.Namespace( db_path=db_path, csv_path=str(csv_path), incremental_batch_tag="", dry_run=False, ) ) output = capsys.readouterr().out assert "added=1" in output assert "history_new=1" in output assert "pending_tasks_created=1" in output assert "payload_storage_issues=0" in output catalog_row = fetch_catalog_row(db_path, "com.example.new") assert catalog_row["last_updated"] == "2026/04/15" assert catalog_row["country_code"] == "JP" assert catalog_row["device_type"] == "physical" payload = json.loads(catalog_row["task_payload_json"]) assert payload["last_updated"] == "2026/04/15" assert payload["country_code"] == "JP" assert payload["device_type"] == "physical" assert payload["available_sources"] == ["google_play", "local"] assert payload["original_row"]["custom_column"] == "kept" assert payload["original_row"]["last_updated"] == "2026/04/15" tasks = fetch_collection_tasks(db_path, "com.example.new") assert len(tasks) == 1 assert tasks[0]["task_status"] == "pending" assert tasks[0]["is_new_app"] == 1 def test_sync_from_csv_counts_added_against_full_history(db_path, tmp_path, capsys): seed_catalog_rows( db_path, [ { "package_name": "com.example.active", "source_order": 0, "catalog_active": 1, "collection_status": "qualified", "latest_status": "success", }, { "package_name": "com.example.reactivated", "source_order": 1, "catalog_active": 0, "collection_status": "qualified", "latest_status": "success", }, ], ) csv_path = tmp_path / "catalog.csv" write_catalog_csv( csv_path, ["com.example.active", "com.example.reactivated", "com.example.new"], ) manage_app_catalog.sync_from_csv( argparse.Namespace( db_path=db_path, csv_path=str(csv_path), incremental_batch_tag="batch-new", dry_run=False, ) ) output = capsys.readouterr().out assert "added=1" in output assert "history_new=1" in output assert "reactivated=1" in output assert "pending_tasks_created=1" in output assert "payload_storage_issues=0" in output assert "incremental_marked=3" in output assert "batch-new" in fetch_summary_row(db_path, "com.example.new")["incremental_batch_tags"] assert "batch-new" in fetch_summary_row(db_path, "com.example.reactivated")["incremental_batch_tags"] assert fetch_summary_row(db_path, "com.example.reactivated")["catalog_active"] is True def test_sync_from_csv_dry_run_counts_added_against_full_history(db_path, tmp_path, capsys): seed_catalog_rows( db_path, [ { "package_name": "com.example.reactivated", "source_order": 1, "catalog_active": 0, "collection_status": "qualified", "latest_status": "success", }, ], ) csv_path = tmp_path / "catalog.csv" write_catalog_csv(csv_path, ["com.example.reactivated", "com.example.new"]) manage_app_catalog.sync_from_csv( argparse.Namespace( db_path=db_path, csv_path=str(csv_path), incremental_batch_tag="batch-new", dry_run=True, ) ) output = capsys.readouterr().out assert "added=1" in output assert "history_new=1" in output assert "reactivated=1" in output assert "would_create_pending_tasks=1" in output assert "incremental_marked=2" in output assert fetch_summary_row(db_path, "com.example.reactivated")["catalog_active"] is False def test_sync_from_csv_dry_run_counts_version_updates_against_full_history(db_path, tmp_path, capsys): seed_catalog_rows( db_path, [ { "package_name": "com.example.reactivated_update", "source_order": 1, "catalog_active": 0, "collection_status": "qualified", "latest_status": "success", "last_updated": "2026/04/01", }, ], ) csv_path = tmp_path / "catalog.csv" write_catalog_csv( csv_path, [{"package_name": "com.example.reactivated_update", "last_updated": "2026/04/15"}], ) manage_app_catalog.sync_from_csv( argparse.Namespace( db_path=db_path, csv_path=str(csv_path), incremental_batch_tag="batch-new", dry_run=True, ) ) output = capsys.readouterr().out assert "added=0" in output assert "history_new=0" in output assert "reactivated=1" in output assert "version_updates=1" in output assert "would_create_pending_tasks=1" in output assert fetch_summary_row(db_path, "com.example.reactivated_update")["catalog_active"] is False def test_sync_from_csv_marks_reactivated_history_version_update_as_app_update(db_path, tmp_path): seed_catalog_rows( db_path, [ { "package_name": "com.example.reactivated_update", "source_order": 1, "catalog_active": 0, "collection_status": "qualified", "latest_status": "success", "last_updated": "2026/04/01", "collection_task_type": "new_app", }, ], ) csv_path = tmp_path / "catalog.csv" write_catalog_csv( csv_path, [{"package_name": "com.example.reactivated_update", "last_updated": "2026/04/15"}], ) manage_app_catalog.sync_from_csv( argparse.Namespace( db_path=db_path, csv_path=str(csv_path), incremental_batch_tag="batch-new", dry_run=False, ) ) row = fetch_summary_row(db_path, "com.example.reactivated_update") assert row["catalog_active"] is True assert row["collection_task_type"] == "app_update" assert row["collection_status"] == "pending" assert row["collection_status_reason"] == "catalog_version_update:2026/04/01->2026/04/15" assert row["last_updated"] == "2026/04/15" assert row["last_update_interval_days"] > 0 tasks = fetch_collection_tasks(db_path, "com.example.reactivated_update") assert len(tasks) == 2 assert tasks[-1]["task_status"] == "pending" assert tasks[-1]["is_new_app"] == 0 @pytest.mark.parametrize( "argv", [ ["set-pending-by-error-type", "APP_ERROR/1", "--tag", "batch-a", "--top-n", "10"], ["set-pending-success-zero-self-ratio", "--tag", "batch-a", "--top-n", "10"], ["set-pending-retryable", "--tag", "batch-a", "--top-n", "10"], ["set-existing-non-retryable-light-restricted", "--tag", "batch-a", "--top-n", "10"], ["export-by-error-type", "APP_ERROR/1", "--tag", "batch-a", "--top-n", "10", "--output", "/tmp/out.csv"], ], ) def test_build_parser_accepts_scope_filters(argv): parser = manage_app_catalog.build_parser() args = parser.parse_args(["--db-path", "/tmp/test.db", *argv]) assert getattr(args, "tag", "") == "batch-a" assert getattr(args, "top_n", None) == 10 def test_set_pending_by_error_type_respects_tag_and_top_n(db_path): seed_catalog_rows( db_path, [ { "package_name": "com.example.target", "source_order": 1, "incremental_batch_tag": "batch-a", "collection_status": "failed_terminal", "latest_failure_type": "APP_ERROR/1", }, { "package_name": "com.example.out_of_topn", "source_order": 5, "incremental_batch_tag": "batch-a", "collection_status": "failed_terminal", "latest_failure_type": "APP_ERROR/1", }, { "package_name": "com.example.other_tag", "source_order": 0, "incremental_batch_tag": "batch-b", "collection_status": "failed_terminal", "latest_failure_type": "APP_ERROR/1", }, ], ) manage_app_catalog.set_pending_by_error_type( argparse.Namespace( db_path=db_path, error_types=["APP_ERROR/1"], error_type_prefix="", all_non_retryable=False, non_retryable_only=False, pending_only=False, reason="manual_test", device_type="", tag="batch-a", top_n=3, ) ) assert fetch_summary_row(db_path, "com.example.target")["collection_status"] == "pending" assert count_pending_task_rows(db_path, "com.example.target") == 1 assert fetch_summary_row(db_path, "com.example.out_of_topn")["collection_status"] == "failed_terminal" assert fetch_summary_row(db_path, "com.example.other_tag")["collection_status"] == "failed_terminal" def test_set_pending_success_zero_self_ratio_respects_tag_and_top_n(db_path): seed_catalog_rows( db_path, [ { "package_name": "com.example.target", "source_order": 1, "incremental_batch_tag": "batch-a", "collection_status": "qualified", "latest_status": "success", "self_ratio": 0.0, "num_nodes": 0, }, { "package_name": "com.example.out_of_topn", "source_order": 5, "incremental_batch_tag": "batch-a", "collection_status": "qualified", "latest_status": "success", "self_ratio": 0.0, "num_nodes": 0, }, { "package_name": "com.example.other_tag", "source_order": 0, "incremental_batch_tag": "batch-b", "collection_status": "qualified", "latest_status": "success", "self_ratio": 0.0, "num_nodes": 0, }, ], ) manage_app_catalog.set_pending_success_zero_self_ratio( argparse.Namespace( db_path=db_path, max_self_ratio=0.0, reason="manual_test", tag="batch-a", top_n=3, ) ) assert fetch_summary_row(db_path, "com.example.target")["collection_status"] == "pending" assert count_pending_task_rows(db_path, "com.example.target") == 1 assert fetch_summary_row(db_path, "com.example.out_of_topn")["collection_status"] == "qualified" assert fetch_summary_row(db_path, "com.example.other_tag")["collection_status"] == "qualified" def test_set_pending_retryable_respects_tag_and_top_n(db_path): seed_catalog_rows( db_path, [ { "package_name": "com.example.target", "source_order": 1, "incremental_batch_tag": "batch-a", "collection_status": "pending", "latest_status": "failed", "latest_failure_type": "DOWNLOAD_ERROR/5", }, { "package_name": "com.example.out_of_topn", "source_order": 5, "incremental_batch_tag": "batch-a", "collection_status": "pending", "latest_status": "failed", "latest_failure_type": "DOWNLOAD_ERROR/5", }, { "package_name": "com.example.other_tag", "source_order": 0, "incremental_batch_tag": "batch-b", "collection_status": "pending", "latest_status": "failed", "latest_failure_type": "DOWNLOAD_ERROR/5", }, ], ) assert count_pending_task_rows(db_path, "com.example.target") == 0 manage_app_catalog.set_pending_retryable( argparse.Namespace( db_path=db_path, current_status="pending", reason="manual_test", tag="batch-a", top_n=3, ) ) assert fetch_summary_row(db_path, "com.example.target")["collection_status"] == "pending" assert count_pending_task_rows(db_path, "com.example.target") == 1 assert count_pending_task_rows(db_path, "com.example.out_of_topn") == 0 assert count_pending_task_rows(db_path, "com.example.other_tag") == 0 def test_set_existing_non_retryable_light_restricted_respects_tag_and_top_n(db_path): seed_catalog_rows( db_path, [ { "package_name": "com.example.target", "source_order": 1, "incremental_batch_tag": "batch-a", "collection_status": "failed_terminal", "latest_failure_type": "APP_ERROR/1", "num_nodes": 0, }, { "package_name": "com.example.out_of_topn", "source_order": 5, "incremental_batch_tag": "batch-a", "collection_status": "failed_terminal", "latest_failure_type": "APP_ERROR/1", "num_nodes": 0, }, { "package_name": "com.example.other_tag", "source_order": 0, "incremental_batch_tag": "batch-b", "collection_status": "failed_terminal", "latest_failure_type": "APP_ERROR/1", "num_nodes": 0, }, ], ) manage_app_catalog.set_existing_non_retryable_light_restricted( argparse.Namespace( db_path=db_path, min_num_nodes=0, reason="manual_test", tag="batch-a", top_n=3, ) ) assert fetch_summary_row(db_path, "com.example.target")["collection_status"] == "pending" assert count_pending_task_rows(db_path, "com.example.target") == 1 assert fetch_summary_row(db_path, "com.example.out_of_topn")["collection_status"] == "failed_terminal" assert fetch_summary_row(db_path, "com.example.other_tag")["collection_status"] == "failed_terminal" def test_export_apps_by_error_type_respects_tag_and_top_n(db_path, tmp_path): seed_catalog_rows( db_path, [ { "package_name": "com.example.target", "source_order": 1, "incremental_batch_tag": "batch-a", "latest_failure_type": "APP_ERROR/1", "latest_task_detail": "target detail", "downloads": 123, }, { "package_name": "com.example.out_of_topn", "source_order": 5, "incremental_batch_tag": "batch-a", "latest_failure_type": "APP_ERROR/1", "latest_task_detail": "out of topn detail", "downloads": 456, }, { "package_name": "com.example.other_tag", "source_order": 0, "incremental_batch_tag": "batch-b", "latest_failure_type": "APP_ERROR/1", "latest_task_detail": "other tag detail", "downloads": 789, }, ], ) output_path = tmp_path / "export.csv" manage_app_catalog.export_apps_by_error_type( argparse.Namespace( db_path=db_path, error_types=["APP_ERROR/1"], error_type_prefix="", output=str(output_path), non_retryable_only=False, tag="batch-a", top_n=3, ) ) with output_path.open("r", encoding="utf-8-sig", newline="") as handle: rows = list(csv.DictReader(handle)) assert [row["package_name"] for row in rows] == ["com.example.target"]