-
Notifications
You must be signed in to change notification settings - Fork 112
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* Package update from watchgod to watchfiles Signed-off-by: Jitendra Gundaniya <[email protected]> * lint fix Signed-off-by: Jitendra Gundaniya <[email protected]> * Test fix Signed-off-by: Jitendra Gundaniya <[email protected]> * Test fixes Signed-off-by: Jitendra Gundaniya <[email protected]> * lint fixes Signed-off-by: Jitendra Gundaniya <[email protected]> * test fix Signed-off-by: Jitendra Gundaniya <[email protected]> * test added Signed-off-by: Jitendra Gundaniya <[email protected]> * lint fix Signed-off-by: Jitendra Gundaniya <[email protected]> * Release note added Signed-off-by: Jitendra Gundaniya <[email protected]> * AutoreloadFileFilter added for accurate auto reload files Signed-off-by: Jitendra Gundaniya <[email protected]> * Test fix Signed-off-by: Jitendra Gundaniya <[email protected]> * Test fix Signed-off-by: Jitendra Gundaniya <[email protected]> * lint and unit test fix Signed-off-by: Jitendra Gundaniya <[email protected]> * Test fix Signed-off-by: Jitendra Gundaniya <[email protected]> * Lint fix Signed-off-by: Jitendra Gundaniya <[email protected]> * Lint fix Signed-off-by: Jitendra Gundaniya <[email protected]> * Fix import issue Signed-off-by: Jitendra Gundaniya <[email protected]> * Extra import removed Signed-off-by: Jitendra Gundaniya <[email protected]> * tmp_path fix for tests Signed-off-by: Jitendra Gundaniya <[email protected]> * Lint fix Signed-off-by: Jitendra Gundaniya <[email protected]> * Tests with docstring Signed-off-by: Jitendra Gundaniya <[email protected]> * Moved to GitIgnoreSpec class Signed-off-by: Jitendra Gundaniya <[email protected]> --------- Signed-off-by: Jitendra Gundaniya <[email protected]>
- Loading branch information
Showing
11 changed files
with
291 additions
and
32 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| """ | ||
| This module provides a custom file filter for autoreloading that filters out files based on allowed | ||
| file extensions and patterns specified in a .gitignore file. | ||
| """ | ||
|
|
||
| import logging | ||
| from pathlib import Path | ||
| from typing import Optional, Set | ||
|
|
||
| from pathspec import GitIgnoreSpec | ||
| from watchfiles import Change, DefaultFilter | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| class AutoreloadFileFilter(DefaultFilter): | ||
| """ | ||
| Custom file filter for autoreloading that extends DefaultFilter. | ||
| Filters out files based on allowed file extensions and patterns specified in a .gitignore file. | ||
| """ | ||
|
|
||
| allowed_extensions: Set[str] = {".py", ".yml", ".yaml", ".json"} | ||
|
|
||
| def __init__(self, base_path: Optional[Path] = None): | ||
| """ | ||
| Initialize the AutoreloadFileFilter. | ||
| Args: | ||
| base_path (Optional[Path]): The base path to set as the current working directory | ||
| for the filter. | ||
| """ | ||
| self.cwd = base_path or Path.cwd() | ||
|
|
||
| # Call the superclass constructor | ||
| super().__init__() | ||
|
|
||
| # Load .gitignore patterns | ||
| gitignore_path = self.cwd / ".gitignore" | ||
| try: | ||
| with open(gitignore_path, "r", encoding="utf-8") as gitignore_file: | ||
| ignore_patterns = gitignore_file.read().splitlines() | ||
| self.gitignore_spec: Optional[GitIgnoreSpec] = GitIgnoreSpec.from_lines( | ||
| "gitwildmatch", ignore_patterns | ||
| ) | ||
| except FileNotFoundError: | ||
| self.gitignore_spec = None | ||
|
|
||
| def __call__(self, change: Change, path: str) -> bool: | ||
| """ | ||
| Determine whether a file change should be processed. | ||
| Args: | ||
| change (Change): The type of change detected. | ||
| path (str): The path to the file that changed. | ||
| Returns: | ||
| bool: True if the file should be processed, False otherwise. | ||
| """ | ||
| if not super().__call__(change, path): | ||
| logger.debug("Filtered out by DefaultFilter: %s", path) | ||
| return False | ||
|
|
||
| path_obj = Path(path) | ||
|
|
||
| # Exclude files matching .gitignore patterns | ||
| try: | ||
| relative_path = path_obj.resolve().relative_to(self.cwd.resolve()) | ||
| except ValueError: | ||
| logger.debug("Path not relative to CWD: %s", path) | ||
| return False | ||
|
|
||
| try: | ||
| if self.gitignore_spec and self.gitignore_spec.match_file( | ||
| str(relative_path) | ||
| ): | ||
| logger.debug("Filtered out by .gitignore: %s", relative_path) | ||
| return False | ||
| # ruff: noqa: BLE001 | ||
| except Exception as exc: | ||
| logger.debug("Exception during .gitignore matching: %s", exc) | ||
| return True # Pass the file if .gitignore matching fails | ||
|
|
||
| # Include only files with allowed extensions | ||
| if path_obj.suffix in self.allowed_extensions: | ||
| logger.debug("Allowed file: %s", path) | ||
| return True | ||
| logger.debug("Filtered out by allowed_extensions: %s", path_obj.suffix) | ||
| return False |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,141 @@ | ||
| import logging | ||
| import shutil | ||
| import tempfile | ||
| from pathlib import Path | ||
| from unittest.mock import patch | ||
|
|
||
| import pytest | ||
| from watchfiles import Change, DefaultFilter | ||
|
|
||
| from kedro_viz.autoreload_file_filter import AutoreloadFileFilter | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| @pytest.fixture | ||
| def file_filter(tmp_path): | ||
| """ | ||
| Fixture to create a temporary .gitignore file and initialize the AutoreloadFileFilter | ||
| with the test directory as the base path. | ||
| """ | ||
| # Create a .gitignore file | ||
| gitignore_path = tmp_path / ".gitignore" | ||
| gitignore_path.write_text("ignored.py\n") | ||
|
|
||
| # Initialize the filter with the test directory as base_path | ||
| return AutoreloadFileFilter(base_path=tmp_path) | ||
|
|
||
|
|
||
| def test_no_gitignore(tmp_path): | ||
| """ | ||
| Test that a file passes the filter when the .gitignore file is missing. | ||
| """ | ||
| gitignored_file = tmp_path / "ignored.py" | ||
| gitignored_file.touch() | ||
|
|
||
| # Initialize the filter without a .gitignore file | ||
| gitignore_path = tmp_path / ".gitignore" | ||
| if gitignore_path.exists(): | ||
| gitignore_path.unlink() | ||
| file_filter = AutoreloadFileFilter(base_path=tmp_path) | ||
|
|
||
| result = file_filter(Change.modified, str(gitignored_file)) | ||
| assert result, "File should pass the filter when .gitignore is missing" | ||
|
|
||
|
|
||
| def test_gitignore_exception(file_filter, tmp_path): | ||
| """ | ||
| Test that a file passes the filter if an exception occurs during .gitignore matching. | ||
| """ | ||
| allowed_file = tmp_path / "test.py" | ||
| allowed_file.touch() | ||
|
|
||
| with patch( | ||
| "pathspec.PathSpec.match_file", side_effect=Exception("Mocked exception") | ||
| ): | ||
| result = file_filter(Change.modified, str(allowed_file)) | ||
| assert result, "Filter should pass the file if .gitignore matching fails" | ||
|
|
||
|
|
||
| def test_allowed_file(file_filter, tmp_path): | ||
| """ | ||
| Test that a file with an allowed extension passes the filter. | ||
| """ | ||
| allowed_file = tmp_path / "test.py" | ||
| allowed_file.touch() | ||
|
|
||
| result = file_filter(Change.modified, str(allowed_file)) | ||
| assert result, "Allowed file should pass the filter" | ||
|
|
||
|
|
||
| def test_disallowed_file(file_filter, tmp_path): | ||
| """ | ||
| Test that a file with a disallowed extension does not pass the filter. | ||
| """ | ||
| disallowed_file = tmp_path / "test.txt" | ||
| disallowed_file.touch() | ||
|
|
||
| result = file_filter(Change.modified, str(disallowed_file)) | ||
| assert not result, "Disallowed file should not pass the filter" | ||
|
|
||
|
|
||
| def test_gitignored_file(file_filter, tmp_path): | ||
| """ | ||
| Test that a file listed in the .gitignore file does not pass the filter. | ||
| """ | ||
| gitignored_file = tmp_path / "ignored.py" | ||
| gitignored_file.touch() | ||
|
|
||
| result = file_filter(Change.modified, str(gitignored_file)) | ||
| assert not result, "Gitignored file should not pass the filter" | ||
|
|
||
|
|
||
| def test_non_relative_path(file_filter): | ||
| """ | ||
| Test that a file outside the current working directory does not pass the filter. | ||
| """ | ||
| original_cwd = Path.cwd().parent # Go up one directory | ||
| outside_file = original_cwd / "outside.py" | ||
| outside_file.touch() | ||
|
|
||
| result = file_filter(Change.modified, str(outside_file)) | ||
| assert not result, "File outside the CWD should not pass the filter" | ||
|
|
||
| # Cleanup | ||
| outside_file.unlink() | ||
|
|
||
|
|
||
| def test_no_allowed_extension(file_filter, tmp_path): | ||
| """ | ||
| Test that a file without an allowed extension does not pass the filter. | ||
| """ | ||
| no_extension_file = tmp_path / "no_extension" | ||
| no_extension_file.touch() | ||
|
|
||
| result = file_filter(Change.modified, str(no_extension_file)) | ||
| assert not result, "File without allowed extension should not pass the filter" | ||
|
|
||
|
|
||
| def test_directory_path(file_filter, tmp_path): | ||
| """ | ||
| Test that a directory does not pass the filter. | ||
| """ | ||
| directory_path = tmp_path / "some_directory" | ||
| directory_path.mkdir() | ||
|
|
||
| result = file_filter(Change.modified, str(directory_path)) | ||
| assert not result, "Directories should not pass the filter" | ||
|
|
||
|
|
||
| def test_filtered_out_by_default_filter(file_filter, tmp_path, mocker): | ||
| """ | ||
| Test that a file is filtered out by the DefaultFilter. | ||
| """ | ||
| filtered_file = tmp_path / "filtered.py" | ||
| filtered_file.touch() | ||
|
|
||
| # Mock the super().__call__ method to return False | ||
| mocker.patch.object(DefaultFilter, "__call__", return_value=False) | ||
|
|
||
| result = file_filter(Change.modified, str(filtered_file)) | ||
| assert not result, "File should be filtered out by DefaultFilter" |
Oops, something went wrong.