khodowany
173 lines
9.7 KiB
Python
173 lines
9.7 KiB
Python
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)
|
|
|
|
def test_solo_variant_metadata_and_required_pairs(self):
|
|
for enabled in (True, False):
|
|
document = bundle(part='Acoustic')
|
|
document['music']['metadata'].update(song_id='tune', solos_enabled=enabled)
|
|
document['music']['staves'][0]['arranged'] = True
|
|
write_score(self.root, 'songs/tune_acoustic' + ('' if enabled else '_improv') + '.json', document)
|
|
generator.catalog.build_catalog(self.root)
|
|
self.assertEqual([row['solos_enabled'] for row in self.rows('songs')], [1, 0])
|
|
saved = (self.root / 'catalog.db').read_bytes()
|
|
alternate = self.root / 'songs/tune_acoustic_improv.json'
|
|
document = json.loads(alternate.read_text())
|
|
for value in ('false', 0, None):
|
|
document['music']['metadata']['solos_enabled'] = value
|
|
alternate.write_text(json.dumps(document))
|
|
with self.assertRaisesRegex(ValueError, 'boolean'):
|
|
generator.catalog.build_catalog(self.root)
|
|
document['music']['metadata']['solos_enabled'] = True
|
|
alternate.write_text(json.dumps(document))
|
|
with self.assertRaisesRegex(ValueError, 'solo variants'):
|
|
generator.catalog.build_catalog(self.root)
|
|
alternate.unlink()
|
|
with self.assertRaisesRegex(ValueError, 'solo variants'):
|
|
generator.catalog.build_catalog(self.root)
|
|
self.assertEqual((self.root / 'catalog.db').read_bytes(), saved)
|