From 516f6f61595c75bce5f36ac761791ca9bc7dceeb Mon Sep 17 00:00:00 2001 From: Mihai Maruseac Date: Tue, 16 Jul 2024 05:04:31 -0800 Subject: [PATCH] Cleanup testing: refactor and standardize. (#237) * Cleanup testing: refactor and standardize. Rename `fixtures_constants` to `test_support` as it will also contain test utilities that can be used to make tests more readable. Tests should now follow a similar structure (test file, test model changes, test folder, test deep folder, test empty, test special), with similar test names and similar spacing between the lines (in general arrange-act-assert, except for tests such as `test_model_with_empty_folder_hashes_differently_than_with_empty_file` which need to create an empty dir, compute a digest, delete the dir, create an empty file, compute a new digest, compare the 2 digests). Some tests are not duplicated in the other class as they don't make sense (e.g. itemized serializers produce no result on empty folders). Some tests have slightly different names in the other class due to behavior differences (e.g., corner cases around empty files). Remove the `test_special_file` that tests for FIFOs in every test suite as we only need to test it once (this way, the special way this test is written is not replicated multiple times). The code refactoring in the next PR will also remove some of the corner cases around these tests. Since Critique gets confused by lines changed in the PR (as tests got reordered), it might be better to just review the file in its entirety Signed-off-by: Mihai Maruseac * Remove unused imports Signed-off-by: Mihai Maruseac * Fix line length Signed-off-by: Mihai Maruseac * Simplify tests Signed-off-by: Mihai Maruseac --------- Signed-off-by: Mihai Maruseac --- model_signing/serialization/fixtures.py | 4 +- .../serialization/fixtures_constants.py | 18 - .../serialize_by_file_shard_test.py | 351 ++++++--------- .../serialization/serialize_by_file_test.py | 412 ++++++++---------- model_signing/serialization/test_support.py | 70 +++ 5 files changed, 395 insertions(+), 460 deletions(-) delete mode 100644 model_signing/serialization/fixtures_constants.py create mode 100644 model_signing/serialization/test_support.py diff --git a/model_signing/serialization/fixtures.py b/model_signing/serialization/fixtures.py index a64eb250..797515d7 100644 --- a/model_signing/serialization/fixtures.py +++ b/model_signing/serialization/fixtures.py @@ -16,14 +16,14 @@ import pytest -from model_signing.serialization import fixtures_constants +from model_signing.serialization import test_support # Note: Don't make fixtures with global scope as we are altering the models! @pytest.fixture def sample_model_file(tmp_path_factory): file = tmp_path_factory.mktemp("model") / "file" - file.write_bytes(fixtures_constants.KNOWN_MODEL_TEXT) + file.write_bytes(test_support.KNOWN_MODEL_TEXT) return file diff --git a/model_signing/serialization/fixtures_constants.py b/model_signing/serialization/fixtures_constants.py deleted file mode 100644 index d7917a72..00000000 --- a/model_signing/serialization/fixtures_constants.py +++ /dev/null @@ -1,18 +0,0 @@ -# Copyright 2024 The Sigstore Authors -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -"""Constants used in test fixtures and tests. Not part of the public API.""" - -KNOWN_MODEL_TEXT: bytes = b"This is a simple model" -ANOTHER_MODEL_TEXT: bytes = b"This is another simple model" diff --git a/model_signing/serialization/serialize_by_file_shard_test.py b/model_signing/serialization/serialize_by_file_shard_test.py index a631c933..0824ad5e 100644 --- a/model_signing/serialization/serialize_by_file_shard_test.py +++ b/model_signing/serialization/serialize_by_file_shard_test.py @@ -12,62 +12,19 @@ # See the License for the specific language governing permissions and # limitations under the License. -import os import pathlib -import pytest from model_signing.hashing import file from model_signing.hashing import memory from model_signing.manifest import manifest -from model_signing.serialization import fixtures_constants from model_signing.serialization import serialize_by_file_shard +from model_signing.serialization import test_support # Load fixtures from serialization/fixtures.py pytest_plugins = ("model_signing.serialization.fixtures",) -def _extract_digests_from_manifest( - manifest: manifest.FileLevelManifest, -) -> list[str]: - """Extracts the hex digest for every subject in a manifest. - - Used in multiple tests to check that we obtained the expected digests. - """ - return [d.digest_hex for d in manifest._item_to_digest.values()] - - -def _extract_items_from_manifest( - manifest: manifest.FileLevelManifest, -) -> dict[str, str]: - """Builds a dictionary representation of the items in a manifest. - - Every item is mapped to its digest. - - Used in multiple tests to check that we obtained the expected manifest. - """ - return { - str(path): digest.digest_hex - for path, digest in manifest._item_to_digest.items() - } - - -def _get_first_directory(path: pathlib.Path) -> pathlib.Path: - """Returns the first directory that is a children of path. - - It is assumed that there is always such a path. - """ - return [d for d in path.iterdir() if d.is_dir()][0] - - -def _get_first_file(path: pathlib.Path) -> pathlib.Path: - """Returns the first file that is a children of path. - - It is assumed that there is always such a path. - """ - return [f for f in path.iterdir() if f.is_file()][0] - - class TestShardedDFSSerializer: def _hasher_factory( @@ -88,12 +45,12 @@ def test_known_file(self, sample_model_file): serializer = serialize_by_file_shard.ShardedDFSSerializer( self._hasher_factory, memory.SHA256() ) - - manifest = serializer.serialize(sample_model_file) - expected = ( "2ca48c47d5311a9b2f9305519cd5f927dcef09404fc32ef7886abe8f11450eff" ) + + manifest = serializer.serialize(sample_model_file) + assert manifest.digest.digest_hex == expected def test_file_hash_is_not_same_as_hash_of_content(self, sample_model_file): @@ -102,11 +59,11 @@ def test_file_hash_is_not_same_as_hash_of_content(self, sample_model_file): ) manifest = serializer.serialize(sample_model_file) + digest = memory.SHA256(test_support.KNOWN_MODEL_TEXT).compute() - digest = memory.SHA256(fixtures_constants.KNOWN_MODEL_TEXT).compute() assert manifest.digest.digest_hex != digest.digest_hex - def test_file_model_hash_is_same_if_model_is_moved(self, sample_model_file): + def test_file_manifest_unchanged_when_model_moved(self, sample_model_file): serializer = serialize_by_file_shard.ShardedDFSSerializer( self._hasher_factory, memory.SHA256() ) @@ -126,7 +83,7 @@ def test_file_model_hash_changes_if_content_changes( ) manifest = serializer.serialize(sample_model_file) - sample_model_file.write_bytes(fixtures_constants.ANOTHER_MODEL_TEXT) + sample_model_file.write_bytes(test_support.ANOTHER_MODEL_TEXT) new_manifest = serializer.serialize(sample_model_file) assert manifest.digest.algorithm == new_manifest.digest.algorithm @@ -136,27 +93,24 @@ def test_directory_model_with_only_known_file(self, sample_model_file): serializer = serialize_by_file_shard.ShardedDFSSerializer( self._hasher_factory, memory.SHA256() ) + manifest_file = serializer.serialize(sample_model_file) + content_digest = memory.SHA256(test_support.KNOWN_MODEL_TEXT).compute() - model = sample_model_file.parent - manifest = serializer.serialize(model) + manifest = serializer.serialize(sample_model_file.parent) - expected = ( - "c030412c4c9e7f46396b591b1b6c4a4e40c15d9b9ca0b3512af8b20f3219c07f" - ) - assert manifest.digest.digest_hex == expected - digest = memory.SHA256(fixtures_constants.KNOWN_MODEL_TEXT).compute() - assert manifest.digest.digest_hex != digest.digest_hex + assert manifest_file != manifest + assert manifest.digest.digest_hex != content_digest.digest_hex def test_known_folder(self, sample_model_folder): serializer = serialize_by_file_shard.ShardedDFSSerializer( self._hasher_factory, memory.SHA256() ) - - manifest = serializer.serialize(sample_model_folder) - expected = ( "d22e0441cfa5ac2bc09715ddd88c802a7f97e29c93dc50f5498bab2954958ebb" ) + + manifest = serializer.serialize(sample_model_folder) + assert manifest.digest.digest_hex == expected def test_folder_model_hash_is_same_if_model_is_moved( @@ -173,148 +127,166 @@ def test_folder_model_hash_is_same_if_model_is_moved( assert manifest == new_manifest - def test_empty_file(self, empty_model_file): + def test_folder_model_empty_folder_gets_included(self, sample_model_folder): serializer = serialize_by_file_shard.ShardedDFSSerializer( self._hasher_factory, memory.SHA256() ) + manifest = serializer.serialize(sample_model_folder) - manifest = serializer.serialize(empty_model_file) + altered_dir = test_support.get_first_directory(sample_model_folder) + new_empty_dir = altered_dir / "empty" + new_empty_dir.mkdir() + new_manifest = serializer.serialize(sample_model_folder) - expected = ( - "5f2d126b0d3540c17481fdf724e31cf03b4436a2ebabaa1d2e94fe09831be64d" - ) - assert manifest.digest.digest_hex == expected + assert manifest != new_manifest - def test_directory_model_with_only_empty_file(self, empty_model_file): + def test_folder_model_empty_file_gets_included(self, sample_model_folder): serializer = serialize_by_file_shard.ShardedDFSSerializer( self._hasher_factory, memory.SHA256() ) - model = empty_model_file.parent + manifest = serializer.serialize(sample_model_folder) - manifest = serializer.serialize(model) + altered_dir = test_support.get_first_directory(sample_model_folder) + new_empty_file = altered_dir / "empty" + new_empty_file.write_text("") + new_manifest = serializer.serialize(sample_model_folder) - expected = ( - "74e81d0062f0a0674014c2f0e4b79985d5015f98a64089e7106a44d32e9ff11f" - ) - assert manifest.digest.digest_hex == expected + assert manifest != new_manifest - def test_empty_folder(self, empty_model_folder): + def test_folder_model_rename_file(self, sample_model_folder): serializer = serialize_by_file_shard.ShardedDFSSerializer( self._hasher_factory, memory.SHA256() ) + manifest = serializer.serialize(sample_model_folder) - manifest = serializer.serialize(empty_model_folder) + altered_dir = test_support.get_first_directory(sample_model_folder) + file_to_rename = test_support.get_first_file(altered_dir) + new_name = file_to_rename.with_name("new-file") + file_to_rename.rename(new_name) + new_manifest = serializer.serialize(sample_model_folder) - expected = ( - "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" - ) - assert manifest.digest.digest_hex == expected + assert manifest != new_manifest - def test_folder_model_empty_entry(self, sample_model_folder): + def test_folder_model_rename_dir(self, sample_model_folder): serializer = serialize_by_file_shard.ShardedDFSSerializer( self._hasher_factory, memory.SHA256() ) + manifest = serializer.serialize(sample_model_folder) - # Alter first directory within the model - dirs = [d for d in sample_model_folder.iterdir() if d.is_dir()] - altered_dir = dirs[0] + dir_to_rename = test_support.get_first_directory(sample_model_folder) + new_name = dir_to_rename.with_name("new-dir") + dir_to_rename.rename(new_name) + new_manifest = serializer.serialize(sample_model_folder) - new_empty_dir = altered_dir / "empty" - new_empty_dir.mkdir() - manifest1 = serializer.serialize(sample_model_folder) + assert manifest != new_manifest - new_empty_dir.rmdir() + def test_folder_model_replace_file_empty_folder(self, sample_model_folder): + serializer = serialize_by_file_shard.ShardedDFSSerializer( + self._hasher_factory, memory.SHA256() + ) + manifest = serializer.serialize(sample_model_folder) - new_empty_file = altered_dir / "empty" - new_empty_file.write_text("") - manifest2 = serializer.serialize(sample_model_folder) + altered_dir = test_support.get_first_directory(sample_model_folder) + file_to_replace = test_support.get_first_file(altered_dir) + file_to_replace.unlink() + file_to_replace.mkdir() + new_manifest = serializer.serialize(sample_model_folder) - assert manifest1.digest != manifest2.digest + assert manifest != new_manifest - def test_folder_model_rename_file(self, sample_model_folder): + def test_folder_model_change_file(self, sample_model_folder): serializer = serialize_by_file_shard.ShardedDFSSerializer( self._hasher_factory, memory.SHA256() ) - manifest1 = serializer.serialize(sample_model_folder) + manifest = serializer.serialize(sample_model_folder) - # Alter first directory within the model - dirs = [d for d in sample_model_folder.iterdir() if d.is_dir()] - altered_dir = dirs[0] + altered_dir = test_support.get_first_directory(sample_model_folder) + file_to_change = test_support.get_first_file(altered_dir) + file_to_change.write_bytes(test_support.KNOWN_MODEL_TEXT) + new_manifest = serializer.serialize(sample_model_folder) - # Alter first file in the altered_dir - files = [f for f in altered_dir.iterdir() if f.is_file()] - file_to_rename = files[0] + assert manifest != new_manifest - new_name = file_to_rename.with_name("new-file") - file_to_rename.rename(new_name) + def test_deep_folder(self, deep_model_folder): + serializer = serialize_by_file_shard.ShardedDFSSerializer( + self._hasher_factory, memory.SHA256() + ) + expected = ( + "52fa3c459aec58bc5f9702c73cb3c6b8fd19e9342aa3e4db851e1bde69ab1727" + ) - manifest2 = serializer.serialize(sample_model_folder) - assert manifest1.digest != manifest2.digest + manifest = serializer.serialize(deep_model_folder) - def test_folder_model_rename_dir(self, sample_model_folder): + assert manifest.digest.digest_hex == expected + + def test_empty_file(self, empty_model_file): serializer = serialize_by_file_shard.ShardedDFSSerializer( self._hasher_factory, memory.SHA256() ) - manifest1 = serializer.serialize(sample_model_folder) - - # Alter first directory within the model - dirs = [d for d in sample_model_folder.iterdir() if d.is_dir()] - dir_to_rename = dirs[0] + expected = ( + "5f2d126b0d3540c17481fdf724e31cf03b4436a2ebabaa1d2e94fe09831be64d" + ) - new_name = dir_to_rename.with_name("new-dir") - dir_to_rename.rename(new_name) + manifest = serializer.serialize(empty_model_file) - manifest2 = serializer.serialize(sample_model_folder) - assert manifest1.digest != manifest2.digest + assert manifest.digest.digest_hex == expected - def test_folder_model_replace_file_empty_folder(self, sample_model_folder): + def test_empty_folder(self, empty_model_folder): serializer = serialize_by_file_shard.ShardedDFSSerializer( self._hasher_factory, memory.SHA256() ) - manifest1 = serializer.serialize(sample_model_folder) - - # Alter first directory within the model - dirs = [d for d in sample_model_folder.iterdir() if d.is_dir()] - altered_dir = dirs[0] + expected = ( + "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + ) - # Replace first file in the altered_dir - files = [f for f in altered_dir.iterdir() if f.is_file()] - file_to_replace = files[0] - file_to_replace.unlink() - file_to_replace.mkdir() + manifest = serializer.serialize(empty_model_folder) - manifest2 = serializer.serialize(sample_model_folder) - assert manifest1.digest != manifest2.digest + assert manifest.digest.digest_hex == expected - def test_folder_model_change_file(self, sample_model_folder): + def test_directory_model_with_only_empty_file(self, empty_model_file): serializer = serialize_by_file_shard.ShardedDFSSerializer( self._hasher_factory, memory.SHA256() ) - manifest1 = serializer.serialize(sample_model_folder) - - # Alter first directory within the model - dirs = [d for d in sample_model_folder.iterdir() if d.is_dir()] - altered_dir = dirs[0] + expected = ( + "74e81d0062f0a0674014c2f0e4b79985d5015f98a64089e7106a44d32e9ff11f" + ) - # Alter first file in the altered_dir - files = [f for f in altered_dir.iterdir() if f.is_file()] - file_to_change = files[0] - file_to_change.write_bytes(fixtures_constants.KNOWN_MODEL_TEXT) + manifest = serializer.serialize(empty_model_file.parent) - manifest2 = serializer.serialize(sample_model_folder) - assert manifest1.digest != manifest2.digest + assert manifest.digest.digest_hex == expected - def test_deep_folder(self, deep_model_folder): + def test_empty_folder_hashes_differently_than_empty_file( + self, empty_model_file, empty_model_folder + ): serializer = serialize_by_file_shard.ShardedDFSSerializer( self._hasher_factory, memory.SHA256() ) - manifest = serializer.serialize(deep_model_folder) + folder_manifest = serializer.serialize(empty_model_folder) + file_manifest = serializer.serialize(empty_model_file) - expected = ( - "52fa3c459aec58bc5f9702c73cb3c6b8fd19e9342aa3e4db851e1bde69ab1727" + assert folder_manifest != file_manifest + + def test_model_with_empty_folder_hashes_differently_than_with_empty_file( + self, sample_model_folder + ): + serializer = serialize_by_file_shard.ShardedDFSSerializer( + self._hasher_factory, memory.SHA256() ) - assert manifest.digest.digest_hex == expected + + # Compute digest of model with empty folder + altered_dir = test_support.get_first_directory(sample_model_folder) + new_empty_dir = altered_dir / "empty" + new_empty_dir.mkdir() + folder_manifest = serializer.serialize(sample_model_folder) + + # Compute digest of model with empty file + new_empty_dir.rmdir() + new_empty_file = altered_dir / "empty" + new_empty_file.write_text("") + file_manifest = serializer.serialize(sample_model_folder) + + assert folder_manifest != file_manifest def test_max_workers_does_not_change_digest(self, sample_model_folder): serializer1 = serialize_by_file_shard.ShardedDFSSerializer( @@ -342,35 +314,6 @@ def test_shard_size_changes_digests(self, sample_model_folder): assert manifest1.digest.digest_value != manifest2.digest.digest_value - def test_special_file(self, sample_model_folder): - # Alter first directory within the model - dirs = [d for d in sample_model_folder.iterdir() if d.is_dir()] - altered_dir = dirs[0] - - # Create a pipe in the altered_dir - pipe = altered_dir / "pipe" - - try: - os.mkfifo(pipe) - except AttributeError: - # On Windows, `os.mkfifo` does not exist (it should not). - return # trivially pass the test - - serializer = serialize_by_file_shard.ShardedDFSSerializer( - self._hasher_factory, memory.SHA256() - ) - - with pytest.raises( - ValueError, match="Cannot use .* as file or directory" - ): - serializer.serialize(sample_model_folder) - - # Also do the same for the pipe itself - with pytest.raises( - ValueError, match="Cannot use .* as file or directory" - ): - serializer.serialize(pipe) - def _extract_shard_items_from_manifest( manifest: manifest.ShardLevelManifest, @@ -413,7 +356,7 @@ def test_known_file(self, sample_model_file): ] manifest = serializer.serialize(sample_model_file) - digests = _extract_digests_from_manifest(manifest) + digests = test_support.extract_digests_from_manifest(manifest) assert digests == expected @@ -434,25 +377,29 @@ def test_file_manifest_changes_if_content_changes(self, sample_model_file): self._hasher_factory ) manifest = serializer.serialize(sample_model_file) - digests = set(_extract_digests_from_manifest(manifest)) + digests = set(test_support.extract_digests_from_manifest(manifest)) - sample_model_file.write_bytes(fixtures_constants.ANOTHER_MODEL_TEXT) + sample_model_file.write_bytes(test_support.ANOTHER_MODEL_TEXT) new_manifest = serializer.serialize(sample_model_file) - new_digests = set(_extract_digests_from_manifest(new_manifest)) + new_digests = set( + test_support.extract_digests_from_manifest(new_manifest) + ) assert manifest != new_manifest assert len(digests) == len(new_digests) assert digests != new_digests - def test_directory_model_with_one_single_file(self, sample_model_file): + def test_directory_model_with_only_known_file(self, sample_model_file): serializer = serialize_by_file_shard.ShardedFilesSerializer( self._hasher_factory ) manifest_file = serializer.serialize(sample_model_file) - digests_file = set(_extract_digests_from_manifest(manifest_file)) + digests_file = set( + test_support.extract_digests_from_manifest(manifest_file) + ) manifest = serializer.serialize(sample_model_file.parent) - digests = set(_extract_digests_from_manifest(manifest)) + digests = set(test_support.extract_digests_from_manifest(manifest)) assert manifest != manifest_file # different paths assert digests == digests_file @@ -541,7 +488,7 @@ def test_folder_model_empty_folder_not_included(self, sample_model_folder): ) manifest = serializer.serialize(sample_model_folder) - altered_dir = _get_first_directory(sample_model_folder) + altered_dir = test_support.get_first_directory(sample_model_folder) new_empty_dir = altered_dir / "empty" new_empty_dir.mkdir() new_manifest = serializer.serialize(sample_model_folder) @@ -554,7 +501,7 @@ def test_folder_model_empty_file_not_included(self, sample_model_folder): ) manifest = serializer.serialize(sample_model_folder) - altered_dir = _get_first_directory(sample_model_folder) + altered_dir = test_support.get_first_directory(sample_model_folder) new_empty_file = altered_dir / "empty" new_empty_file.write_text("") new_manifest = serializer.serialize(sample_model_folder) @@ -592,8 +539,8 @@ def test_folder_model_rename_file_only_changes_path_part( ) manifest = serializer.serialize(sample_model_folder) - altered_dir = _get_first_directory(sample_model_folder) - file_to_rename = _get_first_file(altered_dir) + altered_dir = test_support.get_first_directory(sample_model_folder) + file_to_rename = test_support.get_first_file(altered_dir) old_name = file_to_rename.relative_to(sample_model_folder) old_name = pathlib.PurePosixPath(old_name) # canonicalize to Posix new_name = file_to_rename.with_name("new-file") @@ -640,7 +587,7 @@ def test_folder_model_rename_dir_only_changes_path_part( ) manifest = serializer.serialize(sample_model_folder) - dir_to_rename = _get_first_directory(sample_model_folder) + dir_to_rename = test_support.get_first_directory(sample_model_folder) old_name = dir_to_rename.name new_name = dir_to_rename.with_name("new-dir") dir_to_rename.rename(new_name) @@ -656,8 +603,8 @@ def test_folder_model_replace_file_empty_folder(self, sample_model_folder): ) manifest = serializer.serialize(sample_model_folder) - altered_dir = _get_first_directory(sample_model_folder) - file_to_replace = _get_first_file(altered_dir) + altered_dir = test_support.get_first_directory(sample_model_folder) + file_to_replace = test_support.get_first_file(altered_dir) file_to_replace.unlink() file_to_replace.mkdir() new_manifest = serializer.serialize(sample_model_folder) @@ -695,9 +642,9 @@ def test_folder_model_change_file(self, sample_model_folder): ) manifest = serializer.serialize(sample_model_folder) - altered_dir = _get_first_directory(sample_model_folder) - file_to_change = _get_first_file(altered_dir) - file_to_change.write_bytes(fixtures_constants.KNOWN_MODEL_TEXT) + altered_dir = test_support.get_first_directory(sample_model_folder) + file_to_change = test_support.get_first_file(altered_dir) + file_to_change.write_bytes(test_support.KNOWN_MODEL_TEXT) changed_entry = file_to_change.relative_to(sample_model_folder) changed_entry = pathlib.PurePosixPath(changed_entry) # canonicalize new_manifest = serializer.serialize(sample_model_folder) @@ -754,30 +701,6 @@ def test_empty_folder(self, empty_model_folder): manifest = serializer.serialize(empty_model_folder) assert not manifest._item_to_digest - def test_special_file(self, sample_model_folder): - serializer = serialize_by_file_shard.ShardedFilesSerializer( - self._hasher_factory - ) - - altered_dir = _get_first_directory(sample_model_folder) - pipe = altered_dir / "pipe" - - try: - os.mkfifo(pipe) - except AttributeError: - # On Windows, `os.mkfifo` does not exist (it should not). - return # trivially pass the test - - with pytest.raises( - ValueError, match="Cannot use .* as file or directory" - ): - serializer.serialize(sample_model_folder) - - with pytest.raises( - ValueError, match="Cannot use .* as file or directory" - ): - serializer.serialize(pipe) - def test_max_workers_does_not_change_digest(self, sample_model_folder): serializer1 = serialize_by_file_shard.ShardedFilesSerializer( self._hasher_factory diff --git a/model_signing/serialization/serialize_by_file_test.py b/model_signing/serialization/serialize_by_file_test.py index 7581924a..ee88873a 100644 --- a/model_signing/serialization/serialize_by_file_test.py +++ b/model_signing/serialization/serialize_by_file_test.py @@ -19,78 +19,44 @@ from model_signing.hashing import file from model_signing.hashing import memory from model_signing.manifest import manifest -from model_signing.serialization import fixtures_constants from model_signing.serialization import serialize_by_file +from model_signing.serialization import test_support # Load fixtures from serialization/fixtures.py pytest_plugins = ("model_signing.serialization.fixtures",) -_UNUSED_PATH = pathlib.Path("unused") - - -def _extract_digests_from_manifest( - manifest: manifest.FileLevelManifest, -) -> list[str]: - """Extracts the hex digest for every subject in a manifest. - - Used in multiple tests to check that we obtained the expected digests. - """ - return [d.digest_hex for d in manifest._item_to_digest.values()] - - -def _extract_items_from_manifest( - manifest: manifest.FileLevelManifest, -) -> dict[str, str]: - """Builds a dictionary representation of the items in a manifest. - - Every item is mapped to its digest. - - Used in multiple tests to check that we obtained the expected manifest. - """ - return { - str(path): digest.digest_hex - for path, digest in manifest._item_to_digest.items() - } - - -def _get_first_directory(path: pathlib.Path) -> pathlib.Path: - """Returns the first directory that is a children of path. - - It is assumed that there is always such a path. - """ - return [d for d in path.iterdir() if d.is_dir()][0] - - -def _get_first_file(path: pathlib.Path) -> pathlib.Path: - """Returns the first file that is a children of path. - - It is assumed that there is always such a path. - """ - return [f for f in path.iterdir() if f.is_file()][0] - - class TestDFSSerializer: def test_known_file(self, sample_model_file): - file_hasher = file.SimpleFileHasher(_UNUSED_PATH, memory.SHA256()) + file_hasher = file.SimpleFileHasher( + test_support.UNUSED_PATH, memory.SHA256() + ) serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) - manifest = serializer.serialize(sample_model_file) expected = ( "3aab065c7181a173b5dd9e9d32a9f79923440b413be1e1ffcdba26a7365f719b" ) + + manifest = serializer.serialize(sample_model_file) + assert manifest.digest.digest_hex == expected def test_file_hash_is_same_as_hash_of_content(self, sample_model_file): - file_hasher = file.SimpleFileHasher(_UNUSED_PATH, memory.SHA256()) + file_hasher = file.SimpleFileHasher( + test_support.UNUSED_PATH, memory.SHA256() + ) serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) + manifest = serializer.serialize(sample_model_file) - digest = memory.SHA256(fixtures_constants.KNOWN_MODEL_TEXT).compute() + digest = memory.SHA256(test_support.KNOWN_MODEL_TEXT).compute() + assert manifest.digest.digest_hex == digest.digest_hex - def test_file_model_hash_is_same_if_model_is_moved(self, sample_model_file): - file_hasher = file.SimpleFileHasher(_UNUSED_PATH, memory.SHA256()) + def test_file_manifest_unchanged_when_model_moved(self, sample_model_file): + file_hasher = file.SimpleFileHasher( + test_support.UNUSED_PATH, memory.SHA256() + ) serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) manifest = serializer.serialize(sample_model_file) @@ -100,47 +66,52 @@ def test_file_model_hash_is_same_if_model_is_moved(self, sample_model_file): assert manifest == new_manifest - def test_file_model_hash_changes_if_content_changes( - self, sample_model_file - ): - file_hasher = file.SimpleFileHasher(_UNUSED_PATH, memory.SHA256()) + def test_file_manifest_changes_if_content_changes(self, sample_model_file): + file_hasher = file.SimpleFileHasher( + test_support.UNUSED_PATH, memory.SHA256() + ) serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) manifest = serializer.serialize(sample_model_file) - sample_model_file.write_bytes(fixtures_constants.ANOTHER_MODEL_TEXT) + sample_model_file.write_bytes(test_support.ANOTHER_MODEL_TEXT) new_manifest = serializer.serialize(sample_model_file) + assert manifest != new_manifest assert manifest.digest.algorithm == new_manifest.digest.algorithm assert manifest.digest.digest_value != new_manifest.digest.digest_value def test_directory_model_with_only_known_file(self, sample_model_file): - file_hasher = file.SimpleFileHasher(_UNUSED_PATH, memory.SHA256()) + file_hasher = file.SimpleFileHasher( + test_support.UNUSED_PATH, memory.SHA256() + ) serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) + manifest_file = serializer.serialize(sample_model_file) + content_digest = memory.SHA256(test_support.KNOWN_MODEL_TEXT).compute() - model = sample_model_file.parent - manifest = serializer.serialize(model) - - expected = ( - "a0865eb7e299e3bca3951e24930c56dcf1533ecff63bda06a9be67906773c628" - ) - assert manifest.digest.digest_hex == expected + manifest = serializer.serialize(sample_model_file.parent) - digest = memory.SHA256(fixtures_constants.KNOWN_MODEL_TEXT).compute() - assert manifest.digest.digest_hex != digest.digest_hex + assert manifest_file != manifest + assert manifest.digest.digest_hex != content_digest.digest_hex def test_known_folder(self, sample_model_folder): - file_hasher = file.SimpleFileHasher(_UNUSED_PATH, memory.SHA256()) + file_hasher = file.SimpleFileHasher( + test_support.UNUSED_PATH, memory.SHA256() + ) serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) - manifest = serializer.serialize(sample_model_folder) expected = ( "310af4fc4c52bf63cd1687c67076ed3e56bc5480a1b151539e6c550506ae0301" ) + + manifest = serializer.serialize(sample_model_folder) + assert manifest.digest.digest_hex == expected def test_folder_model_hash_is_same_if_model_is_moved( self, sample_model_folder ): - file_hasher = file.SimpleFileHasher(_UNUSED_PATH, memory.SHA256()) + file_hasher = file.SimpleFileHasher( + test_support.UNUSED_PATH, memory.SHA256() + ) serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) manifest = serializer.serialize(sample_model_folder) @@ -150,171 +121,178 @@ def test_folder_model_hash_is_same_if_model_is_moved( assert manifest == new_manifest - def test_empty_file(self, empty_model_file): - file_hasher = file.SimpleFileHasher(_UNUSED_PATH, memory.SHA256()) - serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) - manifest = serializer.serialize(empty_model_file) - expected = ( - "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + def test_folder_model_empty_folder_gets_included(self, sample_model_folder): + file_hasher = file.SimpleFileHasher( + test_support.UNUSED_PATH, memory.SHA256() ) - assert manifest.digest.digest_hex == expected - - def test_directory_model_with_only_empty_file(self, empty_model_file): - file_hasher = file.SimpleFileHasher(_UNUSED_PATH, memory.SHA256()) - serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) - manifest = serializer.serialize(empty_model_file) - model = empty_model_file.parent - manifest = serializer.serialize(model) - expected = ( - "8a587b2129fdecfbea38d5152b626299f5994d9b99d36b321aea356f69b38c61" - ) - assert manifest.digest.digest_hex == expected - - def test_empty_folder(self, empty_model_folder): - file_hasher = file.SimpleFileHasher(_UNUSED_PATH, memory.SHA256()) - serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) - manifest = serializer.serialize(empty_model_folder) - expected = ( - "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" - ) - assert manifest.digest.digest_hex == expected - - def test_empty_folder_hashes_the_same_as_empty_file( - self, empty_model_file, empty_model_folder - ): - file_hasher = file.SimpleFileHasher(_UNUSED_PATH, memory.SHA256()) serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) - folder_manifest = serializer.serialize(empty_model_folder) - file_manifest = serializer.serialize(empty_model_file) - assert ( - folder_manifest.digest.digest_hex == file_manifest.digest.digest_hex - ) - - def test_folder_model_empty_entry(self, sample_model_folder): - file_hasher = file.SimpleFileHasher(_UNUSED_PATH, memory.SHA256()) - serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) - - # Alter first directory within the model - dirs = [d for d in sample_model_folder.iterdir() if d.is_dir()] - altered_dir = dirs[0] + manifest = serializer.serialize(sample_model_folder) + altered_dir = test_support.get_first_directory(sample_model_folder) new_empty_dir = altered_dir / "empty" new_empty_dir.mkdir() - manifest1 = serializer.serialize(sample_model_folder) + new_manifest = serializer.serialize(sample_model_folder) - new_empty_dir.rmdir() + assert manifest != new_manifest + def test_folder_model_empty_file_gets_included(self, sample_model_folder): + file_hasher = file.SimpleFileHasher( + test_support.UNUSED_PATH, memory.SHA256() + ) + serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) + manifest = serializer.serialize(sample_model_folder) + + altered_dir = test_support.get_first_directory(sample_model_folder) new_empty_file = altered_dir / "empty" new_empty_file.write_text("") - manifest2 = serializer.serialize(sample_model_folder) + new_manifest = serializer.serialize(sample_model_folder) - assert manifest1.digest != manifest2.digest + assert manifest != new_manifest def test_folder_model_rename_file(self, sample_model_folder): - file_hasher = file.SimpleFileHasher(_UNUSED_PATH, memory.SHA256()) + file_hasher = file.SimpleFileHasher( + test_support.UNUSED_PATH, memory.SHA256() + ) serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) - manifest1 = serializer.serialize(sample_model_folder) - - # Alter first directory within the model - dirs = [d for d in sample_model_folder.iterdir() if d.is_dir()] - altered_dir = dirs[0] - - # Alter first file in the altered_dir - files = [f for f in altered_dir.iterdir() if f.is_file()] - file_to_rename = files[0] + manifest = serializer.serialize(sample_model_folder) + altered_dir = test_support.get_first_directory(sample_model_folder) + file_to_rename = test_support.get_first_file(altered_dir) new_name = file_to_rename.with_name("new-file") file_to_rename.rename(new_name) + new_manifest = serializer.serialize(sample_model_folder) - manifest2 = serializer.serialize(sample_model_folder) - assert manifest1.digest != manifest2.digest + assert manifest != new_manifest def test_folder_model_rename_dir(self, sample_model_folder): - file_hasher = file.SimpleFileHasher(_UNUSED_PATH, memory.SHA256()) + file_hasher = file.SimpleFileHasher( + test_support.UNUSED_PATH, memory.SHA256() + ) serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) - manifest1 = serializer.serialize(sample_model_folder) - - # Alter first directory within the model - dirs = [d for d in sample_model_folder.iterdir() if d.is_dir()] - dir_to_rename = dirs[0] + manifest = serializer.serialize(sample_model_folder) + dir_to_rename = test_support.get_first_directory(sample_model_folder) new_name = dir_to_rename.with_name("new-dir") dir_to_rename.rename(new_name) + new_manifest = serializer.serialize(sample_model_folder) - manifest2 = serializer.serialize(sample_model_folder) - assert manifest1.digest != manifest2.digest + assert manifest != new_manifest def test_folder_model_replace_file_empty_folder(self, sample_model_folder): - file_hasher = file.SimpleFileHasher(_UNUSED_PATH, memory.SHA256()) + file_hasher = file.SimpleFileHasher( + test_support.UNUSED_PATH, memory.SHA256() + ) serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) - manifest1 = serializer.serialize(sample_model_folder) - - # Alter first directory within the model - dirs = [d for d in sample_model_folder.iterdir() if d.is_dir()] - altered_dir = dirs[0] + manifest = serializer.serialize(sample_model_folder) - # Replace first file in the altered_dir - files = [f for f in altered_dir.iterdir() if f.is_file()] - file_to_replace = files[0] + altered_dir = test_support.get_first_directory(sample_model_folder) + file_to_replace = test_support.get_first_file(altered_dir) file_to_replace.unlink() file_to_replace.mkdir() + new_manifest = serializer.serialize(sample_model_folder) - manifest2 = serializer.serialize(sample_model_folder) - assert manifest1.digest != manifest2.digest + assert manifest != new_manifest def test_folder_model_change_file(self, sample_model_folder): - file_hasher = file.SimpleFileHasher(_UNUSED_PATH, memory.SHA256()) + file_hasher = file.SimpleFileHasher( + test_support.UNUSED_PATH, memory.SHA256() + ) serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) - manifest1 = serializer.serialize(sample_model_folder) - - # Alter first directory within the model - dirs = [d for d in sample_model_folder.iterdir() if d.is_dir()] - altered_dir = dirs[0] + manifest = serializer.serialize(sample_model_folder) - # Alter first file in the altered_dir - files = [f for f in altered_dir.iterdir() if f.is_file()] - file_to_change = files[0] - file_to_change.write_bytes(fixtures_constants.KNOWN_MODEL_TEXT) + altered_dir = test_support.get_first_directory(sample_model_folder) + file_to_change = test_support.get_first_file(altered_dir) + file_to_change.write_bytes(test_support.KNOWN_MODEL_TEXT) + new_manifest = serializer.serialize(sample_model_folder) - manifest2 = serializer.serialize(sample_model_folder) - assert manifest1.digest != manifest2.digest + assert manifest != new_manifest def test_deep_folder(self, deep_model_folder): - file_hasher = file.SimpleFileHasher(_UNUSED_PATH, memory.SHA256()) + file_hasher = file.SimpleFileHasher( + test_support.UNUSED_PATH, memory.SHA256() + ) serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) - manifest = serializer.serialize(deep_model_folder) expected = ( "36eed9389ebbbe15ac15d33c81dabb60ccb7c945ff641d78f59db9aa9dc47ac9" ) + + manifest = serializer.serialize(deep_model_folder) + assert manifest.digest.digest_hex == expected - def test_special_file(self, sample_model_folder): - # Alter first directory within the model - dirs = [d for d in sample_model_folder.iterdir() if d.is_dir()] - altered_dir = dirs[0] + def test_empty_file(self, empty_model_file): + file_hasher = file.SimpleFileHasher( + test_support.UNUSED_PATH, memory.SHA256() + ) + serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) + expected = ( + "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + ) - # Create a pipe in the altered_dir - pipe = altered_dir / "pipe" + manifest = serializer.serialize(empty_model_file) - try: - os.mkfifo(pipe) - except AttributeError: - # On Windows, `os.mkfifo` does not exist (it should not). - return # trivially pass the test + assert manifest.digest.digest_hex == expected - file_hasher = file.SimpleFileHasher(_UNUSED_PATH, memory.SHA256()) + def test_empty_folder(self, empty_model_folder): + file_hasher = file.SimpleFileHasher( + test_support.UNUSED_PATH, memory.SHA256() + ) serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) + expected = ( + "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + ) - with pytest.raises( - ValueError, match="Cannot use .* as file or directory" - ): - serializer.serialize(sample_model_folder) + manifest = serializer.serialize(empty_model_folder) - # Also do the same for the pipe itself - with pytest.raises( - ValueError, match="Cannot use .* as file or directory" - ): - serializer.serialize(pipe) + assert manifest.digest.digest_hex == expected + + def test_directory_model_with_only_empty_file(self, empty_model_file): + file_hasher = file.SimpleFileHasher( + test_support.UNUSED_PATH, memory.SHA256() + ) + serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) + expected = ( + "8a587b2129fdecfbea38d5152b626299f5994d9b99d36b321aea356f69b38c61" + ) + + manifest = serializer.serialize(empty_model_file.parent) + + assert manifest.digest.digest_hex == expected + + def test_empty_folder_hashes_differently_than_empty_file( + self, empty_model_file, empty_model_folder + ): + file_hasher = file.SimpleFileHasher( + test_support.UNUSED_PATH, memory.SHA256() + ) + serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) + + folder_manifest = serializer.serialize(empty_model_folder) + file_manifest = serializer.serialize(empty_model_file) + + assert folder_manifest != file_manifest + + def test_model_with_empty_folder_hashes_differently_than_with_empty_file( + self, sample_model_folder + ): + file_hasher = file.SimpleFileHasher( + test_support.UNUSED_PATH, memory.SHA256() + ) + serializer = serialize_by_file.DFSSerializer(file_hasher, memory.SHA256) + + # Compute digest of model with empty folder + altered_dir = test_support.get_first_directory(sample_model_folder) + new_empty_dir = altered_dir / "empty" + new_empty_dir.mkdir() + folder_manifest = serializer.serialize(sample_model_folder) + + # Compute digest of model with empty file + new_empty_dir.rmdir() + new_empty_file = altered_dir / "empty" + new_empty_file.write_text("") + file_manifest = serializer.serialize(sample_model_folder) + + assert folder_manifest != file_manifest class TestFilesSerializer: @@ -329,7 +307,7 @@ def test_known_file(self, sample_model_file): ] manifest = serializer.serialize(sample_model_file) - digests = _extract_digests_from_manifest(manifest) + digests = test_support.extract_digests_from_manifest(manifest) assert digests == expected @@ -346,23 +324,27 @@ def test_file_manifest_unchanged_when_model_moved(self, sample_model_file): def test_file_manifest_changes_if_content_changes(self, sample_model_file): serializer = serialize_by_file.FilesSerializer(self._hasher_factory) manifest = serializer.serialize(sample_model_file) - digests = set(_extract_digests_from_manifest(manifest)) + digests = set(test_support.extract_digests_from_manifest(manifest)) - sample_model_file.write_bytes(fixtures_constants.ANOTHER_MODEL_TEXT) + sample_model_file.write_bytes(test_support.ANOTHER_MODEL_TEXT) new_manifest = serializer.serialize(sample_model_file) - new_digests = set(_extract_digests_from_manifest(new_manifest)) + new_digests = set( + test_support.extract_digests_from_manifest(new_manifest) + ) assert manifest != new_manifest assert digests != new_digests assert len(digests) == len(new_digests) - def test_directory_model_with_one_single_file(self, sample_model_file): + def test_directory_model_with_only_known_file(self, sample_model_file): serializer = serialize_by_file.FilesSerializer(self._hasher_factory) manifest_file = serializer.serialize(sample_model_file) - digests_file = set(_extract_digests_from_manifest(manifest_file)) + digests_file = set( + test_support.extract_digests_from_manifest(manifest_file) + ) manifest = serializer.serialize(sample_model_file.parent) - digests = set(_extract_digests_from_manifest(manifest)) + digests = set(test_support.extract_digests_from_manifest(manifest)) assert manifest != manifest_file # different paths assert digests == digests_file @@ -385,7 +367,7 @@ def test_known_folder(self, sample_model_folder): # Re-enable lint, so pylint: enable=line-too-long manifest = serializer.serialize(sample_model_folder) - items = _extract_items_from_manifest(manifest) + items = test_support.extract_items_from_manifest(manifest) assert items == expected_items @@ -405,7 +387,7 @@ def test_folder_model_empty_folder_not_included(self, sample_model_folder): serializer = serialize_by_file.FilesSerializer(self._hasher_factory) manifest = serializer.serialize(sample_model_folder) - altered_dir = _get_first_directory(sample_model_folder) + altered_dir = test_support.get_first_directory(sample_model_folder) new_empty_dir = altered_dir / "empty" new_empty_dir.mkdir() new_manifest = serializer.serialize(sample_model_folder) @@ -416,7 +398,7 @@ def test_folder_model_empty_file_gets_included(self, sample_model_folder): serializer = serialize_by_file.FilesSerializer(self._hasher_factory) manifest = serializer.serialize(sample_model_folder) - altered_dir = _get_first_directory(sample_model_folder) + altered_dir = test_support.get_first_directory(sample_model_folder) new_empty_file = altered_dir / "empty" new_empty_file.write_text("") new_manifest = serializer.serialize(sample_model_folder) @@ -456,8 +438,8 @@ def test_folder_model_rename_file_only_changes_path_part( serializer = serialize_by_file.FilesSerializer(self._hasher_factory) manifest = serializer.serialize(sample_model_folder) - altered_dir = _get_first_directory(sample_model_folder) - file_to_rename = _get_first_file(altered_dir) + altered_dir = test_support.get_first_directory(sample_model_folder) + file_to_rename = test_support.get_first_file(altered_dir) old_name = file_to_rename.relative_to(sample_model_folder) old_name = pathlib.PurePosixPath(old_name) # canonicalize to Posix new_name = file_to_rename.with_name("new-file") @@ -501,7 +483,7 @@ def test_folder_model_rename_dir_only_changes_path_part( serializer = serialize_by_file.FilesSerializer(self._hasher_factory) manifest = serializer.serialize(sample_model_folder) - dir_to_rename = _get_first_directory(sample_model_folder) + dir_to_rename = test_support.get_first_directory(sample_model_folder) old_name = dir_to_rename.name new_name = dir_to_rename.with_name("new-dir") dir_to_rename.rename(new_name) @@ -515,8 +497,8 @@ def test_folder_model_replace_file_empty_folder(self, sample_model_folder): serializer = serialize_by_file.FilesSerializer(self._hasher_factory) manifest = serializer.serialize(sample_model_folder) - altered_dir = _get_first_directory(sample_model_folder) - file_to_replace = _get_first_file(altered_dir) + altered_dir = test_support.get_first_directory(sample_model_folder) + file_to_replace = test_support.get_first_file(altered_dir) file_to_replace.unlink() file_to_replace.mkdir() new_manifest = serializer.serialize(sample_model_folder) @@ -550,9 +532,9 @@ def test_folder_model_change_file(self, sample_model_folder): serializer = serialize_by_file.FilesSerializer(self._hasher_factory) manifest = serializer.serialize(sample_model_folder) - altered_dir = _get_first_directory(sample_model_folder) - file_to_change = _get_first_file(altered_dir) - file_to_change.write_bytes(fixtures_constants.KNOWN_MODEL_TEXT) + altered_dir = test_support.get_first_directory(sample_model_folder) + file_to_change = test_support.get_first_file(altered_dir) + file_to_change.write_bytes(test_support.KNOWN_MODEL_TEXT) changed_entry = file_to_change.relative_to(sample_model_folder) changed_entry = pathlib.PurePosixPath(changed_entry) # canonicalize new_manifest = serializer.serialize(sample_model_folder) @@ -573,7 +555,7 @@ def test_deep_folder(self, deep_model_folder): # Re-enable lint, so pylint: enable=line-too-long manifest = serializer.serialize(deep_model_folder) - items = _extract_items_from_manifest(manifest) + items = test_support.extract_items_from_manifest(manifest) assert items == expected_items @@ -584,7 +566,7 @@ def test_empty_file(self, empty_model_file): ] manifest = serializer.serialize(empty_model_file) - digests = _extract_digests_from_manifest(manifest) + digests = test_support.extract_digests_from_manifest(manifest) assert digests == expected @@ -593,28 +575,6 @@ def test_empty_folder(self, empty_model_folder): manifest = serializer.serialize(empty_model_folder) assert not manifest._item_to_digest - def test_special_file(self, sample_model_folder): - serializer = serialize_by_file.FilesSerializer(self._hasher_factory) - - altered_dir = _get_first_directory(sample_model_folder) - pipe = altered_dir / "pipe" - - try: - os.mkfifo(pipe) - except AttributeError: - # On Windows, `os.mkfifo` does not exist (it should not). - return # trivially pass the test - - with pytest.raises( - ValueError, match="Cannot use .* as file or directory" - ): - serializer.serialize(sample_model_folder) - - with pytest.raises( - ValueError, match="Cannot use .* as file or directory" - ): - serializer.serialize(pipe) - def test_max_workers_does_not_change_digest(self, sample_model_folder): serializer1 = serialize_by_file.FilesSerializer(self._hasher_factory) serializer2 = serialize_by_file.FilesSerializer( diff --git a/model_signing/serialization/test_support.py b/model_signing/serialization/test_support.py new file mode 100644 index 00000000..bc7cb3f5 --- /dev/null +++ b/model_signing/serialization/test_support.py @@ -0,0 +1,70 @@ +# Copyright 2024 The Sigstore Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Helpers and constants used in fixtures and tests. Not in the public API.""" + +import pathlib + +from model_signing.manifest import manifest + + +# Model contents +KNOWN_MODEL_TEXT: bytes = b"This is a simple model" +ANOTHER_MODEL_TEXT: bytes = b"This is another simple model" + + +# Constant for unused path when we need to pass a path as argument to internal +# API but we don't use it. +UNUSED_PATH = pathlib.Path("unused") + + +def get_first_directory(path: pathlib.Path) -> pathlib.Path: + """Returns the first directory that is a children of path. + + It is assumed that there is always such a path. + """ + return [d for d in path.iterdir() if d.is_dir()][0] + + +def get_first_file(path: pathlib.Path) -> pathlib.Path: + """Returns the first file that is a children of path. + + It is assumed that there is always such a path. + """ + return [f for f in path.iterdir() if f.is_file()][0] + + +def extract_digests_from_manifest( + manifest: manifest.FileLevelManifest, +) -> list[str]: + """Extracts the hex digest for every subject in a manifest. + + Used in multiple tests to check that we obtained the expected digests. + """ + return [d.digest_hex for d in manifest._item_to_digest.values()] + + +def extract_items_from_manifest( + manifest: manifest.FileLevelManifest, +) -> dict[str, str]: + """Builds a dictionary representation of the items in a manifest. + + Every item is mapped to its digest. + + Used in multiple tests to check that we obtained the expected manifest. + """ + return { + str(path): digest.digest_hex + for path, digest in manifest._item_to_digest.items() + }