khodowany
204 lines
11 KiB
Python
204 lines
11 KiB
Python
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='<accidental smufl="unavailableGlyph">other</accidental>')))
|
|
with self.assertRaisesRegex(EngravingError, 'unavailableGlyph'):
|
|
generate_assets(args)
|
|
self.assertEqual(original, self.files(output))
|
|
self.assertFalse(list(output.parent.glob('.vamp-assets-*')))
|