Merge branch 'feature/video-search' into feature/video-comments

Author: aninhasalesp

tests/commands/conftest.py

@@ -0,0 +1,29 @@
+import pytest
+
+
+@pytest.fixture
+def channels_urls():
+    return [
+        "https://www.youtube.com/@Turicas/featured",
+        "https://www.youtube.com/c/PythonicCaf%C3%A9"
+    ]
+
+
+@pytest.fixture
+def videos_ids():
+    return [
+        "video_id_1",
+        "video_id_2"
+    ]
+
+
+@pytest.fixture
+def videos_urls(videos_ids):
+    return [
+        f"https://www.youtube.com/?v={video_id}" for video_id in videos_ids
+    ]
+
+
+@pytest.fixture
+def usernames():
+    return ["Turicas", "PythonicCafe"]

tests/commands/test_base.py

@@ -2,8 +2,6 @@
 import argparse
 import pytest
 
-from io import StringIO
-from datetime import datetime
 from pathlib import Path
 from unittest.mock import MagicMock, patch, mock_open
 from youtool.commands import Command
@@ -83,7 +81,7 @@ def test_data_from_csv_column_not_found(mock_csv_file):
             file_path = Path("tests/resources/csv_column_not_found.csv")
             with pytest.raises(Exception) as exc_info:
                 Command.data_from_csv(file_path, "NonExistentColumn")
-            assert f"Column NonExistentColumn not found on {file_path}" in str(exc_info.value)
+            assert "Column NonExistentColumn not found on tests/resources/csv_column_not_found.csv" in str(exc_info.value)
 
 
 @pytest.fixture
@@ -126,3 +124,23 @@ def test_data_to_csv_output(tmp_path):
     assert Path(output_file_path).is_file()
     assert expected_output == Path(output_file_path).read_text()
     assert str(output_file_path) == result
+
+def test_filter_fields():
+    channel_info = {
+        'channel_id': '123456',
+        'channel_name': 'Test Channel',
+        'subscribers': 1000,
+        'videos': 50,
+        'category': 'Tech'
+    }
+    
+    info_columns = ['channel_id', 'channel_name', 'subscribers']
+    filtered_info = Command.filter_fields(channel_info, info_columns)
+    
+    expected_result = {
+        'channel_id': '123456',
+        'channel_name': 'Test Channel',
+        'subscribers': 1000
+    }
+    
+    assert filtered_info == expected_result, f"Expected {expected_result}, but got {filtered_info}"

tests/commands/test_channel_info.py

@@ -0,0 +1,50 @@
+import pytest
+
+from unittest.mock import patch, Mock, call
+
+from youtool.commands.channel_info import ChannelInfo, YouTube
+
+
+def test_filter_fields():
+    channel_info = {
+        'channel_id': '123456',
+        'channel_name': 'Test Channel',
+        'subscribers': 1000,
+        'videos': 50,
+        'category': 'Tech'
+    }
+    
+    info_columns = ['channel_id', 'channel_name', 'subscribers']
+    filtered_info = ChannelInfo.filter_fields(channel_info, info_columns)
+    
+    expected_result = {
+        'channel_id': '123456',
+        'channel_name': 'Test Channel',
+        'subscribers': 1000
+    }
+    
+    assert filtered_info == expected_result, f"Expected {expected_result}, but got {filtered_info}"
+
+
+def test_channel_ids_from_urls_and_usernames(mocker, channels_urls, usernames):
+    ids_from_urls_mock = "id_from_url"
+    ids_from_usernames_mock = "id_from_username"
+    youtube_mock = mocker.patch("youtool.commands.channel_info.YouTube")
+
+    channel_id_from_url_mock = Mock(return_value=ids_from_urls_mock)
+    channel_id_from_username_mock = Mock(return_value=ids_from_usernames_mock)
+    channels_infos_mock = Mock(return_value=[])
+
+    youtube_mock.return_value.channel_id_from_url = channel_id_from_url_mock
+    youtube_mock.return_value.channel_id_from_username = channel_id_from_username_mock
+    youtube_mock.return_value.channels_infos = channels_infos_mock
+
+    ChannelInfo.execute(urls=channels_urls, usernames=usernames)
+
+    channel_id_from_url_mock.assert_has_calls(
+        [call(url) for url in channels_urls]
+    )
+    channel_id_from_username_mock.assert_has_calls(
+        [call(username) for username in usernames]
+    )
+    channels_infos_mock.assert_called_once_with([ids_from_urls_mock, ids_from_usernames_mock])

tests/commands/test_video_info.py

@@ -0,0 +1,80 @@
+import csv
+import pytest
+
+from unittest.mock import Mock
+from pathlib import Path
+from youtool.commands import VideoInfo
+
+
+@pytest.fixture
+def youtube_mock(mocker, mock_video_info):
+    mock = mocker.patch("youtool.commands.video_info.YouTube")
+    mock_instance = mock.return_value
+    mock_instance.videos_infos = Mock(return_value=mock_video_info)
+    return mock_instance
+
+@pytest.fixture
+def mock_video_info():
+    return [
+        {"id": "tmrhPou85HQ", "title": "Title 1", "description": "Description 1", "published_at": "2021-01-01", "view_count": 100, "like_count": 10, "comment_count": 5},
+        {"id": "qoI_x9fylaw", "title": "Title 2", "description": "Description 2", "published_at": "2021-02-01", "view_count": 200, "like_count": 20, "comment_count": 10} 
+    ]
+
+def test_execute_with_ids_and_urls(youtube_mock, mocker, tmp_path, mock_video_info):
+    ids = ["tmrhPou85HQ", "qoI_x9fylaw"]
+    urls = ["https://www.youtube.com/watch?v=tmrhPou85HQ&ab_channel=Turicas", "https://www.youtube.com/watch?v=qoI_x9fylaw&ab_channel=PythonicCaf%C3%A9"]
+    output_file_path = tmp_path / "output.csv"
+
+    VideoInfo.execute(ids=ids, urls=urls, output_file_path=str(output_file_path), api_key="test_api_key")
+
+    assert Path(output_file_path).is_file()
+    with open(output_file_path, 'r') as f:
+        reader = csv.DictReader(f)
+        csv_data = list(reader)
+
+    assert csv_data[0]["id"] == "tmrhPou85HQ"
+    assert csv_data[1]["id"] == "qoI_x9fylaw"
+
+def test_execute_missing_arguments():
+    with pytest.raises(Exception) as exc_info:
+        VideoInfo.execute(api_key="test_api_key")
+    
+    assert str(exc_info.value) == "Either 'ids' or 'urls' must be provided for the video-info command"
+
+def test_execute_with_input_file_path(youtube_mock, mocker, tmp_path, mock_video_info):
+    input_csv_content = """video_id,video_url
+    tmrhPou85HQ,https://www.youtube.com/watch?v=tmrhPou85HQ&ab_channel=Turicas
+    qoI_x9fylaw,https://www.youtube.com/watch?v=qoI_x9fylaw&ab_channel=PythonicCaf%C3%A9
+    """
+    input_file_path = tmp_path / "input.csv"
+    output_file_path = tmp_path / "output.csv"
+
+    with open(input_file_path, 'w') as f:
+        f.write(input_csv_content)
+
+    VideoInfo.execute(input_file_path=str(input_file_path), output_file_path=str(output_file_path), api_key="test_api_key")
+
+    assert Path(output_file_path).is_file()
+    with open(output_file_path, 'r') as f:
+        reader = csv.DictReader(f)
+        csv_data = list(reader)
+      
+    assert csv_data[0]["id"] == "tmrhPou85HQ"
+    assert csv_data[1]["id"] == "qoI_x9fylaw"
+
+
+def test_execute_with_info_columns(youtube_mock, mocker, tmp_path, mock_video_info):
+    ids = ["tmrhPou85HQ", "qoI_x9fylaw"]
+    output_file_path = tmp_path / "output.csv"
+
+    VideoInfo.execute(ids=ids, output_file_path=str(output_file_path), api_key="test_api_key", info_columns="id,title")
+
+    assert Path(output_file_path).is_file()
+    with open(output_file_path, 'r') as f:
+        reader = csv.DictReader(f)
+        csv_data = list(reader)
+
+    assert csv_data[0]["id"] == "tmrhPou85HQ"
+    assert csv_data[0]["title"] == "Title 1"
+    assert csv_data[1]["id"] == "qoI_x9fylaw"
+    assert csv_data[1]["title"] == "Title 2"

tests/commands/test_video_search.py

@@ -0,0 +1,66 @@
+import csv
+import pytest
+
+from io import StringIO
+from unittest.mock import Mock, call
+from datetime import datetime
+
+from youtool.commands.video_search import VideoSearch
+
+
+def test_video_search_string_output(mocker, videos_ids, videos_urls):
+    youtube_mock = mocker.patch("youtool.commands.video_search.YouTube")
+    expected_videos_infos = [
+        {
+            column: f"v_{index}" for column in VideoSearch.INFO_COLUMNS
+        } for index, _ in enumerate(videos_ids)
+    ]
+
+    csv_file = StringIO()
+    csv_writer = csv.DictWriter(csv_file, fieldnames=VideoSearch.INFO_COLUMNS)
+    csv_writer.writeheader()
+    csv_writer.writerows(expected_videos_infos)
+
+    videos_infos_mock = Mock(return_value=expected_videos_infos)
+    youtube_mock.return_value.videos_infos = videos_infos_mock
+
+    result = VideoSearch.execute(ids=videos_ids, urls=videos_urls)
+
+    videos_infos_mock.assert_called_once_with(list(set(videos_ids)))
+    assert result == csv_file.getvalue()
+
+
+def test_video_search_file_output(mocker, videos_ids, videos_urls, tmp_path):
+    youtube_mock = mocker.patch("youtool.commands.video_search.YouTube")
+    expected_videos_infos = [
+        {
+            column: f"v_{index}" for column in VideoSearch.INFO_COLUMNS
+        } for index, _ in enumerate(videos_ids)
+    ]
+
+    expected_csv_file = StringIO()
+    csv_writer = csv.DictWriter(expected_csv_file, fieldnames=VideoSearch.INFO_COLUMNS)
+    csv_writer.writeheader()
+    csv_writer.writerows(expected_videos_infos)
+
+    timestamp = datetime.now().strftime("%f")
+    output_file_name = f"output_{timestamp}.csv"
+    output_file_path = tmp_path / output_file_name
+
+    videos_infos_mock = Mock(return_value=expected_videos_infos)
+    youtube_mock.return_value.videos_infos = videos_infos_mock
+
+    result_file_path = VideoSearch.execute(
+        ids=videos_ids, urls=videos_urls, output_file_path=output_file_path
+    )
+
+    with open(result_file_path, "r") as result_csv_file:
+        result_csv = result_csv_file.read()
+
+    videos_infos_mock.assert_called_once_with(list(set(videos_ids)))
+    assert result_csv.replace("\r", "") == expected_csv_file.getvalue().replace("\r", "")
+
+
+def test_video_search_no_id_and_url_error():
+    with pytest.raises(Exception, match="Either 'ids' or 'urls' must be provided"):
+        VideoSearch.execute(ids=None, urls=None)

youtool/commands/base.py

@@ -50,6 +50,24 @@ def parse_arguments(cls, subparsers: argparse._SubParsersAction) -> None:
             parser.add_argument(argument_name, **argument_copy)
         parser.set_defaults(func=cls.execute)
 
+    @staticmethod
+    def filter_fields(video_info: Dict, info_columns: Optional[List] = None) -> Dict:
+        """Filters the fields of a dictionary containing video information based on specified columns.
+
+        Args:
+            video_info (Dict): A dictionary containing video information.
+            info_columns (Optional[List], optional): A list specifying which fields to include in the filtered output. 
+            If None, returns the entire video_info dictionary. Defaults to None.
+
+        Returns:
+            A dictionary containing only the fields specified in info_columns (if provided) 
+            or the entire video_info dictionary if info_columns is None.
+        """
+        return {
+            field: value for field, value in video_info.items() if field in info_columns
+        } if info_columns else video_info
+
+
     @classmethod
     def execute(cls, **kwargs) -> str:  # noqa: D417
         """Executes the command.
@@ -87,7 +105,7 @@ def data_from_csv(file_path: Path, data_column_name: Optional[str] = None) -> Li
 
             if fieldnames is None:
                 raise ValueError("Fieldnames is None")
-            
+
             if data_column_name not in fieldnames:
                 raise Exception(f"Column {data_column_name} not found on {file_path}")
             for row in reader:

youtool/commands/channel_info.py

@@ -108,7 +108,9 @@ def execute(cls: Self, **kwargs) -> str:
         ] + [
             youtube.channel_id_from_username(username) for username in (usernames or []) if username
         ]
-        channel_ids = [channel_id for channel_id in channels_ids if channel_id]
+        channel_ids = list(
+            set([channel_id for channel_id in channels_ids if channel_id])
+        )
 
         return cls.data_to_csv(
             data=[
@@ -117,4 +119,4 @@ def execute(cls: Self, **kwargs) -> str:
                 ) for channel_info in (youtube.channels_infos(channel_ids) or [])
             ],
             output_file_path=output_file_path
-        )
+        )

youtool/commands/video_info.py

@@ -24,23 +24,6 @@ class VideoInfo(Command):
         "id", "title", "description", "published_at", "view_count", "like_count", "comment_count"
     ]
 
-    @staticmethod
-    def filter_fields(video_info: Dict, info_columns: Optional[List] = None) -> Dict:
-        """Filters the fields of a dictionary containing video information based on specified columns.
-
-        Args:
-            video_info (Dict): A dictionary containing video information.
-            info_columns (Optional[List], optional): A list specifying which fields to include in the filtered output. 
-            If None, returns the entire video_info dictionary. Defaults to None.
-
-        Returns:
-            A dictionary containing only the fields specified in info_columns (if provided) 
-            or the entire video_info dictionary if info_columns is None.
-        """
-        return {
-            field: value for field, value in video_info.items() if field in info_columns
-        } if info_columns else video_info
-
     @classmethod
     def execute(cls: Self, **kwargs) -> str:
         """