"""현장 녹음 변환의 경계와 실제 FFmpeg 출력을 작은 PCM fixture로 검사한다.""" import math from pathlib import Path import sys import tempfile import unittest import wave from array import array sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "scripts")) import ambience_audio as audio class AmbienceConversion(unittest.TestCase): """긴 원본을 번들하지 않고 짧은 무클릭 파생본만 만드는 계약이다.""" def test_convert_format_fade_normalization_and_reproducibility(self): """스테레오 48kHz 입력을 실제로 내려 섞고 길이·양끝·안전 피크를 확인한다.""" with tempfile.TemporaryDirectory() as directory: source = Path(directory) / "source.wav" target = Path(directory) / "target.wav" samples = array("h", (round(2000 * math.sin(i / 20)) for i in range(48000 * 2 * 14))) with wave.open(str(source), "wb") as stream: stream.setparams((2, 2, 48000, 0, "NONE", "not compressed")) stream.writeframes(samples.tobytes()) first = audio.convert(source, target, 1, 12) self.assertEqual(first["duration"], 12) self.assertGreater(first["rms"], 0.003) self.assertLessEqual(first["peak"], 0.85) with wave.open(str(target)) as stream: self.assertEqual((stream.getframerate(), stream.getnchannels(), stream.getsampwidth()), (24000, 1, 2)) output = array("h", stream.readframes(stream.getnframes())) self.assertEqual((output[0], output[-1]), (0, 0)) self.assertEqual(first, audio.convert(source, target, 1, 12)) with self.assertRaises(ValueError): audio.convert(source, target, 0, 120) def test_silence_fails_before_publishing(self): """무음을 정상 현장 녹음으로 포장하지 않는다.""" with tempfile.TemporaryDirectory() as directory: source = Path(directory) / "silent.wav" target = Path(directory) / "target.wav" with wave.open(str(source), "wb") as stream: stream.setparams((1, 2, 24000, 0, "NONE", "not compressed")) stream.writeframes(bytes(24000 * 2 * 12)) with self.assertRaises(ValueError): audio.convert(source, target, 0, 12) self.assertFalse(target.exists()) def test_short_input_seek_retries_full_decode_without_padding_or_truncating(self): import subprocess from unittest.mock import patch with tempfile.TemporaryDirectory() as directory: source = Path(directory) / "source.wav" target = Path(directory) / "target.wav" samples = array("h", (round(2000 * math.sin(i / 20)) for i in range(24000 * 14))) with wave.open(str(source), "wb") as stream: stream.setparams((1, 2, 24000, 0, "NONE", "not compressed")) stream.writeframes(samples.tobytes()) expected = audio.convert(source, target, 1) target.unlink() real_run = subprocess.run calls = [] def shortened_seek(command, **kwargs): calls.append(command) result = real_run(command, **kwargs) if len(calls) == 1: result.stdout = result.stdout[:-191 * 4] return result with patch.object(audio.subprocess, "run", side_effect=shortened_seek): actual = audio.convert(source, target, 1) self.assertEqual(expected["conversion"]["seekMethod"], "input seek") self.assertEqual(actual["conversion"]["seekMethod"], "full decode then sample trim") expected["conversion"]["seekMethod"] = actual["conversion"]["seekMethod"] self.assertEqual(actual, expected) self.assertEqual(len(calls), 2) self.assertLess(calls[0].index("-ss"), calls[0].index("-i")) self.assertNotIn("-ss", calls[1]) self.assertIn("aresample=24000,atrim=start_sample=24000:end_sample=312000,asetpts=N/SR/TB", calls[1]) target.unlink() # A genuinely exhausted source must still fail; no silence padding is allowed. with self.assertRaises(ValueError): audio.convert(source, target, 3) self.assertFalse(target.exists()) class SourceGuard(unittest.TestCase): """승인 목록은 입력이며 생성기가 현재 파일로 덮어쓰지 못한다.""" def setUp(self): import json from unittest.mock import Mock self.temporary = tempfile.TemporaryDirectory() self.addCleanup(self.temporary.cleanup) self.root = Path(self.temporary.name) self.sources = self.root / "sources" self.sources.mkdir() self.output = self.root / "ambience" self.output.mkdir() (self.output / "old").write_bytes(b"prior release") self.plan = json.loads((audio.ROOT / ".agent/AMBIENCE-SOURCES-v0.2.3.json").read_text()) published = json.loads((audio.ROOT / "public/ambience/manifest.json").read_text()) starts = {e["environmentId"]: e["conversion"]["startSeconds"] for e in published["files"]} for asset in self.plan["assets"]: environment = asset["environmentId"] asset.setdefault("selectedSegmentStartSeconds", starts[environment]) extension = ".mp3" if environment in ("theater", "department") else ".flac" if environment == "rain" else ".ogg" source = self.sources / (environment + extension) source.write_bytes(environment.encode()) asset["expectedSourceSha256"] = audio.digest(source) asset["sourceBytes"] = source.stat().st_size self.converter = Mock(side_effect=AssertionError("conversion must not begin")) def test_all_sources_checked_before_first_conversion(self): import copy for damage in ("hash", "bytes", "missing", "duplicate", "inventory"): with self.subTest(damage=damage): plan = copy.deepcopy(self.plan) if damage == "hash": plan["assets"][-1]["expectedSourceSha256"] = "0" * 64 if damage == "bytes": plan["assets"][-1]["sourceBytes"] += 1 if damage == "missing": plan["assets"][-1].pop("expectedSourceSha256") if damage == "duplicate": plan["assets"][-1] = plan["assets"][0] if damage == "inventory": plan["assets"].pop() before = copy.deepcopy(plan) with self.assertRaises(ValueError): audio.generate(plan, self.sources, self.output, converter=self.converter) self.converter.assert_not_called() self.assertEqual(plan, before) self.assertEqual((self.output / "old").read_bytes(), b"prior release") def test_invalid_segment_plan_fails_before_any_conversion_without_mutation(self): import copy cases = [("selectedSegmentStartSeconds", value) for value in (None, -1, True, "36", float("nan"), float("inf"), 1788.001)] cases += [("directMediaDurationSeconds", value) for value in (None, 0, True, "1800", float("nan"), float("inf"), 191.999)] for key, value in cases: with self.subTest(key=key, value=value): plan = copy.deepcopy(self.plan) if value is None: plan["assets"][-1].pop(key) else: plan["assets"][-1][key] = value before = copy.deepcopy(plan) with self.assertRaises(ValueError): audio.generate(plan, self.sources, self.output, converter=self.converter) self.converter.assert_not_called() self.assertEqual(plan, before) self.assertEqual((self.output / "old").read_bytes(), b"prior release") class AtomicPublication(unittest.TestCase): """실제 디렉터리 교환과 장애 주입으로 이전 릴리스 보존을 확인한다.""" def setUp(self): import json import shutil SourceGuard.setUp(self) self.manifest = json.loads((audio.ROOT / "public/ambience/manifest.json").read_text()) self.info = {e["environmentId"]: e for e in self.manifest["files"]} self.stage_paths = [] def copy_pcm(source, destination, start): import copy self.stage_paths.append(destination) entry = self.info[source.stem] shutil.copyfile(audio.ROOT / "public/ambience" / entry["file"], destination) result = copy.deepcopy({k: entry[k] for k in ("sha256", "bytes", "duration", "rms", "peak", "conversion")}) result["conversion"]["startSeconds"] = start return result self.converter = copy_pcm def run_generate(self, converter=None): return audio.generate(self.plan, self.sources, self.output, converter=converter or self.converter) def assert_prior(self): self.assertEqual({p.name: p.read_bytes() for p in self.output.iterdir()}, {"old": b"prior release"}) def test_atomic_success_removes_stale_files_and_stages_notice_on_same_filesystem(self): import json self.run_generate() expected = {f"{a['environmentId']}.wav" for a in self.plan["assets"]} | {"manifest.json", "AMBIENCE_NOTICES.md"} self.assertEqual({p.name for p in self.output.iterdir()}, expected) self.assertTrue(all(self.output.parent in p.parents for p in self.stage_paths)) self.assertFalse((self.output.parent / "AMBIENCE_NOTICES.md").exists()) manifest = json.loads((self.output / "manifest.json").read_text()) notice = (self.output / "AMBIENCE_NOTICES.md").read_text() for entry in manifest["files"]: section = next(s for s in notice.split("## ") if s.startswith(f"{entry['displayName']} — {entry['environmentId']}\n")) for line in (f"Creator: {entry['creator']}", f"Source: {entry['pageUrl']}", f"Download: {entry['directUrl']}", f"License: {entry['license']} — {entry['licenseUrl']}", f"Output license: {entry['outputLicense']}", f"Changes: {entry['conversion']['operations']}"): self.assertIn(line, section) self.assertEqual(list(self.root.glob(".ambience-stage-*")), []) def test_all_eight_assets_validated_before_publication(self): for damaged in self.info: with self.subTest(asset=damaged): def broken(source, destination, start): info = self.converter(source, destination, start) if source.stem == damaged: destination.write_bytes(b"broken WAV") return info with self.assertRaises(ValueError): self.run_generate(broken) self.assert_prior() def test_plan_alone_controls_selection_and_is_preserved_in_manifest_and_notice(self): import copy from unittest.mock import Mock # Arbitrary fractional starts and an exact end boundary detect a hidden release map. for index, asset in enumerate(self.plan["assets"]): asset["selectedSegmentStartSeconds"] = index + 1.25 self.plan["assets"][-1]["selectedSegmentStartSeconds"] = 1788 before = copy.deepcopy(self.plan) plan_file = audio.ROOT / ".agent/AMBIENCE-SOURCES-v0.2.3.json" plan_bytes = plan_file.read_bytes() converter = Mock(side_effect=self.converter) manifest = self.run_generate(converter) self.assertEqual(self.plan, before) self.assertEqual(plan_file.read_bytes(), plan_bytes) notice = (self.output / "AMBIENCE_NOTICES.md").read_text() self.assertEqual(converter.call_count, 8) for asset, entry, call in zip(self.plan["assets"], manifest["files"], converter.call_args_list): start = asset["selectedSegmentStartSeconds"] self.assertEqual(call.args[2], start) self.assertEqual(entry["selectedSegmentStartSeconds"], start) self.assertEqual(entry["conversion"]["startSeconds"], start) self.assertIn(f"source seconds {start}–{start + 12}.", notice) def test_conversion_cannot_override_approved_segment(self): def wrong_segment(source, destination, start): result = self.converter(source, destination, start) result["conversion"]["startSeconds"] = start + 1 return result with self.assertRaises(ValueError): self.run_generate(wrong_segment) self.assert_prior() def test_conversion_failure_and_keyboard_interrupt_preserve_prior_directory(self): for index in range(8): for error in (OSError, KeyboardInterrupt): with self.subTest(index=index, error=error): calls = [] def broken(source, destination, start): calls.append(source) if len(calls) == index + 1: raise error("injected") return self.converter(source, destination, start) with self.assertRaises(error): self.run_generate(broken) self.assert_prior() def test_manifest_and_notice_write_failures_preserve_prior_directory(self): from unittest.mock import patch write = Path.write_text for name in ("manifest.json", "AMBIENCE_NOTICES.md"): with self.subTest(file=name): def fail(path, *args, **kwargs): if path.name == name: raise OSError("write failure") return write(path, *args, **kwargs) with patch.object(Path, "write_text", fail): with self.assertRaises(OSError): self.run_generate() self.assert_prior() def test_symlink_output_is_rejected_before_conversion(self): from unittest.mock import Mock for dangling in (False, True): with self.subTest(dangling=dangling): link = self.root / f"link-{dangling}" link.symlink_to(self.root / "missing" if dangling else self.output, target_is_directory=True) converter = Mock() with self.assertRaises(ValueError): audio.generate(self.plan, self.sources, link, converter=converter) converter.assert_not_called() self.assert_prior() def test_staged_manifest_notice_and_inventory_tampering_are_rejected(self): from unittest.mock import patch import json validate = audio.validate_bundle for damage in ("manifest", "notice", "extra", "symlink", "metadata"): with self.subTest(damage=damage): def tamper(directory, plan): if damage == "manifest": (directory / "manifest.json").write_text("{}") if damage == "notice": (directory / "AMBIENCE_NOTICES.md").write_text("truncated") if damage == "extra": (directory / "original.ogg").write_bytes(b"extra") if damage == "symlink": (directory / "street.wav").unlink() (directory / "street.wav").symlink_to(audio.ROOT / "public/ambience/street.wav") if damage == "metadata": manifest = json.loads((directory / "manifest.json").read_text()) manifest["files"][0]["rms"] = 0.5 (directory / "manifest.json").write_text(json.dumps(manifest)) return validate(directory, plan) with patch.object(audio, "validate_bundle", tamper): with self.assertRaises(ValueError): self.run_generate() self.assert_prior() def test_exchange_failure_preserves_prior_directory(self): from unittest.mock import patch import ctypes libc = ctypes.CDLL(None) with patch.object(audio.ctypes, "CDLL") as mocked: mocked.return_value.renameat2.return_value = -1 mocked.return_value.renameat2.argtypes = libc.renameat2.argtypes with self.assertRaises(OSError): self.run_generate() self.assert_prior() def test_first_publication_and_real_exchange(self): import shutil shutil.rmtree(self.output) self.run_generate() self.assertTrue((self.output / "AMBIENCE_NOTICES.md").is_file()) (self.output / "stale.wav").write_bytes(b"stale") self.run_generate() self.assertFalse((self.output / "stale.wav").exists()) def test_sigterm_before_commit_preserves_prior_directory(self): import multiprocessing import signal from unittest.mock import patch context = multiprocessing.get_context("fork") reached = context.Event() def child(): def pause(*args): reached.set() signal.pause() with patch.object(audio, "atomic_publish", pause): self.run_generate() process = context.Process(target=child) process.start() try: self.assertTrue(reached.wait(10), "validated stage must reach commit") process.terminate() process.join(5) self.assertEqual(process.exitcode, -signal.SIGTERM) self.assert_prior() finally: if process.is_alive(): process.kill() process.join() def test_readers_observe_whole_directory_during_real_exchange(self): import os import threading staged = self.root / "ready" staged.mkdir() for directory, value in ((self.output, b"old"), (staged, b"new")): for name in ("manifest.json", "AMBIENCE_NOTICES.md", "street.wav"): (directory / name).write_bytes(value) stopped, started = threading.Event(), threading.Event() failures, observed = [], [] def reader(): while not stopped.is_set(): try: # 한 번 연 디렉터리 fd는 교환 중에도 같은 세대의 snapshot을 가리킨다. fd = os.open(self.output, os.O_RDONLY | os.O_DIRECTORY) try: values = [] for name in ("manifest.json", "AMBIENCE_NOTICES.md", "street.wav"): child = os.open(name, os.O_RDONLY, dir_fd=fd) with os.fdopen(child, "rb") as stream: values.append(stream.read()) observed.append(values) if len(set(values)) != 1: failures.append(values) finally: os.close(fd) except OSError as error: failures.append(str(error)) started.set() thread = threading.Thread(target=reader) thread.start() try: self.assertTrue(started.wait(5)) for _ in range(100): audio.atomic_publish(staged, self.output) finally: stopped.set() thread.join(5) self.assertTrue(observed) self.assertEqual(failures, []) def test_unsupported_atomic_exchange_preserves_prior_directory(self): from unittest.mock import patch with patch.object(audio.ctypes, "CDLL", return_value=object()): with self.assertRaises(OSError): self.run_generate() self.assert_prior()