Merge pull request #3403 from rommapp/claude/exciting-heisenberg-61PF9

Integrate Hypothesis property-based testing for the backend

Author: gantoine

.github/workflows/pytest.yml

@@ -80,6 +80,7 @@ jobs:
           ROMM_DB_DRIVER: ${{ matrix.db }}
           REDIS_HOST: 127.0.0.1
           REDIS_PORT: ${{ job.services.valkey.ports['6379'] }}
+          HYPOTHESIS_PROFILE: ci
         run: |
           cd backend
           uv run pytest -vv --maxfail=10 --junitxml=pytest-report.xml --cov --cov-report xml:coverage.xml --cov-config=.coveragerc .

.gitignore

@@ -54,6 +54,7 @@ romm_mock
 backend/romm_test/resources
 backend/romm_test/logs
 .pytest_cache
+.hypothesis
 /romm_test
 .coverage
 pytest-report.xml

backend/tests/conftest.py

@@ -1,7 +1,9 @@
+import os
 from datetime import datetime, timedelta, timezone
 
 import alembic.config
 import pytest
+from hypothesis import settings
 from joserfc import jwt
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
@@ -30,6 +32,10 @@
 engine = create_engine(ConfigManager.get_db_engine(), pool_pre_ping=True)
 session = sessionmaker(bind=engine, expire_on_commit=False)
 
+settings.register_profile("ci", max_examples=200, deadline=None)
+settings.register_profile("dev", max_examples=50, deadline=None)
+settings.load_profile(os.getenv("HYPOTHESIS_PROFILE", "dev"))
+
 
 @pytest.fixture(scope="session", autouse=True)
 def setup_database():

backend/tests/handler/filesystem/test_roms_handler.py

@@ -1,12 +1,19 @@
 import os
 import shutil
+import tempfile
 from pathlib import Path
 from unittest.mock import Mock, patch
 
 import pytest
+from hypothesis import assume, given
+from hypothesis import strategies as st
 from tests._zipfile_shim import reload_zipfile
 
 from config.config_manager import LIBRARY_BASE_PATH, Config
+from handler.filesystem.base_handler import (
+    LANGUAGES_BY_SHORTCODE,
+    REGIONS_BY_SHORTCODE,
+)
 from handler.filesystem.roms_handler import (
     FileHash,
     FSRomsHandler,
@@ -1546,3 +1553,99 @@ def test_extract_chd_hash_max_sha1(self, tmp_path):
         assert result
         assert result == "f" * 40
         assert len(result) == 40
+
+
+KNOWN_REGION_NAMES = frozenset(REGIONS_BY_SHORTCODE.values())
+KNOWN_LANGUAGE_NAMES = frozenset(LANGUAGES_BY_SHORTCODE.values())
+
+region_code = st.sampled_from(sorted(REGIONS_BY_SHORTCODE))
+language_code = st.sampled_from(sorted(LANGUAGES_BY_SHORTCODE))
+
+
+class TestParseTagsProperties:
+    """Property-based tests for FSRomsHandler.parse_tags."""
+
+    handler = FSRomsHandler()
+
+    @given(st.text())
+    def test_never_raises_on_arbitrary_input(self, fs_name: str):
+        self.handler.parse_tags(fs_name)
+
+    @given(st.text())
+    def test_is_deterministic(self, fs_name: str):
+        assert self.handler.parse_tags(fs_name) == self.handler.parse_tags(fs_name)
+
+    @given(st.lists(region_code), st.lists(language_code))
+    def test_known_codes_map_to_known_names(self, regions, languages):
+        fs_name = "Game"
+        for code in regions:
+            fs_name += f"({code})"
+        for code in languages:
+            fs_name += f"({code})"
+        fs_name += ".rom"
+
+        parsed = self.handler.parse_tags(fs_name)
+
+        assert set(parsed.regions) <= KNOWN_REGION_NAMES
+        assert set(parsed.languages) <= KNOWN_LANGUAGE_NAMES
+        # Every supplied code resolves to its mapped full name.
+        assert {REGIONS_BY_SHORTCODE[c] for c in regions} <= set(parsed.regions)
+        assert {LANGUAGES_BY_SHORTCODE[c] for c in languages} <= set(parsed.languages)
+
+
+def _chd_header_bytes(version: int, sha1: bytes) -> bytes:
+    """Build a 124-byte CHD header with the given version and combined SHA1."""
+    header = bytearray(124)
+    header[0:8] = b"MComprHD"
+    header[12:16] = version.to_bytes(4, "big")
+    header[84:104] = sha1
+    return bytes(header)
+
+
+def _run_extract(data: bytes) -> str:
+    fd, path = tempfile.mkstemp(suffix=".chd")
+    try:
+        with os.fdopen(fd, "wb") as f:
+            f.write(data)
+        return extract_chd_hash(Path(path))
+    finally:
+        os.unlink(path)
+
+
+class TestExtractCHDHashProperties:
+    """Property-based tests for extract_chd_hash."""
+
+    @given(sha1=st.binary(min_size=20, max_size=20))
+    def test_valid_v5_returns_embedded_sha1(self, sha1: bytes):
+        result = _run_extract(_chd_header_bytes(5, sha1))
+        assert result == sha1.hex()
+        assert len(result) == 40
+
+    @given(
+        version=st.integers(min_value=0, max_value=2**32 - 1),
+        sha1=st.binary(min_size=20, max_size=20),
+    )
+    def test_non_v5_version_returns_empty(self, version: int, sha1: bytes):
+        assume(version != 5)
+        assert _run_extract(_chd_header_bytes(version, sha1)) == ""
+
+    @given(data=st.binary(min_size=16))
+    def test_wrong_signature_returns_empty(self, data: bytes):
+        assume(data[:8] != b"MComprHD")
+        assert _run_extract(data) == ""
+
+    @given(data=st.binary(max_size=15))
+    def test_too_short_for_signature_returns_empty(self, data: bytes):
+        assert _run_extract(data) == ""
+
+    @given(
+        sha1=st.binary(min_size=20, max_size=20),
+        truncate_len=st.integers(min_value=16, max_value=103),
+    )
+    def test_truncated_before_sha1_returns_empty(self, sha1: bytes, truncate_len: int):
+        truncated = _chd_header_bytes(5, sha1)[:truncate_len]
+        assert _run_extract(truncated) == ""
+
+    @given(data=st.binary())
+    def test_never_raises_on_arbitrary_bytes(self, data: bytes):
+        _run_extract(data)

backend/tests/handler/metadata/test_launchbox_utils.py

@@ -0,0 +1,88 @@
+"""Property-based tests for the LaunchBox metadata parsing helpers."""
+
+from datetime import datetime
+
+from hypothesis import assume, given
+from hypothesis import strategies as st
+
+from handler.metadata.launchbox_handler.utils import (
+    dedupe_words,
+    parse_list,
+    parse_release_date,
+    sanitize_filename,
+)
+
+LB_INVALID_CHARS = set("\\/|<>\"?*:'")
+
+
+class TestParseList:
+    @given(st.text() | st.none())
+    def test_elements_are_stripped_and_non_empty(self, value):
+        result = parse_list(value)
+        assert isinstance(result, list)
+        for item in result:
+            assert item == item.strip()
+            assert item != ""
+
+    @given(st.lists(st.text(alphabet=st.characters(blacklist_characters=";,"))))
+    def test_roundtrip_via_semicolon_join(self, items):
+        cleaned = [i.strip() for i in items if i.strip()]
+        assert parse_list(";".join(cleaned)) == cleaned
+
+
+class TestDedupeWords:
+    @given(st.lists(st.text() | st.none()))
+    def test_no_duplicate_lowercased_keys(self, values):
+        result = dedupe_words(values)
+        keys = [v.lower() for v in result]
+        assert len(keys) == len(set(keys))
+
+    @given(st.lists(st.text() | st.none()))
+    def test_result_is_subset_of_stripped_inputs(self, values):
+        stripped = {v.strip() for v in values if v is not None and v.strip()}
+        result = dedupe_words(values)
+        for item in result:
+            assert item in stripped
+
+
+class TestParseReleaseDate:
+    @given(st.text() | st.none())
+    def test_returns_none_or_int(self, value):
+        result = parse_release_date(value)
+        assert result is None or isinstance(result, int)
+
+    @given(
+        st.datetimes(
+            min_value=datetime(1971, 1, 1),
+            max_value=datetime(2100, 1, 1),
+        )
+    )
+    def test_valid_iso_dates_parse_to_timestamp(self, dt):
+        result = parse_release_date(dt.isoformat())
+        assert result == int(dt.timestamp())
+
+    @given(st.dates(min_value=datetime(1971, 1, 1).date()))
+    def test_date_only_format_parses(self, d):
+        result = parse_release_date(d.isoformat())
+        expected = int(
+            datetime(d.year, d.month, d.day).timestamp(),
+        )
+        assert result == expected
+
+
+class TestSanitizeFilename:
+    @given(st.text())
+    def test_output_contains_no_invalid_characters(self, stem):
+        result = sanitize_filename(stem)
+        assert not (set(result) & LB_INVALID_CHARS)
+
+    @given(st.text())
+    def test_is_idempotent(self, stem):
+        once = sanitize_filename(stem)
+        assert sanitize_filename(once) == once
+
+    @given(st.text())
+    def test_no_leading_or_trailing_space_or_dot(self, stem):
+        result = sanitize_filename(stem)
+        assume(result != "")
+        assert result == result.strip(" .")

backend/tests/utils/test_filesystem.py

@@ -5,8 +5,12 @@
 from unittest.mock import patch
 
 import pytest
+from hypothesis import assume, given
+from hypothesis import strategies as st
 
-from utils.filesystem import link_or_copy_file
+from utils.filesystem import link_or_copy_file, sanitize_filename
+
+INVALID_AFTER_SANITIZE = set('\\/:|*?"<>+\0')
 
 
 class TestLinkOrCopyFile:
@@ -180,3 +184,35 @@ def test_helper_uses_os_link_first(self, tmp_path):
         assert args[0] == source
         assert args[1].parent == dest.parent
         assert args[1].name.startswith(".romm_link_tmp_")
+
+
+class TestSanitizeFilename:
+    """Property-based tests for the cross-filesystem filename sanitizer."""
+
+    def _sanitize_or_none(self, name: str) -> str | None:
+        try:
+            return sanitize_filename(name)
+        except ValueError:
+            return None
+
+    @given(st.text())
+    def test_output_contains_no_invalid_characters(self, name: str):
+        result = self._sanitize_or_none(name)
+        assume(result is not None)
+        assert result is not None
+        assert not (set(result) & INVALID_AFTER_SANITIZE)
+
+    @given(st.text())
+    def test_output_is_stripped(self, name: str):
+        result = self._sanitize_or_none(name)
+        assume(result is not None)
+        assert result is not None
+        assert result == result.strip()
+        assert result != ""
+
+    @given(st.text())
+    def test_is_idempotent(self, name: str):
+        once = self._sanitize_or_none(name)
+        assume(once is not None)
+        assert once is not None
+        assert sanitize_filename(once) == once

backend/tests/utils/test_hashing.py

@@ -0,0 +1,20 @@
+import re
+
+from hypothesis import given
+from hypothesis import strategies as st
+
+from utils.hashing import crc32_to_hex
+
+HEX_8 = re.compile(r"^[0-9a-f]{8}$")
+
+
+@given(st.integers(min_value=0, max_value=2**32 - 1))
+def test_crc32_to_hex_is_8_lowercase_hex_chars(value: int):
+    result = crc32_to_hex(value)
+    assert HEX_8.match(result)
+
+
+@given(st.integers())
+def test_crc32_to_hex_roundtrips_modulo_32_bits(value: int):
+    result = crc32_to_hex(value)
+    assert int(result, 16) == value & 0xFFFFFFFF