diff --git a/.gitea/workflows/ci-db.yml b/.gitea/workflows/ci-db.yml new file mode 100644 index 0000000..4fb0c29 --- /dev/null +++ b/.gitea/workflows/ci-db.yml @@ -0,0 +1,52 @@ +# Интеграционные проверки SQL на настоящем PostgreSQL + pgvector: миграции вверх / вниз / вверх, +# запросы сканера, отпечатков, шаблонов и очереди (Go), загрузка каталога, профиля и обучение ALS (Python). +name: ci / db + +on: + push: + branches: [main] + paths: + - 'services/api/**' + - 'services/recommender/**' + - '.gitea/workflows/ci-db.yml' + pull_request: + paths: + - 'services/api/**' + - 'services/recommender/**' + - '.gitea/workflows/ci-db.yml' + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +jobs: + db: + runs-on: ubuntu-latest + services: + postgres: + image: pgvector/pgvector:0.8.7-pg17 + env: + POSTGRES_USER: pipu + POSTGRES_PASSWORD: pipu + POSTGRES_DB: pipu + options: >- + --health-cmd "pg_isready -U pipu" + --health-interval 5s + --health-timeout 5s + --health-retries 20 + env: + TEST_DATABASE_URL: postgres://pipu:pipu@postgres:5432/pipu?sslmode=disable + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-go@v5 + with: { go-version: "1.24", cache-dependency-path: services/api/go.sum } + - name: Go — миграции и запросы store + working-directory: services/api + run: go test -tags integration -count=1 -v ./internal/dbtest/ + - uses: astral-sh/setup-uv@v6 + with: { version: "0.10.2", enable-cache: true, cache-dependency-glob: services/recommender/uv.lock } + - name: Python — каталог, профиль, обучение ALS + working-directory: services/recommender + run: | + uv sync --frozen + uv run pytest -q tests/test_db_integration.py diff --git a/services/api/internal/dbtest/integration_test.go b/services/api/internal/dbtest/integration_test.go new file mode 100644 index 0000000..553bd24 --- /dev/null +++ b/services/api/internal/dbtest/integration_test.go @@ -0,0 +1,228 @@ +//go:build integration + +// Package dbtest — интеграционные проверки SQL на настоящем PostgreSQL с pgvector (CI-job `ci / db`). +// Локально: TEST_DATABASE_URL=postgres://… go test -tags integration ./internal/dbtest/ +package dbtest + +import ( + "context" + "crypto/sha256" + "database/sql" + "log/slog" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/jackc/pgx/v5/stdlib" + "github.com/pressly/goose/v3" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "pipu/api/internal/library" + "pipu/api/internal/library/meta" + "pipu/api/internal/platform/db" + "pipu/api/internal/uploads" + "pipu/api/migrations" +) + +func connect(t *testing.T) *pgxpool.Pool { + url := os.Getenv("TEST_DATABASE_URL") + if url == "" { + t.Skip("TEST_DATABASE_URL is not set") + } + ctx := context.Background() + pool, err := db.Connect(ctx, url) + require.NoError(t, err) + t.Cleanup(pool.Close) + return pool +} + +// Все миграции применяются, откатываются до нуля и применяются снова — Down-секции рабочие. +func TestMigrationsUpDownUp(t *testing.T) { + pool := connect(t) + ctx := context.Background() + require.NoError(t, db.Migrate(ctx, pool)) + + sqlDB := stdlib.OpenDBFromPool(pool) + defer func(d *sql.DB) { _ = d.Close() }(sqlDB) + provider, err := goose.NewProvider(goose.DialectPostgres, sqlDB, migrations.FS) + require.NoError(t, err) + _, err = provider.DownTo(ctx, 0) + require.NoError(t, err, "down migrations") + _, err = provider.Up(ctx) + require.NoError(t, err, "up again") +} + +// fakeProber — теги из имени файла «Исполнитель - Название.flac», длительность 200 с. +type fakeProber struct{} + +func (fakeProber) Probe(_ context.Context, abs string) (library.Probe, error) { + base := strings.TrimSuffix(filepath.Base(abs), filepath.Ext(abs)) + artist, title, _ := strings.Cut(base, " - ") + album := filepath.Base(filepath.Dir(abs)) + return library.Probe{ + Format: "flac", Codec: "flac", Bitrate: 900_000, SampleRate: 44100, BitDepth: 16, Channels: 2, DurationMs: 200_000, + Tags: meta.Fields{Artist: artist, Title: title, Album: album, Year: "1988", Track: "1"}, + }, nil +} + +func seedUser(t *testing.T, pool *pgxpool.Pool, name, role string) uuid.UUID { + id := uuid.Must(uuid.NewV7()) + _, err := pool.Exec(context.Background(), `INSERT INTO users (id, username, password_hash, role) VALUES ($1, $2, 'x', $3)`, + id, name+"-"+id.String()[:8], role) + require.NoError(t, err) + return id +} + +// Сканер, каталог, отпечатки, шаблоны записи и очередь на добавление — запросы на настоящей схеме. +func TestLibraryAndUploads(t *testing.T) { + pool := connect(t) + ctx := context.Background() + require.NoError(t, db.Migrate(ctx, pool)) + log := slog.New(slog.DiscardHandler) + lib := library.NewStore(pool) + + // --- Хранилище и точечный скан --- + root := t.TempDir() + album := filepath.Join(root, "Кино", "Группа крови") + require.NoError(t, os.MkdirAll(album, 0o755)) + for _, name := range []string{"Кино - Группа крови.flac", "Кино - Закрой за мной дверь.flac"} { + require.NoError(t, os.WriteFile(filepath.Join(album, name), []byte(name), 0o644)) + } + tpl := "{albumartist}/{album}/{title}.{ext}" + src, err := lib.SaveSource(ctx, library.Source{ + ID: uuid.Must(uuid.NewV7()), Name: "it", Type: library.SourceLocal, TagsFirst: true, Status: "mounted", + NFSOptions: "ro", AcceptsUploads: true, WriteTemplate: &tpl, + }) + require.NoError(t, err) + assert.True(t, src.AcceptsUploads) + require.NotNil(t, src.WriteTemplate) + + scanner := library.NewScanner(lib, fakeProber{}, library.NewProgressBoard(), log) + var fileIDs []uuid.UUID + for _, rel := range []string{"Кино/Группа крови/Кино - Группа крови.flac", "Кино/Группа крови/Кино - Закрой за мной дверь.flac"} { + id, err := scanner.ScanFile(ctx, src, root, rel) + require.NoError(t, err) + fileIDs = append(fileIDs, id) + } + again, err := scanner.ScanFile(ctx, src, root, "Кино/Группа крови/Кино - Группа крови.flac") + require.NoError(t, err) + assert.Equal(t, fileIDs[0], again, "rescan keeps the file row") + + // --- Отпечатки --- + todo, err := lib.FilesToFingerprint(ctx, true, uuid.Nil, 10) + require.NoError(t, err) + require.Len(t, todo, 2) + sum := sha256.Sum256([]byte("Кино - Группа крови.flac")) + fp := &library.Fingerprint{DurationMs: 200_000, Points: []uint32{1, 2, 3, 4294967295}} + require.NoError(t, lib.SaveFingerprint(ctx, todo[0].ID, sum[:], fp, true)) + todo, err = lib.FilesToFingerprint(ctx, true, uuid.Nil, 10) + require.NoError(t, err) + assert.Len(t, todo, 1) + + // --- Шаблон записи и популярные треки --- + samples, err := lib.TemplateSamples(ctx, src.ID, 50) + require.NoError(t, err) + assert.Len(t, samples, 2) + var artistID uuid.UUID + require.NoError(t, pool.QueryRow(ctx, `SELECT id FROM artists WHERE name = 'Кино'`).Scan(&artistID)) + top, err := lib.ArtistTopTracks(ctx, artistID, 10) + require.NoError(t, err) + assert.Len(t, top, 2) + + // --- Очередь на добавление --- + up := uploads.NewStore(pool) + author := seedUser(t, pool, "moder", "moderator") + settings, err := up.Settings(ctx) + require.NoError(t, err) + settings.QueueLimitBytes = 1000 + require.NoError(t, up.SaveSettings(ctx, settings)) + settings, err = up.Settings(ctx) + require.NoError(t, err) + assert.EqualValues(t, 1000, settings.QueueLimitBytes) + + fresh := sha256.Sum256([]byte("new file")) + batch, res, err := up.CreateBatch(ctx, author, nil, []uploads.NewFile{ + {ClientID: "a", Name: "a.flac", Size: 400, SHA256: fresh[:]}, + {ClientID: "dup", Name: "b.flac", Size: 100, SHA256: sum[:]}, // уже в библиотеке + {ClientID: "big", Name: "c.flac", Size: 900, SHA256: []byte("cccccccccccccccccccccccccccccccc")}, + }, settings) + require.NoError(t, err) + assert.NotEqual(t, uuid.Nil, batch) + byClient := map[string]uploads.Reserved{} + for _, r := range res { + byClient[r.ClientID] = r + } + assert.Empty(t, byClient["a"].Rejected) + assert.Equal(t, uploads.CodeDuplicate, byClient["dup"].Rejected) + assert.Equal(t, uploads.CodeQueueFull, byClient["big"].Rejected) + used, userUsed, err := up.Usage(ctx, author) + require.NoError(t, err) + assert.EqualValues(t, 400, used) + assert.EqualValues(t, 400, userUsed) + + fileID := byClient["a"].FileID + require.NoError(t, up.SetReceived(ctx, fileID, 400)) + require.NoError(t, up.Transition(ctx, fileID, []uploads.Status{uploads.StatusUploading}, uploads.StatusProcessing, nil)) + next, err := up.NextWithStatus(ctx, uploads.StatusProcessing) + require.NoError(t, err) + assert.Equal(t, fileID, next.ID) + require.ErrorIs(t, up.Transition(ctx, fileID, []uploads.Status{uploads.StatusUploading}, uploads.StatusExpired, nil), + uploads.ErrWrongStatus) + + // Обработка: метаданные, отпечаток, путь; затем отклонение освобождает квоту. + require.NoError(t, up.Transition(ctx, fileID, []uploads.Status{uploads.StatusProcessing}, uploads.StatusPendingReview, + func(tx pgx.Tx) error { + _, err := tx.Exec(ctx, `UPDATE upload_files SET metadata = '{"artist":"Кино","album":"Группа крови","disc":2}', + fingerprint = '{1,2,3}', duration_ms = 200000 WHERE id = $1`, fileID) + return err + })) + f, err := up.File(ctx, fileID) + require.NoError(t, err) + assert.Equal(t, uploads.StatusPendingReview, f.Status) + require.NotNil(t, f.Meta) + assert.Equal(t, "Кино", f.Meta.Artist) + + files, nextCursor, err := up.List(ctx, uploads.ListFilter{AuthorID: &author, Statuses: []uploads.Status{uploads.StatusPendingReview}, Limit: 1}) + require.NoError(t, err) + assert.Len(t, files, 1) + assert.Empty(t, nextCursor) + + _, dup, err := up.DuplicateBySHA(ctx, sum[:], uuid.Nil) + require.NoError(t, err) + assert.True(t, dup) + cands, err := up.FingerprintCandidates(ctx, 200_000, 5000, uuid.Nil) + require.NoError(t, err) + assert.Len(t, cands, 2, "library print and the queued one") + match, err := up.MetadataMatch(ctx, meta.Key("Кино"), meta.Key("Группа крови"), 200_000, 2000) + require.NoError(t, err) + assert.NotNil(t, match) + ar, al, _, err := up.MatchCatalog(ctx, meta.Key("Кино"), meta.Key("Группа крови")) + require.NoError(t, err) + assert.NotNil(t, ar) + assert.NotNil(t, al) + discs, err := up.BatchMaxDisc(ctx, batch, "Группа крови") + require.NoError(t, err) + assert.Equal(t, 2, discs) + inUse, err := up.StagingInUse(ctx) + require.NoError(t, err) + assert.True(t, inUse[f.StagingPath]) + stale, err := up.Stale(ctx, time.Now().Add(time.Hour)) + require.NoError(t, err) + assert.Empty(t, stale, "only uploading files expire") + up.Audit(ctx, &author, "upload.test", fileID, map[string]any{"ok": true}) + + // Импорт: связь с файлом библиотеки, место освобождено. + require.NoError(t, up.Transition(ctx, fileID, []uploads.Status{uploads.StatusPendingReview}, uploads.StatusApproved, nil)) + require.NoError(t, up.Transition(ctx, fileID, []uploads.Status{uploads.StatusApproved}, uploads.StatusImporting, nil)) + require.NoError(t, up.Transition(ctx, fileID, []uploads.Status{uploads.StatusImporting}, uploads.StatusImported, + func(tx pgx.Tx) error { return uploads.MarkImported(ctx, tx, fileID, fileIDs[1]) })) + used, _, err = up.Usage(ctx, author) + require.NoError(t, err) + assert.EqualValues(t, 0, used) +} diff --git a/services/recommender/pyproject.toml b/services/recommender/pyproject.toml index 7ce2bea..1c2be52 100644 --- a/services/recommender/pyproject.toml +++ b/services/recommender/pyproject.toml @@ -41,6 +41,10 @@ select = ["E", "F", "W", "I", "B", "UP", "SIM", "RUF"] # Комментарии и докстринги на русском: «с», «у», «−» — не опечатки в латинице. ignore = ["RUF001", "RUF002", "RUF003"] +[tool.ruff.lint.per-file-ignores] +# SQL-вставки тестовых данных читаются лучше одной строкой. +"tests/test_db_integration.py" = ["E501"] + [tool.mypy] strict = true python_version = "3.12" diff --git a/services/recommender/tests/test_db_integration.py b/services/recommender/tests/test_db_integration.py new file mode 100644 index 0000000..6d7e2b9 --- /dev/null +++ b/services/recommender/tests/test_db_integration.py @@ -0,0 +1,141 @@ +"""SQL рекомендателя на настоящей схеме (CI-job `ci / db`; схему заранее накатывают миграции api). + +Локально: TEST_DATABASE_URL=postgresql://… uv run pytest tests/test_db_integration.py +""" + +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_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