diff --git a/pyproject.toml b/pyproject.toml index 2a570668..19527e13 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -74,7 +74,7 @@ changelog_file = "CHANGELOG.md" build_command = "pip install poetry && poetry build" [tool.pytest.ini_options] -addopts = "--strict-markers" +addopts = "--cov=chaturbate_poller --cov-report=xml --cov-fail-under=80" generate_report_on_test = "True" asyncio_mode = "strict" diff --git a/tests/test___main__.py b/tests/test___main__.py index 1d736c51..0503eb92 100644 --- a/tests/test___main__.py +++ b/tests/test___main__.py @@ -2,6 +2,7 @@ """Tests for the __main__ module.""" import asyncio +import logging import subprocess import sys from contextlib import suppress @@ -40,3 +41,14 @@ def test_start_polling() -> None: pytest.raises(ValueError, match="Unauthorized access. Verify the username and token."), ): asyncio.run(start_polling("username", "token", 10, testbed=False, verbose=False)) + + +def test_start_polling_verbose() -> None: + """Test the start_polling function with verbose output.""" + with ( # noqa: PT012 + suppress(KeyboardInterrupt), + pytest.raises(ValueError, match="Unauthorized access. Verify the username and token."), + ): + asyncio.run(start_polling("username", "token", 10, testbed=False, verbose=True)) + + assert logging.getLogger().level == logging.DEBUG diff --git a/tests/test_docker_integration.py b/tests/test_docker_integration.py index 2f123865..dd35dec4 100644 --- a/tests/test_docker_integration.py +++ b/tests/test_docker_integration.py @@ -5,35 +5,23 @@ import subprocess -def test_docker_image_build(mocker) -> None: # noqa: ANN001 - """Test building the Docker image.""" - docker_executable = shutil.which("docker") # Find the full path of the Docker executable - assert docker_executable is not None, "Docker is not installed or not found in PATH" - - mock_subprocess = mocker.patch("subprocess.run") - mock_subprocess.return_value.returncode = 0 - - result = subprocess.run( - [docker_executable, "build", "-t", "chaturbate_poller:latest", "."], check=True +def test_docker_build_and_run(mocker) -> None: # noqa: ANN001 + """Test the full Docker build and run process.""" + docker_executable = shutil.which("docker") + assert docker_executable is not None, "Docker is not installed." + + # Mock subprocess for Docker build + mocker.patch( + "subprocess.run", + return_value=subprocess.CompletedProcess(args=["docker", "build"], returncode=0), ) + subprocess.run([docker_executable, "build", "-t", "chaturbate_poller:latest", "."], check=True) - assert result.returncode == 0 - mock_subprocess.assert_called_once_with( - [docker_executable, "build", "-t", "chaturbate_poller:latest", "."], check=True + # Mock subprocess for Docker run + mocker.patch( + "subprocess.run", + return_value=subprocess.CompletedProcess(args=["docker", "run"], returncode=0), ) + subprocess.run([docker_executable, "run", "chaturbate_poller:latest"], check=True) - -def test_docker_image_run(mocker) -> None: # noqa: ANN001 - """Test running the Docker container.""" - docker_executable = shutil.which("docker") # Find the full path of the Docker executable - assert docker_executable is not None, "Docker is not installed or not found in PATH" - - mock_subprocess = mocker.patch("subprocess.run") - mock_subprocess.return_value.returncode = 0 - - result = subprocess.run([docker_executable, "run", "chaturbate_poller:latest"], check=True) - - assert result.returncode == 0 - mock_subprocess.assert_called_once_with( - [docker_executable, "run", "chaturbate_poller:latest"], check=True - ) + assert True diff --git a/tests/test_influxdb_handler.py b/tests/test_influxdb_handler.py index e8f64f57..318dd424 100644 --- a/tests/test_influxdb_handler.py +++ b/tests/test_influxdb_handler.py @@ -11,82 +11,69 @@ from chaturbate_poller.influxdb_client import InfluxDBHandler -class TestInfluxDBHandler: - """Tests for InfluxDBHandler.""" - - def test_flatten_dict(self) -> None: - """Test the dictionary flattening method.""" - handler = InfluxDBHandler() - nested_dict = {"level1": {"level2": "value", "level2b": {"level3": "value3"}}} - flattened_dict = handler.flatten_dict(nested_dict) - expected_dict = {"level1.level2": "value", "level1.level2b.level3": "value3"} - - assert flattened_dict == expected_dict - - @pytest.mark.parametrize( - ("input_dict", "expected_dict"), - [ - ( - {"key": "value", "nested": {"subkey": "subvalue"}}, - {"key": "value", "nested.subkey": "subvalue"}, - ), - ( - { - "key": "value", - "nested": {"subkey": "subvalue", "subkey2": {"subsubkey": "value"}}, - }, - {"key": "value", "nested.subkey": "subvalue", "nested.subkey2.subsubkey": "value"}, - ), - ], - ) - def test_flatten_dict_with_params(self, input_dict: dict, expected_dict: dict) -> None: - """Test flatten_dict method with various inputs.""" - handler = InfluxDBHandler() - assert handler.flatten_dict(input_dict) == expected_dict - - def test_write_event(self, mocker) -> None: # noqa: ANN001 - """Test writing an event to InfluxDB.""" - handler = InfluxDBHandler() - mock_write_api = mocker.patch.object(handler.write_api, "write", autospec=True) - event_data = {"event": "data"} - handler.write_event("test_measurement", event_data) - mock_write_api.assert_called_once() - - def test_write_event_failure(self, mocker, caplog) -> None: # noqa: ANN001 - """Test writing an event to InfluxDB failure.""" - handler = InfluxDBHandler() - mocker.patch.object(handler.write_api, "write", side_effect=ApiException) - event_data = {"event": "data"} - - with caplog.at_level(logging.ERROR), pytest.raises(ApiException): - handler.write_event("test_measurement", event_data) - - assert "Failed to write data to InfluxDB" in caplog.text - - def test_close_handler(self, mocker) -> None: # noqa: ANN001 - """Test closing the InfluxDB handler.""" - handler = InfluxDBHandler() - mock_close = mocker.patch.object(handler.client, "close", autospec=True) - handler.close() - mock_close.assert_called_once() - - @mock.patch.dict( +@pytest.fixture(scope="module") +def influxdb_handler() -> InfluxDBHandler: + """Fixture for InfluxDBHandler.""" + return InfluxDBHandler() + + +def test_flatten_dict(influxdb_handler) -> None: # noqa: ANN001 + """Test flatten_dict method.""" + nested_dict = {"level1": {"level2": "value", "level2b": {"level3": "value3"}}} + flattened_dict = influxdb_handler.flatten_dict(nested_dict) + expected_dict = {"level1.level2": "value", "level1.level2b.level3": "value3"} + assert flattened_dict == expected_dict + + +@pytest.mark.parametrize( + ("input_dict", "expected_dict"), + [ + ( + {"key": "value", "nested": {"subkey": "subvalue"}}, + {"key": "value", "nested.subkey": "subvalue"}, + ), + ( + {"key": "value", "nested": {"subkey": "subvalue", "subkey2": {"subsubkey": "value"}}}, + {"key": "value", "nested.subkey": "subvalue", "nested.subkey2.subsubkey": "value"}, + ), + ], +) +def test_flatten_dict_with_params(influxdb_handler, input_dict, expected_dict) -> None: # noqa: ANN001 + """Test flatten_dict method with different parameters.""" + assert influxdb_handler.flatten_dict(input_dict) == expected_dict + + +def test_write_event_success(influxdb_handler, mocker) -> None: # noqa: ANN001 + """Test write_event method success.""" + mock_write_api = mocker.patch.object(influxdb_handler.write_api, "write", autospec=True) + event_data = {"event": "data"} + influxdb_handler.write_event("test_measurement", event_data) + mock_write_api.assert_called_once() + + +def test_write_event_failure(influxdb_handler, mocker, caplog) -> None: # noqa: ANN001 + """Test write_event method when write fails.""" + mocker.patch.object(influxdb_handler.write_api, "write", side_effect=ApiException) + event_data = {"event": "data"} + + with caplog.at_level(logging.ERROR), pytest.raises(ApiException): + influxdb_handler.write_event("test_measurement", event_data) + + assert "Failed to write data to InfluxDB" in caplog.text + + +def test_close_handler(influxdb_handler, mocker) -> None: # noqa: ANN001 + """Test close method.""" + mock_close = mocker.patch.object(influxdb_handler.client, "close", autospec=True) + influxdb_handler.close() + mock_close.assert_called_once() + + +def test_influxdb_handler_init() -> None: + """Test InfluxDBHandler initialization.""" + with mock.patch.dict( os.environ, {"INFLUXDB_URL": "http://localhost:8086", "INFLUXDB_TOKEN": "test_token"} - ) - def test_init_handler(self) -> None: - """Test initializing the InfluxDB handler.""" + ): handler = InfluxDBHandler() assert handler.client.url == "http://localhost:8086" assert handler.client.token == "test_token" # noqa: S105 - - def test_flatten_dict_with_complex_structure(self) -> None: - """Test flatten_dict method with a more complex nested structure.""" - handler = InfluxDBHandler() - complex_dict = { - "level1": {"level2": {"level3": {"level4": "value"}}}, - "another_level1": "another_value", - } - flattened_dict = handler.flatten_dict(complex_dict) - expected_dict = {"level1.level2.level3.level4": "value", "another_level1": "another_value"} - - assert flattened_dict == expected_dict diff --git a/tests/test_models.py b/tests/test_models.py index 3b5577f1..3e30b912 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -1,5 +1,5 @@ # ruff: noqa: S101 -"""Tests for Pydantic models.""" +"""Tests for the models module.""" import pytest from pydantic import ValidationError @@ -7,72 +7,81 @@ from chaturbate_poller.models import Event, EventData, Gender, Media, MediaType, Message, Tip, User -class TestModels: - """Tests for the models.""" - - def test_user_model(self) -> None: - """Test the User model.""" - user = User( +@pytest.mark.parametrize( + ("username", "in_fanclub", "gender", "has_tokens", "recent_tips", "is_mod"), + [ + ("example_user", False, Gender.MALE, True, "none", False), + ("test_user", True, Gender.FEMALE, False, "few", True), + ], +) +def test_user_model(username, in_fanclub, gender, has_tokens, recent_tips, is_mod) -> None: # noqa: ANN001, PLR0913 + """Test User model.""" + user = User( + username=username, + inFanclub=in_fanclub, + gender=gender, + hasTokens=has_tokens, + recentTips=recent_tips, + isMod=is_mod, + ) + assert user.username == username + assert user.in_fanclub == in_fanclub + assert user.gender == gender + assert user.has_tokens == has_tokens + assert user.recent_tips == recent_tips + assert user.is_mod == is_mod + + +def test_invalid_user_model() -> None: + """Test invalid User model.""" + with pytest.raises(ValidationError): + User( username="example_user", - inFanclub=False, + inFanclub="not_a_boolean", # type: ignore[arg-type] gender=Gender.MALE, hasTokens=True, recentTips="none", isMod=False, ) - assert user.username == "example_user" - assert user.in_fanclub is False - assert user.gender == Gender.MALE - assert user.has_tokens is True - assert user.recent_tips == "none" - assert user.is_mod is False - - def test_invalid_user_model(self) -> None: - """Test the User model with invalid data.""" - with pytest.raises(ValidationError): - User( - username="example_user", - inFanclub="not_a_boolean", # type: ignore[arg-type] - gender=Gender.MALE, - hasTokens=True, - recentTips="none", - isMod=False, - ) - def test_message_model(self) -> None: - """Test the Message model.""" - message = Message( + +def test_message_model() -> None: + """Test Message model.""" + message = Message( + fromUser="example_user", + message="example message", + color="example_color", + font="example_font", + toUser="user", + bgColor="example_bg_color", + ) + assert message.from_user == "example_user" + assert message.message == "example message" + assert message.color == "example_color" + assert message.font == "example_font" + assert message.to_user == "user" + assert message.bg_color == "example_bg_color" + + +def test_invalid_message_model() -> None: + """Test invalid Message model.""" + with pytest.raises(ValidationError): + Message( fromUser="example_user", - message="example message", + message=123, # type: ignore[arg-type] color="example_color", font="example_font", toUser="user", bgColor="example_bg_color", ) - assert message.from_user == "example_user" - assert message.message == "example message" - assert message.color == "example_color" - assert message.font == "example_font" - assert message.to_user == "user" - assert message.bg_color == "example_bg_color" - - def test_invalid_message_model(self) -> None: - """Test the Message model with invalid data.""" - with pytest.raises(ValidationError): - Message( - fromUser="example_user", - message=123, # type: ignore[arg-type] - color="example_color", - font="example_font", - toUser="user", - bgColor="example_bg_color", - ) - def test_event_data_model(self) -> None: - """Test the EventData model.""" - event_data = EventData( - broadcaster="example_broadcaster", - user=User( + +@pytest.mark.parametrize( + ("broadcaster", "user", "tip", "media", "subject", "message"), + [ + ( + "example_broadcaster", + User( username="example_user", inFanclub=False, gender=Gender.MALE, @@ -80,11 +89,50 @@ def test_event_data_model(self) -> None: recentTips="none", isMod=False, ), - tip=Tip( - tokens=100, + Tip(tokens=100, message="example message", isAnon=False), + Media(id=1, name="photoset1", type=MediaType.PHOTOS, tokens=25), + "example subject", + Message( + fromUser="example_user", message="example message", - isAnon=False, + color="example_color", + font="example_font", + toUser="user", + bgColor="example_bg_color", ), + ), + (None, None, None, None, None, None), + ], +) +def test_event_data_model(broadcaster, user, tip, media, subject, message) -> None: # noqa: ANN001, PLR0913 + """Test EventData model.""" + event_data = EventData( + broadcaster=broadcaster, + user=user, + tip=tip, + media=media, + subject=subject, + message=message, + ) + if user: + assert event_data.user.username == user.username # type: ignore[union-attr] + if tip: + assert event_data.tip.tokens == tip.tokens # type: ignore[union-attr] + if media: + assert event_data.media.id == media.id # type: ignore[union-attr] + if message: + assert event_data.message.from_user == message.from_user # type: ignore[union-attr] + assert event_data.broadcaster == broadcaster + assert event_data.subject == subject + + +def test_invalid_event_data_model() -> None: + """Test invalid EventData model.""" + with pytest.raises(ValidationError): + EventData( + broadcaster="example_broadcaster", + user="not_a_user", # type: ignore[arg-type] + tip=Tip(tokens=100, message="example message", isAnon=False), media=Media(id=1, name="photoset1", type=MediaType.PHOTOS, tokens=25), subject="example subject", message=Message( @@ -96,100 +144,66 @@ def test_event_data_model(self) -> None: bgColor="example_bg_color", ), ) - assert event_data.broadcaster == "example_broadcaster" - assert event_data.user.username == "example_user" # type: ignore[union-attr] - assert event_data.tip.tokens == 100 # type: ignore[union-attr] # noqa: PLR2004 - assert event_data.media.name == "photoset1" # type: ignore[union-attr] - assert event_data.subject == "example subject" - assert event_data.message.message == "example message" # type: ignore[union-attr] - - def test_invalid_event_data_model(self) -> None: - """Test the EventData model with invalid data.""" - with pytest.raises(ValidationError): - EventData( - broadcaster="example_broadcaster", - user="not_a_user", # type: ignore[arg-type] - tip=Tip( - tokens=100, - message="example message", - isAnon=False, - ), - media=Media(id=1, name="photoset1", type=MediaType.PHOTOS, tokens=25), - subject="example subject", - message=Message( - fromUser="example_user", - message="example message", - color="example_color", - font="example_font", - toUser="user", - bgColor="example_bg_color", - ), - ) - - def test_event_model(self) -> None: - """Test the Event model.""" - event = Event( - method="userEnter", - object=EventData( - broadcaster="example_broadcaster", - user=User( - username="example_user", - inFanclub=False, - gender=Gender.MALE, - hasTokens=True, - recentTips="none", - isMod=False, - ), + + +def test_event_model() -> None: + """Test Event model.""" + event = Event( + method="userEnter", + object=EventData( + broadcaster="example_broadcaster", + user=User( + username="example_user", + inFanclub=False, + gender=Gender.MALE, + hasTokens=True, + recentTips="none", + isMod=False, ), + ), + id="UNIQUE_EVENT_ID", + ) + assert event.method == "userEnter" + assert event.object.broadcaster == "example_broadcaster" + if event.object.user is not None: + assert event.object.user.username == "example_user" + assert event.id == "UNIQUE_EVENT_ID" + + +def test_invalid_event_model() -> None: + """Test invalid Event model.""" + with pytest.raises(ValidationError): + Event( + method="userEnter", + object="not_an_event_data", # type: ignore[arg-type] id="UNIQUE_EVENT_ID", ) - assert event.method == "userEnter" - assert event.object.broadcaster == "example_broadcaster" - assert event.object.user.username == "example_user" # type: ignore[union-attr] - assert event.id == "UNIQUE_EVENT_ID" - - def test_invalid_event_model(self) -> None: - """Test the Event model with invalid data.""" - with pytest.raises(ValidationError): - Event( - method="userEnter", - object="not_an_event_data", # type: ignore[arg-type] - id="UNIQUE_EVENT_ID", - ) - - def test_tip_model(self) -> None: - """Test the Tip model.""" - tip = Tip(tokens=100, message="example message", isAnon=False) - assert tip.tokens == 100 # noqa: PLR2004 - assert tip.message == "example message" - assert tip.is_anon is False - - def test_invalid_tip_model(self) -> None: - """Test the Tip model with invalid data.""" - with pytest.raises(ValidationError): - Tip(tokens=-1, message="example message", isAnon=False) - - def test_media_model(self) -> None: - """Test the Media model.""" - media = Media(id=1, name="photoset1", type=MediaType.PHOTOS, tokens=25) - assert media.id == 1 - assert media.name == "photoset1" - assert media.type == MediaType.PHOTOS - assert media.tokens == 25 # noqa: PLR2004 - - def test_invalid_media_model(self) -> None: - """Test the Media model with invalid data.""" - with pytest.raises(ValidationError): - Media(id=1, name="photoset1", type="invalid_type", tokens=25) # type: ignore[arg-type] - - def test_enum_gender(self) -> None: - """Test the Gender enum.""" - assert Gender.MALE.value == "m" - assert Gender.FEMALE.value == "f" - assert Gender.TRANS.value == "t" - assert Gender.COUPLE.value == "c" - - def test_enum_media_type(self) -> None: - """Test the MediaType enum.""" - assert MediaType.PHOTOS.value == "photos" - assert MediaType.VIDEOS.value == "videos" + + +def test_tip_model() -> None: + """Test Tip model.""" + tip = Tip(tokens=100, message="example message", isAnon=False) + assert tip.tokens == 100 # noqa: PLR2004 + assert tip.message == "example message" + assert tip.is_anon is False + + +def test_invalid_tip_model() -> None: + """Test invalid Tip model.""" + with pytest.raises(ValidationError): + Tip(tokens=-1, message="example message", isAnon=False) + + +def test_media_model() -> None: + """Test Media model.""" + media = Media(id=1, name="photoset1", type=MediaType.PHOTOS, tokens=25) + assert media.id == 1 + assert media.name == "photoset1" + assert media.type == MediaType.PHOTOS + assert media.tokens == 25 # noqa: PLR2004 + + +def test_invalid_media_model() -> None: + """Test invalid Media model.""" + with pytest.raises(ValidationError): + Media(id="not_an_integer", name="photoset1", type=MediaType.PHOTOS, tokens=25) # type: ignore[arg-type]