File size: 4,955 Bytes
330f477
 
bff0fe1
330f477
 
 
bff0fe1
330f477
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
bff0fe1
 
 
330f477
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
bff0fe1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
from __future__ import annotations

import numpy as np
from fastapi.testclient import TestClient

from openmusic_analysis.api import create_app
from openmusic_analysis.settings import Settings, TemporalAudioConfig

from .conftest import FailingAudioEncoder, make_service


def client_for(service) -> TestClient:
    return TestClient(create_app(service=service, settings=Settings()))


def upload(content: bytes = b"valid fake audio"):
    return {"audio": ("track.mp3", content, "audio/mpeg")}


def test_valid_audio_only_request(service_bundle):
    service, _, _ = service_bundle
    response = client_for(service).post("/v1/tracks/analyze", files=upload())
    assert response.status_code == 200
    assert set(response.json()["representations"]) == {"audio.global", "audio.temporal"}


def test_audio_and_multilingual_lyrics(service_bundle):
    service, _, _ = service_bundle
    response = client_for(service).post(
        "/v1/tracks/analyze",
        files=upload(),
        data={"lyrics": "[Verse]\nHello world\n\n[Припев]\nПривет, мир!"},
    )
    assert response.status_code == 200
    assert "lyrics.global" in response.json()["representations"]


def test_requested_representations_avoid_unrequested_work(service_bundle):
    service, decoder, encoder = service_bundle
    response = client_for(service).post(
        "/v1/tracks/analyze",
        files=upload(),
        data={
            "lyrics": "lyrics that must not be encoded",
            "requested_representations": "audio.global",
        },
    )
    assert response.status_code == 200
    assert list(response.json()["representations"]) == ["audio.global"]
    assert decoder.calls == 1
    assert len(encoder.calls) == 1


def test_invalid_file_returns_structured_decode_error(service_bundle):
    service, _, _ = service_bundle
    response = client_for(service).post(
        "/v1/tracks/analyze", files=upload(b"bad bytes")
    )
    assert response.status_code == 422
    assert response.json()["error"]["code"] == "AUDIO_DECODE_FAILED"


def test_missing_audio_is_structured(service_bundle):
    service, _, _ = service_bundle
    response = client_for(service).post("/v1/tracks/analyze", data={"lyrics": "text"})
    assert response.status_code == 422
    assert response.json()["error"]["code"] == "MISSING_AUDIO"


def test_unsupported_representation(service_bundle):
    service, _, _ = service_bundle
    response = client_for(service).post(
        "/v1/tracks/analyze",
        files=upload(),
        data={"requested_representations": "audio.emotion"},
    )
    assert response.status_code == 422
    assert response.json()["error"]["code"] == "UNSUPPORTED_REPRESENTATION"


def test_unsupported_extension(service_bundle):
    service, _, _ = service_bundle
    response = client_for(service).post(
        "/v1/tracks/analyze", files={"audio": ("track.txt", b"data", "text/plain")}
    )
    assert response.status_code == 415
    assert response.json()["error"]["code"] == "UNSUPPORTED_AUDIO_FORMAT"


def test_explicit_lyrics_representation_requires_lyrics(service_bundle):
    service, _, _ = service_bundle
    response = client_for(service).post(
        "/v1/tracks/analyze",
        files=upload(),
        data={"requested_representations": "lyrics.global"},
    )
    assert response.status_code == 422
    assert response.json()["error"]["code"] == "LYRICS_REQUIRED"


def test_invalid_empty_lyrics(service_bundle):
    service, _, _ = service_bundle
    response = client_for(service).post(
        "/v1/tracks/analyze", files=upload(), data={"lyrics": "   \n "}
    )
    assert response.status_code == 422
    assert response.json()["error"]["code"] == "INVALID_LYRICS"


def test_model_failure_does_not_leak_internal_exception():
    service, _, _ = make_service(audio_encoder=FailingAudioEncoder())
    response = client_for(service).post(
        "/v1/tracks/analyze",
        files=upload(),
        data={"requested_representations": "audio.global"},
    )
    assert response.status_code == 500
    assert response.json()["error"]["code"] == "MODEL_INFERENCE_FAILED"
    assert "internal model detail" not in response.text


def test_temporal_api_response_uses_adjacent_transition_index():
    service, _, _ = make_service(waveform=np.arange(150, dtype=np.float32))
    service.registry.analyzer("audio.temporal").config = TemporalAudioConfig(
        sample_rate=10,
        window_seconds=1,
        hop_seconds=1,
        max_segments=15,
        minimum_audio_seconds=1,
    )
    response = client_for(service).post(
        "/v1/tracks/analyze",
        files=upload(),
        data={"requested_representations": "audio.temporal"},
    )
    assert response.status_code == 200
    temporal = response.json()["representations"]["audio.temporal"]
    assert len(temporal["segments"]) == 15
    assert temporal["summary"]["number_of_segments"] == 15
    assert 0 <= temporal["summary"]["largest_transition_index"] <= 13