diff --git a/src/models/config.py b/src/models/config.py index dbcef69f..1598d16b 100644 --- a/src/models/config.py +++ b/src/models/config.py @@ -229,11 +229,27 @@ class UserDataCollection(ConfigurationBase): @model_validator(mode="after") def check_storage_location_is_set_when_needed(self) -> Self: """Check that storage_location is set when enabled.""" - if self.feedback_enabled and self.feedback_storage is None: - raise ValueError("feedback_storage is required when feedback is enabled") - if self.transcripts_enabled and self.transcripts_storage is None: - raise ValueError( - "transcripts_storage is required when transcripts is enabled" + if self.feedback_enabled: + if self.feedback_storage is None: + raise ValueError( + "feedback_storage is required when feedback is enabled" + ) + checks.directory_check( + Path(self.feedback_storage), + desc="Check directory to store feedback", + must_exists=False, + must_be_writable=True, + ) + if self.transcripts_enabled: + if self.transcripts_storage is None: + raise ValueError( + "transcripts_storage is required when transcripts is enabled" + ) + checks.directory_check( + Path(self.transcripts_storage), + desc="Check directory to store transcripts", + must_exists=False, + must_be_writable=True, ) return self diff --git a/src/utils/checks.py b/src/utils/checks.py index 509115b1..d30b183f 100644 --- a/src/utils/checks.py +++ b/src/utils/checks.py @@ -57,6 +57,36 @@ def file_check(path: FilePath, desc: str) -> None: raise InvalidConfigurationError(f"{desc} '{path}' is not readable") +def directory_check( + path: FilePath, must_exists: bool, must_be_writable: bool, desc: str +) -> None: + """ + Ensure the given path is an existing directory. + + If the path is not a directory, raises InvalidConfigurationError. + + Parameters: + path (FilePath): Filesystem path to validate. + must_exists (bool): Should the directory exists? + must_be_writable (bool): Should the check test if directory is writable? + desc (str): Short description of the value being checked; used in error + messages. + + Raises: + InvalidConfigurationError: If `path` does not point to a directory or + is not writable when required. + """ + if not os.path.exists(path): + if must_exists: + raise InvalidConfigurationError(f"{desc} '{path}' does not exist") + return + if not os.path.isdir(path): + raise InvalidConfigurationError(f"{desc} '{path}' is not a directory") + if must_be_writable: + if not os.access(path, os.W_OK): + raise InvalidConfigurationError(f"{desc} '{path}' is not writable") + + def import_python_module(profile_name: str, profile_path: str) -> ModuleType | None: """Import a Python module from a file path.""" if not profile_path.endswith(".py"): diff --git a/src/utils/transcripts.py b/src/utils/transcripts.py index 74b20fdb..c6346cc8 100644 --- a/src/utils/transcripts.py +++ b/src/utils/transcripts.py @@ -80,7 +80,11 @@ def store_transcript( # pylint: disable=too-many-arguments,too-many-positional- # stores feedback in a file under unique uuid transcript_file_path = transcripts_path / f"{get_suid()}.json" - with open(transcript_file_path, "w", encoding="utf-8") as transcript_file: - json.dump(data_to_store, transcript_file) + try: + with open(transcript_file_path, "w", encoding="utf-8") as transcript_file: + json.dump(data_to_store, transcript_file) + except (IOError, OSError) as e: + logger.error("Failed to store transcript into %s: %s", transcript_file_path, e) + raise logger.info("Transcript successfully stored at: %s", transcript_file_path) diff --git a/tests/unit/models/config/test_user_data_collection.py b/tests/unit/models/config/test_user_data_collection.py index dd0d7057..602ce810 100644 --- a/tests/unit/models/config/test_user_data_collection.py +++ b/tests/unit/models/config/test_user_data_collection.py @@ -3,6 +3,7 @@ import pytest from models.config import UserDataCollection +from utils.checks import InvalidConfigurationError def test_user_data_collection_feedback_enabled() -> None: @@ -39,3 +40,18 @@ def test_user_data_collection_transcripts_disabled() -> None: match="transcripts_storage is required when transcripts is enabled", ): UserDataCollection(transcripts_enabled=True, transcripts_storage=None) + + +def test_user_data_collection_wrong_directory_path() -> None: + """Test the UserDataCollection constructor for wrong directory path.""" + with pytest.raises( + InvalidConfigurationError, + match="Check directory to store feedback '/root' is not writable", + ): + _ = UserDataCollection(feedback_enabled=True, feedback_storage="/root") + + with pytest.raises( + InvalidConfigurationError, + match="Check directory to store transcripts '/root' is not writable", + ): + _ = UserDataCollection(transcripts_enabled=True, transcripts_storage="/root") diff --git a/tests/unit/utils/test_checks.py b/tests/unit/utils/test_checks.py index 89bd0271..65f35d22 100644 --- a/tests/unit/utils/test_checks.py +++ b/tests/unit/utils/test_checks.py @@ -19,6 +19,14 @@ def input_file_fixture(tmp_path): return filename +@pytest.fixture(name="input_directory") +def input_directory_fixture(tmp_path): + """Create directory manually using the tmp_path fixture.""" + dirname = os.path.join(tmp_path, "mydir") + os.mkdir(dirname) + return dirname + + def test_get_attribute_from_file_no_record(): """Test the get_attribute_from_file function when record is not in dictionary.""" # no data @@ -81,6 +89,44 @@ def test_file_check_not_readable_file(input_file): checks.file_check(input_file, "description") +def test_directory_check_non_existing_directory(): + """Test the function directory_check skips non-existing directory.""" + # just call the function, it should not raise an exception + checks.directory_check( + "/foo/bar/baz", must_exists=False, must_be_writable=False, desc="foobar" + ) + with pytest.raises(checks.InvalidConfigurationError): + checks.directory_check( + "/foo/bar/baz", must_exists=True, must_be_writable=False, desc="foobar" + ) + + +def test_directory_check_existing_writable_directory(input_directory): + """Test the function directory_check checks directory.""" + # just call the function, it should not raise an exception + checks.directory_check( + input_directory, must_exists=True, must_be_writable=True, desc="foobar" + ) + + +def test_directory_check_non_a_directory(input_file): + """Test the function directory_check checks directory.""" + # pass a filename not a directory name + with pytest.raises(checks.InvalidConfigurationError): + checks.directory_check( + input_file, must_exists=True, must_be_writable=True, desc="foobar" + ) + + +def test_directory_check_existing_non_writable_directory(input_directory): + """Test the function directory_check checks directory.""" + with patch("os.access", return_value=False): + with pytest.raises(checks.InvalidConfigurationError): + checks.directory_check( + input_directory, must_exists=True, must_be_writable=True, desc="foobar" + ) + + def test_import_python_module_success(): """Test importing a Python module.""" module_path = "tests/profiles/test/profile.py"