diff --git a/test/deltares_testbench/src/suite/test_case.py b/test/deltares_testbench/src/suite/test_case.py index caf96729da..8c77d4887c 100644 --- a/test/deltares_testbench/src/suite/test_case.py +++ b/test/deltares_testbench/src/suite/test_case.py @@ -6,6 +6,8 @@ import copy import os import time +from dataclasses import dataclass, field +from datetime import datetime, timezone from typing import ClassVar, Dict, List, Tuple from src.config.test_case_config import TestCaseConfig @@ -14,8 +16,25 @@ from src.utils.paths import Paths -# Test case handler (compare or reference) +@dataclass +class DirectoryState: + """Snapshot of a directory's state used to detect added/changed files. + + Attributes + ---------- + files : Dict[str, datetime] + Mapping of filename to last modification time (UTC). + size : int + Total size in bytes of all files in the directory at snapshot time. + """ + + files: Dict[str, datetime] = field(default_factory=dict) + size: int = 0 + + class TestCase: + """Test Case instance (compare or reference).""" + __test__: ClassVar[bool] = False # constructor @@ -23,27 +42,27 @@ class TestCase: def __init__(self, config: TestCaseConfig, logger: ILogger) -> None: self.__config = config self.__logger = logger - self.__maxRunTime: float = self.__config.max_run_time + self.__max_run_time: float = self.__config.max_run_time self.__programs: List[Tuple[int, Program]] = [] self.__errors: list[Exception] = [] - logger.debug(f"Initializing test case ({self.__config.name}), max runtime : {str(self.__maxRunTime)}") + logger.debug(f"Initializing test case ({self.__config.name}), max runtime : {str(self.__max_run_time)}") self.__config.run_file_name = os.path.join(self.__config.absolute_test_case_path, "_tb3_char.run") refrunfile = os.path.join(config.absolute_test_case_reference_path, "_tb3_char.run") if os.path.exists(refrunfile): - refruntime = self.__findCharacteristicsRunTime__(refrunfile) + refruntime = self.__find_characteristics_run_time(refrunfile) if refruntime: self.__config.ref_run_time = refruntime if not self.__config.overrule_ref_max_run_time: # set maxRunTime to 1.5 * reference runtime and add a few seconds (some systems start slow) # The variation in runtimes vary a lot (different machines, other processes) - self.__maxRunTime = refruntime * 1.5 + 10.0 - logger.info(f"Overwriting max run time via reference _tb3_char.run ({str(self.__maxRunTime)})") + self.__max_run_time = refruntime * 1.5 + 10.0 + logger.info(f"Overwriting max run time via reference _tb3_char.run ({str(self.__max_run_time)})") - self.__maxRunTime = max(self.__maxRunTime, 120.0) * 5.0 + 300.0 - logger.debug(f"maxRunTime increased to {str(self.__maxRunTime)}") + self.__max_run_time = max(self.__max_run_time, 120.0) * 5.0 + 300.0 + logger.debug(f"maxRunTime increased to {str(self.__max_run_time)}") def run(self, programs: List[Program]) -> None: """Execute a Test Case. @@ -63,12 +82,14 @@ def run(self, programs: List[Program]) -> None: # prepare the programs for running logger = self.__logger - self.__initializeProgramList__(programs) + self.__initialize_program_list(programs) logger.debug("Starting test case") # prepare presets for testbench run file - input_files, size = self.__get_initial_state() + pre_run_state = self.__get_state_directory(self.__config.absolute_test_case_path) + pre_run_files = pre_run_state.files + size = pre_run_state.size start_time = time.time() logger.debug(f"Test case start time {str(time.ctime(int(start_time)))}") @@ -91,34 +112,39 @@ def run(self, programs: List[Program]) -> None: with open(self.__config.run_file_name, "w") as runfile: runfile.write("Start_size:" + str(size) + "\n") runfile.write("Runtime:" + str(elapsed_time) + "\n") - for allfile in os.listdir(self.__config.absolute_test_case_path): + post_run_state = self.__get_state_directory(self.__config.absolute_test_case_path) + for post_file in post_run_state.files: # collect all added and changed files in the working directory (after running, compare to initial list) - if allfile not in {}.fromkeys(input_files, 0): - runfile.write("Output_added:" + str(allfile) + "\n") - size = size + os.path.getsize(os.path.join(self.__config.absolute_test_case_path, allfile)) + if post_file not in {}.fromkeys(pre_run_files, 0): + runfile.write("Output_added:" + str(post_file) + "\n") + size = size + os.path.getsize(os.path.join(self.__config.absolute_test_case_path, post_file)) else: - ftime = os.path.getmtime(os.path.join(self.__config.absolute_test_case_path, allfile)) - if ftime != input_files[allfile]: - runfile.write("Output_changed:" + str(allfile) + "\n") + ftime = post_run_state.files[post_file] + if ftime != pre_run_files[post_file]: + runfile.write("Output_changed:" + str(post_file) + "\n") runfile.write("End_size:" + str(size) + "\n") - def __get_initial_state(self) -> Tuple[Dict[str, float], int]: - inputfiles: Dict[str, float] = {} + def __get_state_directory(self, directory: str) -> DirectoryState: + files: Dict[str, datetime] = {} size: int = 0 # collect all initial files in the working directory before running - for infile in os.listdir(self.__config.absolute_test_case_path): - inputfiles[infile] = os.path.getmtime(os.path.join(self.__config.absolute_test_case_path, infile)) - size = size + os.path.getsize(os.path.join(self.__config.absolute_test_case_path, infile)) + for infile in os.listdir(directory): + files[infile] = datetime.fromtimestamp( + os.path.getmtime(os.path.join(directory, infile)), + tz=timezone.utc, + ) + size = size + os.path.getsize(os.path.join(directory, infile)) - return inputfiles, size + return DirectoryState(files=files, size=size) # get errors from Test Case - # output: list of Errors (type), can be None - def getErrors(self): + # output: list of Errors (type) + def get_errors(self) -> list[Exception]: + """Retrieve errors encountered during Test Case execution.""" return self.__errors - def __initializeProgramList__(self, programs: List[Program]): + def __initialize_program_list(self, programs: List[Program]) -> None: """Prepare programs from configuration.""" # programs are loaded by the manager shell_arguments = " ".join(self.__config.shell_arguments) @@ -150,7 +176,7 @@ def __initializeProgramList__(self, programs: List[Program]): # retrieve runtime or none from _tb3_char.run file # input: path to _tb3_char.run file # output: actual runtime value (float) - def __findCharacteristicsRunTime__(self, filename): + def __find_characteristics_run_time(self, filename: str) -> float | None: with open(filename) as f: retval = None for line in f: diff --git a/test/deltares_testbench/src/suite/test_set_runner.py b/test/deltares_testbench/src/suite/test_set_runner.py index a9c2c66c16..8b3facf653 100644 --- a/test/deltares_testbench/src/suite/test_set_runner.py +++ b/test/deltares_testbench/src/suite/test_set_runner.py @@ -5,6 +5,7 @@ import multiprocessing import os +import shutil import sys from abc import ABC, abstractmethod from datetime import datetime, timedelta @@ -23,7 +24,7 @@ from src.suite.test_bench_settings import TestBenchSettings from src.suite.test_case import TestCase from src.suite.test_case_result import TestCaseResult -from src.utils.common import log_header, log_separator, log_sub_header +from src.utils.common import delete_directory, log_header, log_separator, log_sub_header from src.utils.errors.test_bench_error import TestBenchError from src.utils.handlers.handler_factory import HandlerFactory from src.utils.handlers.resolve_handler import ResolveHandler @@ -240,8 +241,8 @@ def run_test_case( logger.info("Testcase not executed (ignored)...\n") # Check for errors during execution of testcase - if len(testcase.getErrors()) > 0: - errstr = ",".join(str(e) for e in testcase.getErrors()) + if len(testcase.get_errors()) > 0: + errstr = ",".join(str(e) for e in testcase.get_errors()) logger.error("Errors during testcase: " + errstr) raise TestCaseFailure("Errors during testcase: " + errstr) @@ -559,6 +560,9 @@ def __process_test_case_locations(self, config: TestCaseConfig, logger: ILogger) remote_path = self.__build_remote_path(config, location) local_path = self.__build_local_path(config, location) self.__download_location_with_retries(config, location, remote_path, local_path, logger) + if location.type == PathType.INPUT: + self.__copy_to_work_folder(local_path, logger) + self.__set_absolute_paths(config, location.type, local_path) def __validate_location(self, config: TestCaseConfig, location: Location) -> None: @@ -637,6 +641,25 @@ def __download_location_with_retries( error_message = f"Unable to download testcase: {error}" raise TestBenchError(error_message) from e + def __copy_to_work_folder(self, local_path: str, logger: ILogger) -> None: + """Copy downloaded files to work folder if needed.""" + copy_path = local_path + "_work" + if not os.path.exists(local_path): + logger.warning(f"Work path does not exist, cannot create work copy: {local_path}") + return + + # Clean work directory if it exists + if os.path.exists(copy_path): + delete_directory(copy_path, logger) + + # Copy input to work directory + logger.debug(f"Copying input from {local_path} to {copy_path}") + if os.path.isdir(local_path): + shutil.copytree(local_path, copy_path, symlinks=False, ignore_dangling_symlinks=True) + else: + os.makedirs(os.path.dirname(copy_path) or ".", exist_ok=True) + shutil.copy2(local_path, copy_path) + def __download_single_location( self, config: TestCaseConfig, location: Location, remote_path: str, local_path: str, logger: ILogger ) -> None: @@ -655,7 +678,7 @@ def __get_destination_directory(self, location_type: PathType) -> Optional[str]: def __set_absolute_paths(self, config: TestCaseConfig, location_type: PathType, local_path: str) -> None: """Set absolute paths on the config based on location type.""" if location_type == PathType.INPUT: - config.absolute_test_case_path = local_path + config.absolute_test_case_path = local_path + "_work" elif location_type == PathType.REFERENCE: config.absolute_test_case_reference_path = local_path diff --git a/test/deltares_testbench/test/suite/test_ComparisonRunner.py b/test/deltares_testbench/test/suite/test_ComparisonRunner.py index 2cbfddb7bf..10ed125c2d 100644 --- a/test/deltares_testbench/test/suite/test_ComparisonRunner.py +++ b/test/deltares_testbench/test/suite/test_ComparisonRunner.py @@ -2,6 +2,7 @@ import os import pathlib as pl from datetime import datetime, timezone +from enum import Enum from typing import List from unittest.mock import MagicMock, PropertyMock, call @@ -16,6 +17,7 @@ from src.config.test_case_path import TestCasePath from src.config.types.path_type import PathType from src.suite.comparison_runner import ComparisonRunner +from src.suite.run_data import RunData from src.suite.test_bench_settings import TestBenchSettings from src.utils.common import get_default_logging_folder_path from src.utils.comparers.end_result import EndResult @@ -25,6 +27,47 @@ from src.utils.xml_config_parser import XmlConfigParser +class FakeDownloadMode(Enum): + ALL = "all" + REFS_ONLY = "refs_only" + FILES = "files" + OVERWRITE = "overwrite" + + +def patch_fake_download(mocker: MockerFixture, fs: FakeFilesystem, mode: FakeDownloadMode) -> None: + def _fake_download( + from_path: str, + to_path: str, + programs, + logger, + credentials, + version, + ) -> None: + match mode: + case FakeDownloadMode.ALL: + fs.makedirs(to_path, exist_ok=True) + return + case FakeDownloadMode.REFS_ONLY: + if to_path.startswith("/refs"): + fs.makedirs(to_path, exist_ok=True) + return + case FakeDownloadMode.FILES: + if to_path.startswith("/refs"): + fs.makedirs(to_path, exist_ok=True) + elif to_path.startswith("/cases"): + fs.makedirs(to_path, exist_ok=True) + fs.makedirs(f"{to_path}/sub", exist_ok=True) + fs.create_file(f"{to_path}/sub/real.txt", contents="hello") + case FakeDownloadMode.OVERWRITE: + if to_path.startswith("/refs"): + fs.makedirs(to_path, exist_ok=True) + elif to_path.startswith("/cases"): + fs.makedirs(to_path, exist_ok=True) + fs.create_file(f"{to_path}/file.txt", contents="new") + + mocker.patch("src.suite.test_set_runner.HandlerFactory.download", side_effect=_fake_download) + + class TestComparisonRunner: @pytest.mark.usefixtures("fs") # Use fake filesystem. def test_run_tests_and_debug_log_downloaded_file(self, mocker: MockerFixture) -> None: @@ -306,6 +349,136 @@ def test_run_tests_sequentially__run_multiple__continue_on_error( assert failed.results[0][-1].result == EndResult.ERROR assert not succeeded.results # No `EndResult.ERROR` in `results` means the comparison can potentially succeed. + def test_case_preperation_makes_copy_of_case_dir(self, mocker: MockerFixture, fs: FakeFilesystem) -> None: + # Arrange + settings = TestBenchSettings() + settings.command_line_settings.skip_download = [] + settings.local_paths = LocalPaths(cases_path="/cases", references_path="/refs") + ref_location = TestComparisonRunner.create_location(name="reference", location_type=PathType.REFERENCE) + case_location = TestComparisonRunner.create_location(name="case", location_type=PathType.INPUT) + config = TestComparisonRunner.create_test_case_config( + "Banana_1", ignore_testcase=True, locations=[ref_location, case_location] + ) + logger = MagicMock(spec=ConsoleLogger) + runner = ComparisonRunner(settings, logger) + run_data = RunData(1, 10) + + patch_fake_download(mocker, fs, FakeDownloadMode.ALL) + + expected_work_path = "/cases/win64/Banana_1_work" + + # Act + runner.run_test_case(config=config, run_data=run_data) + + # Assert + assert fs.exists(expected_work_path) + + def test_copy_to_work_folder__missing_source__warns_and_returns( + self, mocker: MockerFixture, fs: FakeFilesystem + ) -> None: + # Arrange + settings = TestBenchSettings() + settings.command_line_settings.skip_download = [] + settings.command_line_settings.skip_run = True + settings.command_line_settings.skip_post_processing = True + settings.local_paths = LocalPaths(cases_path="/cases", references_path="/refs") + + ref_location = TestComparisonRunner.create_location(name="reference", location_type=PathType.REFERENCE) + case_location = TestComparisonRunner.create_location(name="case", location_type=PathType.INPUT) + config = TestComparisonRunner.create_test_case_config( + "Name_1", ignore_testcase=True, locations=[ref_location, case_location] + ) + + logger = MagicMock(spec=ConsoleLogger) + testcase_logger = MagicMock() + logger.create_test_case_logger.return_value = testcase_logger + runner = ComparisonRunner(settings, logger) + run_data = RunData(1, 1) + + expected_local_input_path = "/cases/win64/Name_1" + expected_work_path = expected_local_input_path + "_work" + + patch_fake_download(mocker, fs, FakeDownloadMode.REFS_ONLY) + + # Act + runner.run_test_case(config=config, run_data=run_data) + + # Assert + assert any( + "Work path does not exist" in str(call.args[0]) for call in testcase_logger.warning.call_args_list + ), "Expected warning about missing source work path" + assert not fs.exists(expected_work_path) + + def test_copy_to_work_folder__copies_directory(self, mocker: MockerFixture, fs: FakeFilesystem) -> None: + # Arrange + settings = TestBenchSettings() + settings.command_line_settings.skip_download = [] + settings.command_line_settings.skip_run = True + settings.command_line_settings.skip_post_processing = True + settings.local_paths = LocalPaths(cases_path="/cases", references_path="/refs") + + ref_location = TestComparisonRunner.create_location(name="reference", location_type=PathType.REFERENCE) + case_location = TestComparisonRunner.create_location(name="case", location_type=PathType.INPUT) + config = TestComparisonRunner.create_test_case_config( + "Name_1", ignore_testcase=True, locations=[ref_location, case_location] + ) + + logger = MagicMock(spec=ConsoleLogger) + testcase_logger = MagicMock() + logger.create_test_case_logger.return_value = testcase_logger + runner = ComparisonRunner(settings, logger) + run_data = RunData(1, 1) + + expected_local_input_path = "/cases/win64/Name_1" + expected_work_path = expected_local_input_path + "_work" + + patch_fake_download(mocker, fs, FakeDownloadMode.FILES) + + # Act + runner.run_test_case(config=config, run_data=run_data) + + # Assert + assert fs.exists(f"{expected_work_path}/sub/real.txt") + + def test_copy_to_work_folder__overwrites_existing_work_folder( + self, mocker: MockerFixture, fs: FakeFilesystem + ) -> None: + # Arrange + settings = TestBenchSettings() + settings.command_line_settings.skip_download = [] + settings.command_line_settings.skip_run = True + settings.command_line_settings.skip_post_processing = True + settings.local_paths = LocalPaths(cases_path="/cases", references_path="/refs") + + ref_location = TestComparisonRunner.create_location(name="reference", location_type=PathType.REFERENCE) + case_location = TestComparisonRunner.create_location(name="case", location_type=PathType.INPUT) + config = TestComparisonRunner.create_test_case_config( + "Name_1", ignore_testcase=True, locations=[ref_location, case_location] + ) + + logger = MagicMock(spec=ConsoleLogger) + testcase_logger = MagicMock() + logger.create_test_case_logger.return_value = testcase_logger + runner = ComparisonRunner(settings, logger) + run_data = RunData(1, 1) + + expected_local_input_path = "/cases/win64/Name_1" + expected_work_path = expected_local_input_path + "_work" + + fs.makedirs(expected_work_path, exist_ok=True) + fs.create_file(f"{expected_work_path}/file.txt", contents="old") + fs.create_file(f"{expected_work_path}/old.txt", contents="should be removed") + + patch_fake_download(mocker, fs, FakeDownloadMode.OVERWRITE) + + # Act + runner.run_test_case(config=config, run_data=run_data) + + # Assert + with open(f"{expected_work_path}/file.txt") as f: + assert f.read() == "new" + assert not fs.exists(f"{expected_work_path}/old.txt") + @staticmethod def create_test_case_config( name: str, diff --git a/test/deltares_testbench/test/suite/test_TestCase.py b/test/deltares_testbench/test/suite/test_TestCase.py index 3824047b1a..264e59bfc3 100644 --- a/test/deltares_testbench/test/suite/test_TestCase.py +++ b/test/deltares_testbench/test/suite/test_TestCase.py @@ -4,8 +4,10 @@ from pyfakefs.fake_filesystem import FakeFilesystem from pytest_mock import MockerFixture +from src.config.program_config import ProgramConfig from src.config.test_case_config import TestCaseConfig from src.suite.program import Program +from src.suite.test_bench_settings import TestBenchSettings from src.suite.test_case import TestCase from src.utils.logging.file_logger import FileLogger @@ -17,16 +19,25 @@ def test_run__tb3_char_output_file(self, mocker: MockerFixture, fs: FakeFilesyst logger = mocker.Mock(spec=FileLogger) config = self.create_test_case_config("name_1", platform=platform) fs.create_file(f"{config.absolute_test_case_path}/input1.input", contents="input data") + program_config = ProgramConfig() + program_config.name = "program_1" + program_config.path = "program_1" + program_config.absolute_bin_path = "/bin/program_1" + program_config.sequence = 0 + config.program_configs = [program_config] + program = Program(program_config, TestBenchSettings()) test_case = TestCase(config, logger) - mocked_program = mocker.Mock(spec=Program) - mocked_program.run.side_effect = self.create_file_side_effect(f"{config.absolute_test_case_path}/out1.out") - mocked_program.name.return_value = "program_1" - test_case._TestCase__programs = [[0, mocked_program]] - mocker.patch("src.suite.test_case.TestCase.__initializeProgramList__") + def run_side_effect(self, _logger) -> None: + with open(f"{config.absolute_test_case_path}/out1.out", "w") as file: + file.write("File content") + self._Program__last_return_code = 0 + self._Program__error = None + + mocker.patch("src.suite.program.Program.run", new=run_side_effect) # Act - test_case.run([]) + test_case.run([program]) # Assert assert fs.exists(f"{config.absolute_test_case_path}/_tb3_char.run"), "during run no _tb3_char.run was created" @@ -40,6 +51,45 @@ def test_run__tb3_char_output_file(self, mocker: MockerFixture, fs: FakeFilesyst else: assert line == reference, "_tb3_char.run content does not match" + def test_run__changes_runfile(self, mocker: MockerFixture, fs: FakeFilesystem) -> None: + # Arrange + logger = mocker.Mock(spec=FileLogger) + config = self.create_test_case_config("name_2") + fs.create_dir(config.absolute_test_case_path) + input_path = f"{config.absolute_test_case_path}/input.mdu" + fs.create_file(input_path, contents="input data") + program_config = ProgramConfig() + program_config.name = "program_1" + program_config.path = "program_1" + program_config.absolute_bin_path = "/bin/program_1" + program_config.sequence = 0 + config.program_configs = [program_config] + program = Program(program_config, TestBenchSettings()) + test_case = TestCase(config, logger) + + def run_side_effect(self, _logger) -> None: + # Modify existing file. + with open(input_path, "w") as file: + file.write("updated input") + # Add new file. + with open(f"{config.absolute_test_case_path}/new.out", "w") as file: + file.write("new output") + self._Program__last_return_code = 0 + self._Program__error = None + + mocker.patch("src.suite.program.Program.run", new=run_side_effect) + + # Act + test_case.run([program]) + + # Assert + run_file = f"{config.absolute_test_case_path}/_tb3_char.run" + assert fs.exists(run_file) + with open(run_file, "r") as file: + lines = [line.strip() for line in file.readlines()] + assert any(line == "Output_changed:input.mdu" for line in lines) + assert any(line == "Output_added:new.out" for line in lines) + def create_test_case_config(self, name: str, platform: Optional[str] = "lnx64") -> TestCaseConfig: config = TestCaseConfig() config.name = name diff --git a/test/deltares_testbench/test/utils/test_paths.py b/test/deltares_testbench/test/utils/test_paths.py index 0d6f4c8990..bbaad92f30 100644 --- a/test/deltares_testbench/test/utils/test_paths.py +++ b/test/deltares_testbench/test/utils/test_paths.py @@ -1,15 +1,124 @@ +import os +from pathlib import Path + import pytest from src.utils.paths import Paths +@pytest.mark.parametrize( + ("path", "expected"), + [ + pytest.param("//server/folder/rest/sub", ("server", "folder", f"rest{os.sep}sub{os.sep}"), id="fwd"), + pytest.param( + "//server/folder/rest/sub/", + ("server", "folder", f"rest{os.sep}sub{os.sep}"), + id="fwd-trailing", + ), + pytest.param( + r"\\server\folder\rest\sub", + ("server", "folder", f"rest{os.sep}sub{os.sep}"), + id="back", + ), + pytest.param( + r"\\server\folder\rest\sub\\", + ("server", "folder", f"rest{os.sep}sub{os.sep}"), + id="back-trailing", + ), + ], +) +def test_split_network_path(path: str, expected: tuple[str, str, str]) -> None: + # Arrange + paths = Paths() + + # Act + server, folder, rest = paths.splitNetworkPath(path) + + # Assert + assert (server, folder, rest) == expected + + +@pytest.mark.parametrize( + ("path", "expected"), + [ + pytest.param("/etc/path", True, id="linux-absolute"), + pytest.param("/etc/path/file.txt", True, id="linux-file"), + pytest.param("relative/path", True, id="linux-relative"), + pytest.param("justone/", True, id="single-slash-end"), + pytest.param("/justone", False, id="single-slash-begin"), + pytest.param("http://example.com/a/b", False, id="url-http"), + pytest.param("https://example.com/a/b", False, id="url-https"), + pytest.param(r"C:\\temp\\file.txt", True, id="windows-path"), + ], +) +def test_is_path(path: str, expected: bool) -> None: + # Arrange + paths = Paths() + + # Act + result = paths.isPath(path) + + # Assert + assert result is expected + + +@pytest.mark.parametrize( + ("path", "expected"), + [ + pytest.param("/etc/path", True, id="linux-absolute"), + pytest.param("relative/path", False, id="linux-relative"), + pytest.param(r"C:\\temp\\file.txt", True, id="windows-absolute"), + ], +) +def test_is_absolute(path: str, expected: bool) -> None: + # Arrange + paths = Paths() + + # Act + result = paths.isAbsolute(path) + + # Assert + assert result is expected + + +def test_rebuild_to_local_path_normalizes_mixed_separators() -> None: + # Arrange + paths = Paths() + mixed = "/a/b\\c/d" + + # Act + result = paths.rebuildToLocalPath(mixed) + + # Assert + assert result == os.path.join("/", "a", "b", "c", "d") + + +def test_rebuild_to_local_path_preserves_drive_letter() -> None: + # Arrange + paths = Paths() + mixed = "C:/folder/sub" + + # Act + result = paths.rebuildToLocalPath(mixed) + + # Assert + assert result == os.path.join("C:\\", "folder", "sub") + + @pytest.mark.parametrize( ("left", "right", "expected"), [ pytest.param(None, "fruit", "fruit", id="none-left"), pytest.param("", "fruit", "fruit", id="empty-left"), + pytest.param("/etc/path", None, "/etc/path", id="none-right"), + pytest.param("/etc/path", "", "/etc/path", id="empty-right"), pytest.param("/etc/path", "child", "/etc/path/child", id="linux-path"), pytest.param(r"C:\user\documents", "child", r"C:\user\documents\child", id="windows-path"), + pytest.param("/etc/path", "/child", "/etc/path/child", id="trim-right-fwd"), + pytest.param(r"C:\root", r"\child", r"C:\root\child", id="trim-right-back"), + pytest.param(r"C:\root\\", "child", r"C:\root\child", id="trim-left-back"), + pytest.param("root", "sub/child", "root/sub/child", id="right-fwd-no-left-slash"), + pytest.param("root", r"sub\child", r"root\sub\child", id="right-back-no-left-slash"), ], ) def test_merge_path_elements(left: str | None, right: str, expected: str) -> None: @@ -30,6 +139,7 @@ def test_merge_path_elements(left: str | None, right: str, expected: str) -> Non pytest.param("", ("fruit", "apple"), "fruit/apple", id="empty-base"), pytest.param("/etc", ("sub1", "sub2"), "/etc/sub1/sub2", id="linux-base"), pytest.param(r"C:\user", ("documents",), r"C:\user\documents", id="windows-base"), + pytest.param(None, ("", None), "", id="all-empty"), ], ) def test_merge_full_path(left: str | None, segments: tuple[str, ...], expected: str) -> None: @@ -41,3 +151,39 @@ def test_merge_full_path(left: str | None, segments: tuple[str, ...], expected: # Assert assert result == expected + + +def test_find_all_sub_files_returns_relative_files(tmp_path: Path) -> None: + # Arrange + paths = Paths() + root = tmp_path / "root" + root.mkdir() + (root / "b.txt").write_text("root") + (root / "a").mkdir() + (root / "a" / "file1.txt").write_text("nested") + + # Act + result = paths.findAllSubFiles(str(root)) + + # Assert + assert sorted(result) == sorted(["b.txt", os.path.join("a", "file1.txt")]) + + +def test_find_all_sub_folders_respects_exclude(tmp_path: Path) -> None: + # Arrange + paths = Paths() + root = tmp_path / "root" + root.mkdir() + (root / "keep").mkdir() + (root / "skip").mkdir() + (root / "skip" / "child").mkdir() + + # Act + result = paths.findAllSubFolders(str(root), "skip") + + # Assert + expected = { + os.path.abspath(str(root)), + os.path.abspath(str(root / "keep")), + } + assert set(result) == expected