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'))), 3)
    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_*')))