import argparse import contextlib import copy import hashlib import io import json from pathlib import Path import tempfile import unittest from unittest.mock import patch from generator.__main__ import generate_assets from generator.engraving import EngravingError, Fonts, Style, engrave_tree, layout_score from generator.engraving.cache import renderer_fingerprint, renderer_namespace, restore_pages, score_fingerprint from generator.score import validate_generated_tree from tests.test_musicxml import parse, xml_note, xml_score class EngravingCacheTests(unittest.TestCase): def test_ci_snapshot_tracks_all_generator_inputs_and_keeps_fallback(self): workflow = Path('.gitea/workflows/ci.yml').read_text() self.assertIn("hashFiles('generator/**/*.py', 'generator/*.py', 'data/practice/*.xml', 'data/songs/*.xml')", workflow) self.assertIn('python3 -m generator --renderer-fingerprint', workflow) self.assertIn('restore-keys: product-vamp-pages-${{ steps.renderer.outputs.fingerprint }}-', workflow) self.assertIn('path: build/build_common/engraving-cache/${{ steps.renderer.outputs.fingerprint }}', workflow) self.assertIn("if: steps.engraving-cache.outputs.cache-hit != 'true'", workflow) self.assertIn('key: ${{ steps.engraving-cache.outputs.cache-primary-key }}', workflow) def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.addCleanup(self.temporary.cleanup) self.root = Path(self.temporary.name) self.cache = self.root / 'cache' self.fonts = Path('package/assets/fonts/petaluma') self.document = parse(xml_score(xml_note())) def write_scores(self, root, documents): root.mkdir(parents=True, exist_ok=True) for index, document in enumerate(documents): (root / f'{index:03}.json').write_text(json.dumps(document), encoding='utf-8') def files(self, root): return {p.relative_to(root): p.read_bytes() for p in root.rglob('*') if p.is_file()} def seed_cache(self): root = self.root / 'seed' self.write_scores(root, [self.document]) engrave_tree(root, self.fonts, cache_dir=self.cache) return next(self.cache.rglob('pages.json')).parent def test_cache_hit_preserves_bundle_bytes_across_output_directories(self): cold, warm = self.root / 'cold', self.root / 'warm' self.write_scores(cold, [self.document]) engrave_tree(cold, self.fonts, cache_dir=self.cache) self.write_scores(warm, [self.document]) with patch('generator.engraving.layout_score', side_effect=AssertionError('cache miss')): with patch('generator.engraving.rasterize_page', side_effect=AssertionError('cache miss')): engrave_tree(warm, self.fonts, cache_dir=self.cache) self.assertEqual(self.files(cold), self.files(warm)) def test_playback_is_not_restored_from_page_cache(self): self.seed_cache() changed = copy.deepcopy(self.document) changed['playback'] = {'new-playback': 'independent of page rendering'} root = self.root / 'warm' self.write_scores(root, [changed]) with patch('generator.engraving.layout_score', side_effect=AssertionError('cache miss')): engrave_tree(root, self.fonts, cache_dir=self.cache) self.assertEqual(json.loads((root / '000.json').read_text())['playback'], changed['playback']) def test_changed_music_and_renderer_invalidate_pages(self): self.seed_cache() changed = parse(xml_score(xml_note('D')).replace('Fixture', 'Changed')) self.assertNotEqual(score_fingerprint(self.document, 'renderer'), score_fingerprint(changed, 'renderer')) self.assertNotEqual(score_fingerprint(self.document, 'renderer'), score_fingerprint(self.document, 'new-renderer')) root = self.root / 'changed' self.write_scores(root, [changed]) with patch('generator.engraving.layout_score', wraps=layout_score) as render: engrave_tree(root, self.fonts, cache_dir=self.cache) render.assert_called_once() self.assertEqual(len(list(self.cache.rglob('pages.json'))), 2) def test_renderer_fingerprint_tracks_fonts_style_code_and_dependencies(self): fonts, style = Fonts(self.fonts), Style() expected = renderer_fingerprint(fonts, style) self.assertEqual(expected, renderer_fingerprint(fonts, style)) self.assertNotEqual(expected, renderer_fingerprint(fonts, Style(margin_x=29))) with patch.object(fonts, 'fingerprint', return_value='changed-fonts'): self.assertNotEqual(expected, renderer_fingerprint(fonts, style)) with patch('generator.engraving.cache.importlib.metadata.version', return_value='new-version'): self.assertNotEqual(expected, renderer_fingerprint(fonts, style)) with patch('generator.engraving.cache.platform.python_version', return_value='0.0.0'): self.assertNotEqual(expected, renderer_fingerprint(fonts, style)) read_bytes = Path.read_bytes with patch.object(Path, 'read_bytes', lambda path: read_bytes(path) + (b'\n' if path.name == 'raster.py' else b'')): self.assertNotEqual(expected, renderer_fingerprint(fonts, style)) def test_corrupt_or_missing_pages_are_regenerated_and_repaired(self): entry = self.seed_cache() page = entry / '0001.png' original = page.read_bytes() for corruption in ('missing', 'checksum', 'undecodable', 'manifest', 'page_directory', 'manifest_directory'): with self.subTest(corruption=corruption): if corruption == 'missing': page.unlink() elif corruption == 'checksum': page.write_bytes(original + b'changed') elif corruption == 'undecodable': page.write_bytes(original[:24]) (entry / 'pages.json').write_text(json.dumps([{'path': page.name, 'sha256': hashlib.sha256(page.read_bytes()).hexdigest()}])) elif corruption == 'manifest': (entry / 'pages.json').write_text('invalid json') else: target = page if corruption == 'page_directory' else entry / 'pages.json' target.unlink() target.mkdir() root = self.root / corruption self.write_scores(root, [self.document]) with patch('generator.engraving.layout_score', wraps=layout_score) as render: engrave_tree(root, self.fonts, cache_dir=self.cache) render.assert_called_once() self.assertEqual(page.read_bytes(), original) self.assertIsNotNone(restore_pages(entry, self.root / 'restored')) def test_cache_manifest_cannot_escape_entry_or_inject_bundle_fields(self): entry = self.seed_cache() original = json.loads((entry / 'pages.json').read_text()) for invalid in ([], {}, [{'path': '../outside.png', 'sha256': original[0]['sha256']}], [{**original[0], 'extra': True}]): with self.subTest(manifest=invalid): (entry / 'pages.json').write_text(json.dumps(invalid)) self.assertIsNone(restore_pages(entry, self.root / 'output')) (entry / 'pages.json').write_text(json.dumps(original)) page = entry / '0001.png' external = self.root / 'outside.png' page.rename(external) page.symlink_to(external) self.assertIsNone(restore_pages(entry, self.root / 'output')) alias = self.root / 'alias' alias.symlink_to(entry, target_is_directory=True) self.assertIsNone(restore_pages(alias, self.root / 'output')) def test_parallel_output_and_shared_cache_are_deterministic(self): documents = [self.document, parse(xml_score(xml_note('D')))] * 4 serial, parallel, warm = (self.root / name for name in ('serial', 'parallel', 'warm')) self.write_scores(serial, documents) engrave_tree(serial, self.fonts) for root in (parallel, warm): self.write_scores(root, documents) engrave_tree(root, self.fonts, cache_dir=self.cache, jobs=2) self.assertEqual(self.files(serial), self.files(root)) self.assertEqual(len(list(self.cache.rglob('pages.json'))), 2) self.assertFalse(list(self.cache.rglob('.engraving-*'))) def test_environment_change_gets_new_namespace_and_clear_progress(self): old = self.seed_cache() root = self.root / 'new-environment' self.write_scores(root, [self.document]) progress = io.StringIO() with patch('generator.engraving.cache.platform.python_version', return_value='0.0.0'), contextlib.redirect_stdout(progress): engrave_tree(root, self.fonts, cache_dir=self.cache, progress=True) self.assertIn('cold namespace', progress.getvalue()) self.assertIn('0 reused, 1 missing, 0 rejected', progress.getvalue()) self.assertTrue((old / 'pages.json').is_file()) self.assertEqual(len(list(self.cache.iterdir())), 2) self.write_scores(root, [self.document]) progress = io.StringIO() with patch('generator.engraving.cache.platform.python_version', return_value='0.0.0'), contextlib.redirect_stdout(progress), patch('generator.engraving.layout_score', side_effect=AssertionError('cache miss')): engrave_tree(root, self.fonts, cache_dir=self.cache, progress=True) self.assertIn('1 reused, 0 missing, 0 rejected', progress.getvalue()) def test_namespace_symlink_is_rejected(self): self.cache.mkdir() (self.cache / 'renderer').symlink_to(self.root, target_is_directory=True) with self.assertRaisesRegex(ValueError, 'must not be symlinks'): renderer_namespace(self.cache, 'renderer') def test_only_changed_score_is_rendered_and_validation_reads_json_once(self): root = self.root / 'scores' self.write_scores(root, [self.document, self.document]) engrave_tree(root, self.fonts, cache_dir=self.cache) changed = parse(xml_score(xml_note('D'))) self.write_scores(root, [self.document, changed]) with patch('generator.engraving.layout_score', wraps=layout_score) as render: engrave_tree(root, self.fonts, cache_dir=self.cache) render.assert_called_once() read_text = Path.read_text reads = [] def read(path, *args, **kwargs): reads.append(path) return read_text(path, *args, **kwargs) with patch.object(Path, 'read_text', read): validate_generated_tree(root) self.assertCountEqual(reads, sorted(root.glob('*.json'))) def test_invalid_worker_count_and_overlapping_cache_are_rejected(self): root = self.root / 'assets' self.write_scores(root, [self.document]) for jobs in (0, -1, 17, True): with self.assertRaisesRegex(ValueError, 'between 1 and 16'): engrave_tree(root, self.fonts, jobs=jobs) for cache in (root, root / 'cache', self.root): with self.assertRaisesRegex(ValueError, 'must not overlap'): engrave_tree(root, self.fonts, cache_dir=cache) def test_parallel_generation_preserves_assets_on_worker_failure(self): data, output = self.root / 'data', self.root / 'output' (data / 'songs').mkdir(parents=True) (data / 'practice').mkdir() source = data / 'practice/001.xml' source.write_text(xml_score(xml_note())) args = argparse.Namespace(data_dir=data, output_dir=output, static_assets_dir=Path('package/assets'), check=False, engraving_cache=self.cache, jobs=2) generate_assets(args) original = self.files(output) generate_assets(args) self.assertEqual(original, self.files(output)) source.write_text(xml_score(xml_note(extra='other'))) with self.assertRaisesRegex(EngravingError, 'unavailableGlyph'): generate_assets(args) self.assertEqual(original, self.files(output)) self.assertFalse(list(output.parent.glob('.vamp-assets-*')))