Merge branch 'fix/broken_fatfsgen_tests' into 'master'

fix(storage/fatfsgen): broken test_apps scripts for fatfsgen

Closes IDFCI-2644 and IDFCI-2645

See merge request espressif/esp-idf!36317

tools/test_apps/storage/fatfsgen/pytest_fatfsgen_example.py

@@ -1,13 +1,14 @@
-# SPDX-FileCopyrightText: 2022-2024 Espressif Systems (Shanghai) CO LTD
+# SPDX-FileCopyrightText: 2022-2025 Espressif Systems (Shanghai) CO LTD
 # SPDX-License-Identifier: Unlicense OR CC0-1.0
 import os
 import re
 import shutil
 import sys
+import typing as t
 from datetime import datetime
 from subprocess import run
 from subprocess import STDOUT
-from typing import List
+from time import sleep
 
 import pytest
 from pytest_embedded import Dut
@@ -20,35 +21,103 @@ sys.path.append(parttool_dir)
 from parttool import PartitionName, ParttoolTarget  # noqa E402  # pylint: disable=C0413
 
 
-def file_(x: str, content_: str = 'hey this is a test') -> dict:
+FileStructure = t.Dict[str, t.Union[t.Optional[str], 'FileStructure']]
-    return {
-        'type': 'file',
-        'name': x,
-        'content': content_
-    }
 
 
-def generate_local_folder_structure(structure_: dict, path_: str) -> None:
+class DirectoryStructureError(Exception):
-    if structure_['type'] == 'folder':
+    '''Base exception for directory structure errors.'''
-        new_path_ = os.path.join(path_, structure_['name'])
+    pass
-        os.makedirs(new_path_)
+
-        for item_ in structure_['content']:
+
-            generate_local_folder_structure(item_, new_path_)
+class MissingStructureError(DirectoryStructureError):
+    '''Raised when a directory or file is missing.'''
+    pass
+
+
+class ContentMismatchError(DirectoryStructureError):
+    '''Raised when a file's content does not match the expected content.'''
+    pass
+
+
+class IOErrorInStructure(DirectoryStructureError):
+    '''Raised when an I/O error occurs during structure validation.'''
+    pass
+
+
+def validate_directory_structure(base_path: str,
+                                 expected_structure: FileStructure) -> None:
+    '''
+    Checks if the directory structure
+    and file contents match the expected structure.
+
+    :param base_path: The root directory to check.
+    :param expected_structure: A nested dictionary representing
+                               the expected structure.
+                               Keys are folder names or file names.
+
+                               For files, values are either:
+                                   - None (skip content check) or
+                                   - string (expected content).
+
+                               For folders values are dictionaries defining
+                               their structure.
+
+    :raises DirectoryStructureError: If the structure or contents do not match.
+    '''
+    def normalize_case(items: t.List[str]) -> t.Dict[str, str]:
+        return {item.lower(): item for item in items}
+
+    def escape_output(text: str) -> str:
+        return (text.replace('"', '\"')
+                    .replace('\n', '\\n')
+                    .replace('\t', '\\t')
+                    .replace('\r', '\\r'))
+
+    def validate_structure(current_path: str,
+                           structure: FileStructure) -> None:
+        current_items = os.listdir(current_path)
+        normalized_items = normalize_case(current_items)
+
+        for name, substructure in structure.items():
+            expected_name = normalized_items.get(name.lower())
+            if not expected_name:
+                raise MissingStructureError(
+                    f'Missing item: {os.path.join(current_path, name)}')
+
+            full_path = os.path.join(current_path, expected_name)
+
+            # File
+            if isinstance(substructure, str) or substructure is None:
+                if not os.path.isfile(full_path):
+                    raise MissingStructureError(f'Missing file: {full_path}')
+
+                if isinstance(substructure, str):  # Check file content
+                    try:
+                        with open(full_path, 'r', encoding='utf-8') as file:
+                            content = file.read()
+                        if content != substructure:
+                            raise ContentMismatchError(
+                                f'Content mismatch in file: {full_path}\n'
+                                f'Expected: "{escape_output(substructure)}"\n'
+                                f'Found: "{escape_output(content)}"'
+                            )
+                    except OSError as e:
+                        raise IOErrorInStructure(
+                            f'Error reading file {full_path}: {e}')
+
+            # Folder
+            elif isinstance(substructure, dict):
+                if not os.path.isdir(full_path):
+                    raise MissingStructureError(
+                        f'Missing directory: {full_path}')
+                # Recursively check the subdirectory
+                validate_structure(full_path, substructure)
+
             else:
-        new_path_ = os.path.join(path_, structure_['name'])
+                raise DirectoryStructureError(
-        with open(new_path_, 'w') as f_:
+                    f'Invalid structure definition for {name}.')
-            f_.write(structure_['content'])
 
-
+    validate_structure(base_path, expected_structure)
-def compare_folders(fp1: str, fp2: str) -> bool:
-    if os.path.isdir(fp1) != os.path.isdir(fp2):
-        return False
-    if os.path.isdir(fp1):
-        if set(os.listdir(fp1)) != set(os.listdir(fp2)):
-            return False
-        return all([compare_folders(os.path.join(fp1, path_), os.path.join(fp2, path_)) for path_ in os.listdir(fp1)])
-    with open(fp1, 'rb') as f1_, open(fp2, 'rb') as f2_:
-        return f1_.read() == f2_.read()
 
 
 @pytest.mark.esp32
@@ -63,164 +132,109 @@ def compare_folders(fp1: str, fp2: str) -> bool:
                                     'test_read_write_partition_gen_ln_default_dt',
                                     ], indirect=True)
 def test_examples_fatfsgen(config: str, dut: Dut) -> None:
+    # Default timeout
+    timeout = 20
+
+    def expect(msg: str, timeout: int = timeout) -> None:
+        dut.expect(msg, timeout=timeout)
+
     # Expects list of strings sequentially
-    def expect_all(msg_list: List[str], to: int) -> None:
+    def expect_all(msg_list: t.List[str], timeout: int = timeout) -> None:
         for msg in msg_list:
-            dut.expect(msg, timeout=to)
+            expect(msg, timeout)
 
     # Expects prefix string followed by date in the format 'yyyy-mm-dd'
-    def expect_date(prefix: str, to: int) -> datetime:
+    def expect_date(prefix: str, timeout: int = timeout) -> datetime:
         expect_str = prefix + '(\\d+)-(\\d+)-(\\d+)'
-        match_ = dut.expect(re.compile(str.encode(expect_str)), timeout=to)
+        match_ = dut.expect(re.compile(str.encode(expect_str)),
+                            timeout=timeout)
         year_ = int(match_[1].decode())
         month_ = int(match_[2].decode())
         day_ = int(match_[3].decode())
         return datetime(year_, month_, day_)
 
-    # Calculates absolute difference in days between date_reference and date_actual.
+    # Calculates absolute difference in days
-    # Raises exception if difference exceeds tolerance
+    # between date_reference and date_actual.
-    def evaluate_dates(date_reference: datetime, date_actual: datetime, days_tolerance: int) -> None:
+    # Raises exception if difference exceeds date_tolerance
+    # 30 days by default,
+    def check_dates(date_reference: datetime,
+                    date_actual: datetime,
+                    days_date_tolerance: int = 30) -> None:
         td = date_actual - date_reference
-        if abs(td.days) > days_tolerance:
+        if abs(td.days) > days_date_tolerance:
-            raise Exception(f'Too big date difference. Actual: {date_actual}, reference: {date_reference}, tolerance: {days_tolerance} day(s)')
+            raise Exception(f'Too big date difference. \
+                Actual: {date_actual}, \
+                reference: {date_reference}, \
+                date_tolerance: {days_date_tolerance} day(s)')
 
-    # Expect timeout
+    fatfs_parser_path = os.path.join(idf_path, 'components',
-    timeout = 20
+                                     'fatfs', 'fatfsparse.py')
 
-    # We tolerate 30 days difference between actual file creation and date when test was executed.
+    config_read_only = '_read_only' in config
-    tolerance = 30
+    config_long_names = '_ln' in config
-    filename_ln = 'sublongnames/testlongfilenames.txt'
+    config_default_date = '_default_dt' in config
-    filename_sn = 'sub/test.txt'
-    date_modified = datetime.today()
-    date_default = datetime(1980, 1, 1)
-    fatfs_parser_path = os.path.join(idf_path, 'components', 'fatfs', 'fatfsparse.py')
 
-    if config in ['test_read_write_partition_gen', 'test_read_write_partition_gen_default_dt']:
+    expect('example: Mounting FAT filesystem')
-        filename = filename_sn
+
-        filename_expected = f'/spiflash/{filename}'
+    if not config_read_only:
-        date_ref = date_default if config == 'test_read_write_partition_gen_default_dt' else date_modified
+        expect_all(['example: Opening file',
-        expect_all(['example: Mounting FAT filesystem',
-                    'example: Opening file',
                     'example: File written',
                     'example: Reading file',
-                    'example: Read from file: \'This is written by the device\'',
+                    'example: Read from file: '
-                    'example: Reading file'], timeout)
+                    + '\'This is written by the device\''], 30)
-        date_act = expect_date(f'The file \'{filename_expected}\' was modified at date: ', timeout)
-        evaluate_dates(date_ref, date_act, tolerance)
-        expect_all(['example: Read from file: \'This is generated on the host\'',
-                    'example: Unmounting FAT filesystem',
-                    'example: Done'], timeout)
 
-        target = ParttoolTarget(dut.port)
+    expect('example: Reading file')
+
+    stat_filename = 'sublongnames/testlongfilenames.txt' \
+                    if config_long_names \
+                    else 'sub/test.txt'
+
+    modification_date = expect_date(
+        f'The file \'/spiflash/{stat_filename}\' was modified at date: ')
+
+    date_ref = datetime(1980, 1, 1) \
+        if config_default_date \
+        else datetime.today()
+
+    check_dates(date_ref, modification_date)
+
+    expect_all(['example: Unmounting FAT filesystem',
+                'example: Done'])
+
+    dut.serial.close()
+    sleep(1)
+
+    target = ParttoolTarget(dut.serial.port, 1843200)
     target.read_partition(PartitionName('storage'), 'temp.img')
-        run(['python', fatfs_parser_path, '--wear-leveling', 'temp.img'], stderr=STDOUT)
+    if config_long_names:
-        folder_ = {
+        run(['python', fatfs_parser_path, '--long-name-support', 'temp.img'],
-            'type': 'folder',
+            stderr=STDOUT)
-            'name': 'SUB',
+    else:
-            'content': [
+        run(['python', fatfs_parser_path, 'temp.img'],
-                file_('TEST.TXT', content_='this is test\n'),
+            stderr=STDOUT)
-            ]
+
+    long_names_read_only: FileStructure = {
+        'sublongnames': {
+            'testlongfilenames.txt': 'this is test; long name it has\n',
+        },
+        'hellolongname.txt': 'This is generated on the host; long name it has\n',
     }
-        struct_: dict = {
+
-            'type': 'folder',
+    short_names_read_only: FileStructure = {
-            'name': 'testf',
+        'sub': {
-            'content': [
+            'test.txt': 'this is test\n',
-                file_('HELLO.TXT', content_='This is generated on the host\n'),
+        },
-                file_('INNER.TXT', content_='This is written by the device'),
+        'hello.txt': 'This is generated on the host\n',
-                folder_
-            ]
     }
-        generate_local_folder_structure(struct_, path_='.')
+
+    file_structure: FileStructure = long_names_read_only \
+        if config_long_names \
+        else short_names_read_only
+
+    if not config_read_only and not config_long_names:
+        file_structure['inner.txt'] = 'This is written by the device'
+
     try:
-            assert compare_folders('testf', 'Espressif')
+        validate_directory_structure('Espressif', file_structure)
     finally:
         shutil.rmtree('Espressif', ignore_errors=True)
-            shutil.rmtree('testf', ignore_errors=True)
-
-    elif config in ['test_read_only_partition_gen', 'test_read_only_partition_gen_default_dt']:
-        filename = filename_sn
-        filename_expected = f'/spiflash/{filename}'
-        date_ref = date_default if config == 'test_read_only_partition_gen_default_dt' else date_modified
-        expect_all(['example: Mounting FAT filesystem',
-                    'example: Reading file'], timeout)
-        date_act = expect_date(f'The file \'{filename_expected}\' was modified at date: ', timeout)
-        evaluate_dates(date_ref, date_act, tolerance)
-        expect_all(['example: Read from file: \'this is test\'',
-                    'example: Unmounting FAT filesystem',
-                    'example: Done'], timeout)
-        target = ParttoolTarget(dut.port)
-        target.read_partition(PartitionName('storage'), 'temp.img')
-        run(['python', fatfs_parser_path, '--long-name-support', 'temp.img'], stderr=STDOUT)
-        folder_ = {
-            'type': 'folder',
-            'name': 'sublongnames',
-            'content': [
-                file_('testlongfilenames.txt', content_='this is test; long name it has\n'),
-            ]
-        }
-        struct_ = {
-            'type': 'folder',
-            'name': 'testf',
-            'content': [
-                file_('hellolongname.txt', content_='This is generated on the host; long name it has\n'),
-                folder_
-            ]
-        }
-        generate_local_folder_structure(struct_, path_='.')
-        try:
-            assert compare_folders('testf', 'Espressif')
-        finally:
-            shutil.rmtree('Espressif', ignore_errors=True)
-            shutil.rmtree('testf', ignore_errors=True)
-
-    elif config in ['test_read_write_partition_gen_ln', 'test_read_write_partition_gen_ln_default_dt']:
-        filename = filename_ln
-        filename_expected = f'/spiflash/{filename}'
-        date_ref = date_default if config == 'test_read_write_partition_gen_ln_default_dt' else date_modified
-        expect_all(['example: Mounting FAT filesystem',
-                    'example: Opening file',
-                    'example: File written',
-                    'example: Reading file',
-                    'example: Read from file: \'This is written by the device\'',
-                    'example: Reading file'], timeout)
-        date_act = expect_date(f'The file \'{filename_expected}\' was modified at date: ', timeout)
-        evaluate_dates(date_ref, date_act, tolerance)
-        expect_all(['example: Read from file: \'This is generated on the host; long name it has\'',
-                    'example: Unmounting FAT filesystem',
-                    'example: Done'], timeout)
-
-    elif config in ['test_read_only_partition_gen_ln', 'test_read_only_partition_gen_ln_default_dt']:
-        filename = filename_ln
-        filename_expected = f'/spiflash/{filename}'
-        date_ref = date_default if config == 'test_read_only_partition_gen_ln_default_dt' else date_modified
-        expect_all(['example: Mounting FAT filesystem',
-                    'example: Reading file'], timeout)
-        date_act = expect_date(f'The file \'{filename_expected}\' was modified at date: ', timeout)
-        evaluate_dates(date_ref, date_act, tolerance)
-        expect_all(['example: Read from file: \'this is test; long name it has\'',
-                    'example: Unmounting FAT filesystem',
-                    'example: Done'], timeout)
-        target = ParttoolTarget(dut.port)
-        target.read_partition(PartitionName('storage'), 'temp.img')
-        run(['python', fatfs_parser_path, 'temp.img'], stderr=STDOUT)
-        folder_ = {
-            'type': 'folder',
-            'name': 'SUB',
-            'content': [
-                file_('TEST.TXT', content_='this is test\n'),
-            ]
-        }
-        struct_ = {
-            'type': 'folder',
-            'name': 'testf',
-            'content': [
-                file_('HELLO.TXT', content_='This is generated on the host\n'),
-                folder_
-            ]
-        }
-        generate_local_folder_structure(struct_, path_='.')
-        try:
-            assert compare_folders('testf', 'Espressif')
-        finally:
-            shutil.rmtree('Espressif', ignore_errors=True)
-            shutil.rmtree('testf', ignore_errors=True)

tools/test_apps/storage/fatfsgen/sdkconfig.ci.test_read_write_partition_gen

@@ -2,4 +2,4 @@ CONFIG_EXAMPLE_FATFS_MODE_READ_ONLY=n
 CONFIG_FATFS_LFN_HEAP=n
 CONFIG_FATFS_LFN_STACK=n
 CONFIG_FATFS_LFN_NONE=y
-CONFIG_EXAMPLE_FATFS_WRITE_COUNT=300
+CONFIG_EXAMPLE_FATFS_WRITE_COUNT=30

tools/test_apps/storage/fatfsgen/sdkconfig.ci.test_read_write_partition_gen_default_dt

@@ -3,4 +3,4 @@ CONFIG_FATFS_LFN_HEAP=n
 CONFIG_FATFS_LFN_STACK=n
 CONFIG_FATFS_LFN_NONE=y
 CONFIG_EXAMPLE_FATFS_DEFAULT_DATETIME=y
-CONFIG_EXAMPLE_FATFS_WRITE_COUNT=300
+CONFIG_EXAMPLE_FATFS_WRITE_COUNT=30

tools/test_apps/storage/fatfsgen/sdkconfig.ci.test_read_write_partition_gen_ln

@@ -2,4 +2,4 @@ CONFIG_EXAMPLE_FATFS_MODE_READ_ONLY=n
 CONFIG_FATFS_LFN_HEAP=y
 CONFIG_FATFS_LFN_NONE=n
 CONFIG_FATFS_LFN_STACK=n
-CONFIG_EXAMPLE_FATFS_WRITE_COUNT=300
+CONFIG_EXAMPLE_FATFS_WRITE_COUNT=30

tools/test_apps/storage/fatfsgen/sdkconfig.ci.test_read_write_partition_gen_ln_default_dt

@@ -3,4 +3,4 @@ CONFIG_FATFS_LFN_HEAP=y
 CONFIG_FATFS_LFN_NONE=n
 CONFIG_FATFS_LFN_STACK=n
 CONFIG_EXAMPLE_FATFS_DEFAULT_DATETIME=y
-CONFIG_EXAMPLE_FATFS_WRITE_COUNT=300
+CONFIG_EXAMPLE_FATFS_WRITE_COUNT=30