Files
product-vamp/tests/test_engraving_cache.py
khodowany 150394d598
ci / checks (push) Failing after 24m18s
Toggle Solos
2026-10-08 22:11:34 -05:00

379 lines
23 KiB
Python

import generator.score_io
import tests.production_fixtures
import argparse
import contextlib
import copy
import hashlib
import io
import json
import pathlib
import shutil
import tempfile
import unittest
import unittest.mock
import PIL.Image
import PIL.ImageChops
import generator.__main__
import generator.engraving.cache
import generator.engraving.fonts
import generator.engraving.layout
import generator.engraving.pipeline
import generator.engraving.worker
import generator.engraving.raster
import generator.engraving.types
import generator.engraving.validation
import generator.pico_artwork
import tests.solo_fixtures
import tests.test_musicxml
class EngravingCacheTests(unittest.TestCase):
def test_ci_snapshot_tracks_all_generator_inputs_and_keeps_fallback(self):
workflow = pathlib.Path('.gitea/workflows/ci.yml').read_text()
self.assertIn("hashFiles('generator/**/*.py', 'generator/*.py', 'generator/palette.json', 'data/songs/*.xml', 'data/composition/*.json', 'data/models/*.json', 'data/training/*.json', 'data/training/*.xml')", workflow)
self.assertIn('python3 modulus.py python -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):
tests.production_fixtures.install(self)
self.temporary = tempfile.TemporaryDirectory()
self.addCleanup(self.temporary.cleanup)
self.root = pathlib.Path(self.temporary.name)
self.cache = self.root / 'cache'
self.fonts = pathlib.Path('package/assets/fonts/petaluma')
self.document = tests.test_musicxml.parse(tests.test_musicxml.xml_score(tests.test_musicxml.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])
generator.engraving.pipeline.engrave_tree(root, self.fonts, cache_dir=self.cache)
return next(self.cache.rglob('pages.json')).parent
def titled(self, title):
document = copy.deepcopy(self.document)
document['metadata']['title'] = title
return document
def body_calls(self, render):
return [call for call in render.call_args_list if any(p.role == 'staff' for p in call.args[1])]
def test_cold_titles_group_before_dispatch_and_render_body_once(self):
root = self.root / 'cold'
self.write_scores(root, [self.titled(f'Fixture {index}') for index in range(8)])
progress = io.StringIO()
with unittest.mock.patch('generator.engraving.layout.layout_systems', wraps=generator.engraving.layout.layout_systems) as layout, unittest.mock.patch('generator.engraving.raster.rasterize_band', wraps=generator.engraving.raster.rasterize_band) as render, unittest.mock.patch('generator.engraving.pipeline.concurrent.futures.ProcessPoolExecutor', side_effect=AssertionError('one body must have one owner')), contextlib.redirect_stdout(progress):
generator.engraving.pipeline.engrave_tree(root, self.fonts, jobs=4, progress=True)
layout.assert_called_once()
self.assertEqual(len(self.body_calls(render)), 1)
self.assertIn('fullscore 0, bodyhits 7, bodies 1, renderedpages 8', progress.getvalue())
self.assertIn('Engraving timings:', progress.getvalue())
for phase in ('prepare', 'layout', 'pagination', 'raster', 'cache', 'validation', 'wall'):
self.assertIn(phase + ' ', progress.getvalue())
def test_title_change_reuses_persistent_notation_and_matches_full_page(self):
self.seed_cache()
document = self.titled('Fixture A')
root = self.root / 'title'
self.write_scores(root, [document])
with unittest.mock.patch('generator.engraving.raster.rasterize_band', wraps=generator.engraving.raster.rasterize_band) as render:
generator.engraving.pipeline.engrave_tree(root, self.fonts, cache_dir=self.cache)
self.assertFalse(self.body_calls(render))
self.assertEqual(len(list(self.cache.rglob('band.json'))), 1)
family = generator.engraving.fonts.Fonts(self.fonts)
layout = generator.engraving.layout.layout_score(document, generator.engraving.types.Style(), family)
with PIL.Image.open(root / '000.pages/0001.png') as actual:
reference = generator.engraving.raster.rasterize_page(layout.pages[0], family)
self.assertIsNone(PIL.ImageChops.difference(actual, reference).getbbox())
def test_duplicate_fullscores_share_rendering_without_persistent_cache(self):
root = self.root / 'duplicate'
self.write_scores(root, [self.document] * 8)
with unittest.mock.patch('generator.engraving.raster.compose_page', wraps=generator.engraving.raster.compose_page) as render, unittest.mock.patch('generator.engraving.validation.check_content', wraps=generator.engraving.validation.check_content) as decode:
trusted = generator.engraving.pipeline.engrave_tree(root, self.fonts, jobs=4)
render.assert_called_once()
decode.assert_called_once()
self.assertEqual(len(trusted), 1)
def test_full_page_validation_is_once_per_content_but_every_file_is_read(self):
self.seed_cache()
root = self.root / 'validation'
self.write_scores(root, [self.document] * 4)
with unittest.mock.patch('generator.engraving.validation.check_content', wraps=generator.engraving.validation.check_content) as decode:
trusted = generator.engraving.pipeline.engrave_tree(root, self.fonts, cache_dir=self.cache)
decode.assert_called_once()
read_bytes = pathlib.Path.read_bytes
reads = []
def read(path):
reads.append(path)
return read_bytes(path)
pages = sorted(root.glob('*.pages/*.png'))
digest = next(iter(trusted))
with unittest.mock.patch.object(pathlib.Path, 'read_bytes', read):
for path in pages:
generator.engraving.validation.check_page(path, trusted, expected_sha256=digest)
self.assertCountEqual(reads, pages)
decode.assert_called_once()
pages[-1].write_bytes(pages[-1].read_bytes() + b'changed')
with self.assertRaisesRegex(ValueError, 'checksum'):
generator.engraving.validation.check_page(pages[-1], trusted, expected_sha256=digest)
pages[-1].unlink()
pages[-1].symlink_to(pages[0])
with self.assertRaisesRegex(ValueError, 'regular PNG'):
generator.engraving.validation.check_page(pages[-1], trusted, expected_sha256=digest)
def test_font_content_change_creates_new_shared_notation_namespace(self):
self.seed_cache()
fonts = self.root / 'fonts'
shutil.copytree(self.fonts, fonts)
metadata = fonts / 'petaluma_metadata.json'
metadata.write_bytes(metadata.read_bytes() + b'\n')
root = self.root / 'font-change'
self.write_scores(root, [self.document])
with unittest.mock.patch('generator.engraving.raster.rasterize_band', wraps=generator.engraving.raster.rasterize_band) as render:
generator.engraving.pipeline.engrave_tree(root, fonts, cache_dir=self.cache)
self.assertEqual(len(self.body_calls(render)), 1)
self.assertEqual(len(list(self.cache.iterdir())), 2)
self.assertEqual(len(list(self.cache.rglob('band.json'))), 2)
def test_corrupt_shared_bands_are_rejected_and_repaired(self):
self.seed_cache()
manifest = next(self.cache.rglob('band.json'))
asset = manifest.with_name('band.png')
original, metadata = asset.read_bytes(), manifest.read_bytes()
for index, corruption in enumerate(('checksum', 'undecodable', 'blank', 'wrong_size', 'manifest', 'context', 'symlink', 'directory')):
with self.subTest(corruption=corruption):
if corruption in ('checksum', 'undecodable'):
asset.write_bytes(original + b'changed' if corruption == 'checksum' else original[:24])
elif corruption in ('blank', 'wrong_size'):
dimensions = json.loads(metadata)
size = (dimensions['width'], dimensions['height']) if corruption == 'blank' else (10, 10)
PIL.Image.new('RGB', size, 'white').save(asset)
elif corruption == 'manifest':
manifest.write_text('invalid json')
elif corruption == 'context':
manifest.write_text(json.dumps({**json.loads(metadata), 'context': 'different'}))
elif corruption == 'symlink':
external = self.root / 'external.png'
external.write_bytes(original)
asset.unlink()
asset.symlink_to(external)
else:
asset.unlink()
asset.mkdir()
if corruption in ('undecodable', 'blank', 'wrong_size'):
manifest.write_text(json.dumps({**json.loads(metadata), 'sha256': hashlib.sha256(asset.read_bytes()).hexdigest()}))
root = self.root / f'corrupt-band-{index}'
self.write_scores(root, [self.titled(f'Fixture A {index}')])
with unittest.mock.patch('generator.engraving.raster.rasterize_band', wraps=generator.engraving.raster.rasterize_band) as render:
generator.engraving.pipeline.engrave_tree(root, self.fonts, cache_dir=self.cache)
self.assertEqual(len(self.body_calls(render)), 1)
self.assertEqual(asset.read_bytes(), original)
self.assertEqual(json.loads(manifest.read_bytes()), json.loads(metadata))
def test_band_eviction_reloads_final_pixels_without_rerendering(self):
family = generator.engraving.fonts.Fonts(self.fonts)
style = generator.engraving.types.Style()
documents = [self.document, tests.test_musicxml.parse(tests.test_musicxml.xml_score(tests.test_musicxml.xml_note('D')))]
pages = [generator.engraving.layout.layout_score(document, style, family).pages[0] for document in documents]
result = generator.engraving.worker.GroupResult()
bands = generator.engraving.worker.BodyBands(self.root / 'spool', None, result, limit=1)
self.addCleanup(bands.close)
original = None
for index in (0, 1, 0):
page = pages[index]
notation = generator.engraving.cache.notation_fingerprint(documents[index], 'renderer')
image = bands.get(notation, page, generator.engraving.raster.page_bands(page)[1], family)
if index == 0:
if original is None:
original = image.tobytes()
else:
self.assertEqual(image.tobytes(), original)
self.assertEqual(len(bands.images), 1)
self.assertEqual(result.bodies, 2)
self.assertEqual(result.bodyhits, 1)
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])
generator.engraving.pipeline.engrave_tree(cold, self.fonts, cache_dir=self.cache)
self.write_scores(warm, [self.document])
with unittest.mock.patch('generator.engraving.layout.layout_systems', side_effect=AssertionError('cache miss')):
with unittest.mock.patch('generator.engraving.raster.rasterize_page', side_effect=AssertionError('cache miss')):
generator.engraving.pipeline.engrave_tree(warm, self.fonts, cache_dir=self.cache)
self.assertEqual(self.files(cold), self.files(warm))
def test_invalid_cached_page_timing_is_rebuilt(self):
entry = self.seed_cache()
manifest = entry / 'pages.json'
pages = json.loads(manifest.read_text())
pages[0]['measure_count'] += 1
manifest.write_text(json.dumps(pages))
root = self.root / 'warm'
self.write_scores(root, [self.document])
generator.engraving.pipeline.engrave_tree(root, self.fonts, cache_dir=self.cache)
bundle = json.loads((root / '000.json').read_text())
self.assertEqual(bundle['presentation']['pages'][0]['measure_count'], 1)
def test_changed_music_and_renderer_invalidate_pages(self):
self.seed_cache()
changed = tests.test_musicxml.parse(tests.test_musicxml.xml_score(tests.test_musicxml.xml_note('D')).replace('Fixture', 'Changed'))
self.assertNotEqual(generator.engraving.cache.score_fingerprint(self.document, 'renderer'), generator.engraving.cache.score_fingerprint(changed, 'renderer'))
self.assertNotEqual(generator.engraving.cache.score_fingerprint(self.document, 'renderer'), generator.engraving.cache.score_fingerprint(self.document, 'new-renderer'))
root = self.root / 'changed'
self.write_scores(root, [changed])
with unittest.mock.patch('generator.engraving.layout.layout_systems', wraps=generator.engraving.layout.layout_systems) as render:
generator.engraving.pipeline.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 = generator.engraving.fonts.Fonts(self.fonts), generator.engraving.types.Style()
expected = generator.engraving.cache.renderer_fingerprint(fonts, style)
self.assertEqual(expected, generator.engraving.cache.renderer_fingerprint(fonts, style))
self.assertNotEqual(expected, generator.engraving.cache.renderer_fingerprint(fonts, generator.engraving.types.Style(margin_x=29)))
with unittest.mock.patch.object(fonts, 'fingerprint', return_value='changed-fonts'):
self.assertNotEqual(expected, generator.engraving.cache.renderer_fingerprint(fonts, style))
with unittest.mock.patch('generator.engraving.cache.importlib.metadata.version', return_value='new-version'):
self.assertNotEqual(expected, generator.engraving.cache.renderer_fingerprint(fonts, style))
with unittest.mock.patch('generator.engraving.cache.platform.python_version', return_value='0.0.0'):
self.assertNotEqual(expected, generator.engraving.cache.renderer_fingerprint(fonts, style))
read_bytes = pathlib.Path.read_bytes
with unittest.mock.patch.object(pathlib.Path, 'read_bytes', lambda path: read_bytes(path) + (b'\n' if path.name == 'raster.py' else b'')):
self.assertNotEqual(expected, generator.engraving.cache.renderer_fingerprint(fonts, style))
with unittest.mock.patch.object(pathlib.Path, 'read_bytes', lambda path: read_bytes(path) + (b'\n' if path.name == 'palette.json' else b'')):
self.assertNotEqual(expected, generator.engraving.cache.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 unittest.mock.patch('generator.engraving.layout.layout_systems', wraps=generator.engraving.layout.layout_systems) as render:
generator.engraving.pipeline.engrave_tree(root, self.fonts, cache_dir=self.cache)
render.assert_called_once()
self.assertEqual(page.read_bytes(), original)
self.assertIsNotNone(generator.engraving.cache.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}], [{**original[0], 'sha256': None}], [{**original[0], 'sha256': 1}]):
with self.subTest(manifest=invalid):
(entry / 'pages.json').write_text(json.dumps(invalid))
self.assertIsNone(generator.engraving.cache.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(generator.engraving.cache.restore_pages(entry, self.root / 'output'))
alias = self.root / 'alias'
alias.symlink_to(entry, target_is_directory=True)
self.assertIsNone(generator.engraving.cache.restore_pages(alias, self.root / 'output'))
def test_parallel_output_and_shared_cache_are_deterministic(self):
documents = []
for index in range(4):
for letter in ('C', 'D'):
document = tests.test_musicxml.parse(tests.test_musicxml.xml_score(tests.test_musicxml.xml_note(letter)))
document['metadata']['title'] = f'Fixture {index}'
documents.append(document)
serial, parallel, warm = (self.root / name for name in ('serial', 'parallel', 'warm'))
self.write_scores(serial, documents)
generator.engraving.pipeline.engrave_tree(serial, self.fonts)
for root in (parallel, warm):
self.write_scores(root, documents)
progress = io.StringIO()
with contextlib.redirect_stdout(progress):
generator.engraving.pipeline.engrave_tree(root, self.fonts, cache_dir=self.cache, jobs=2, progress=True)
self.assertEqual(self.files(serial), self.files(root))
if root == parallel:
self.assertIn('fullscore 0, bodyhits 6, bodies 2, renderedpages 8', progress.getvalue())
else:
self.assertIn('fullscore 8, bodyhits 0, bodies 0, renderedpages 0', progress.getvalue())
self.assertEqual(len(list(self.cache.rglob('pages.json'))), 8)
self.assertEqual(len(list(self.cache.rglob('band.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 unittest.mock.patch('generator.engraving.cache.platform.python_version', return_value='0.0.0'), contextlib.redirect_stdout(progress):
generator.engraving.pipeline.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 unittest.mock.patch('generator.engraving.cache.platform.python_version', return_value='0.0.0'), contextlib.redirect_stdout(progress), unittest.mock.patch('generator.engraving.layout.layout_systems', side_effect=AssertionError('cache miss')):
generator.engraving.pipeline.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'):
generator.engraving.cache.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 / 'songs', [self.document, self.document])
generator.engraving.pipeline.engrave_tree(root, self.fonts, cache_dir=self.cache)
changed = tests.test_musicxml.parse(tests.test_musicxml.xml_score(tests.test_musicxml.xml_note('D')))
self.write_scores(root / 'songs', [self.document, changed])
with unittest.mock.patch('generator.engraving.layout.layout_systems', wraps=generator.engraving.layout.layout_systems) as render:
generator.engraving.pipeline.engrave_tree(root, self.fonts, cache_dir=self.cache)
render.assert_called_once()
generator.pico_artwork.generate_titles(root, self.fonts)
read_text = pathlib.Path.read_text
reads = []
def read(path, *args, **kwargs):
reads.append(path)
return read_text(path, *args, **kwargs)
with unittest.mock.patch.object(pathlib.Path, 'read_text', read):
generator.score_io.validate_generated_tree(root)
self.assertCountEqual(reads, sorted(root.rglob('*.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'):
generator.engraving.pipeline.engrave_tree(root, self.fonts, jobs=jobs)
for cache in (root, root / 'cache', self.root):
with self.assertRaisesRegex(ValueError, 'must not overlap'):
generator.engraving.pipeline.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)
source = data / 'songs/fixture.xml'
source.write_text(tests.test_musicxml.xml_score('<harmony><root><root-step>C</root-step></root><kind>major</kind></harmony>' + tests.test_musicxml.xml_note()))
args = argparse.Namespace(data_dir=data, output_dir=output, static_assets_dir=pathlib.Path('package/assets'), check=False, engraving_cache=self.cache, jobs=2)
generator.__main__.generate_assets(args)
original = self.files(output)
(output / 'solo_xml').mkdir()
(output / 'solo_xml/stale.xml').write_text('obsolete solo export')
(output / 'solo_xml/manifest.solo').write_text('{}')
(output / 'print/stale.pdf').write_bytes(b'obsolete print part')
generator.__main__.generate_assets(args)
self.assertEqual(original, self.files(output))
self.assertFalse((output / 'solo_xml').exists())
self.assertFalse(list(output.rglob('*.xml')))
self.assertEqual(len(list(output.glob('print/*.pdf'))), 5)
with unittest.mock.patch('generator.printing.generate_tree', side_effect=ValueError('print rendering failed')):
with self.assertRaisesRegex(ValueError, 'print rendering failed'):
generator.__main__.generate_assets(args)
self.assertEqual(original, self.files(output))
source.write_text(tests.test_musicxml.xml_score('<harmony><root><root-step>C</root-step></root><kind>major</kind></harmony>' + tests.test_musicxml.xml_note(extra='<accidental smufl="unavailableGlyph">other</accidental>')))
with self.assertRaisesRegex(generator.engraving.types.EngravingError, 'unavailableGlyph'):
generator.__main__.generate_assets(args)
self.assertEqual(original, self.files(output))
self.assertFalse(list(output.parent.glob('.vamp_assets_*')))