Christian Krakau-Louis
ce97e71239
Add generic MCP CRUD access for mapped datastore objects and include song usage frequency in song query results.
212 lines
7.5 KiB
Python
212 lines
7.5 KiB
Python
"""Tests for agent automation services."""
|
|
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
os.environ.setdefault("SECRET_KEY", "test-secret-key-for-testing-only")
|
|
os.environ.setdefault("AUTOMATION_TOKEN", "test-automation-token-for-testing")
|
|
|
|
from musicround.models import Song, SongTag, Tag, User, db
|
|
from musicround.services import automation
|
|
|
|
|
|
def _create_user(username="agentuser", email="agent@example.com"):
|
|
user = User(username=username, email=email)
|
|
user.password = "AgentPass123!"
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
return user
|
|
|
|
|
|
def _create_song(title="Song", artist="Artist", **kwargs):
|
|
song = Song(title=title, artist=artist, **kwargs)
|
|
db.session.add(song)
|
|
db.session.commit()
|
|
return song
|
|
|
|
|
|
class TestSongAutomation:
|
|
"""Tests for catalog lookup and mutation."""
|
|
|
|
def test_find_songs_by_query(self, app):
|
|
with app.app_context():
|
|
_create_song(
|
|
title="Blue Monday",
|
|
artist="New Order",
|
|
genre="Synthpop",
|
|
used_count=3,
|
|
)
|
|
|
|
result = automation.find_songs(query="blue")
|
|
|
|
assert result["count"] == 1
|
|
assert result["songs"][0]["title"] == "Blue Monday"
|
|
assert result["songs"][0]["used_count"] == 3
|
|
assert result["songs"][0]["usage_frequency"] == 3
|
|
assert "last_used" in result["songs"][0]
|
|
|
|
def test_add_song_reuses_existing_by_isrc_and_adds_tags(self, app):
|
|
with app.app_context():
|
|
_create_song(title="Old Title", artist="Old Artist", isrc="ABC123")
|
|
|
|
result = automation.add_song(
|
|
title="New Title",
|
|
artist="New Artist",
|
|
isrc="ABC123",
|
|
tags=["warmup", "classic"],
|
|
)
|
|
|
|
assert result["created"] is False
|
|
assert Song.query.count() == 1
|
|
song = Song.query.first()
|
|
assert song.title == "New Title"
|
|
assert sorted(tag.name for tag in song.tags) == ["classic", "warmup"]
|
|
|
|
|
|
class TestRoundAutomation:
|
|
"""Tests for round creation and naming."""
|
|
|
|
def test_create_and_rename_manual_round(self, app):
|
|
with app.app_context():
|
|
song_one = _create_song(title="One", artist="A")
|
|
song_two = _create_song(title="Two", artist="B")
|
|
|
|
created = automation.create_round(
|
|
name="Initial Name",
|
|
round_type="manual",
|
|
song_ids=[song_one.id, song_two.id],
|
|
)
|
|
renamed = automation.rename_round(created["round"]["id"], "Final Name")
|
|
|
|
assert created["round"]["song_ids"] == [song_one.id, song_two.id]
|
|
assert renamed["round"]["name"] == "Final Name"
|
|
assert Song.query.get(song_one.id).used_count == 1
|
|
|
|
|
|
class TestAssetInspection:
|
|
"""Tests for generated asset quality checks."""
|
|
|
|
def test_inspect_pdf_quality(self, app):
|
|
with app.app_context(), tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp:
|
|
tmp.write(b"%PDF-1.4\n")
|
|
tmp.write(b"1 0 obj << /Type /Page >> endobj\n" * 40)
|
|
tmp.write(b"%%EOF")
|
|
tmp_path = tmp.name
|
|
|
|
try:
|
|
with app.app_context():
|
|
result = automation.inspect_pdf_quality(path=tmp_path)
|
|
|
|
assert result["ok"] is True
|
|
assert result["page_count_estimate"] > 0
|
|
finally:
|
|
os.remove(tmp_path)
|
|
|
|
def test_inspect_mp3_quality_existing_fixture(self, app):
|
|
if not shutil.which("ffprobe"):
|
|
pytest.skip("ffprobe is required for MP3 inspection")
|
|
|
|
fixture_path = os.path.abspath("musicround/mp3/intro.mp3")
|
|
|
|
with app.app_context():
|
|
result = automation.inspect_mp3_quality(path=fixture_path)
|
|
|
|
assert result["duration_seconds"] > 0
|
|
assert result["channels"] >= 1
|
|
assert "ok" in result
|
|
|
|
|
|
class TestTTSAutomation:
|
|
"""Tests for TTS snippet assignment."""
|
|
|
|
def test_generate_tts_snippet_updates_user_audio_path(self, app):
|
|
with app.app_context():
|
|
user = _create_user()
|
|
|
|
with patch("musicround.services.automation.generate_tts_mp3") as mock_tts:
|
|
mock_tts.return_value = "custommp3/agentuser/intro.mp3"
|
|
result = automation.generate_tts_snippet(
|
|
user_id=user.id,
|
|
mp3_type="intro",
|
|
text="Welcome to the quiz",
|
|
service="openai",
|
|
)
|
|
|
|
assert result["path"] == "custommp3/agentuser/intro.mp3"
|
|
assert User.query.get(user.id).intro_mp3 == "custommp3/agentuser/intro.mp3"
|
|
|
|
|
|
class TestDatastoreCrudAutomation:
|
|
"""Tests for generic datastore CRUD operations exposed through MCP."""
|
|
|
|
def test_datastore_schema_lists_mapped_models(self, app):
|
|
with app.app_context():
|
|
schema = automation.datastore_schema()
|
|
|
|
assert "song" in schema["object_types"]
|
|
assert "round" in schema["object_types"]
|
|
assert "user" in schema["object_types"]
|
|
song_schema = next(
|
|
item for item in schema["objects"] if item["object_type"] == "song"
|
|
)
|
|
assert song_schema["primary_key"] == ["id"]
|
|
assert any(column["name"] == "title" for column in song_schema["columns"])
|
|
|
|
def test_crud_lifecycle_for_single_primary_key_object(self, app):
|
|
with app.app_context():
|
|
created = automation.create_datastore_object("tag", {"name": "warmup"})
|
|
tag_id = created["object"]["id"]
|
|
|
|
listed = automation.list_datastore_objects(
|
|
"tag", filters={"name": "warmup"}, order_by="id"
|
|
)
|
|
fetched = automation.get_datastore_object("tag", tag_id)
|
|
updated = automation.update_datastore_object(
|
|
"tag", tag_id, {"name": "opener"}
|
|
)
|
|
deleted = automation.delete_datastore_object("tag", tag_id)
|
|
|
|
assert created["object_type"] == "tag"
|
|
assert listed["total"] == 1
|
|
assert fetched["object"]["name"] == "warmup"
|
|
assert updated["object"]["name"] == "opener"
|
|
assert deleted["deleted"] is True
|
|
assert Tag.query.get(tag_id) is None
|
|
|
|
def test_crud_supports_composite_primary_keys(self, app):
|
|
with app.app_context():
|
|
song = _create_song(title="Composite", artist="Key")
|
|
tag = Tag(name="linked")
|
|
db.session.add(tag)
|
|
db.session.commit()
|
|
|
|
created = automation.create_datastore_object(
|
|
"song_tag", {"song_id": song.id, "tag_id": tag.id}
|
|
)
|
|
fetched = automation.get_datastore_object(
|
|
"song_tag", {"song_id": song.id, "tag_id": tag.id}
|
|
)
|
|
deleted = automation.delete_datastore_object(
|
|
"song_tag", {"song_id": song.id, "tag_id": tag.id}
|
|
)
|
|
|
|
assert created["object"]["song_id"] == song.id
|
|
assert fetched["object"]["tag_id"] == tag.id
|
|
assert deleted["deleted"] is True
|
|
assert SongTag.query.count() == 0
|
|
|
|
def test_user_sensitive_fields_are_redacted_by_default(self, app):
|
|
with app.app_context():
|
|
user = _create_user()
|
|
user.spotify_token = "secret-token"
|
|
db.session.commit()
|
|
|
|
result = automation.get_datastore_object("user", user.id)
|
|
|
|
assert result["object"]["spotify_token"] == "[redacted]"
|
|
assert result["object"]["password_hash"] == "[redacted]"
|