mrleo1nid
75ae2092ad
ci / api / api (push) Successful in 56s
ci / contract / contract (сгенерированный код совпадает с openapi.yaml) (push) Successful in 1m23s
ci / db / db (push) Successful in 1m8s
ci / recommender / recommender (push) Failing after 40s
ci / mobile / mobile (Metro-бандл Android) (push) Successful in 1m25s
ci / ts / ts (packages + web) (push) Successful in 7m20s
251 lines
11 KiB
Python
251 lines
11 KiB
Python
"""SQL рекомендателя на настоящей схеме (CI-job `ci / db`; схему заранее накатывают миграции api).
|
||
|
||
Локально: TEST_DATABASE_URL=postgresql://… uv run pytest tests/test_db_integration.py
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import time
|
||
import uuid
|
||
from typing import Any
|
||
|
||
import numpy as np
|
||
import psycopg
|
||
import pytest
|
||
|
||
from pipu_rec.analysis.features import Features
|
||
from pipu_rec.analysis.worker import Worker
|
||
from pipu_rec.train.pipeline import load_interactions, train_all
|
||
from pipu_rec.wave.engine import Profile, Session, build_batch
|
||
from pipu_rec.wave.store import load_index, load_profile
|
||
|
||
DSN = os.environ.get("TEST_DATABASE_URL", "")
|
||
pytestmark = pytest.mark.skipif(not DSN, reason="TEST_DATABASE_URL is not set")
|
||
|
||
|
||
def uid() -> str:
|
||
return str(uuid.uuid4())
|
||
|
||
|
||
@pytest.fixture()
|
||
def world() -> dict[str, Any]:
|
||
"""Хранилище, 2 исполнителя × 10 треков, 6 слушателей с событиями и реакциями."""
|
||
with psycopg.connect(DSN, autocommit=True) as conn:
|
||
src = uid()
|
||
conn.execute(
|
||
"INSERT INTO library_sources (id, name, type, local_path, status) VALUES (%s, 'it', 'local', '', 'mounted')",
|
||
(src,),
|
||
)
|
||
tracks: list[str] = []
|
||
for a in range(2):
|
||
artist = uid()
|
||
conn.execute(
|
||
"INSERT INTO artists (id, name, sort_name, name_norm) VALUES (%s, %s, %s, %s)",
|
||
(artist, f"A{a}", f"A{a}", f"a{a}-{artist[:6]}"),
|
||
)
|
||
album = uid()
|
||
conn.execute(
|
||
"INSERT INTO albums (id, title, title_norm, album_artist_id) VALUES (%s, %s, %s, %s)",
|
||
(album, f"Al{a}", f"al{a}", artist),
|
||
)
|
||
for i in range(10):
|
||
t, f = uid(), uid()
|
||
conn.execute(
|
||
"INSERT INTO tracks (id, title, title_norm, album_id, duration_ms) VALUES (%s, %s, %s, %s, 200000)",
|
||
(t, f"T{a}{i}", f"t{a}{i}", album),
|
||
)
|
||
conn.execute(
|
||
"INSERT INTO media_files (id, source_id, rel_path, size_bytes, mtime, format, duration_ms, track_id) "
|
||
"VALUES (%s, %s, %s, 1, now(), 'flac', 200000, %s)",
|
||
(f, src, f"{a}/{i}.flac", t),
|
||
)
|
||
conn.execute("UPDATE tracks SET primary_file_id = %s WHERE id = %s", (f, t))
|
||
conn.execute(
|
||
"INSERT INTO track_artists (track_id, artist_id, role, position) VALUES (%s, %s, 'main', 0)",
|
||
(t, artist),
|
||
)
|
||
tracks.append(t)
|
||
users = []
|
||
for u in range(6):
|
||
user = uid()
|
||
conn.execute(
|
||
"INSERT INTO users (id, username, password_hash) VALUES (%s, %s, 'x')",
|
||
(user, f"u{u}-{user[:6]}"),
|
||
)
|
||
users.append(user)
|
||
group = tracks[:10] if u % 2 == 0 else tracks[10:]
|
||
for k, t in enumerate(group):
|
||
kind = "play_end" if k < 8 else "skip"
|
||
conn.execute(
|
||
"INSERT INTO play_events (id, user_id, track_id, type, occurred_at, played_ms, client_type) "
|
||
"VALUES (%s, %s, %s, %s, now() - interval '20 days', %s, 'web')",
|
||
(uid(), user, t, kind, 200000 if kind == "play_end" else 3000),
|
||
)
|
||
conn.execute(
|
||
"INSERT INTO user_track_stats (user_id, track_id, play_count, complete_count, last_played_at) "
|
||
"VALUES (%s, %s, 1, %s, now() - interval '20 days')",
|
||
(user, t, 1 if kind == "play_end" else 0),
|
||
)
|
||
conn.execute(
|
||
"INSERT INTO track_reactions (user_id, track_id, reaction) VALUES (%s, %s, 'like')",
|
||
(user, group[0]),
|
||
)
|
||
return {"tracks": tracks, "users": users}
|
||
|
||
|
||
def test_analysis_worker_saves_features(world: dict[str, Any]) -> None:
|
||
w = Worker(DSN, "/library", "/library-host", 1)
|
||
emb = np.zeros(256, dtype=np.float32)
|
||
emb[0] = 1.0
|
||
feats = Features(
|
||
bpm=120.0,
|
||
bpm_confidence=0.9,
|
||
key="C",
|
||
scale="major",
|
||
loudness_lufs=-12.0,
|
||
energy=0.6,
|
||
danceability=0.5,
|
||
valence=0.5,
|
||
acousticness=0.2,
|
||
intro_ms=1000,
|
||
embedding=emb,
|
||
)
|
||
with psycopg.connect(DSN, autocommit=True) as conn:
|
||
jobs = w.claim(conn, 5)
|
||
assert jobs, "every new track is queued for analysis by the trigger"
|
||
w.save(conn, jobs[0].track_id, feats)
|
||
w.fail(conn, jobs[1].track_id, "boom")
|
||
w.requeue_stale(conn)
|
||
row = conn.execute(
|
||
"SELECT bpm FROM track_features WHERE track_id = %s", (jobs[0].track_id,)
|
||
).fetchone()
|
||
assert row is not None and row[0] == pytest.approx(120.0)
|
||
|
||
|
||
def test_analysis_worker_skips_excluded_folders_and_uses_allowed_copy(world: dict[str, Any]) -> None:
|
||
worker = Worker(DSN, "/library", "/library-host", 1)
|
||
blocked, alternative, allowed = world["tracks"][:3]
|
||
with psycopg.connect(DSN, autocommit=True) as conn:
|
||
source = conn.execute(
|
||
"SELECT source_id FROM media_files WHERE track_id = %s", (blocked,)
|
||
).fetchone()
|
||
assert source is not None
|
||
conn.execute(
|
||
"UPDATE library_sources SET excluded_dirs = ARRAY['Downloads_PumpEars'] WHERE id = %s",
|
||
(source[0],),
|
||
)
|
||
for track, rel in [(blocked, "Downloads_PumpEars/nested/loop.wav"), (alternative, "Downloads_PumpEars/loop.wav"), (allowed, "Downloads_PumpEars-copy/song.flac")]:
|
||
conn.execute("UPDATE media_files SET rel_path = %s WHERE track_id = %s", (rel, track))
|
||
conn.execute(
|
||
"INSERT INTO media_files (id, source_id, rel_path, size_bytes, mtime, format, duration_ms, track_id) "
|
||
"VALUES (%s, %s, 'Music/allowed-copy.flac', 1, now(), 'flac', 200000, %s)",
|
||
(uid(), source[0], alternative),
|
||
)
|
||
conn.execute(
|
||
"UPDATE analysis_jobs SET status = 'queued' WHERE track_id = ANY(%s::uuid[])",
|
||
([blocked, alternative, allowed],),
|
||
)
|
||
jobs = {job.track_id: job.path for job in worker.claim(conn, 10000)}
|
||
assert blocked not in jobs
|
||
assert jobs[alternative].replace("\\", "/").endswith("/Music/allowed-copy.flac")
|
||
assert jobs[allowed].replace("\\", "/").endswith("/Downloads_PumpEars-copy/song.flac")
|
||
row = conn.execute("SELECT attempts, status FROM analysis_jobs WHERE track_id = %s", (blocked,)).fetchone()
|
||
assert row == (0, 'queued'), "excluded files must not consume retry attempts"
|
||
|
||
|
||
def test_training_and_wave_queries(world: dict[str, Any]) -> None:
|
||
with psycopg.connect(DSN) as conn:
|
||
assert len(load_interactions(conn)) >= 60
|
||
result = train_all(DSN, now=time.time())
|
||
assert result.get("activated") is True, result
|
||
with psycopg.connect(DSN) as conn:
|
||
index = load_index(conn)
|
||
assert set(world["tracks"]) <= set(index.ids)
|
||
assert index.has_cf.sum() >= 20
|
||
profile: Profile = load_profile(conn, world["users"][0])
|
||
assert profile.taste_cf is not None
|
||
assert profile.recent_likes
|
||
active = conn.execute(
|
||
"SELECT count(*) FROM model_versions WHERE kind = 'als' AND is_active"
|
||
).fetchone()
|
||
assert active is not None and active[0] == 1
|
||
picks = build_batch(index, profile, Session(), [], n=5, rng=np.random.default_rng(0))
|
||
assert len(picks) == 5
|
||
|
||
|
||
def test_onboarding_seeds_taste(world: dict[str, Any]) -> None:
|
||
user = uid()
|
||
with psycopg.connect(DSN, autocommit=True) as conn:
|
||
conn.execute(
|
||
"INSERT INTO users (id, username, password_hash) VALUES (%s, %s, 'x')", (user, f"new-{user[:6]}")
|
||
)
|
||
artist = conn.execute(
|
||
"SELECT artist_id::text FROM track_artists WHERE track_id = %s", (world["tracks"][0],)
|
||
).fetchone()
|
||
assert artist is not None
|
||
conn.execute(
|
||
"INSERT INTO user_profiles_rec (user_id, onboarding) VALUES (%s, %s)",
|
||
(user, json.dumps({"artist_ids": [artist[0]], "genres": []})),
|
||
)
|
||
profile = load_profile(conn, user)
|
||
assert len(profile.scores) >= 10, "chosen artist's tracks become the initial taste"
|
||
assert not profile.played
|
||
|
||
|
||
def test_impression_features_roundtrip(world: dict[str, Any]) -> None:
|
||
from pipu_rec.train.ranker import load_rows, train_and_save
|
||
from pipu_rec.wave.engine import FEATURE_NAMES
|
||
|
||
session, user, track = uid(), world["users"][0], world["tracks"][0]
|
||
with psycopg.connect(DSN, autocommit=True) as conn:
|
||
conn.execute("INSERT INTO wave_sessions (id, user_id) VALUES (%s, %s)", (session, user))
|
||
conn.execute(
|
||
"INSERT INTO wave_impressions (wave_session_id, position, track_id, candidate_source, outcome) "
|
||
"VALUES (%s, 0, %s, 'audio_sim', 'complete')",
|
||
(session, track),
|
||
)
|
||
conn.execute(
|
||
"INSERT INTO rec_impression_features (wave_session_id, track_id, features) VALUES (%s, %s, %s)",
|
||
(session, track, [0.5] * len(FEATURE_NAMES)),
|
||
)
|
||
rows = load_rows(conn)
|
||
assert any(r[0] == session and r[2] == "complete" for r in rows)
|
||
result = train_and_save(conn, time.time())
|
||
assert "skipped" in result["ranker"], "too few impressions — v1 stays"
|
||
|
||
|
||
def test_demographic_prior_advances_age_and_preserves_own_feedback(world: dict[str, Any]) -> None:
|
||
user = uid()
|
||
with psycopg.connect(DSN, autocommit=True) as conn:
|
||
conn.execute(
|
||
"INSERT INTO users (id, username, password_hash) VALUES (%s, %s, 'x')",
|
||
(user, f"cohort-{user[:6]}"),
|
||
)
|
||
# Target selected 13–19 twenty years ago; matching listeners are now 30–39.
|
||
conn.execute(
|
||
"INSERT INTO user_profiles_rec (user_id, onboarding) VALUES (%s, "
|
||
"jsonb_build_object('age_group', '13_19', 'gender', 'male', "
|
||
"'age_group_year', EXTRACT(YEAR FROM CURRENT_DATE)::int - 20))",
|
||
(user,),
|
||
)
|
||
for peer in world["users"][:3]:
|
||
conn.execute(
|
||
"INSERT INTO user_profiles_rec (user_id, onboarding) VALUES (%s, %s) "
|
||
"ON CONFLICT (user_id) DO UPDATE SET onboarding = EXCLUDED.onboarding",
|
||
(peer, json.dumps({"gender": "male", "age_group": "30_39"})),
|
||
)
|
||
profile = load_profile(conn, user)
|
||
track = world["tracks"][0] # two matching listeners liked it
|
||
assert 0 < profile.scores[track] <= 0.25
|
||
assert not profile.played
|
||
conn.execute(
|
||
"INSERT INTO track_reactions (user_id, track_id, reaction) VALUES (%s, %s, 'dislike')",
|
||
(user, track),
|
||
)
|
||
assert track not in load_profile(conn, user).scores
|
||
conn.execute(
|
||
"UPDATE user_profiles_rec SET onboarding = %s WHERE user_id = %s",
|
||
(json.dumps({"gender": "female"}), user),
|
||
)
|
||
assert not load_profile(conn, user).scores, "no invented preferences when the cohort has no evidence"
|