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): @classmethod def setUpClass(cls): Base.metadata.create_all(bind=engine) cls.client = TestClient(app) @classmethod 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()