Spaces:
Running
Running
| import sys | |
| import os | |
| import unittest | |
| from unittest.mock import MagicMock | |
| # Force environment variables for tests | |
| os.environ["SUPABASE_JWT_SECRET"] = "supersecretkey_change_me_in_production" | |
| os.environ["SECRET_KEY"] = "supersecretkey_change_me_in_production" | |
| os.environ["REDIS_URL"] = "rpc://" | |
| # Mock heavy dependencies | |
| sys.modules['librosa'] = MagicMock() | |
| sys.modules['torch'] = MagicMock() | |
| sys.modules['yt_dlp'] = MagicMock() | |
| sys.modules['pretty_midi'] = MagicMock() | |
| sys.modules['numpy'] = MagicMock() | |
| sys.modules['demucs'] = MagicMock() | |
| sys.modules['demucs.separate'] = MagicMock() | |
| # Include MelodixAPI in python path | |
| sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) | |
| from fastapi.testclient import TestClient | |
| from sqlalchemy import create_engine | |
| from sqlalchemy.orm import sessionmaker | |
| from app.database import Base, get_db | |
| from app.models.user import User, Device | |
| from app.models.lyrics import Group, UserGroup, Lyric, Repertorio | |
| from app.models.process import Task | |
| from main import app | |
| TEST_DATABASE_URL = "sqlite:///./test_group_perms.db" | |
| engine = create_engine(TEST_DATABASE_URL, connect_args={"check_same_thread": False}) | |
| TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) | |
| def override_get_db(): | |
| db = TestingSessionLocal() | |
| try: | |
| yield db | |
| finally: | |
| db.close() | |
| app.dependency_overrides[get_db] = override_get_db | |
| class TestGroupPermissions(unittest.TestCase): | |
| def setUpClass(cls): | |
| Base.metadata.create_all(bind=engine) | |
| cls.client = TestClient(app) | |
| def tearDownClass(cls): | |
| Base.metadata.drop_all(bind=engine) | |
| if os.path.exists("./test_group_perms.db"): | |
| try: | |
| os.remove("./test_group_perms.db") | |
| except Exception: | |
| pass | |
| def setUp(self): | |
| db = TestingSessionLocal() | |
| db.query(Task).delete() | |
| db.query(UserGroup).delete() | |
| db.query(Lyric).delete() | |
| db.query(Repertorio).delete() | |
| db.query(Group).delete() | |
| db.query(Device).delete() | |
| db.query(User).delete() | |
| db.commit() | |
| db.close() | |
| def _create_user_and_token(self, username, email): | |
| payload = {"username": username, "email": email, "password": "password123"} | |
| self.client.post("/auth/register", json=payload) | |
| login_res = self.client.post("/auth/token", data={ | |
| "username": email, | |
| "password": "password123", | |
| "device_id": f"dev_{username}", | |
| "device_name": "Test Device" | |
| }) | |
| token = login_res.json()["access_token"] | |
| user_id = login_res.json()["user"]["id"] | |
| return {"Authorization": f"Bearer {token}"}, user_id | |
| def test_lyrics_sharing_permissions(self): | |
| """Test sharing lyrics based on perm_compartir_letras granular permission""" | |
| headers_a, id_a = self._create_user_and_token("user_a", "a@example.com") | |
| headers_b, id_b = self._create_user_and_token("user_b", "b@example.com") | |
| # User A creates group | |
| g_res = self.client.post("/lyrics/groups", json={"name": "Grupo A"}, headers=headers_a) | |
| self.assertEqual(g_res.status_code, 200) | |
| group_id = g_res.json()["id"] | |
| # Generate invitation and join B | |
| inv_res = self.client.post(f"/lyrics/groups/{group_id}/invite", headers=headers_a) | |
| code = inv_res.json()["code"] | |
| self.client.post(f"/lyrics/join/{code}", headers=headers_b) | |
| # User B creates a lyric | |
| l_res = self.client.post("/lyrics", json={ | |
| "title": "Lyric B", "artist": "Artist B", "content": "Content B", "group_id": None | |
| }, headers=headers_b) | |
| self.assertEqual(l_res.status_code, 200) | |
| lyric_id = l_res.json()["id"] | |
| # By default B can share | |
| share_res = self.client.post(f"/lyrics/{lyric_id}/share", json={"group_id": group_id, "action": "share"}, headers=headers_b) | |
| self.assertEqual(share_res.status_code, 200) | |
| # Unshare | |
| unshare_res = self.client.post(f"/lyrics/{lyric_id}/share", json={"group_id": group_id, "action": "unshare"}, headers=headers_b) | |
| self.assertEqual(unshare_res.status_code, 200) | |
| # Disable perm_compartir_letras for User B | |
| requests_json = { | |
| "perm_compartir_pistas": True, | |
| "perm_compartir_letras": False, # Disabled | |
| "perm_crud_repertorios": False, | |
| "perm_crud_letras": True, | |
| "perm_ver_pistas": True, | |
| "perm_ver_letras": True, | |
| "perm_ver_repertorios": True, | |
| "perm_gestionar_solicitudes": False | |
| } | |
| update_perm = self.client.put(f"/alabanza/grupos/{group_id}/miembros/{id_b}/permisos", json=requests_json, headers=headers_a) | |
| self.assertEqual(update_perm.status_code, 200) | |
| # User B tries to share -> 403 Forbidden | |
| share_res = self.client.post(f"/lyrics/{lyric_id}/share", json={"group_id": group_id, "action": "share"}, headers=headers_b) | |
| self.assertEqual(share_res.status_code, 403) | |
| def test_lyrics_view_permissions(self): | |
| """Test viewing lyrics in group and get_lyric based on perm_ver_letras""" | |
| headers_a, id_a = self._create_user_and_token("user_a", "a@example.com") | |
| headers_b, id_b = self._create_user_and_token("user_b", "b@example.com") | |
| # User A creates group | |
| g_res = self.client.post("/lyrics/groups", json={"name": "Grupo A"}, headers=headers_a) | |
| group_id = g_res.json()["id"] | |
| # Join B | |
| inv_res = self.client.post(f"/lyrics/groups/{group_id}/invite", headers=headers_a) | |
| code = inv_res.json()["code"] | |
| self.client.post(f"/lyrics/join/{code}", headers=headers_b) | |
| # User A creates and shares a lyric | |
| l_res = self.client.post("/lyrics", json={ | |
| "title": "Lyric A", "artist": "Artist A", "content": "Content A", "group_id": group_id | |
| }, headers=headers_a) | |
| lyric_id = l_res.json()["id"] | |
| # User B can view by default | |
| get_res = self.client.get(f"/lyrics/{lyric_id}", headers=headers_b) | |
| self.assertEqual(get_res.status_code, 200) | |
| list_res = self.client.get(f"/alabanza/grupos/{group_id}/letras", headers=headers_b) | |
| self.assertEqual(list_res.status_code, 200) | |
| self.assertEqual(len(list_res.json()), 1) | |
| # Disable perm_ver_letras | |
| requests_json = { | |
| "perm_compartir_pistas": True, | |
| "perm_compartir_letras": True, | |
| "perm_crud_repertorios": False, | |
| "perm_crud_letras": True, | |
| "perm_ver_pistas": True, | |
| "perm_ver_letras": False, # Disabled | |
| "perm_ver_repertorios": True, | |
| "perm_gestionar_solicitudes": False | |
| } | |
| update_perm = self.client.put(f"/alabanza/grupos/{group_id}/miembros/{id_b}/permisos", json=requests_json, headers=headers_a) | |
| self.assertEqual(update_perm.status_code, 200) | |
| # User B tries to view group lyrics -> 403 | |
| list_res = self.client.get(f"/alabanza/grupos/{group_id}/letras", headers=headers_b) | |
| self.assertEqual(list_res.status_code, 403) | |
| # User B tries to view individual lyric -> 403 | |
| get_res = self.client.get(f"/lyrics/{lyric_id}", headers=headers_b) | |
| self.assertEqual(get_res.status_code, 403) | |
| def test_repertoire_crud_permissions(self): | |
| """Test creating, editing, and deleting repertoires based on perm_crud_repertorios""" | |
| headers_a, id_a = self._create_user_and_token("user_a", "a@example.com") | |
| headers_b, id_b = self._create_user_and_token("user_b", "b@example.com") | |
| # User A creates group | |
| g_res = self.client.post("/lyrics/groups", json={"name": "Grupo A"}, headers=headers_a) | |
| group_id = g_res.json()["id"] | |
| # Join B | |
| inv_res = self.client.post(f"/lyrics/groups/{group_id}/invite", headers=headers_a) | |
| code = inv_res.json()["code"] | |
| self.client.post(f"/lyrics/join/{code}", headers=headers_b) | |
| # User A creates a repertoire | |
| rep_res = self.client.post("/lyrics/repertorios", json={ | |
| "name": "Rep Admin", "group_id": group_id, "lyrics_ids": [] | |
| }, headers=headers_a) | |
| rep_id = rep_res.json()["id"] | |
| # User B has perm_crud_repertorios = False by default -> should get 403 on editing | |
| edit_res = self.client.put(f"/lyrics/repertorios/{rep_id}", json={ | |
| "name": "Rep Mod", "group_id": group_id, "lyrics_ids": [] | |
| }, headers=headers_b) | |
| self.assertEqual(edit_res.status_code, 403) | |
| # Enable perm_crud_repertorios | |
| requests_json = { | |
| "perm_compartir_pistas": True, | |
| "perm_compartir_letras": True, | |
| "perm_crud_repertorios": True, # Enabled! | |
| "perm_crud_letras": True, | |
| "perm_ver_pistas": True, | |
| "perm_ver_letras": True, | |
| "perm_ver_repertorios": True, | |
| "perm_gestionar_solicitudes": False | |
| } | |
| self.client.put(f"/alabanza/grupos/{group_id}/miembros/{id_b}/permisos", json=requests_json, headers=headers_a) | |
| # User B edits -> 200 OK | |
| edit_res = self.client.put(f"/lyrics/repertorios/{rep_id}", json={ | |
| "name": "Rep Mod", "group_id": group_id, "lyrics_ids": [] | |
| }, headers=headers_b) | |
| self.assertEqual(edit_res.status_code, 200) | |
| def test_repertoire_view_permissions(self): | |
| """Test viewing repertoires based on perm_ver_repertorios""" | |
| headers_a, id_a = self._create_user_and_token("user_a", "a@example.com") | |
| headers_b, id_b = self._create_user_and_token("user_b", "b@example.com") | |
| # User A creates group | |
| g_res = self.client.post("/lyrics/groups", json={"name": "Grupo A"}, headers=headers_a) | |
| group_id = g_res.json()["id"] | |
| # Join B | |
| inv_res = self.client.post(f"/lyrics/groups/{group_id}/invite", headers=headers_a) | |
| code = inv_res.json()["code"] | |
| self.client.post(f"/lyrics/join/{code}", headers=headers_b) | |
| # User A creates a repertoire | |
| rep_res = self.client.post("/lyrics/repertorios", json={ | |
| "name": "Rep Admin", "group_id": group_id, "lyrics_ids": [] | |
| }, headers=headers_a) | |
| rep_id = rep_res.json()["id"] | |
| # User B can view by default | |
| get_res = self.client.get(f"/lyrics/repertorios/{rep_id}", headers=headers_b) | |
| self.assertEqual(get_res.status_code, 200) | |
| list_res = self.client.get(f"/alabanza/grupos/{group_id}/repertorios", headers=headers_b) | |
| self.assertEqual(list_res.status_code, 200) | |
| # Disable perm_ver_repertorios | |
| requests_json = { | |
| "perm_compartir_pistas": True, | |
| "perm_compartir_letras": True, | |
| "perm_crud_repertorios": True, | |
| "perm_crud_letras": True, | |
| "perm_ver_pistas": True, | |
| "perm_ver_letras": True, | |
| "perm_ver_repertorios": False, # Disabled | |
| "perm_gestionar_solicitudes": False | |
| } | |
| self.client.put(f"/alabanza/grupos/{group_id}/miembros/{id_b}/permisos", json=requests_json, headers=headers_a) | |
| # User B tries to view group repertoires -> 403 | |
| list_res = self.client.get(f"/alabanza/grupos/{group_id}/repertorios", headers=headers_b) | |
| self.assertEqual(list_res.status_code, 403) | |
| # User B tries to view individual repertoire -> 403 | |
| get_res = self.client.get(f"/lyrics/repertorios/{rep_id}", headers=headers_b) | |
| self.assertEqual(get_res.status_code, 403) | |
| def test_songs_view_permissions(self): | |
| """Test viewing tracks/songs and get_stems/harmony_status based on perm_ver_pistas""" | |
| headers_a, id_a = self._create_user_and_token("user_a", "a@example.com") | |
| headers_b, id_b = self._create_user_and_token("user_b", "b@example.com") | |
| # User A creates group | |
| g_res = self.client.post("/lyrics/groups", json={"name": "Grupo A"}, headers=headers_a) | |
| group_id = g_res.json()["id"] | |
| # Join B | |
| inv_res = self.client.post(f"/lyrics/groups/{group_id}/invite", headers=headers_a) | |
| code = inv_res.json()["code"] | |
| self.client.post(f"/lyrics/join/{code}", headers=headers_b) | |
| # Seed a Task in DB directly | |
| db = TestingSessionLocal() | |
| task = Task( | |
| task_id="task_test_123", | |
| user_id=id_a, | |
| song_name="Song A", | |
| status="success", | |
| telegram_file_ids='{"vocals": "stems/task_test_123/vocals.flac"}' | |
| ) | |
| db.add(task) | |
| db.commit() | |
| # Share it | |
| db_group = db.query(Group).filter(Group.id == group_id).first() | |
| task.groups_shared.append(db_group) | |
| db.commit() | |
| db.close() | |
| # User B can view by default | |
| get_res = self.client.get("/songs/tasks/task_test_123", headers=headers_b) | |
| self.assertEqual(get_res.status_code, 200) | |
| stems_res = self.client.get("/songs/stems/task_test_123", headers=headers_b) | |
| self.assertEqual(stems_res.status_code, 200) | |
| harm_res = self.client.get("/songs/harmony-status/task_test_123", headers=headers_b) | |
| self.assertEqual(harm_res.status_code, 200) | |
| list_res = self.client.get(f"/alabanza/grupos/{group_id}/pistas", headers=headers_b) | |
| self.assertEqual(list_res.status_code, 200) | |
| # Disable perm_ver_pistas | |
| requests_json = { | |
| "perm_compartir_pistas": True, | |
| "perm_compartir_letras": True, | |
| "perm_crud_repertorios": True, | |
| "perm_crud_letras": True, | |
| "perm_ver_pistas": False, # Disabled | |
| "perm_ver_letras": True, | |
| "perm_ver_repertorios": True, | |
| "perm_gestionar_solicitudes": False | |
| } | |
| self.client.put(f"/alabanza/grupos/{group_id}/miembros/{id_b}/permisos", json=requests_json, headers=headers_a) | |
| # User B tries to view group tracks -> 403 | |
| list_res = self.client.get(f"/alabanza/grupos/{group_id}/pistas", headers=headers_b) | |
| self.assertEqual(list_res.status_code, 403) | |
| # User B tries to view individual track -> 403 | |
| get_res = self.client.get("/songs/tasks/task_test_123", headers=headers_b) | |
| self.assertEqual(get_res.status_code, 403) | |
| # User B tries to view stems -> 403 | |
| stems_res = self.client.get("/songs/stems/task_test_123", headers=headers_b) | |
| self.assertEqual(stems_res.status_code, 403) | |
| # User B tries to check harmony status -> 403 | |
| harm_res = self.client.get("/songs/harmony-status/task_test_123", headers=headers_b) | |
| self.assertEqual(harm_res.status_code, 403) | |
| if __name__ == "__main__": | |
| unittest.main() | |