You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
224 lines
8.3 KiB
224 lines
8.3 KiB
"""Tests for political axis orientation validation.
|
|
|
|
Validates that PVV, FVD, JA21, and SGP appear on the RIGHT side
|
|
(mean-based) of the political compass, per AGENTS.md convention.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
duckdb = pytest.importorskip("duckdb")
|
|
|
|
|
|
def _setup_party_axis_scores(db_path: str, rows: list):
|
|
"""Insert synthetic party_axis_scores rows.
|
|
|
|
Args:
|
|
db_path: Path to DuckDB database.
|
|
rows: List of (party_abbrev, window_id, x_axis_aligned, y_axis_aligned).
|
|
"""
|
|
conn = duckdb.connect(db_path)
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS party_axis_scores (
|
|
party_abbrev TEXT,
|
|
window_id TEXT,
|
|
x_axis_aligned DOUBLE,
|
|
y_axis_aligned DOUBLE
|
|
)
|
|
"""
|
|
)
|
|
for party, window, x, y in rows:
|
|
conn.execute(
|
|
"INSERT INTO party_axis_scores (party_abbrev, window_id, x_axis_aligned, y_axis_aligned) VALUES (?, ?, ?, ?)",
|
|
(party, window, x, y),
|
|
)
|
|
conn.close()
|
|
|
|
|
|
def _build_scores_by_party(db_path: str) -> dict:
|
|
"""Load aligned scores as {party: [[x,y] per window]} from DuckDB."""
|
|
from analysis.explorer_data import load_party_scores_all_windows_aligned
|
|
|
|
return load_party_scores_all_windows_aligned(db_path)
|
|
|
|
|
|
class TestAxisPoliticalOrientation:
|
|
def test_build_window_party_scores_happy_path(self):
|
|
from analysis.explorer_data import build_window_party_scores
|
|
|
|
data = {
|
|
"PVV": [[0.5, 0.3], [0.6, 0.4]],
|
|
"FVD": [[0.4, 0.2], [0.5, 0.3]],
|
|
"SP": [[-0.4, -0.2], [-0.5, -0.3]],
|
|
"DENK": [[-0.3, -0.1], [-0.4, -0.2]],
|
|
}
|
|
result = build_window_party_scores(data, 0)
|
|
assert result == {
|
|
"PVV": [0.5, 0.3],
|
|
"FVD": [0.4, 0.2],
|
|
"SP": [-0.4, -0.2],
|
|
"DENK": [-0.3, -0.1],
|
|
}
|
|
|
|
result = build_window_party_scores(data, 1)
|
|
assert result == {
|
|
"PVV": [0.6, 0.4],
|
|
"FVD": [0.5, 0.3],
|
|
"SP": [-0.5, -0.3],
|
|
"DENK": [-0.4, -0.2],
|
|
}
|
|
|
|
def test_build_window_party_scores_out_of_range(self):
|
|
from analysis.explorer_data import build_window_party_scores
|
|
|
|
data = {"PVV": [[0.5, 0.3]], "SP": [[-0.4, -0.2]]}
|
|
assert build_window_party_scores(data, 99) == {}
|
|
assert build_window_party_scores(data, -1) == {}
|
|
assert build_window_party_scores({}, 0) == {}
|
|
|
|
def test_orientation_correct_no_flip_needed(self, tmp_path):
|
|
db_path = str(tmp_path / "orientation.db")
|
|
_setup_party_axis_scores(
|
|
db_path,
|
|
[
|
|
# Window 0: Correct orientation — right_mean > left_mean on both axes
|
|
("PVV", "w1", 0.8, 0.2),
|
|
("FVD", "w1", 0.6, 0.1),
|
|
("JA21", "w1", 0.5, 0.0),
|
|
("SGP", "w1", 0.4, 0.0),
|
|
("SP", "w1", -0.6, -0.2),
|
|
("DENK", "w1", -0.4, -0.1),
|
|
("PvdA", "w1", -0.5, -0.1),
|
|
("Volt", "w1", -0.3, -0.0),
|
|
# Window 1: Same correct orientation
|
|
("PVV", "w2", 0.7, 0.3),
|
|
("FVD", "w2", 0.5, 0.2),
|
|
("JA21", "w2", 0.4, 0.1),
|
|
("SGP", "w2", 0.3, 0.0),
|
|
("SP", "w2", -0.5, -0.2),
|
|
("DENK", "w2", -0.3, -0.1),
|
|
("PvdA", "w2", -0.4, -0.1),
|
|
("Volt", "w2", -0.2, 0.0),
|
|
],
|
|
)
|
|
|
|
scores_by_party = _build_scores_by_party(db_path)
|
|
from analysis.explorer_data import build_window_party_scores
|
|
from analysis.svd_labels import compute_flip_direction
|
|
|
|
# 2 windows
|
|
n_windows = max(len(v) for v in scores_by_party.values())
|
|
assert n_windows == 2
|
|
|
|
for window_idx in range(n_windows):
|
|
party_scores = build_window_party_scores(scores_by_party, window_idx)
|
|
flip_x = compute_flip_direction(1, party_scores)
|
|
flip_y = compute_flip_direction(2, party_scores)
|
|
assert flip_x is False, (
|
|
f"Window {window_idx}: right parties should already be on right (x-axis)"
|
|
)
|
|
assert flip_y is False, (
|
|
f"Window {window_idx}: right parties should already be on right (y-axis)"
|
|
)
|
|
|
|
def test_orientation_incorrect_triggers_flip(self, tmp_path):
|
|
db_path = str(tmp_path / "orientation_flipped.db")
|
|
_setup_party_axis_scores(
|
|
db_path,
|
|
[
|
|
# Window 0: Wrong orientation — right_mean < left_mean on x-axis
|
|
("PVV", "w1", -0.8, 0.0), # Right party on left
|
|
("FVD", "w1", -0.6, 0.0),
|
|
("JA21", "w1", -0.5, 0.0),
|
|
("SGP", "w1", -0.4, 0.0),
|
|
("SP", "w1", 0.6, 0.0), # Left party on right
|
|
("DENK", "w1", 0.4, 0.0),
|
|
],
|
|
)
|
|
|
|
scores_by_party = _build_scores_by_party(db_path)
|
|
from analysis.explorer_data import build_window_party_scores
|
|
from analysis.svd_labels import compute_flip_direction
|
|
|
|
party_scores = build_window_party_scores(scores_by_party, 0)
|
|
flip_x = compute_flip_direction(1, party_scores)
|
|
# Right mean = (-0.8 + -0.6 + -0.5 + -0.4) / 4 = -0.575
|
|
# Left mean = (0.6 + 0.4) / 2 = 0.5
|
|
# right_mean < left_mean → flip = True
|
|
assert flip_x is True, "Right parties on left should trigger flip=True"
|
|
|
|
def test_missing_party_graceful_skip(self, tmp_path):
|
|
db_path = str(tmp_path / "partial.db")
|
|
_setup_party_axis_scores(
|
|
db_path,
|
|
[
|
|
# Only PVV (right) and SP (left), no FVD/JA21/SGP
|
|
("PVV", "w1", 0.8, 0.2),
|
|
("SP", "w1", -0.6, -0.2),
|
|
("DENK", "w1", -0.4, -0.1),
|
|
],
|
|
)
|
|
|
|
scores_by_party = _build_scores_by_party(db_path)
|
|
from analysis.explorer_data import build_window_party_scores
|
|
from analysis.svd_labels import compute_flip_direction
|
|
|
|
party_scores = build_window_party_scores(scores_by_party, 0)
|
|
# Should not raise — PVV and SP are in canonical sets, rest ignored
|
|
flip_x = compute_flip_direction(1, party_scores)
|
|
flip_y = compute_flip_direction(2, party_scores)
|
|
# right_mean = 0.8, left_mean = (-0.6 + -0.4) / 2 = -0.5
|
|
# 0.8 > -0.5 → flip = False
|
|
assert flip_x is False
|
|
assert flip_y is False
|
|
|
|
def test_party_name_aliasing_normalized(self, tmp_path):
|
|
"""Test that aliased party names are handled gracefully.
|
|
|
|
DB may return 'GL' while canonical sets use 'GroenLinks-PvdA'.
|
|
The test uses exact canonical names; _PARTY_NORMALIZE handles aliases.
|
|
"""
|
|
db_path = str(tmp_path / "aliased.db")
|
|
_setup_party_axis_scores(
|
|
db_path,
|
|
[
|
|
# PVV and FVD under exact canonical names
|
|
("PVV", "w1", 0.8, 0.2),
|
|
("FVD", "w1", 0.6, 0.1),
|
|
# Left parties under exact canonical names
|
|
("SP", "w1", -0.6, -0.2),
|
|
("DENK", "w1", -0.4, -0.1),
|
|
("Volt", "w1", -0.3, -0.1),
|
|
],
|
|
)
|
|
|
|
scores_by_party = _build_scores_by_party(db_path)
|
|
from analysis.explorer_data import build_window_party_scores
|
|
from analysis.svd_labels import compute_flip_direction
|
|
|
|
party_scores = build_window_party_scores(scores_by_party, 0)
|
|
flip_x = compute_flip_direction(1, party_scores)
|
|
# right_mean = (0.8 + 0.6) / 2 = 0.7
|
|
# left_mean = (-0.6 + -0.4 + -0.3) / 3 = -0.433
|
|
# 0.7 > -0.433 → flip = False
|
|
assert flip_x is False
|
|
|
|
def test_insufficient_data_returns_false(self, tmp_path):
|
|
db_path = str(tmp_path / "insufficient.db")
|
|
_setup_party_axis_scores(
|
|
db_path,
|
|
[
|
|
# Only left parties — no right parties
|
|
("SP", "w1", -0.6, -0.2),
|
|
("DENK", "w1", -0.4, -0.1),
|
|
],
|
|
)
|
|
|
|
scores_by_party = _build_scores_by_party(db_path)
|
|
from analysis.explorer_data import build_window_party_scores
|
|
from analysis.svd_labels import compute_flip_direction
|
|
|
|
party_scores = build_window_party_scores(scores_by_party, 0)
|
|
flip = compute_flip_direction(1, party_scores)
|
|
# No right parties in data → returns False (no flip)
|
|
assert flip is False
|
|
|