import pytest import os import sys import asyncio import logging from dotenv import load_dotenv # Dodanie katalogu nadrzędnego do ścieżki dla importów sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) from src.youtube_utils import ( extract_youtube_urls, extract_video_id, get_transcript, NoTranscriptFound, TranscriptsDisabled, APITokenMissing, AuthorizationError, APIConnectionError, APIResponseError, NoTranscriptLanguagesAvailable, YouTubeUtilsError ) from src.openai_utils import summarize_text, check_openai_api_status, QuotaExceededError # Konfiguracja logowania dla testów logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # Ładowanie zmiennych środowiskowych z pliku .env (jeśli istnieje) load_dotenv() # Zmiana na bardziej popularne wideo, które powinno być bardziej stabilne # "What makes a good life? Lessons from the longest study on happiness | Robert Waldinger" # TED Talk - bardzo popularne wystąpienie z napisami w wielu językach (>42M wyświetleń) TEST_VIDEO_URL = "https://www.youtube.com/watch?v=8KkKuTCFvzI" TEST_VIDEO_ID = extract_video_id(TEST_VIDEO_URL) TEST_LANGUAGES = ["pl", "en"] # Preferujemy polską transkrypcję, potem angielską # Drugi film zapasowy (na wypadek problemów z pierwszym) BACKUP_VIDEO_URL = "https://www.youtube.com/watch?v=jNQXAC9IVRw" # Me at the zoo - pierwszy film na YouTube BACKUP_VIDEO_ID = extract_video_id(BACKUP_VIDEO_URL) # Flaga określająca dostępność API OpenAI, inicjalizowana jako None (sprawdzana przed testami) OPENAI_API_AVAILABLE = None OPENAI_STATUS_INFO = {} # Oznaczamy testy jako "slow", żeby można było je pominąć za pomocą --skip-slow # lub --skip-integration podczas uruchamiania testów pytestmark = [pytest.mark.integration, pytest.mark.slow] # Helper do sprawdzania dostępności API OpenAI przed testami async def ensure_openai_api_status(): """Sprawdza status API OpenAI i ustawia globalną flagę OPENAI_API_AVAILABLE""" global OPENAI_API_AVAILABLE, OPENAI_STATUS_INFO # Jeśli nie ma klucza API, nie ma sensu sprawdzać if not os.environ.get("OPENAI_API_KEY"): logger.warning("Brak klucza API OpenAI, pomijam sprawdzanie statusu API") OPENAI_API_AVAILABLE = False OPENAI_STATUS_INFO = {"error": "Brak klucza API"} return False # Jeśli już sprawdziliśmy, używamy zapisanego wyniku if OPENAI_API_AVAILABLE is not None: return OPENAI_API_AVAILABLE try: logger.info("Sprawdzam dostępność API OpenAI...") OPENAI_API_AVAILABLE, OPENAI_STATUS_INFO = await check_openai_api_status() logger.info(f"Status API OpenAI: {'dostępne' if OPENAI_API_AVAILABLE else 'niedostępne'}") if not OPENAI_API_AVAILABLE: logger.warning(f"API OpenAI niedostępne: {OPENAI_STATUS_INFO.get('error', 'Nieznany błąd')}") return OPENAI_API_AVAILABLE except QuotaExceededError as e: logger.warning(f"Przekroczono limit zapytań API OpenAI: {e}") OPENAI_API_AVAILABLE = False OPENAI_STATUS_INFO = {"error": str(e)} return False except Exception as e: logger.error(f"Błąd podczas sprawdzania statusu API OpenAI: {e}", exc_info=True) OPENAI_API_AVAILABLE = False OPENAI_STATUS_INFO = {"error": str(e)} return False @pytest.mark.asyncio async def test_real_extract_youtube_urls(): # Test na rzeczywistym tekście z linkami YouTube text = """Sprawdź te filmy: - https://www.youtube.com/watch?v=8KkKuTCFvzI - https://youtu.be/iCvmsMzlF7o - Również shorts: https://www.youtube.com/shorts/pFaEGmxQFnM""" urls = extract_youtube_urls(text) assert len(urls) == 3 assert "https://www.youtube.com/watch?v=8KkKuTCFvzI" in urls assert "https://youtu.be/iCvmsMzlF7o" in urls assert "https://www.youtube.com/shorts/pFaEGmxQFnM" in urls @pytest.mark.asyncio async def test_real_extract_video_id(): # Test na rzeczywistych URL-ach assert extract_video_id("https://www.youtube.com/watch?v=8KkKuTCFvzI") == "8KkKuTCFvzI" assert extract_video_id("https://youtu.be/iCvmsMzlF7o") == "iCvmsMzlF7o" assert extract_video_id("https://www.youtube.com/shorts/pFaEGmxQFnM") == "pFaEGmxQFnM" @pytest.mark.asyncio async def test_real_get_transcript_and_title(): """Test pobierania rzeczywistej transkrypcji i tytułu""" # Ten test wymaga dostępu do internetu i tokenu API youtube-transcript.io # Sprawdź czy mamy token API from src.config import YOUTUBE_TRANSCRIPT_API_TOKEN if not YOUTUBE_TRANSCRIPT_API_TOKEN: pytest.skip("Brak tokenu API youtube-transcript.io") # Wypróbuj główne wideo logger.info(f"Próbuję pobrać transkrypcję dla głównego wideo: {TEST_VIDEO_ID}") try: # Spróbuj z pierwszym filmem transcript, title = await get_transcript(TEST_VIDEO_ID, TEST_LANGUAGES) assert len(transcript) > 100 # Upewniamy się, że transkrypcja ma sensowną długość assert len(title) > 0 # Tytuł powinien być niepusty # Wypiszmy tytuł i fragment transkrypcji do weryfikacji logger.info(f"Tytuł filmu {TEST_VIDEO_ID}: {title}") logger.info(f"Fragment transkrypcji: {transcript[:50]}...") except (NoTranscriptFound, TranscriptsDisabled) as e: logger.warning(f"Nie udało się pobrać transkrypcji dla {TEST_VIDEO_ID}: {e}") logger.info(f"Próbuję film zapasowy: {BACKUP_VIDEO_ID}") # Spróbuj z zapasowym filmem try: transcript, title = await get_transcript(BACKUP_VIDEO_ID, TEST_LANGUAGES) assert len(transcript) > 10 # Film zapasowy "Me at the zoo" ma bardzo krótką transkrypcję assert len(title) > 0 # Wypiszmy tytuł i fragment transkrypcji do weryfikacji logger.info(f"Tytuł filmu zapasowego {BACKUP_VIDEO_ID}: {title}") logger.info(f"Fragment transkrypcji zapasowej: {transcript[:50]}...") except (NoTranscriptFound, TranscriptsDisabled) as e: logger.error(f"Nie można pobrać transkrypcji dla żadnego z testowych filmów: {e}") pytest.skip(f"Nie można pobrać transkrypcji dla żadnego z testowych filmów: {e}") except (APITokenMissing, AuthorizationError) as e: logger.error(f"Błąd autoryzacji: {e}") pytest.skip(f"Błąd autoryzacji API: {e}") except (APIConnectionError, APIResponseError) as e: logger.error(f"Błąd API: {e}") pytest.skip(f"Błąd API: {e}") except Exception as e: logger.error(f"Nieoczekiwany błąd podczas pobierania transkrypcji: {e}", exc_info=True) pytest.skip(f"Nieoczekiwany błąd podczas pobierania transkrypcji: {e}") @pytest.mark.asyncio async def test_real_summarize_text(): # Test rzeczywistego streszczania tekstu z OpenAI API # Ten test wymaga klucza API OpenAI i dostępnego API # Sprawdź czy API OpenAI jest dostępne if not await ensure_openai_api_status(): pytest.skip(f"API OpenAI niedostępne: {OPENAI_STATUS_INFO.get('error', 'Nieznany błąd')}") # Użyjmy krótkiej transkrypcji dla oszczędności tokenów short_transcript = """ Chciałbym porozmawiać o tym, co sprawia, że życie jest dobre i wartościowe. Przeprowadziliśmy jedno z najdłuższych badań nad szczęściem - trwające ponad 75 lat. Badaliśmy życie tych samych osób od czasu, gdy byli nastolatkami, aż do starości. Wniosek? Dobre relacje z innymi ludźmi są kluczem do szczęścia i zdrowia. Nie pieniądze, nie sława, nie ciężka praca, ale jakość naszych relacji z bliskimi. Osoby, które miały dobre relacje z rodziną, przyjaciółmi i społecznością, były szczęśliwsze, zdrowsze i żyły dłużej. """ logger.info("Próbuję streszczenie transkrypcji za pomocą OpenAI API") try: summary = await summarize_text(short_transcript) assert summary is not None assert len(summary) > 0 logger.info(f"Wygenerowane streszczenie: {summary}") except QuotaExceededError as e: logger.warning(f"Przekroczono limit zapytań API OpenAI: {e}") pytest.skip(f"Przekroczono limit zapytań API OpenAI: {e}") except Exception as e: logger.error(f"Problem z API OpenAI: {e}", exc_info=True) pytest.skip(f"Problem z API OpenAI: {e}") @pytest.mark.asyncio async def test_end_to_end_integration(): """ Ten test wykonuje pełną integrację od linku YouTube do streszczenia. Wymaga dostępu do internetu, tokenu API youtube-transcript.io oraz klucza API OpenAI. """ # Sprawdź czy mamy token API youtube-transcript.io from src.config import YOUTUBE_TRANSCRIPT_API_TOKEN if not YOUTUBE_TRANSCRIPT_API_TOKEN: pytest.skip("Brak tokenu API youtube-transcript.io") # Sprawdź czy API OpenAI jest dostępne if not await ensure_openai_api_status(): pytest.skip(f"API OpenAI niedostępne: {OPENAI_STATUS_INFO.get('error', 'Nieznany błąd')}") logger.info("Rozpoczynam test integracyjny end-to-end") # Testujemy na klasycznym filmie "Me at the zoo", który jest krótki i ma dostępne transkrypcje text_with_url = f"Sprawdź ten film: {BACKUP_VIDEO_URL}" # 1. Wydobycie URL urls = extract_youtube_urls(text_with_url) assert len(urls) == 1 url = urls[0] logger.info(f"Znaleziono URL: {url}") # 2. Wydobycie ID filmu video_id = extract_video_id(url) assert video_id == BACKUP_VIDEO_ID logger.info(f"ID filmu: {video_id}") # 3. Pobranie transkrypcji i tytułu try: logger.info(f"Próbuję pobrać transkrypcję dla {video_id}") transcript, title = await get_transcript(video_id, TEST_LANGUAGES) logger.info(f"Pobrano tytuł filmu: {title}") logger.info(f"Długość transkrypcji: {len(transcript)} znaków") logger.info(f"Fragment transkrypcji: {transcript[:150]}...") # Weryfikacja tytułu dla "Me at the zoo" assert "Me at the zoo" in title except (NoTranscriptFound, TranscriptsDisabled, APIConnectionError, APIResponseError) as e: logger.warning(f"Nie udało się pobrać transkrypcji dla {video_id}: {e}") # Jeśli wystąpił wyjątek, spróbuj z głównym filmem logger.info(f"Próbuję główny film: {TEST_VIDEO_ID}") try: transcript, title = await get_transcript(TEST_VIDEO_ID, TEST_LANGUAGES) logger.info(f"Pobrano tytuł głównego filmu: {title}") logger.info(f"Długość transkrypcji: {len(transcript)} znaków") logger.info(f"Fragment transkrypcji: {transcript[:150]}...") except (NoTranscriptFound, TranscriptsDisabled, APIConnectionError, APIResponseError) as e: logger.error(f"Nie można pobrać transkrypcji dla żadnego z testowych filmów: {e}") pytest.skip(f"Nie można pobrać transkrypcji dla żadnego z testowych filmów: {e}") # 4. "Me at the zoo" ma wystarczająco krótką transkrypcję, więc możemy użyć całości logger.info(f"Długość oryginalnej transkrypcji: {len(transcript)} znaków") # 5. Wygenerowanie streszczenia try: logger.info("Próbuję wygenerować streszczenie") summary = await summarize_text(transcript) assert summary is not None assert len(summary) > 10 logger.info(f"Wygenerowane streszczenie: {summary}") except QuotaExceededError as e: logger.warning(f"Przekroczono limit zapytań API OpenAI: {e}") pytest.skip(f"Przekroczono limit zapytań API OpenAI: {e}") except Exception as e: logger.error(f"Błąd podczas generowania streszczenia: {e}", exc_info=True) pytest.skip(f"Błąd podczas generowania streszczenia: {e}") if __name__ == "__main__": # Możliwość uruchomienia testów bezpośrednio z tego pliku asyncio.run(test_real_extract_youtube_urls()) asyncio.run(test_real_extract_video_id()) # Testy wymagające połączenia z YouTube try: asyncio.run(test_real_get_transcript_and_title()) except Exception as e: logger.error(f"Problemy z połączeniem z YouTube API: {e}", exc_info=True) # Testy wymagające klucza API OpenAI if os.environ.get("OPENAI_API_KEY"): asyncio.run(ensure_openai_api_status()) # Sprawdź status API przed uruchomieniem testów if OPENAI_API_AVAILABLE: try: asyncio.run(test_real_summarize_text()) if os.environ.get("YOUTUBE_TRANSCRIPT_API_TOKEN"): asyncio.run(test_end_to_end_integration()) else: logger.warning("Pominięto test end-to-end (brak tokenu API youtube-transcript.io)") except Exception as e: logger.error(f"Problemy z OpenAI API: {e}", exc_info=True) else: logger.warning(f"Pominięto testy OpenAI (API niedostępne: {OPENAI_STATUS_INFO.get('error')})") else: logger.warning("Pominięto testy OpenAI (brak klucza API)")