# PLAY CHANNELS API regression tests (auth, content, favorites, history, vpn, admin CRUD/config) import os import uuid import pytest import requests from pathlib import Path def _base(): url = os.environ.get("EXPO_PUBLIC_BACKEND_URL") or os.environ.get("EXPO_BACKEND_URL") if not url: for line in (Path("/app/frontend/.env").read_text().splitlines()): if line.startswith("EXPO_PUBLIC_BACKEND_URL="): url = line.split("=", 1)[1].strip().strip('"') return url.rstrip("/") BASE_URL = _base() API = f"{BASE_URL}/api" def login(path, u, p): return requests.post(f"{API}{path}", json={"username": u, "password": p}, timeout=20) @pytest.fixture(scope="module") def user_h(): r = login("/auth/login", "teste", "teste123") if r.status_code != 200: pytest.skip("user login failed") return {"Authorization": f"Bearer {r.json()['token']}"} @pytest.fixture(scope="module") def admin_h(): r = login("/admin/login", "admin", "admin123") if r.status_code != 200: pytest.skip("admin login failed") return {"Authorization": f"Bearer {r.json()['token']}"} # ---- Auth ---- class TestAuth: def test_user_login(self): r = login("/auth/login", "teste", "teste123") assert r.status_code == 200 d = r.json() assert d["user"]["username"] == "teste" and d["user"]["role"] == "user" assert "password_hash" not in d["user"] and d["token"] def test_user_login_wrong(self): r = login("/auth/login", "teste", "wrong") assert r.status_code == 401 def test_admin_login(self): r = login("/admin/login", "admin", "admin123") assert r.status_code == 200 and r.json()["user"]["role"] == "admin" def test_user_on_admin_login_403(self): assert login("/admin/login", "teste", "teste123").status_code == 403 def test_me_and_no_token(self, user_h): assert requests.get(f"{API}/auth/me", headers=user_h).json()["username"] == "teste" assert requests.get(f"{API}/home").status_code == 401 # ---- Content ---- class TestContent: def test_home(self, user_h): d = requests.get(f"{API}/home", headers=user_h).json() assert len(d["banners"]) > 0 and len(d["rows"]) > 0 assert len(d["featured_movies"]) > 0 and len(d["featured_series"]) > 0 def test_content_types(self, user_h): mv = requests.get(f"{API}/content?type=movie", headers=user_h).json() sr = requests.get(f"{API}/content?type=series", headers=user_h).json() assert mv and all(x["type"] == "movie" for x in mv) assert sr and all(x["type"] == "series" for x in sr) assert sr[0].get("seasons") and sr[0]["seasons"][0]["episodes"] def test_content_detail_and_404(self, user_h): mv = requests.get(f"{API}/content?type=movie", headers=user_h).json() r = requests.get(f"{API}/content/{mv[0]['id']}", headers=user_h) assert r.status_code == 200 and "is_favorite" in r.json() and "_id" not in r.json() assert requests.get(f"{API}/content/nonexistent", headers=user_h).status_code == 404 def test_channels_categories(self, user_h): ch = requests.get(f"{API}/channels", headers=user_h).json() cats = requests.get(f"{API}/categories", headers=user_h).json() assert ch and ch[0]["stream_url"] and cats and cats[0]["name"] # ---- Favorites & history ---- class TestFavHistory: def test_favorites_flow(self, user_h): cid = requests.get(f"{API}/content?type=movie", headers=user_h).json()[0]["id"] assert requests.post(f"{API}/favorites/{cid}", headers=user_h).json()["is_favorite"] is True favs = requests.get(f"{API}/favorites", headers=user_h).json() assert any(f["id"] == cid for f in favs) assert requests.get(f"{API}/content/{cid}", headers=user_h).json()["is_favorite"] is True requests.delete(f"{API}/favorites/{cid}", headers=user_h) favs = requests.get(f"{API}/favorites", headers=user_h).json() assert not any(f["id"] == cid for f in favs) def test_history_flow(self, user_h): body = {"content_id": "TEST_c1", "title": "TEST", "stream_url": "http://x", "position": 12, "duration": 100} assert requests.post(f"{API}/history", json=body, headers=user_h).status_code == 200 body["position"] = 30 requests.post(f"{API}/history", json=body, headers=user_h) h = [x for x in requests.get(f"{API}/history", headers=user_h).json() if x["content_id"] == "TEST_c1"] assert len(h) == 1 and h[0]["position"] == 30 assert requests.delete(f"{API}/history", headers=user_h).status_code == 200 assert requests.get(f"{API}/history", headers=user_h).json() == [] # ---- VPN ---- class TestVpn: def test_servers_no_config(self, user_h): s = requests.get(f"{API}/vpn/servers", headers=user_h).json() assert s and all("config" not in x for x in s) def test_config_409(self, user_h): s = requests.get(f"{API}/vpn/servers", headers=user_h).json() r = requests.get(f"{API}/vpn/servers/{s[0]['id']}/config", headers=user_h) assert r.status_code == 409 assert requests.get(f"{API}/vpn/servers/bad/config", headers=user_h).status_code == 404 # ---- Admin ---- SAMPLES = { "banners": {"title": "TEST_banner", "image_url": "http://x/i.jpg", "order": 99}, "categories": {"name": "TEST_cat", "order": 99}, "content": {"type": "movie", "title": "TEST_movie", "stream_url": "http://x/v.mp4", "category_ids": []}, "channels": {"name": "TEST_ch", "number": 999, "stream_url": "http://x/l.m3u8"}, "vpn_servers": {"name": "TEST_vpn", "country": "BR", "protocol": "wireguard", "config": ""}, } class TestAdmin: def test_non_admin_403(self, user_h): assert requests.get(f"{API}/admin/categories", headers=user_h).status_code == 403 assert requests.put(f"{API}/admin/config", json={"x": 1}, headers=user_h).status_code == 403 @pytest.mark.parametrize("col", list(SAMPLES)) def test_crud(self, admin_h, col): r = requests.post(f"{API}/admin/{col}", json=SAMPLES[col], headers=admin_h) assert r.status_code == 200 iid = r.json()["id"] assert any(x["id"] == iid for x in requests.get(f"{API}/admin/{col}", headers=admin_h).json()) u = requests.put(f"{API}/admin/{col}/{iid}", json={"order": 5, "active": False}, headers=admin_h) assert u.status_code == 200 and u.json()["active"] is False assert requests.delete(f"{API}/admin/{col}/{iid}", headers=admin_h).status_code == 200 assert not any(x["id"] == iid for x in requests.get(f"{API}/admin/{col}", headers=admin_h).json()) assert requests.delete(f"{API}/admin/{col}/{iid}", headers=admin_h).status_code == 404 def test_invalid_col(self, admin_h): assert requests.get(f"{API}/admin/bogus", headers=admin_h).status_code == 404 def test_users_crud(self, admin_h): un = f"test_{uuid.uuid4().hex[:6]}" r = requests.post(f"{API}/admin/users", json={"username": un, "password": "secret1", "name": "T"}, headers=admin_h) assert r.status_code == 200 and "password_hash" not in r.json() uid = r.json()["id"] assert requests.post(f"{API}/admin/users", json={"username": un, "password": "secret1"}, headers=admin_h).status_code == 409 assert login("/auth/login", un, "secret1").status_code == 200 assert requests.put(f"{API}/admin/users/{uid}", json={"password": "newpass1"}, headers=admin_h).status_code == 200 assert login("/auth/login", un, "secret1").status_code == 401 assert login("/auth/login", un, "newpass1").status_code == 200 requests.put(f"{API}/admin/users/{uid}", json={"active": False}, headers=admin_h) assert login("/auth/login", un, "newpass1").status_code == 403 assert requests.delete(f"{API}/admin/users/{uid}", headers=admin_h).status_code == 200 assert login("/auth/login", un, "newpass1").status_code == 401 def test_config_get_put(self, admin_h): orig = requests.get(f"{API}/admin/config", headers=admin_h).json() assert orig.get("app_name") r = requests.put(f"{API}/admin/config", json={"app_name": "TEST_NAME"}, headers=admin_h) assert r.status_code == 200 and r.json()["app_name"] == "TEST_NAME" assert requests.get(f"{API}/config").json()["app_name"] == "TEST_NAME" requests.put(f"{API}/admin/config", json={"app_name": orig["app_name"]}, headers=admin_h) assert requests.get(f"{API}/config").json()["app_name"] == orig["app_name"] def test_empty_update_body(self, admin_h): # empty body update should not 500 cats = requests.get(f"{API}/admin/categories", headers=admin_h).json() r = requests.put(f"{API}/admin/categories/{cats[0]['id']}", json={}, headers=admin_h) assert r.status_code != 500