import contextlib import json import pathlib import sqlite3 import subprocess import tempfile import unittest import generator.catalog def bundle(title='Fixture', key='C', part='Lead'): return { 'music': {'metadata': {'title': title, 'composer': 'Composer', 'key_signature': key, 'time_signature': '4/4'}, 'staves': [{'name': part, 'arrangement_part_type': part.lower(), 'measures': [{'voices': []}]}]}, 'presentation': {'width': 976, 'height': 484, 'title_image': '0' * 64, 'pages': [{'path': 'fixture.pages/1.png', 'sha256': '0' * 64}]}, } def write_score(root, relative, document=None): path = root / relative path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(document or bundle()), encoding='utf-8') return path class CatalogTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.addCleanup(self.temporary.cleanup) self.root = pathlib.Path(self.temporary.name) (self.root / 'songs').mkdir() def rows(self, table): with contextlib.closing(sqlite3.connect(self.root / 'catalog.db')) as connection: connection.row_factory = sqlite3.Row return [dict(row) for row in connection.execute(f'SELECT * FROM {table} ORDER BY id')] def test_metadata_relative_paths_and_complete_documents(self): document = bundle('Song', 'F') document['presentation']['title_image'] = '0123456789abcdef' * 4 write_score(self.root, 'songs/tune.json', document) database = generator.catalog.build_catalog(self.root) self.assertEqual(database, self.root / 'catalog.db') song = self.rows('songs')[0] self.assertEqual(json.loads(song['source_json']), document) self.assertEqual((song['id'], song['part'], song['source_path']), ('tune', 'Lead', 'songs/tune.json')) self.assertEqual((song['measure_count'], song['total_pages']), (1, 1)) self.assertEqual(song['title_image'], document['presentation']['title_image']) with contextlib.closing(sqlite3.connect(database)) as connection: self.assertEqual(connection.execute('PRAGMA application_id').fetchone()[0], generator.catalog.APPLICATION_ID) self.assertEqual(connection.execute('PRAGMA integrity_check').fetchall(), [('ok',)]) self.assertEqual(connection.execute("SELECT name FROM sqlite_schema WHERE type='table' ORDER BY name").fetchall(), [('resources',), ('songs',)]) columns = {row[1]: row for row in connection.execute('PRAGMA table_info(songs)')} self.assertEqual((columns['title_image'][2], columns['title_image'][3]), ('TEXT', 1)) self.assertFalse(pathlib.Path(str(database) + '-wal').exists()) self.assertFalse(pathlib.Path(str(database) + '-journal').exists()) def test_rebuild_is_deterministic_relocatable_and_replaces_deleted_rows(self): score = write_score(self.root, 'songs/tune.json') database = generator.catalog.build_catalog(self.root) previous = database.read_bytes() generator.catalog.build_catalog(self.root) self.assertEqual(database.read_bytes(), previous) with tempfile.TemporaryDirectory() as other: root = pathlib.Path(other) write_score(root, 'songs/tune.json') self.assertEqual(generator.catalog.build_catalog(root).read_bytes(), previous) score.rename(score.with_name('renamed.json')) generator.catalog.build_catalog(self.root) self.assertEqual([row['id'] for row in self.rows('songs')], ['renamed']) score.with_name('renamed.json').unlink() generator.catalog.build_catalog(self.root) self.assertEqual(self.rows('songs'), []) def test_invalid_input_preserves_existing_database(self): score = write_score(self.root, 'songs/tune.json') database = generator.catalog.build_catalog(self.root) previous = database.read_bytes() for document in ({}, bundle(title=''), bundle(key=None)): score.write_text(json.dumps(document)) with self.assertRaises(ValueError): generator.catalog.build_catalog(self.root) self.assertEqual(database.read_bytes(), previous) self.assertFalse(list(self.root.glob('.vamp_catalog-*'))) def test_missing_or_invalid_title_image_preserves_existing_database(self): score = write_score(self.root, 'songs/tune.json') database = generator.catalog.build_catalog(self.root) previous = database.read_bytes() invalid = [bundle() for _ in range(9)] invalid[0]['presentation'].pop('title_image') for document, value in zip(invalid[1:], (None, 64, '', '0' * 63, '0' * 65, 'A' * 64, 'g' * 64, '0' * 63 + '\n')): document['presentation']['title_image'] = value for document in invalid: with self.subTest(presentation=document['presentation']): score.write_text(json.dumps(document), encoding='utf-8') with self.assertRaisesRegex(ValueError, 'title image'): generator.catalog.build_catalog(self.root) self.assertEqual(database.read_bytes(), previous) self.assertFalse(list(self.root.glob('.vamp_catalog-*'))) def test_keyless_parts_preserve_their_original_key(self): for part in ('Acoustic', 'Electric', 'Vocal'): write_score(self.root, f'songs/{part.lower()}.json', bundle(key='No key', part=part)) generator.catalog.build_catalog(self.root) self.assertTrue(all(row['key_signature'] == 'No key' for row in self.rows('songs'))) def test_invalid_layout_and_escaping_source_fail(self): with self.assertRaisesRegex(ValueError, 'directory not found'): generator.catalog.build_catalog(self.root / 'missing') self.assertFalse((self.root / 'missing').exists()) score = write_score(self.root, 'songs/nested/tune.json') with self.assertRaisesRegex(ValueError, 'unexpected songs'): generator.catalog.build_catalog(self.root) score.unlink() (self.root / 'songs/escape.json').symlink_to('/etc/passwd') with self.assertRaises(ValueError): generator.catalog.build_catalog(self.root) def test_runtime_asset_packaging_excludes_score_json(self): write_score(self.root, 'songs/tune.json') generator.catalog.build_catalog(self.root) (self.root / 'songs/tune.png').write_bytes(b'page') (self.root / 'print').mkdir() (self.root / 'print/tune.pdf').write_bytes(b'build-only PDF') with tempfile.TemporaryDirectory() as temporary: destination = pathlib.Path(temporary) subprocess.run(['cmake', f'-DSOURCE={self.root}', f'-DDESTINATION={destination}', '-P', 'cmake/runtime_assets.cmake'], check=True, capture_output=True) self.assertTrue((destination / 'catalog.db').exists()) for name in ('vamp.example.json', 'vamp.development.json'): config = json.loads((pathlib.Path('package/config') / name).read_text()) self.assertEqual(pathlib.Path(config['database']['path']).name, 'catalog.db') self.assertTrue((destination / 'songs/tune.png').exists()) self.assertFalse(list(destination.rglob('*.json'))) self.assertFalse((destination / 'print').exists()) self.assertIn('PATTERN "print" EXCLUDE', pathlib.Path('cmake/zero.cmake').read_text()) def test_menu_query_uses_index_and_preserves_binary_identity_order(self): for name, title, composer, part in (('z', 'Same', 'Bea', 'Acoustic'), ('b', 'Same', 'Ada', 'Vocal'), ('a', 'Same', 'Ada', 'Acoustic'), ('c', 'same', 'Ada', 'Electric'), ('d', 'Same', 'Ada', 'Acoustic')): document = bundle(title=title, part=part) document['music']['metadata']['composer'] = composer write_score(self.root, f'songs/{name}.json', document) database = generator.catalog.build_catalog(self.root) query = 'SELECT id, title, composer, part, source_path, measure_count, key_signature, total_pages, title_image FROM songs ORDER BY title, composer, part, id' with contextlib.closing(sqlite3.connect(database)) as connection: self.assertEqual([row[0] for row in connection.execute(query)], ['a', 'd', 'b', 'z', 'c']) plan = ' '.join(row[3] for row in connection.execute('EXPLAIN QUERY PLAN ' + query)) self.assertIn('songs_menu', plan) self.assertNotIn('TEMP B-TREE', plan) def test_failed_population_rolls_back_schema_and_rows(self): write_score(self.root, 'songs/a.json') write_score(self.root, 'songs/b.json', bundle(title='')) with contextlib.closing(sqlite3.connect(':memory:')) as connection: with self.assertRaises(ValueError): generator.catalog.populate(connection, self.root) self.assertFalse(connection.in_transaction) self.assertEqual(connection.execute("SELECT name FROM sqlite_schema").fetchall(), []) self.assertEqual(connection.execute('PRAGMA application_id').fetchone()[0], 0)