Updated the subtitles generation test to run several generation tasks

This commit is contained in:
Erik Thuning 2023-12-12 15:39:19 +01:00
parent a5d9eef356
commit c34ed2cc62

76
test.py

@ -469,39 +469,55 @@ class PipelineTest(DaemonTest):
#@unittest.skip("This test is very slow")
def test_generating_subs(self):
jobspec = self.init_job(source_count=1, poster_count=1)
subsource = next(iter(jobspec['sources']))
jobspec['generate_subtitles'] = {'Generated': {'type': 'whisper',
'source': subsource}}
tracknames = ['Generated_1']
job_id_spec_map = {}
for i in range(3):
jobspec = self.init_job(source_count=1, poster_count=1)
subsource = next(iter(jobspec['sources']))
jobspec['generate_subtitles'] = {}
for name in tracknames:
jobspec['generate_subtitles'][name] = {'type': 'whisper',
'source': subsource}
jobid = self.submit_default_job(jobspec)
job_id_spec_map[jobid] = jobspec
jobid = self.submit_default_job(jobspec)
result = self.wait_for_result(jobid, ['AudioHandler',
'TranscodeHandler',
'SubtitlesWhisperHandler'],
timeout=180)
expected_handlers = ['AudioHandler',
'TranscodeHandler',
'SubtitlesWhisperHandler']
job_result_map = {jobid: (jobspec,
self.wait_for_result(jobid,
expected_handlers,
timeout=540))
for jobid, jobspec
in job_id_spec_map.items()}
package_id = result['package']['pkg_id']
with PackageManager(package_id, path.join(self.pipeconf['packagedir'],
package_id)) as package:
# Check match of saved package and jobspec
for name, source in jobspec['sources'].items():
pkgsource = package['sources'][name]
self.assertEqual(source['playAudio'], pkgsource['playAudio'])
self.assertTrue(path.exists(path.join(package.basedir,
pkgsource['poster'])))
for variant, filename in pkgsource['video'].items():
videopath = path.join(package.basedir, filename)
self.assertTrue(path.exists(videopath))
subspath = path.join(package.basedir,
package['subtitles']['Generated'])
self.assertTrue(path.exists(subspath))
with open(subspath) as f:
print("¤ Generated subs ¤")
print(f.read())
print("¤ End generated subs ¤")
for jobid, (jobspec, result) in job_result_map.items():
package_id = result['package']['pkg_id']
with PackageManager(package_id,
path.join(self.pipeconf['packagedir'],
package_id)) as package:
# Check match of saved package and jobspec
for name, source in jobspec['sources'].items():
pkgsource = package['sources'][name]
self.assertEqual(source['playAudio'],
pkgsource['playAudio'])
self.assertTrue(
path.exists(path.join(package.basedir,
pkgsource['poster'])))
for variant, filename in pkgsource['video'].items():
videopath = path.join(package.basedir, filename)
self.assertTrue(path.exists(videopath))
for name in tracknames:
subspath = path.join(package.basedir,
package['subtitles'][name])
self.assertTrue(path.exists(subspath))
with open(subspath) as f:
print(f"¤ Subtitle track {name} ¤")
print(f.read())
print(f"¤ End subtitle track {name} ¤")
# uldir should be gone
self.assertFalse(path.exists(jobspec['upload_dir']))
# uldir should be gone
self.assertFalse(path.exists(jobspec['upload_dir']))
if __name__ == '__main__':