hiffi-artist-intelligence/test_multiplatform_ranking.py

"""
test_multiplatform_ranking.py
Automated Quality Assurance & Governance Test Suite for HIFFI Multi-Platform Pipeline.
"""

from pathlib import Path
import numpy as np
import pandas as pd
import pytest

DATA_FILE = Path("data/hiffi_multiplatform_500.csv")


@pytest.fixture(scope="module")
def df():
    assert DATA_FILE.exists(), f"Target dataset {DATA_FILE} does not exist."
    return pd.read_csv(DATA_FILE)


def test_cohort_size(df):
    """Dataset must contain exactly 500 artist records."""
    assert len(df) == 500, f"Expected 500 artists, found {len(df)}"


def test_spotify_streaming_governance(df):
    """
    Validates audited public streaming ingestion:
    1. Populated listener counts must be positive integers (> 0).
    2. Status must be explicit (either VERIFIED_PUBLIC_AGGREGATOR or UNAVAILABLE_IN_TOP_5000).
    3. No zero-imputed values allowed.
    """
    valid_statuses = {"VERIFIED_PUBLIC_AGGREGATOR", "UNAVAILABLE_IN_TOP_5000"}
    assert set(df["spotify_status"].unique()).issubset(valid_statuses), (
        f"Found invalid spotify_status values: {df['spotify_status'].unique()}"
    )

    valid_listeners = df["spotify_monthly_listeners"].dropna()
    assert len(valid_listeners) > 0, "Expected verified streaming records to be populated."
    assert (valid_listeners > 0).all(), "Found zero or negative listener values (must be positive integer or NaN)."


def test_entity_resolution_completeness(df):
    """Top 50 established artists must have candidate Spotify IDs resolved (>= 90%)."""
    top_50 = df[df["production_hiffi_rank"] <= 50]
    top_50_resolved = (top_50["spotify_id"].notna()) & (top_50["spotify_id"] != "N/A")
    assert top_50_resolved.mean() >= 0.90, (
        f"Expected >= 90% of top 50 artists to have resolved candidate Spotify IDs, got {top_50_resolved.mean():.1%}"
    )

    valid_mask = (
        df["spotify_id"].isna()
        | (df["spotify_id"] == "N/A")
        | (df["spotify_id"].astype(str).str.len() == 22)
    )
    assert valid_mask.all(), "Found malformed Spotify IDs."


def test_instagram_follower_validity(df):
    """Follower counts must be non-negative numbers or NaN."""
    valid_followers = df["instagram_followers"].dropna()
    assert (valid_followers > 0).all(), "Found zero or negative follower values."
    assert np.isfinite(valid_followers).all(), "Found infinite values."


def test_score_normalization_bounds(df):
    """Production and experimental scores must strictly fall within [0.0, 100.0]."""
    for col in ["production_hiffi_score", "hiffi_xp_score"]:
        valid_scores = df[col].dropna()
        assert (valid_scores >= 0.0).all(), f"Found scores < 0 in {col}"
        assert (valid_scores <= 100.0).all(), f"Found scores > 100 in {col}"


def test_rank_monotonicity(df):
    """Higher or equal scores must produce better (lower numeric) or equal ranks."""
    sorted_df = df.sort_values("hiffi_xp_rank", ascending=True).reset_index(drop=True)
    scores = sorted_df["hiffi_xp_score"].tolist()
    for i in range(len(scores) - 1):
        assert scores[i] >= scores[i + 1], f"Rank inversion detected at row {i}"


def test_schema_provenance_columns(df):
    """Dataset must contain all mandatory provenance and audit timestamps."""
    mandatory_cols = [
        "name", "production_hiffi_rank", "production_hiffi_score",
        "hiffi_xp_rank", "hiffi_xp_score", "wikidata_qid", "spotify_id",
        "spotify_status", "spotify_monthly_listeners", "spotify_data_source",
        "instagram_handle", "instagram_followers", "instagram_collected_at",
        "youtube_data_source", "has_complete_multiplatform_signals"
    ]
    for col in mandatory_cols:
        assert col in df.columns, f"Mandatory column missing: {col}"