Files
product-vamp/tests/test_catalog_database.py
T
khodowany 9fd5abd075
ci / checks (push) Successful in 1h16m10s
Rhythm Selection & Cleanup
2026-09-26 02:50:22 -05:00

143 lines
8.0 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)
for part in ('Acoustic', 'Electric', 'Vocal'):
write_score(self.root, f'practice/0001/{part.lower()}.json', bundle('Exercise', 'Eb', part))
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'])
practice = self.rows('practice_exercises')
self.assertEqual([row['id'] for row in practice], ['0001-acoustic', '0001-electric', '0001-vocal'])
self.assertEqual(practice[0]['exercise_id'], '0001')
self.assertNotIn('practice_type_id', practice[0])
self.assertNotIn('lesson_group_id', practice[0])
self.assertEqual(practice[0]['key_signature'], 'Eb')
self.assertTrue(all(row['title_image'] == '0' * 64 for row in practice))
self.assertTrue(all(json.loads(row['source_json'])['presentation']['title_image'] == row['title_image'] for row in practice))
self.assertNotIn('authored_key_signature', practice[0])
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(), [('practice_exercises',), ('songs',)])
for table in ('songs', 'practice_exercises'):
columns = {row[1]: row for row in connection.execute(f'PRAGMA table_info({table})')}
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_parts_and_metadata_disagreement_fail(self):
for part in ('Acoustic', 'Electric'):
write_score(self.root, f'practice/0001/{part.lower()}.json', bundle(part=part))
with self.assertRaisesRegex(ValueError, 'all three parts'):
generator.catalog.build_catalog(self.root)
write_score(self.root, 'practice/0001/vocal.json', bundle(key='F', part='Vocal'))
with self.assertRaisesRegex(ValueError, 'disagree'):
generator.catalog.build_catalog(self.root)
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'practice/0001/{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('practice_exercises')))
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')
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')))