1172 lines
54 KiB
Python
1172 lines
54 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
r"""
|
|
============================================================================
|
|
AP_DS 0.0.1a3 - Comprehensive CICD Test Suite
|
|
Audio Library By DVS
|
|
============================================================================
|
|
|
|
Automated + interactive test coverage for every module of the ap_ds package:
|
|
|
|
[A] Package Imports / Exports / Error Codes
|
|
[B] Metadata Parsing (WAV/MP3, single file + batch)
|
|
[C] AudioLibrary Initialization
|
|
[D] Playback (file / memory / DAP / error paths)
|
|
[E] Playback Control (pause / resume / stop)
|
|
[F] Volume Control
|
|
[G] Seek
|
|
[H] Fade In / Out
|
|
[I] DAP Recording System
|
|
[J] Metadata Methods
|
|
[K] Helper Methods
|
|
[L] Resource Management
|
|
[M] Top-Level API
|
|
[N] Listening Tests (interactive, requires ears)
|
|
[O] Edge Cases & Error Handling
|
|
[P] __init__.py Module Coverage
|
|
[Q] _sdl2.py Constants / Structures / Bindings
|
|
[R] audio_parser.py Deep Parser Coverage
|
|
[S] Supplementary Cases
|
|
|
|
Usage:
|
|
python cicd_test.py --auto Run automated tests only
|
|
python cicd_test.py --listen Run interactive listening tests only
|
|
python cicd_test.py --full Run everything (default)
|
|
|
|
The suite verifies that every public method returns the documented
|
|
(result, error_code, suggestion) tuple on failure and never raises
|
|
unexpected exceptions for invalid or boundary input.
|
|
============================================================================
|
|
"""
|
|
import os
|
|
import sys
|
|
import io
|
|
import json
|
|
import time
|
|
import struct
|
|
import wave
|
|
import contextlib
|
|
|
|
# ============================================================================
|
|
# Environment
|
|
# ============================================================================
|
|
os.environ.setdefault('AP_DS_SKIP_AUTO_CHECK', '1')
|
|
os.environ.setdefault('AP_DS_SUPPRESS_WARNINGS', '1')
|
|
|
|
# Make ap_ds importable regardless of where this script lives.
|
|
_HERE = os.path.dirname(os.path.abspath(__file__))
|
|
if os.path.basename(_HERE) == 'ap_ds':
|
|
sys.path.insert(0, os.path.dirname(_HERE))
|
|
else:
|
|
sys.path.insert(0, _HERE)
|
|
|
|
import ap_ds
|
|
from ap_ds import (
|
|
AudioLibrary,
|
|
get_audio_duration,
|
|
get_audio_metadata,
|
|
batch_get_metadata,
|
|
batch_get_duration,
|
|
batch_get_metadata_by_type,
|
|
is_full_performance,
|
|
get_runtime_info,
|
|
)
|
|
import ap_ds.player as player
|
|
import ap_ds.audio_parser as audio_parser
|
|
|
|
|
|
def _prompt_mp3():
|
|
"""Ask the user for an MP3 file path. Returns None if skipped/EOF."""
|
|
print("\n" + "=" * 60)
|
|
print(" AP_DS 0.0.1a3 CICD Test Suite")
|
|
print("=" * 60)
|
|
try:
|
|
answer = input(
|
|
"Enter path to an MP3 file for playback / metadata tests\n"
|
|
"(or press Enter to skip MP3-dependent tests): "
|
|
).strip().strip('"').strip("'")
|
|
return answer if answer else None
|
|
except EOFError:
|
|
# Non-interactive execution (e.g. piped stdin) -> skip MP3 tests
|
|
return None
|
|
|
|
|
|
# Prompting happens inside main() (guarded by __main__) so that
|
|
# multiprocessing spawn children never re-run input().
|
|
MP3_FILE = None
|
|
|
|
# ============================================================================
|
|
# Test resources
|
|
# ============================================================================
|
|
TMP_DIR = os.path.join(_HERE, 'cicd_tmp')
|
|
os.makedirs(TMP_DIR, exist_ok=True)
|
|
|
|
WAV_SHORT = os.path.join(TMP_DIR, 't_short.wav') # 2s -> sound-effect mode
|
|
WAV_LONG = os.path.join(TMP_DIR, 't_long.wav') # 10s -> music mode
|
|
WAV_BAD = os.path.join(TMP_DIR, 't_bad.wav') # corrupted
|
|
FAKE_DAP = os.path.join(TMP_DIR, 't.ap-ds-dap') # DAP export artifact
|
|
NODIR = os.path.join(TMP_DIR, 'no_such_dir')
|
|
|
|
|
|
def make_wav(path, seconds, sr=22050, ch=1, sw=2):
|
|
"""Create a valid silent WAV file (PCM). Reuses the file if locked."""
|
|
n = sr * seconds * ch
|
|
try:
|
|
with wave.open(path, 'w') as w:
|
|
w.setnchannels(ch)
|
|
w.setsampwidth(sw)
|
|
w.setframerate(sr)
|
|
if sw == 1:
|
|
data = b'\x80' * (n * sw)
|
|
elif sw == 2:
|
|
data = b''.join(struct.pack('<h', 0) for _ in range(n))
|
|
else:
|
|
data = b''.join(struct.pack('<i', 0) for _ in range(n))
|
|
w.writeframes(data)
|
|
except OSError as e:
|
|
print(f" ! Cannot overwrite {os.path.basename(path)} (locked); reusing existing file: {e}", flush=True)
|
|
return path
|
|
|
|
|
|
def make_bad_wav(path):
|
|
"""Create a WAV file with a broken header."""
|
|
with open(path, 'wb') as f:
|
|
f.write(b'RIFF' + b'\x00' * 100 + b'NOTWAVE' + b'\x00' * 200)
|
|
|
|
|
|
def _make_flac(path, seconds=10, sr=44100, ch=2):
|
|
"""Build a minimal valid FLAC file (header + STREAMINFO block)."""
|
|
total_samples = sr * seconds
|
|
data = bytearray(34)
|
|
data[10] = (sr >> 12) & 0xFF
|
|
data[11] = (sr >> 4) & 0xFF
|
|
d12_hi = sr & 0xF
|
|
data[12] = (d12_hi << 4) | ((ch - 1) << 1)
|
|
ts = total_samples
|
|
data[17] = ts & 0xFF
|
|
data[16] = (ts >> 8) & 0xFF
|
|
data[15] = (ts >> 16) & 0xFF
|
|
data[14] = (ts >> 24) & 0xFF
|
|
data[13] = (ts >> 32) & 0x0F
|
|
header = b'\x80\x00\x00\x22' # is_last=1, type=STREAMINFO(0), size=34
|
|
with open(path, 'wb') as f:
|
|
f.write(b'fLaC' + header + bytes(data))
|
|
return path
|
|
|
|
|
|
# ============================================================================
|
|
# Test framework
|
|
# ============================================================================
|
|
class CICD:
|
|
def __init__(self):
|
|
self.passed = 0
|
|
self.failed = 0
|
|
self.skipped = 0
|
|
self.failures = []
|
|
|
|
def log(self, msg):
|
|
print(msg, flush=True)
|
|
|
|
def check(self, name, cond, detail=""):
|
|
if cond:
|
|
self.passed += 1
|
|
self.log(f" [PASS] {name}" + (f" | {detail}" if detail else ""))
|
|
else:
|
|
self.failed += 1
|
|
self.failures.append((name, detail))
|
|
self.log(f" [FAIL] {name}" + (f" | {detail}" if detail else ""))
|
|
|
|
def skip(self, name, reason):
|
|
self.skipped += 1
|
|
self.log(f" [SKIP] {name} | {reason}")
|
|
|
|
def section(self, title):
|
|
self.log("\n" + "=" * 66)
|
|
self.log(f" {title}")
|
|
self.log("=" * 66)
|
|
|
|
def summary(self):
|
|
self.log("\n" + "=" * 66)
|
|
self.log(" CICD Test Summary")
|
|
self.log("=" * 66)
|
|
self.log(f" Passed : {self.passed}")
|
|
self.log(f" Failed : {self.failed}")
|
|
self.log(f" Skipped: {self.skipped}")
|
|
if self.failures:
|
|
self.log(" --- Failed details ---")
|
|
for name, detail in self.failures:
|
|
self.log(f" [FAIL] {name}: {detail}")
|
|
self.log("=" * 66)
|
|
|
|
|
|
# ============================================================================
|
|
# [A] Package Imports / Exports / Error Codes
|
|
# ============================================================================
|
|
def test_imports(t):
|
|
t.section("[A] Package Imports / Exports / Error Codes")
|
|
t.check("ap_ds version", ap_ds.__version__ == "0.0.1a3", f"v{ap_ds.__version__}")
|
|
t.check("AudioLibrary importable", callable(AudioLibrary))
|
|
t.check("get_audio_duration importable", callable(get_audio_duration))
|
|
t.check("get_audio_metadata importable", callable(get_audio_metadata))
|
|
t.check("batch_get_metadata importable", callable(batch_get_metadata))
|
|
t.check("batch_get_duration importable", callable(batch_get_duration))
|
|
t.check("batch_get_metadata_by_type importable", callable(batch_get_metadata_by_type))
|
|
t.check("is_full_performance importable", callable(is_full_performance))
|
|
t.check("get_runtime_info importable", callable(get_runtime_info))
|
|
required = ["__version__", "AudioLibrary", "get_audio_duration", "get_audio_metadata",
|
|
"batch_get_metadata", "batch_get_duration", "batch_get_metadata_by_type",
|
|
"auto_check_runtime", "check_runtime_mode", "show_tech_manual"]
|
|
missing = [x for x in required if x not in ap_ds.__all__]
|
|
t.check("__all__ complete", not missing, f"missing={missing}")
|
|
# Error-code constants
|
|
t.check("AP_DS_SUCCESS=0", player.AP_DS_SUCCESS == 0)
|
|
t.check("AP_DS_ERR_FILE_NOT_FOUND=1001", player.AP_DS_ERR_FILE_NOT_FOUND == 1001)
|
|
t.check("AP_DS_ERR_INVALID_AID=1002", player.AP_DS_ERR_INVALID_AID == 1002)
|
|
t.check("AP_DS_ERR_AUDIO_LOAD_FAILED=1003", player.AP_DS_ERR_AUDIO_LOAD_FAILED == 1003)
|
|
t.check("AP_DS_ERR_PLAYBACK_FAILED=1004", player.AP_DS_ERR_PLAYBACK_FAILED == 1004)
|
|
t.check("AP_DS_ERR_DAP_INVALID_EXT=1009", player.AP_DS_ERR_DAP_INVALID_EXT == 1009)
|
|
t.check("AP_DS_ERR_DAP_SAVE_FAILED=1010", player.AP_DS_ERR_DAP_SAVE_FAILED == 1010)
|
|
t.check("AP_DS_ERR_METADATA_PARSE_FAILED=1011", player.AP_DS_ERR_METADATA_PARSE_FAILED == 1011)
|
|
t.check("AP_DS_ERR_FADE_NOT_SUPPORTED=1012", player.AP_DS_ERR_FADE_NOT_SUPPORTED == 1012)
|
|
t.check("AP_DS_ERR_AUDIO_NOT_LOADED=1013", player.AP_DS_ERR_AUDIO_NOT_LOADED == 1013)
|
|
t.check("AP_DS_ERR_INVALID_SOURCE=1014", player.AP_DS_ERR_INVALID_SOURCE == 1014)
|
|
t.check("AP_DS_ERR_INVALID_VOLUME=1015", player.AP_DS_ERR_INVALID_VOLUME == 1015)
|
|
t.check("AP_DS_ERR_SEEK_NOT_SUPPORTED=1016", player.AP_DS_ERR_SEEK_NOT_SUPPORTED == 1016)
|
|
t.check("AP_DS_ERR_UNKNOWN=1999", player.AP_DS_ERR_UNKNOWN == 1999)
|
|
# SDL bindings present
|
|
t.check("SDL_Init bound", callable(player.SDL_Init))
|
|
t.check("SDL_GetError bound", callable(player.SDL_GetError))
|
|
t.check("Mix_LoadMUS bound", callable(player.Mix_LoadMUS))
|
|
t.check("Mix_PlayMusic bound", callable(player.Mix_PlayMusic))
|
|
t.check("Mix_SetMusicPosition bound", callable(player.Mix_SetMusicPosition))
|
|
t.check("Mix_FadeInMusicPos bound", callable(player.Mix_FadeInMusicPos))
|
|
t.check("WAV_THRESHOLD=6", player.WAV_THRESHOLD == 6, f"got={player.WAV_THRESHOLD}")
|
|
|
|
|
|
# ============================================================================
|
|
# [B] Metadata Parsing
|
|
# ============================================================================
|
|
def test_metadata(t):
|
|
t.section("[B] Metadata Parsing (WAV/MP3 + batch)")
|
|
make_wav(WAV_SHORT, 2)
|
|
make_wav(WAV_LONG, 10)
|
|
make_bad_wav(WAV_BAD)
|
|
with io.open(FAKE_DAP, 'w', encoding='utf-8') as f:
|
|
json.dump([], f)
|
|
|
|
d1 = get_audio_duration(WAV_SHORT)
|
|
t.check("WAV short duration=2s", d1 == 2, f"got={d1}")
|
|
d2 = get_audio_duration(WAV_LONG)
|
|
t.check("WAV long duration=10s", d2 == 10, f"got={d2}")
|
|
meta = get_audio_metadata(WAV_SHORT)
|
|
t.check("WAV metadata is dict", isinstance(meta, dict))
|
|
if isinstance(meta, dict):
|
|
t.check("WAV sample_rate=22050", meta.get('sample_rate') == 22050, f"got={meta.get('sample_rate')}")
|
|
t.check("WAV channels=1", meta.get('channels') == 1, f"got={meta.get('channels')}")
|
|
t.check("WAV fields complete", all(k in meta for k in ('path', 'format', 'duration', 'length', 'sample_rate', 'channels', 'bitrate')))
|
|
if MP3_FILE and os.path.exists(MP3_FILE):
|
|
dm = get_audio_duration(MP3_FILE)
|
|
t.check("MP3 duration>0", dm > 0, f"got={dm}")
|
|
mm = get_audio_metadata(MP3_FILE)
|
|
t.check("MP3 metadata is dict", isinstance(mm, dict))
|
|
if isinstance(mm, dict):
|
|
t.check("MP3 format=mp3", mm.get('format') == 'mp3', f"got={mm.get('format')}")
|
|
t.check("MP3 duration field>0", mm.get('duration', 0) > 0, f"got={mm.get('duration')}")
|
|
else:
|
|
t.skip("MP3 metadata", "no MP3 file provided")
|
|
|
|
db = get_audio_duration(WAV_BAD)
|
|
t.check("Corrupted WAV duration=0", db == 0, f"got={db}")
|
|
mb = get_audio_metadata(WAV_BAD)
|
|
t.check("Corrupted WAV metadata=None", mb is None, f"got={mb}")
|
|
|
|
files = [WAV_SHORT, WAV_LONG]
|
|
bl = batch_get_metadata(files, max_workers=2)
|
|
t.check("batch_get_metadata returns 2", len(bl) == 2, f"got={len(bl)}")
|
|
bd = batch_get_duration(files, max_workers=2)
|
|
t.check("batch_get_duration returns 2", len(bd) == 2, f"got={bd}")
|
|
bt = batch_get_metadata_by_type(files, 'wav', max_workers=2)
|
|
t.check("batch_by_type filters wav=2", len(bt) == 2, f"got={len(bt)}")
|
|
bl2 = batch_get_metadata([WAV_SHORT, WAV_BAD], max_workers=2)
|
|
t.check("batch with corrupted -> 1 kept", len(bl2) == 1, f"got={len(bl2)}")
|
|
bd3 = batch_get_duration(TMP_DIR, max_workers=2)
|
|
t.check("batch on directory >0", len(bd3) > 0, f"got={len(bd3)}")
|
|
try:
|
|
audio_parser.open_audio(os.path.join(TMP_DIR, 'x.txt'))
|
|
t.check("Unsupported format raises ValueError", False)
|
|
except ValueError:
|
|
t.check("Unsupported format raises ValueError", True)
|
|
|
|
|
|
# ============================================================================
|
|
# [C] AudioLibrary Initialization
|
|
# ============================================================================
|
|
def test_init(t):
|
|
t.section("[C] AudioLibrary Initialization")
|
|
lib = AudioLibrary()
|
|
t.check("Default init ok", lib is not None)
|
|
t.check("AID counter starts at 0", lib._aid_counter == 0)
|
|
t.check("Caches initially empty", lib._audio_cache == {} and lib._music_cache == {})
|
|
t.check("DAP initially empty", lib._dap_recordings == [] and lib._dap_records_set == set())
|
|
t.check("MUS_NO_FADING=0", lib.MUS_NO_FADING == 0)
|
|
lib.cleanup_function()
|
|
|
|
lib2 = AudioLibrary(frequency=48000, channels=1, chunksize=1024)
|
|
t.check("Custom-param init ok", lib2 is not None)
|
|
t.check("frequency stored=48000", lib2._sample_rate == 48000)
|
|
t.check("channels stored=1", lib2._channels == 1)
|
|
lib2.cleanup_function()
|
|
return lib
|
|
|
|
|
|
# ============================================================================
|
|
# [D] Playback
|
|
# ============================================================================
|
|
def test_play(t, lib):
|
|
t.section("[D] Playback")
|
|
aid = lib.play_from_file(WAV_SHORT)
|
|
t.check("play_from_file(short WAV) returns AID", isinstance(aid, int), f"got={aid}")
|
|
if isinstance(aid, int):
|
|
time.sleep(0.3)
|
|
t.check("short WAV is_music_playing=False", lib.is_music_playing() is False)
|
|
t.check("short WAV in audio_cache", WAV_SHORT in lib._audio_cache)
|
|
lib.stop_audio(aid)
|
|
aid2 = lib.play_from_file(WAV_LONG)
|
|
t.check("play_from_file(long WAV) returns AID", isinstance(aid2, int), f"got={aid2}")
|
|
if isinstance(aid2, int):
|
|
time.sleep(0.3)
|
|
t.check("long WAV is_music_playing=True", lib.is_music_playing() is True)
|
|
t.check("long WAV in music_cache", WAV_LONG in lib._music_cache)
|
|
lib.stop_audio(aid2)
|
|
aid3 = lib.play_from_file(WAV_LONG, start_pos=3.0)
|
|
t.check("play_from_file(start_pos=3) returns AID", isinstance(aid3, int), f"got={aid3}")
|
|
if isinstance(aid3, int):
|
|
lib.stop_audio(aid3)
|
|
r = lib.play_from_file(os.path.join(TMP_DIR, 'missing.mp3'))
|
|
t.check("Missing file -> 1001", isinstance(r, tuple) and r[0] == 1001, f"got={r}")
|
|
r = lib.play_from_file(FAKE_DAP)
|
|
t.check(".ap-ds-dap -> 1003", isinstance(r, tuple) and r[0] == 1003, f"got={r}")
|
|
r = lib.play_from_memory(os.path.join(TMP_DIR, 'never_loaded.wav'))
|
|
t.check("play_from_memory(not loaded) -> 1013", isinstance(r, tuple) and r[0] == 1013, f"got={r}")
|
|
aidn = lib.new_aid(WAV_SHORT)
|
|
t.check("new_aid returns AID", isinstance(aidn, int), f"got={aidn}")
|
|
r = lib.play_from_memory(WAV_SHORT)
|
|
t.check("play_from_memory after new_aid ok", isinstance(r, int), f"got={r}")
|
|
if isinstance(r, int):
|
|
lib.stop_audio(r)
|
|
r = lib.new_aid(os.path.join(TMP_DIR, 'missing.wav'))
|
|
t.check("new_aid(missing) -> 1001", isinstance(r, tuple) and r[0] == 1001, f"got={r}")
|
|
return aidn
|
|
|
|
|
|
# ============================================================================
|
|
# [E] Playback Control
|
|
# ============================================================================
|
|
def test_control(t, lib):
|
|
t.section("[E] Playback Control")
|
|
aid = lib.play_from_file(WAV_LONG)
|
|
t.check("Play long WAV ok", isinstance(aid, int), f"got={aid}")
|
|
if not isinstance(aid, int):
|
|
return
|
|
time.sleep(0.3)
|
|
r = lib.pause_audio(aid)
|
|
t.check("pause_audio ok", r[0] == 0, f"got={r}")
|
|
time.sleep(0.2)
|
|
t.check("is_music_paused=True", lib.is_music_paused() is True)
|
|
r = lib.play_audio(aid)
|
|
t.check("play_audio resume ok", r[0] == 0, f"got={r}")
|
|
time.sleep(0.2)
|
|
t.check("is_music_playing=True after resume", lib.is_music_playing() is True)
|
|
r = lib.stop_audio(aid)
|
|
t.check("stop_audio returns float", isinstance(r, float), f"got={r}")
|
|
r = lib.pause_audio(99999)
|
|
t.check("pause_audio(invalid AID) -> 1002", r[0] == 1002, f"got={r}")
|
|
r = lib.play_audio(99999)
|
|
t.check("play_audio(invalid AID) -> 1002", r[0] == 1002, f"got={r}")
|
|
r = lib.stop_audio(99999)
|
|
t.check("stop_audio(invalid AID) -> 1002", r[0] == 1002, f"got={r}")
|
|
r = lib.seek_audio(99999, 1.0)
|
|
t.check("seek_audio(invalid AID) -> 1002", r[0] == 1002, f"got={r}")
|
|
|
|
|
|
# ============================================================================
|
|
# [F] Volume Control
|
|
# ============================================================================
|
|
def test_volume(t, lib):
|
|
t.section("[F] Volume Control")
|
|
aid = lib.play_from_file(WAV_LONG)
|
|
if isinstance(aid, int):
|
|
r = lib.set_volume(aid, 64)
|
|
t.check("set_volume(music,64) ok", r[0] == 0, f"got={r}")
|
|
g = lib.get_volume(aid)
|
|
t.check("get_volume(music) is int", isinstance(g, int), f"got={g}")
|
|
lib.stop_audio(aid)
|
|
aid2 = lib.play_from_file(WAV_SHORT)
|
|
if isinstance(aid2, int):
|
|
r = lib.set_volume(aid2, 100)
|
|
t.check("set_volume(sound,100) ok", r[0] == 0, f"got={r}")
|
|
g = lib.get_volume(aid2)
|
|
t.check("get_volume(sound) is int", isinstance(g, int), f"got={g}")
|
|
lib.stop_audio(aid2)
|
|
aidv = lib.play_from_file(WAV_LONG)
|
|
if isinstance(aidv, int):
|
|
r = lib.set_volume(aidv, -1)
|
|
t.check("set_volume(-1) -> 1015", r[0] == 1015, f"got={r}")
|
|
r = lib.set_volume(aidv, 129)
|
|
t.check("set_volume(129) -> 1015", r[0] == 1015, f"got={r}")
|
|
lib.stop_audio(aidv)
|
|
r = lib.set_volume(99999, 50)
|
|
t.check("set_volume(invalid AID) -> 1002", r[0] == 1002, f"got={r}")
|
|
r = lib.get_volume(99999)
|
|
t.check("get_volume(invalid AID) -> 1002", r[0] == 1002, f"got={r}")
|
|
|
|
|
|
# ============================================================================
|
|
# [G] Seek
|
|
# ============================================================================
|
|
def test_seek(t, lib):
|
|
t.section("[G] Seek")
|
|
aid = lib.play_from_file(WAV_LONG)
|
|
if isinstance(aid, int):
|
|
time.sleep(0.2)
|
|
r = lib.seek_audio(aid, 5.0)
|
|
t.check("seek_audio(music,5s) ok", r[0] == 0, f"got={r}")
|
|
t.check("music still playing after seek", lib.is_music_playing() is True)
|
|
lib.stop_audio(aid)
|
|
aid2 = lib.play_from_file(WAV_SHORT)
|
|
if isinstance(aid2, int):
|
|
r = lib.seek_audio(aid2, 0.5)
|
|
t.check("seek_audio(sound) ok", r[0] == 0, f"got={r}")
|
|
still = any(v['aid'] == aid2 for v in lib._channel_info.values())
|
|
t.check("sound entry retained after seek", still)
|
|
rp = lib.pause_audio(aid2)
|
|
t.check("sound controllable after seek", rp[0] == 0, f"got={rp}")
|
|
lib.stop_audio(aid2)
|
|
|
|
|
|
# ============================================================================
|
|
# [H] Fade In / Out
|
|
# ============================================================================
|
|
def test_fade(t, lib):
|
|
t.section("[H] Fade In / Out")
|
|
aid = lib.play_from_file(WAV_LONG)
|
|
if not isinstance(aid, int):
|
|
return
|
|
time.sleep(0.2)
|
|
r = lib.fadein_music(aid, ms=5000)
|
|
t.check("fadein_music ok", r[0] == 0, f"got={r}")
|
|
if r[0] == 0:
|
|
time.sleep(0.2)
|
|
fading = lib.get_music_fading()
|
|
t.check("get_music_fading returns 0/1/2", fading in (0, 1, 2), f"got={fading}")
|
|
r = lib.fadeout_music(5000)
|
|
t.check("fadeout_music ok", r[0] == 0, f"got={r}")
|
|
time.sleep(0.3)
|
|
r = lib.fadein_music_pos(aid, ms=5000, position=2.0)
|
|
t.check("fadein_music_pos ok", r[0] == 0, f"got={r}")
|
|
lib.stop_audio(aid)
|
|
r = lib.fadein_music(99999)
|
|
t.check("fadein_music(invalid AID) -> 1002", r[0] == 1002, f"got={r}")
|
|
r = lib.fadein_music_pos(99999, ms=5000)
|
|
t.check("fadein_music_pos(invalid AID) -> 1002", r[0] == 1002, f"got={r}")
|
|
|
|
|
|
# ============================================================================
|
|
# [I] DAP Recording System
|
|
# ============================================================================
|
|
def test_dap(t, lib):
|
|
t.section("[I] DAP Recording System")
|
|
lib.clear_dap_recordings()
|
|
t.check("clear_dap empties list", lib.get_dap_recordings() == [])
|
|
lib._add_to_dap_recordings(WAV_SHORT)
|
|
lib._add_to_dap_recordings(WAV_LONG)
|
|
recs = lib.get_dap_recordings()
|
|
t.check("DAP records 2 entries", len(recs) == 2, f"got={len(recs)}")
|
|
lib._add_to_dap_recordings(WAV_SHORT)
|
|
recs = lib.get_dap_recordings()
|
|
t.check("DAP dedupe keeps 2", len(recs) == 2, f"got={len(recs)}")
|
|
if recs:
|
|
t.check("DAP record fields complete",
|
|
all(k in recs[0] for k in ('path', 'duration', 'bitrate', 'channels')))
|
|
buf = io.StringIO()
|
|
with contextlib.redirect_stdout(buf):
|
|
lib._add_to_dap_recordings(os.path.join(TMP_DIR, 'missing.m4a'))
|
|
t.check("DAP record missing file no crash", "set deduplication failed" not in buf.getvalue())
|
|
dap_save = os.path.join(TMP_DIR, 'out.ap-ds-dap')
|
|
r = lib.save_dap_to_json(dap_save)
|
|
t.check("save_dap_to_json ok", r[0] == 0, f"got={r}")
|
|
t.check("DAP file created", os.path.exists(dap_save))
|
|
r = lib.save_dap_to_json(os.path.join(TMP_DIR, 'out.json'))
|
|
t.check("save_dap_to_json(.json) -> 1009", r[0] == 1009, f"got={r}")
|
|
r = lib.save_dap_to_json(r'Z:\no\such\dir\out.ap-ds-dap')
|
|
t.check("save_dap_to_json(bad path) -> 1010", r[0] == 1010, f"got={r}")
|
|
lib.clear_dap_recordings()
|
|
t.check("clear_dap again empties", lib.get_dap_recordings() == [])
|
|
|
|
|
|
# ============================================================================
|
|
# [J] Metadata Methods
|
|
# ============================================================================
|
|
def test_metadata_methods(t, lib):
|
|
t.section("[J] Metadata Methods")
|
|
aid = lib.play_from_file(WAV_LONG)
|
|
if isinstance(aid, int):
|
|
r = lib.get_audio_metadata_by_aid(aid)
|
|
t.check("get_audio_metadata_by_aid is dict", isinstance(r, dict), f"got={type(r).__name__}")
|
|
r = lib.get_audio_metadata_by_path(WAV_LONG)
|
|
t.check("get_audio_metadata_by_path is dict", isinstance(r, dict), f"got={type(r).__name__}")
|
|
r = lib.get_audio_metadata(WAV_LONG, is_file=True)
|
|
t.check("get_audio_metadata(str path) is dict", isinstance(r, dict), f"got={type(r).__name__}")
|
|
r = lib.get_audio_metadata(aid)
|
|
t.check("get_audio_metadata(int AID) is dict", isinstance(r, dict), f"got={type(r).__name__}")
|
|
r = lib.get_audio_metadata(1.5)
|
|
t.check("get_audio_metadata(float) -> 1014", isinstance(r, tuple) and r[0] == 1014, f"got={r}")
|
|
r = lib.get_audio_duration(aid)
|
|
t.check("get_audio_duration(AID)=10", r == 10, f"got={r}")
|
|
r = lib.get_audio_duration(WAV_LONG, is_file=True)
|
|
t.check("get_audio_duration(path)=10", r == 10, f"got={r}")
|
|
r = lib.get_audio_duration(99999)
|
|
t.check("get_audio_duration(invalid AID) -> 1002", isinstance(r, tuple) and r[0] == 1002, f"got={r}")
|
|
sr = lib._get_sample_rate(WAV_LONG)
|
|
t.check("_get_sample_rate=22050", sr == 22050, f"got={sr}")
|
|
ch = lib._get_channels(WAV_LONG)
|
|
t.check("_get_channels=1", ch == 1, f"got={ch}")
|
|
est = lib.simple_mp3_duration_estimation(WAV_LONG)
|
|
t.check("simple_mp3_duration_estimation>0", est > 0, f"got={est}")
|
|
pd = lib._get_playing_duration(aid)
|
|
t.check("_get_playing_duration>=10", pd >= 10, f"got={pd}")
|
|
fd = lib._get_file_duration(WAV_LONG)
|
|
t.check("_get_file_duration=10", fd == 10.0, f"got={fd}")
|
|
r = lib.get_audio_metadata_by_aid(99999)
|
|
t.check("get_audio_metadata_by_aid(invalid) -> 1002", r[0] == 1002, f"got={r}")
|
|
r = lib.get_audio_metadata_by_path(os.path.join(TMP_DIR, 'missing.mp3'))
|
|
t.check("get_audio_metadata_by_path(missing) -> 1001", r[0] == 1001, f"got={r}")
|
|
r = lib.get_audio_duration(os.path.join(TMP_DIR, 'missing.mp3'), is_file=True)
|
|
t.check("get_audio_duration(missing path) -> 1001", r[0] == 1001, f"got={r}")
|
|
lib.stop_audio(aid)
|
|
|
|
|
|
# ============================================================================
|
|
# [K] Helper Methods
|
|
# ============================================================================
|
|
def test_helpers(t, lib):
|
|
t.section("[K] Helper Methods")
|
|
t.check("_is_music_file(.mp3)=True", lib._is_music_file('x.mp3') is True)
|
|
t.check("_is_music_file(.ogg)=True", lib._is_music_file('x.ogg') is True)
|
|
t.check("_is_music_file(.flac)=True", lib._is_music_file('x.flac') is True)
|
|
t.check("_is_music_file(long wav)=True", lib._is_music_file(WAV_LONG) is True)
|
|
t.check("_is_music_file(short wav)=False", lib._is_music_file(WAV_SHORT) is False)
|
|
t.check("_is_music_file(.aif)=False", lib._is_music_file('x.aif') is False)
|
|
t.check("_is_music_file(.txt)=False", lib._is_music_file('x.txt') is False)
|
|
aid = lib.play_from_file(WAV_LONG)
|
|
if isinstance(aid, int):
|
|
ch = lib._find_channel_by_aid(aid)
|
|
t.check("_find_channel_by_aid found", ch is not None, f"got={ch}")
|
|
t.check("_find_channel_by_aid(invalid)=None", lib._find_channel_by_aid(99999) is None)
|
|
fp = lib._get_file_path_by_aid(aid)
|
|
t.check("_get_file_path_by_aid returns path", fp == WAV_LONG, f"got={fp}")
|
|
fp = lib._get_file_path_by_aid(99999)
|
|
t.check("_get_file_path_by_aid(invalid) -> 1002", isinstance(fp, tuple) and fp[0] == 1002, f"got={fp}")
|
|
ga = lib._get_aid_for_music(WAV_LONG)
|
|
t.check("_get_aid_for_music found", ga == aid, f"got={ga}")
|
|
lib.stop_audio(aid)
|
|
aid2 = lib.play_from_file(WAV_SHORT)
|
|
if isinstance(aid2, int):
|
|
ga = lib._get_aid_for_audio(WAV_SHORT)
|
|
t.check("_get_aid_for_audio found", ga == aid2, f"got={ga}")
|
|
ga = lib._get_aid_for_audio(WAV_LONG)
|
|
t.check("_get_aid_for_audio(music file) -> 1002", isinstance(ga, tuple) and ga[0] == 1002, f"got={ga}")
|
|
lib.stop_audio(aid2)
|
|
|
|
|
|
# ============================================================================
|
|
# [L] Resource Management + [M] Top-Level API
|
|
# ============================================================================
|
|
def test_resources(t, lib):
|
|
t.section("[L] Resource Management")
|
|
lib.play_from_file(WAV_LONG)
|
|
lib.play_from_file(WAV_SHORT)
|
|
lib.clear_memory_cache()
|
|
t.check("clear_memory_cache empties caches", lib._audio_cache == {} and lib._music_cache == {})
|
|
lib.cleanup_function()
|
|
t.check("cleanup_function completes", True)
|
|
t.section("[M] Top-Level API")
|
|
info = get_runtime_info()
|
|
t.check("get_runtime_info is dict", isinstance(info, dict), f"got={type(info).__name__}")
|
|
fp = is_full_performance()
|
|
t.check("is_full_performance is bool", isinstance(fp, bool), f"got={fp}")
|
|
|
|
|
|
# ============================================================================
|
|
# [O] Edge Cases & Error Handling
|
|
# ============================================================================
|
|
def test_edge_cases(t):
|
|
t.section("[O] Edge Cases & Error Handling")
|
|
lib = AudioLibrary()
|
|
|
|
# --- O1: invalid argument types -> error tuples, never crash ---
|
|
t.log(" --- O1 invalid argument types ---")
|
|
for bad in (None, 1.5, [], {}, ('a',)):
|
|
r = lib.play_from_file(bad)
|
|
ok = isinstance(r, tuple) and len(r) == 3 and r[0] == 1001
|
|
t.check(f"play_from_file({type(bad).__name__}) -> tuple(1001)", ok, f"got={r!r}")
|
|
for bad in ([], {}):
|
|
r = lib.play_from_memory(bad)
|
|
ok = isinstance(r, tuple) and len(r) == 3
|
|
t.check(f"play_from_memory({type(bad).__name__}) -> tuple", ok, f"got={r!r}")
|
|
for bad in (None, 1.5, []):
|
|
r = lib.new_aid(bad)
|
|
ok = isinstance(r, tuple) and len(r) == 3 and r[0] == 1001
|
|
t.check(f"new_aid({type(bad).__name__}) -> tuple(1001)", ok, f"got={r!r}")
|
|
|
|
# --- O2: seek / volume / fade boundary values ---
|
|
t.log(" --- O2 boundary values ---")
|
|
aid = lib.play_from_file(WAV_LONG)
|
|
if isinstance(aid, int):
|
|
for pos in (None, '5'):
|
|
r = lib.seek_audio(aid, pos)
|
|
t.check(f"seek_audio({pos!r}) -> tuple", isinstance(r, tuple), f"got={r!r}")
|
|
r = lib.seek_audio(aid, 3.0)
|
|
t.check("seek_audio(3.0) -> success", isinstance(r, tuple) and r[0] == 0, f"got={r!r}")
|
|
for v in (None, '50', 1.5):
|
|
r = lib.set_volume(aid, v)
|
|
t.check(f"set_volume({v!r}) -> 1015", isinstance(r, tuple) and r[0] == 1015, f"got={r!r}")
|
|
r = lib.set_volume(aid, 0)
|
|
t.check("set_volume(0) -> success", r[0] == 0, f"got={r}")
|
|
r = lib.set_volume(aid, 128)
|
|
t.check("set_volume(128) -> success", r[0] == 0, f"got={r}")
|
|
r = lib.set_volume(aid, -1)
|
|
t.check("set_volume(-1) -> 1015", r[0] == 1015, f"got={r}")
|
|
r = lib.set_volume(aid, 129)
|
|
t.check("set_volume(129) -> 1015", r[0] == 1015, f"got={r}")
|
|
r = lib.fadein_music(aid, ms=None)
|
|
t.check("fadein_music(ms=None) -> tuple", isinstance(r, tuple), f"got={r!r}")
|
|
r = lib.fadein_music_pos(aid, ms=100, position=None)
|
|
t.check("fadein_music_pos(position=None) -> tuple", isinstance(r, tuple), f"got={r!r}")
|
|
lib.stop_audio(aid)
|
|
|
|
# --- O3: exact error codes ---
|
|
t.log(" --- O3 exact error codes ---")
|
|
_aidv = lib.play_from_file(WAV_LONG)
|
|
_valid = _aidv if isinstance(_aidv, int) else None
|
|
cases = {
|
|
'1001 missing file': (lambda: lib.play_from_file(os.path.join(TMP_DIR, 'no_such_dir.mp3')), 1001),
|
|
'1002 invalid AID': (lambda: lib.pause_audio(99999), 1002),
|
|
'1003 load failure (DAP file)': (lambda: lib.play_from_file(FAKE_DAP), 1003),
|
|
'1009 DAP bad extension': (lambda: lib.save_dap_to_json(os.path.join(TMP_DIR, 'x.json')), 1009),
|
|
'1010 DAP save failure': (lambda: lib.save_dap_to_json(r'Z:\no\such\dir\x.ap-ds-dap'), 1010),
|
|
'1011 metadata parse failure': (lambda: lib.get_audio_metadata_by_path(WAV_BAD), 1011),
|
|
'1013 not loaded in memory': (lambda: lib.play_from_memory(os.path.join(TMP_DIR, 'x.wav')), 1013),
|
|
'1014 invalid source type': (lambda: lib.get_audio_metadata(1.5), 1014),
|
|
'1015 invalid volume': (lambda: lib.set_volume(_valid if _valid is not None else 99999, 200), 1015),
|
|
}
|
|
for name, (fn, expect_code) in cases.items():
|
|
r = fn()
|
|
ok = isinstance(r, tuple) and r[0] == expect_code
|
|
t.check(f"{name} -> {expect_code}", ok, f"got={r!r}")
|
|
if _valid is not None:
|
|
lib.stop_audio(_valid)
|
|
|
|
# --- O4: valid return values ---
|
|
# --- O4: valid return values ---
|
|
t.log(" --- O4 valid return values ---")
|
|
r = lib.play_from_file(WAV_SHORT)
|
|
t.check("play_from_file(short WAV) -> int AID", isinstance(r, int), f"got={r!r}")
|
|
if isinstance(r, int):
|
|
lib.stop_audio(r)
|
|
r = lib.play_from_file(WAV_LONG)
|
|
if isinstance(r, int):
|
|
lib.stop_audio(r)
|
|
d = lib.get_audio_duration(WAV_LONG, is_file=True)
|
|
t.check("get_audio_duration(path) -> valid int>0", isinstance(d, int) and d > 0, f"got={d!r}")
|
|
m = lib.get_audio_metadata_by_path(WAV_SHORT)
|
|
t.check("get_audio_metadata_by_path -> dict complete",
|
|
isinstance(m, dict) and all(k in m for k in ('path', 'format', 'duration', 'length', 'sample_rate', 'channels', 'bitrate')),
|
|
f"got={m!r}")
|
|
lib.clear_dap_recordings()
|
|
lib._add_to_dap_recordings(WAV_SHORT)
|
|
r = lib.save_dap_to_json(os.path.join(TMP_DIR, 'edge.ap-ds-dap'))
|
|
t.check("save_dap_to_json -> (0,'','')", isinstance(r, tuple) and r[0] == 0, f"got={r!r}")
|
|
bl = batch_get_metadata([WAV_SHORT, WAV_LONG], max_workers=2)
|
|
t.check("batch_get_metadata -> valid list",
|
|
isinstance(bl, list) and all(isinstance(x, dict) for x in bl) and len(bl) == 2, f"got={bl!r}")
|
|
r = batch_get_metadata([])
|
|
t.check("batch_get_metadata([]) -> []", r == [], f"got={r!r}")
|
|
r = batch_get_metadata(None)
|
|
t.check("batch_get_metadata(None) -> []", r == [], f"got={r!r}")
|
|
r = batch_get_metadata([WAV_SHORT], max_workers=0)
|
|
t.check("batch max_workers=0 returns error tuple", isinstance(r, tuple) and r[0] == 1999, f"got={r!r}")
|
|
r = batch_get_metadata([WAV_SHORT], max_workers=1000)
|
|
t.check("batch max_workers=1000 returns error tuple", isinstance(r, tuple) and r[0] == 1999, f"got={r!r}")
|
|
|
|
lib.cleanup_function()
|
|
|
|
# ============================================================================
|
|
# [P] __init__.py Module Coverage
|
|
# ============================================================================
|
|
def test_init_module(t):
|
|
t.section("[P] __init__.py Module Coverage")
|
|
import ap_ds as _m
|
|
|
|
# Public API
|
|
t.check("__version__ = 0.0.1a3", _m.__version__ == "0.0.1a3", f"{_m.__version__}")
|
|
for fn_name in ('get_audio_duration', 'get_audio_metadata', 'batch_get_metadata',
|
|
'batch_get_duration', 'batch_get_metadata_by_type',
|
|
'is_full_performance', 'get_runtime_info', 'auto_check_runtime',
|
|
'check_runtime_mode', 'show_tech_manual'):
|
|
t.check(f"top-level {fn_name} callable", callable(getattr(_m, fn_name, None)))
|
|
for name in ('__version__', 'AudioLibrary', 'get_audio_duration', 'get_audio_metadata',
|
|
'batch_get_metadata', 'batch_get_duration', 'batch_get_metadata_by_type',
|
|
'auto_check_runtime', 'check_runtime_mode', 'show_tech_manual'):
|
|
t.check(f"__all__ contains {name}", name in _m.__all__)
|
|
|
|
# show_tech_manual output
|
|
buf = io.StringIO()
|
|
with contextlib.redirect_stdout(buf):
|
|
_m.show_tech_manual()
|
|
out = buf.getvalue()
|
|
t.check("show_tech_manual output >1000 chars", len(out) > 1000, f"{len(out)}")
|
|
for key in ('TECHNICAL MANUAL', 'AudioLibrary', 'AP_DS_WAV_THRESHOLD', 'DAP', 'SDL2'):
|
|
t.check(f"manual contains '{key}'", key in out)
|
|
|
|
# Import-time auto execution + env-var control
|
|
t.check("_RUNTIME_CHECKED=True after import", _m._RUNTIME_CHECKED is True)
|
|
_exp_sup = os.environ.get('AP_DS_SUPPRESS_WARNINGS', '').lower() in ('1', 'true', 'yes', 'on')
|
|
t.check("SUPPRESS_WARNINGS matches env", _m.SUPPRESS_WARNINGS == _exp_sup,
|
|
f"module={_m.SUPPRESS_WARNINGS} env={_exp_sup}")
|
|
_exp_con = os.environ.get('AP_DS_SHOW_CONGRATS', '').lower() not in ('0', 'false', 'no', 'off')
|
|
t.check("SHOW_CONGRATS matches env", _m.SHOW_CONGRATS == _exp_con,
|
|
f"module={_m.SHOW_CONGRATS} env={_exp_con}")
|
|
_exp_skip = os.environ.get('AP_DS_SKIP_AUTO_CHECK', '1').lower() in ('1', 'true', 'yes', 'on')
|
|
t.check("_AUTO_CHECK_SKIP matches env", _m._AUTO_CHECK_SKIP == _exp_skip,
|
|
f"module={_m._AUTO_CHECK_SKIP} env={_exp_skip}")
|
|
|
|
# ensure_runtime_checked
|
|
_saved_rc = _m._RUNTIME_CHECKED
|
|
_m._RUNTIME_CHECKED = False
|
|
_m.ensure_runtime_checked()
|
|
t.check("ensure_runtime_checked sets True", _m._RUNTIME_CHECKED is True)
|
|
_m._RUNTIME_CHECKED = False
|
|
_m.ensure_runtime_checked()
|
|
t.check("ensure_runtime_checked idempotent", _m._RUNTIME_CHECKED is True)
|
|
_m._RUNTIME_CHECKED = _saved_rc
|
|
|
|
# check_runtime_mode
|
|
r = _m.check_runtime_mode()
|
|
t.check("check_runtime_mode returns bool", isinstance(r, bool), f"{r}")
|
|
r2 = _m._check_runtime_mode()
|
|
t.check("_check_runtime_mode returns bool", isinstance(r2, bool), f"{r2}")
|
|
t.check("check_runtime_mode == _check_runtime_mode", r == r2)
|
|
|
|
# SKIP mode (default)
|
|
t.check("auto_check_runtime(SKIP) returns None", _m.auto_check_runtime() is None)
|
|
t.check("_auto_check_runtime(SKIP) returns None", _m._auto_check_runtime() is None)
|
|
t.check("get_runtime_info(SKIP) returns {}", _m.get_runtime_info() == {})
|
|
t.check("is_full_performance(SKIP) returns False", _m.is_full_performance() is False)
|
|
|
|
# Simulate AP_DS_SKIP_AUTO_CHECK=0
|
|
_m._AUTO_CHECK_SKIP = False
|
|
try:
|
|
buf2 = io.StringIO()
|
|
with contextlib.redirect_stdout(buf2):
|
|
info = _m._auto_check_runtime()
|
|
out2 = buf2.getvalue()
|
|
t.check("_auto_check_runtime(SKIP=0) returns dict", isinstance(info, dict))
|
|
expect_keys = ('library_name', 'library_version', 'library_install_path',
|
|
'library_website', 'library_author', 'python_version',
|
|
'gil_enabled', 'has_profiling', 'is_full_performance',
|
|
'cpu_count', 'platform')
|
|
if isinstance(info, dict):
|
|
missing = [k for k in expect_keys if k not in info]
|
|
t.check("auto dict 11 keys complete", not missing, f"missing={missing}")
|
|
t.check("info[library_name]=AP_DS", info.get('library_name') == 'AP_DS')
|
|
t.check("info[library_version]=0.0.1a3", info.get('library_version') == '0.0.1a3')
|
|
t.check("info[gil_enabled] is bool", isinstance(info.get('gil_enabled'), bool))
|
|
t.check("info[has_profiling] is bool", isinstance(info.get('has_profiling'), bool))
|
|
t.check("info[is_full_performance] is bool", isinstance(info.get('is_full_performance'), bool))
|
|
t.check("info[cpu_count]>0", info.get('cpu_count', 0) > 0)
|
|
t.check("info[platform]=win32", info.get('platform') == sys.platform)
|
|
t.check("self-check prints 'Runtime Self-Check'", "Runtime Self-Check" in out2)
|
|
t.check("self-check prints 'Python'", "Python" in out2)
|
|
t.check("self-check prints 'GIL'", "GIL" in out2)
|
|
t.check("self-check prints 'CPU Cores'", "CPU Cores" in out2)
|
|
gi = _m.get_runtime_info()
|
|
t.check("get_runtime_info(SKIP=0) returns dict", isinstance(gi, dict))
|
|
if isinstance(gi, dict):
|
|
t.check("get_runtime_info keys complete", all(k in gi for k in expect_keys), f"keys={list(gi.keys())}")
|
|
t.check("get_runtime_info[library_name]=AP_DS", gi.get('library_name') == 'AP_DS')
|
|
t.check("get_runtime_info same source as auto", gi.get('library_version') == info.get('library_version'))
|
|
fp = _m.is_full_performance()
|
|
t.check("is_full_performance(SKIP=0) is bool", isinstance(fp, bool))
|
|
t.check("is_full_performance matches info", fp == info.get('is_full_performance'))
|
|
finally:
|
|
_m._AUTO_CHECK_SKIP = True
|
|
|
|
# Restored default behaviour
|
|
t.check("after restore auto_check_runtime returns None", _m.auto_check_runtime() is None)
|
|
t.check("after restore get_runtime_info returns {}", _m.get_runtime_info() == {})
|
|
|
|
|
|
# ============================================================================
|
|
# [Q] _sdl2.py Constants / Structures / Bindings
|
|
# ============================================================================
|
|
def test_sdl2_bindings(t):
|
|
t.section("[Q] _sdl2.py Constants / Structures / Bindings")
|
|
import ap_ds._sdl2 as s
|
|
consts = {
|
|
'SDL_TRUE': 1, 'SDL_FALSE': 0,
|
|
'SDL_INIT_TIMER': 1, 'SDL_INIT_AUDIO': 0x10, 'SDL_INIT_VIDEO': 0x20,
|
|
'SDL_INIT_JOYSTICK': 0x200, 'SDL_INIT_HAPTIC': 0x1000,
|
|
'SDL_INIT_GAMECONTROLLER': 0x2000, 'SDL_INIT_EVENTS': 0x4000,
|
|
'AUDIO_U8': 0x8, 'AUDIO_S8': 0x8008, 'AUDIO_U16LSB': 0x10,
|
|
'AUDIO_S16LSB': 0x8010, 'AUDIO_U16MSB': 0x1010, 'AUDIO_S16MSB': 0x9010,
|
|
'AUDIO_S32LSB': 0x8020, 'AUDIO_S32MSB': 0x9020,
|
|
'AUDIO_F32LSB': 0x8120, 'AUDIO_F32MSB': 0x9120,
|
|
'MIX_INIT_FLAC': 1, 'MIX_INIT_MOD': 2, 'MIX_INIT_MP3': 8, 'MIX_INIT_OGG': 0x10,
|
|
'MIX_INIT_MID': 0x20, 'MIX_INIT_OPUS': 0x40,
|
|
'MIX_CHANNEL_POST': -2, 'MIX_DEFAULT_CHANNELS': 2,
|
|
'MUS_NONE': 0, 'MUS_CMD': 1, 'MUS_WAV': 2, 'MUS_MOD': 3, 'MUS_MID': 4,
|
|
'MUS_OGG': 5, 'MUS_MP3': 6, 'MUS_FLAC': 7, 'MUS_OPUS': 8,
|
|
}
|
|
for name, val in consts.items():
|
|
t.check(f"constant {name}={val}", getattr(s, name, None) == val, f"got={getattr(s, name, None)}")
|
|
t.check("MIX_DEFAULT_FORMAT==AUDIO_S16SYS", s.MIX_DEFAULT_FORMAT == s.AUDIO_S16SYS)
|
|
t.check("SDL_INIT_EVERYTHING combination", s.SDL_INIT_EVERYTHING == 0x00007231)
|
|
t.check("AUDIO_S16SYS is S16LSB on little-endian", s.AUDIO_S16SYS == s.AUDIO_S16LSB)
|
|
spec_fields = dict(s.SDL_AudioSpec._fields_)
|
|
for fn in ('freq', 'format', 'channels', 'silence', 'samples', 'padding', 'size', 'callback', 'userdata'):
|
|
t.check(f"SDL_AudioSpec.{fn} field", fn in spec_fields)
|
|
chunk_fields = dict(s.Mix_Chunk._fields_)
|
|
for fn in ('allocated', 'abuf', 'alen', 'volume'):
|
|
t.check(f"Mix_Chunk.{fn} field", fn in chunk_fields)
|
|
t.check("_sdl_lib loaded", s._sdl_lib is not None)
|
|
t.check("_mix_lib loaded", s._mix_lib is not None)
|
|
r = s.import_sdl2()
|
|
t.check("import_sdl2 returns 2-tuple", isinstance(r, tuple) and len(r) == 2)
|
|
t.check("_check_sdl2_loaded() True", s._check_sdl2_loaded() is True)
|
|
pkg = os.path.dirname(s.__file__)
|
|
t.check("_check_sdl_libraries_exist(package dir)", s._check_sdl_libraries_exist(pkg) is True)
|
|
for fn in ('SDL_Init', 'SDL_Quit', 'SDL_GetError', 'SDL_RWFromFile', 'SDL_Delay',
|
|
'Mix_OpenAudio', 'Mix_CloseAudio', 'Mix_LoadWAV', 'Mix_LoadMUS',
|
|
'Mix_FreeChunk', 'Mix_FreeMusic', 'Mix_PlayChannel', 'Mix_PlayMusic',
|
|
'Mix_Pause', 'Mix_PauseMusic', 'Mix_Resume', 'Mix_ResumeMusic',
|
|
'Mix_HaltChannel', 'Mix_HaltMusic', 'Mix_SetMusicPosition',
|
|
'Mix_MusicDuration', 'Mix_Volume', 'Mix_VolumeMusic', 'Mix_AllocateChannels',
|
|
'Mix_GetMusicType', 'Mix_FadingMusic', 'Mix_FadeInMusic', 'Mix_FadeOutMusic',
|
|
'Mix_FadeInChannel', 'Mix_FadeOutChannel', 'Mix_Playing', 'Mix_PlayingMusic',
|
|
'Mix_Paused', 'Mix_PausedMusic', 'Mix_SetPanning', 'Mix_SetDistance',
|
|
'Mix_SetPosition', 'Mix_SetReverseStereo', 'Mix_FadeInMusicPos',
|
|
'_load_from_directory', '_load_from_system', '_load_user_config',
|
|
'_linux_auto_install', '_linux_interactive_setup', '_check_sdl_libraries_exist',
|
|
'_check_sdl2_loaded', 'import_sdl2', '_setup_prototypes'):
|
|
t.check(f"binding {fn} exists", callable(getattr(s, fn, None)))
|
|
t.check("SDL_Init.argtypes=[c_uint32]", list(s._sdl_lib.SDL_Init.argtypes) == [s.c_uint32])
|
|
t.check("SDL_GetError.restype=c_char_p", s._sdl_lib.SDL_GetError.restype == s.c_char_p)
|
|
t.check("Mix_OpenAudio.argtypes set", list(s._mix_lib.Mix_OpenAudio.argtypes) == [s.c_int, s.c_uint16, s.c_int, s.c_int])
|
|
|
|
|
|
# ============================================================================
|
|
# [R] audio_parser.py Deep Parser Coverage
|
|
# ============================================================================
|
|
def test_parser_deep(t):
|
|
t.section("[R] audio_parser.py Deep Parser Coverage")
|
|
import ap_ds.audio_parser as ap
|
|
si = ap.StreamInfo(10.5, 44100, 2, 128000)
|
|
t.check("StreamInfo.length=10.5", si.length == 10.5)
|
|
t.check("StreamInfo.sample_rate=44100", si.sample_rate == 44100)
|
|
t.check("StreamInfo.channels=2", si.channels == 2)
|
|
t.check("StreamInfo.bitrate=128000", si.bitrate == 128000)
|
|
t.check("StreamInfo repr contains StreamInfo", "StreamInfo" in repr(si))
|
|
try:
|
|
ap.FileType(os.path.join(TMP_DIR, 'x.wav'))
|
|
t.check("FileType._parse raises ValueError", False)
|
|
except ValueError:
|
|
t.check("FileType._parse raises ValueError", True)
|
|
t.check("read_u32_be(0x0100)", ap.read_u32_be(io.BytesIO(b'\x00\x00\x01\x00')) == 256)
|
|
t.check("read_u32_le(0x0100)", ap.read_u32_le(io.BytesIO(b'\x00\x01\x00\x00')) == 256)
|
|
t.check("read_u16_le(0x0100)", ap.read_u16_le(io.BytesIO(b'\x00\x01')) == 256)
|
|
t.check("open_audio(.wav) -> WAVFile", isinstance(ap.open_audio(WAV_SHORT), ap.WAVFile))
|
|
if MP3_FILE and os.path.exists(MP3_FILE):
|
|
t.check("open_audio(.mp3) -> MP3File", isinstance(ap.open_audio(MP3_FILE), ap.MP3File))
|
|
else:
|
|
t.skip("open_audio(.mp3)", "no MP3 file provided")
|
|
for bad in ('.txt', '.xyz', '.flac_bad'):
|
|
try:
|
|
ap.open_audio(os.path.join(TMP_DIR, 'x' + bad))
|
|
t.check(f"open_audio({bad}) raises ValueError", False)
|
|
except ValueError:
|
|
t.check(f"open_audio({bad}) raises ValueError", True)
|
|
_st = make_wav(os.path.join(TMP_DIR, 'r_stereo.wav'), 2, 22050, 2)
|
|
for name, path, exp_d, exp_sr, exp_ch in (
|
|
('16bit mono', WAV_SHORT, 2, 22050, 1),
|
|
('16bit stereo', _st, 2, 22050, 2),
|
|
('10s long', WAV_LONG, 10, 22050, 1),
|
|
):
|
|
m = ap.get_audio_metadata(path)
|
|
ok = (isinstance(m, dict) and m['duration'] == exp_d and m['sample_rate'] == exp_sr
|
|
and m['channels'] == exp_ch)
|
|
t.check(f"metadata {name} ({exp_d}s/{exp_sr}/{exp_ch}ch)", ok, f"got={m}")
|
|
flac_path = os.path.join(TMP_DIR, 'r_test.flac')
|
|
_make_flac(flac_path, seconds=10, sr=44100, ch=2)
|
|
m = ap.get_audio_metadata(flac_path)
|
|
ok = (isinstance(m, dict) and m['duration'] == 10 and m['sample_rate'] == 44100
|
|
and m['channels'] == 2 and m['format'] == 'flac')
|
|
t.check("Constructed FLAC parses (10s/44100/2ch)", ok, f"got={m}")
|
|
for ext in ('.flac', '.aac', '.ogg', '.mp3'):
|
|
empty = os.path.join(TMP_DIR, 'empty' + ext)
|
|
with io.open(empty, 'wb'):
|
|
pass
|
|
d = ap.get_audio_duration(empty)
|
|
t.check(f"empty {ext} -> 0", d == 0, f"got={d}")
|
|
bl = batch_get_metadata_by_type([WAV_SHORT, WAV_LONG, flac_path], 'flac', max_workers=2)
|
|
t.check("by_type filters flac -> 1", len(bl) == 1 and bl[0]['format'] == 'flac', f"got={len(bl)}")
|
|
bl2 = batch_get_metadata([WAV_SHORT, WAV_LONG, flac_path], max_workers=2)
|
|
t.check("batch mixed formats -> 3", len(bl2) == 3, f"got={len(bl2)}")
|
|
|
|
|
|
# ============================================================================
|
|
# [S] Supplementary Cases
|
|
# ============================================================================
|
|
def test_supplementary(t):
|
|
t.section("[S] Supplementary Cases")
|
|
|
|
# 1. AudioLibrary initialization boundary parameters
|
|
t.log(" --- 1. init boundary parameters ---")
|
|
for label, kwargs in (
|
|
('frequency=0', {'frequency': 0}),
|
|
('frequency=-44100', {'frequency': -44100}),
|
|
('channels=0', {'channels': 0}),
|
|
('channels=9', {'channels': 9}),
|
|
('channels=32', {'channels': 32}),
|
|
('chunksize=0', {'chunksize': 0}),
|
|
('chunksize=-1', {'chunksize': -1}),
|
|
('format=0', {'format': 0}),
|
|
('format=0xFFFF', {'format': 0xFFFF}),
|
|
):
|
|
try:
|
|
l = AudioLibrary(**kwargs)
|
|
l.cleanup_function()
|
|
t.check(f"init({label}) accepted", True)
|
|
except RuntimeError:
|
|
t.check(f"init({label}) rejected by SDL", True)
|
|
except Exception as e:
|
|
t.check(f"init({label}) no unexpected exception", False, f"{type(e).__name__}: {e}")
|
|
|
|
lib = AudioLibrary()
|
|
|
|
# 2. loops parameter (play_from_file / play_from_memory)
|
|
t.log(" --- 2. loops parameter ---")
|
|
aid = lib.play_from_file(WAV_LONG, loops=-1)
|
|
t.check("play_from_file(loops=-1) returns AID", isinstance(aid, int), f"got={aid!r}")
|
|
if isinstance(aid, int):
|
|
time.sleep(0.2)
|
|
t.check("loops=-1 still playing", lib.is_music_playing() is True)
|
|
lib.stop_audio(aid)
|
|
aid = lib.play_from_file(WAV_LONG, loops=2)
|
|
t.check("play_from_file(loops=2) returns AID", isinstance(aid, int), f"got={aid!r}")
|
|
if isinstance(aid, int):
|
|
lib.stop_audio(aid)
|
|
aid = lib.play_from_file(WAV_SHORT, loops=-1)
|
|
t.check("play_from_file(short WAV, loops=-1) returns AID", isinstance(aid, int), f"got={aid!r}")
|
|
if isinstance(aid, int):
|
|
lib.stop_audio(aid)
|
|
for bad in ('2', None):
|
|
try:
|
|
r = lib.play_from_file(WAV_LONG, loops=bad)
|
|
ok = isinstance(r, (int, tuple))
|
|
t.check(f"play_from_file(loops={bad!r}) no crash", ok, f"got={r!r}")
|
|
except Exception as e:
|
|
t.check(f"play_from_file(loops={bad!r}) no crash", False, f"{type(e).__name__}: {e}")
|
|
aidn = lib.new_aid(WAV_LONG)
|
|
if isinstance(aidn, int):
|
|
r = lib.play_from_memory(WAV_LONG, loops=-1)
|
|
t.check("play_from_memory(loops=-1) returns AID", isinstance(r, int), f"got={r!r}")
|
|
if isinstance(r, int):
|
|
lib.stop_audio(r)
|
|
|
|
# 3. batch show_progress output
|
|
t.log(" --- 3. batch show_progress output ---")
|
|
buf = io.StringIO()
|
|
with contextlib.redirect_stdout(buf):
|
|
bl = batch_get_metadata([WAV_SHORT, WAV_LONG], max_workers=2, show_progress=True)
|
|
out = buf.getvalue()
|
|
t.check("show_progress returns list(2)", isinstance(bl, list) and len(bl) == 2, f"got={len(bl)}")
|
|
t.check("show_progress prints 'Batch parse complete'", "Batch parse complete" in out, out[-80:])
|
|
|
|
# 4. Delay method
|
|
t.log(" --- 4. Delay method ---")
|
|
t0 = time.time()
|
|
lib.Delay(100)
|
|
dt = time.time() - t0
|
|
t.check("Delay(100) ~100ms", 0.08 <= dt < 0.5, f"{dt:.3f}s")
|
|
t0 = time.time()
|
|
lib.Delay(0)
|
|
t.check("Delay(0) fast", (time.time() - t0) < 0.1)
|
|
|
|
# 5. FileType base class via a WAVFile instance
|
|
t.log(" --- 5. FileType base class ---")
|
|
import ap_ds.audio_parser as ap
|
|
f = ap.WAVFile(WAV_LONG)
|
|
t.check("FileType.filename", f.filename == WAV_LONG)
|
|
t.check("FileType.info is StreamInfo", isinstance(f.info, ap.StreamInfo))
|
|
t.check("FileType.length property", isinstance(f.length, float) and f.length == 10.0, f"{f.length}")
|
|
t.check("FileType.sample_rate property", f.sample_rate == 22050, f"{f.sample_rate}")
|
|
t.check("FileType.channels property", f.channels == 1, f"{f.channels}")
|
|
t.check("FileType.bitrate property", isinstance(f.bitrate, int) and f.bitrate > 0, f"{f.bitrate}")
|
|
|
|
# 6. _get_sample_rate / _get_channels
|
|
t.log(" --- 6. _get_sample_rate / _get_channels ---")
|
|
aid = lib.play_from_file(WAV_LONG)
|
|
if isinstance(aid, int):
|
|
t.check("_get_sample_rate(AID)=22050", lib._get_sample_rate(aid) == 22050, f"{lib._get_sample_rate(aid)}")
|
|
t.check("_get_channels(AID)=1", lib._get_channels(aid) == 1, f"{lib._get_channels(aid)}")
|
|
lib.stop_audio(aid)
|
|
t.check("_get_sample_rate(invalid)=44100 fallback", lib._get_sample_rate(None) == 44100)
|
|
t.check("_get_channels(invalid)=2 fallback", lib._get_channels(None) == 2)
|
|
|
|
# 7. _get_aid_for_audio / _get_aid_for_music success paths
|
|
t.log(" --- 7. _get_aid_for_* success paths ---")
|
|
aid_m = lib.play_from_file(WAV_LONG)
|
|
aid_s = lib.play_from_file(WAV_SHORT)
|
|
if isinstance(aid_m, int) and isinstance(aid_s, int):
|
|
r = lib._get_aid_for_music(WAV_LONG)
|
|
t.check("_get_aid_for_music success", r == aid_m, f"got={r}")
|
|
r = lib._get_aid_for_audio(WAV_SHORT)
|
|
t.check("_get_aid_for_audio success", r == aid_s, f"got={r}")
|
|
r = lib._get_aid_for_music(WAV_SHORT)
|
|
t.check("_get_aid_for_music(sound file) -> 1002", isinstance(r, tuple) and r[0] == 1002, f"got={r}")
|
|
r = lib._get_aid_for_audio(WAV_LONG)
|
|
t.check("_get_aid_for_audio(music file) -> 1002", isinstance(r, tuple) and r[0] == 1002, f"got={r}")
|
|
lib.stop_audio(aid_m)
|
|
lib.stop_audio(aid_s)
|
|
|
|
lib.cleanup_function()
|
|
|
|
|
|
# ============================================================================
|
|
# [N] Listening Tests (interactive)
|
|
# ============================================================================
|
|
def test_listen(t):
|
|
t.section("[N] Listening Tests (please put on headphones / turn on speakers)")
|
|
if not MP3_FILE or not os.path.exists(MP3_FILE):
|
|
t.skip("Listening tests", "no MP3 file provided")
|
|
return
|
|
|
|
def ask(q):
|
|
while True:
|
|
try:
|
|
ans = input(f" >>> {q} (y/n): ").strip().lower()
|
|
except EOFError:
|
|
print(" ! Cannot interact; skipping")
|
|
return True
|
|
if ans in ('y', 'n'):
|
|
return ans == 'y'
|
|
print(" Please enter y or n")
|
|
|
|
lib = AudioLibrary()
|
|
try:
|
|
print("\n Now playing at default volume 128 for 3 seconds...")
|
|
aid = lib.play_from_file(MP3_FILE)
|
|
time.sleep(3)
|
|
ok = ask("Did you hear clear, normal-volume music?")
|
|
t.check("Listening 1: normal playback audible", ok)
|
|
lib.stop_audio(aid)
|
|
|
|
print("\n Volume set to 40/128, playing for 2 seconds...")
|
|
aid = lib.play_from_file(MP3_FILE)
|
|
lib.set_volume(aid, 40)
|
|
time.sleep(2)
|
|
ok = ask("Was the volume clearly lower (very quiet)?")
|
|
t.check("Listening 2: volume lowered", ok)
|
|
|
|
print("\n Volume restored to 128, playing for 1 second...")
|
|
lib.set_volume(aid, 128)
|
|
time.sleep(1)
|
|
ok = ask("Did the volume return to loud?")
|
|
t.check("Listening 3: volume restored", ok)
|
|
|
|
print("\n Pausing for 2 seconds (should be silent)...")
|
|
lib.pause_audio(aid)
|
|
time.sleep(2)
|
|
ok = ask("Was it completely silent while paused?")
|
|
t.check("Listening 4: pause is silent", ok)
|
|
|
|
print("\n Resuming for 2 seconds...")
|
|
lib.play_audio(aid)
|
|
time.sleep(2)
|
|
ok = ask("Did playback resume from where it paused?")
|
|
t.check("Listening 5: resume works", ok)
|
|
|
|
print("\n Fade-in test: fadein(5000ms), playing for 6 seconds...")
|
|
lib.fadein_music(aid, ms=5000)
|
|
time.sleep(6)
|
|
ok = ask("Did the sound fade in from silence to full volume?")
|
|
t.check("Listening 6: fade-in effect", ok)
|
|
|
|
print("\n Fade-out test: fadeout(5000ms), waiting 6 seconds...")
|
|
lib.fadeout_music(5000)
|
|
time.sleep(6)
|
|
ok = ask("Did the sound fade out gradually to silence?")
|
|
t.check("Listening 7: fade-out effect", ok)
|
|
|
|
print("\n Seek test: jump to 90s, playing for 2 seconds...")
|
|
lib.fadein_music(aid, ms=0)
|
|
time.sleep(0.3)
|
|
lib.seek_audio(aid, 90.0)
|
|
time.sleep(2)
|
|
ok = ask("Did you hear the middle of the song (not the beginning)?")
|
|
t.check("Listening 8: seek positioning", ok)
|
|
|
|
lib.stop_audio(aid)
|
|
finally:
|
|
lib.cleanup_function()
|
|
print("\n Listening tests finished")
|
|
|
|
|
|
# ============================================================================
|
|
# Main
|
|
# ============================================================================
|
|
def main():
|
|
global MP3_FILE
|
|
mode = sys.argv[1].lower().lstrip('-') if len(sys.argv) > 1 else 'full'
|
|
if mode not in ('auto', 'listen', 'full'):
|
|
print(f"Unknown mode: {mode}, using full")
|
|
mode = 'full'
|
|
MP3_FILE = _prompt_mp3()
|
|
|
|
print(f"\n Python: {sys.version.split()[0]} | Platform: {sys.platform} | Mode: {mode}")
|
|
print(f" ap_ds version: {ap_ds.__version__} | Path: {os.path.dirname(ap_ds.__file__)}")
|
|
print(f" MP3 test file: {MP3_FILE} (exists={bool(MP3_FILE) and os.path.exists(MP3_FILE)})")
|
|
|
|
t = CICD()
|
|
|
|
if mode in ('auto', 'full'):
|
|
test_imports(t)
|
|
test_metadata(t)
|
|
test_init(t)
|
|
lib = AudioLibrary()
|
|
test_play(t, lib)
|
|
test_control(t, lib)
|
|
test_volume(t, lib)
|
|
test_seek(t, lib)
|
|
test_fade(t, lib)
|
|
test_dap(t, lib)
|
|
test_metadata_methods(t, lib)
|
|
test_helpers(t, lib)
|
|
test_resources(t, lib)
|
|
test_edge_cases(t)
|
|
test_init_module(t)
|
|
test_sdl2_bindings(t)
|
|
test_parser_deep(t)
|
|
test_supplementary(t)
|
|
|
|
if mode in ('listen', 'full'):
|
|
test_listen(t)
|
|
|
|
t.summary()
|
|
return 0 if t.failed == 0 else 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|