fieldwitness/tests/test_stegasoo.py
Aaron D. Lee e3bc1cce1f Consolidate stegasoo and verisoo into soosef monorepo
Merge stegasoo (v4.3.0, steganography) and verisoo (v0.1.0, attestation)
as subpackages under soosef.stegasoo and soosef.verisoo. This eliminates
cross-repo coordination and enables atomic changes across the full stack.

- Copy stegasoo (34 modules) and verisoo (15 modules) into src/soosef/
- Convert all verisoo absolute imports to relative imports
- Rewire ~50 import sites across soosef code (cli, web, keystore, tests)
- Replace stegasoo/verisoo pip deps with inlined code + pip extras
  (stego-dct, stego-audio, attest, web, api, cli, fieldkit, all, dev)
- Add _availability.py for runtime feature detection
- Add unified FastAPI mount point at soosef.api
- Copy and adapt tests from both repos (155 pass, 1 skip)
- Drop standalone CLI/web frontends; keep FastAPI as optional modules
- Both source repos tagged pre-monorepo-consolidation on GitHub

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-01 19:06:14 -04:00

682 lines
19 KiB
Python

"""
Stegasoo Library Unit Tests
Tests core functionality: encode/decode, LSB/DCT modes, channel keys, validation.
"""
import io
from pathlib import Path
import pytest
from PIL import Image
import soosef.stegasoo as stegasoo
from soosef.stegasoo import (
decode,
decode_text,
encode,
generate_channel_key,
generate_passphrase,
generate_pin,
has_dct_support,
validate_image,
validate_message,
validate_passphrase,
validate_pin,
)
# Test data paths
TEST_DATA = Path(__file__).parent.parent / "test_data"
CARRIER_PATH = TEST_DATA / "carrier.jpg"
REF_PATH = TEST_DATA / "ref.jpg"
# Test credentials
TEST_PASSPHRASE = "tower booty sunny windy toasty spicy"
TEST_PIN = "727643678"
TEST_MESSAGE = "Hello, Stegasoo!"
@pytest.fixture
def carrier_bytes():
"""Load carrier image as bytes."""
return CARRIER_PATH.read_bytes()
@pytest.fixture
def ref_bytes():
"""Load reference image as bytes."""
return REF_PATH.read_bytes()
@pytest.fixture
def small_image():
"""Create a small test image in memory."""
img = Image.new("RGB", (200, 200), color="blue")
buf = io.BytesIO()
img.save(buf, format="PNG")
return buf.getvalue()
class TestVersion:
"""Test version info."""
def test_version_exists(self):
assert hasattr(stegasoo, "__version__")
assert stegasoo.__version__
def test_version_format(self):
parts = stegasoo.__version__.split(".")
assert len(parts) >= 2
assert all(p.isdigit() for p in parts[:2])
class TestGeneration:
"""Test credential generation."""
def test_generate_passphrase_default(self):
passphrase = generate_passphrase()
words = passphrase.split()
assert len(words) == stegasoo.DEFAULT_PASSPHRASE_WORDS
def test_generate_passphrase_custom_length(self):
passphrase = generate_passphrase(words=8)
words = passphrase.split()
assert len(words) == 8
def test_generate_pin_default(self):
pin = generate_pin()
assert pin.isdigit()
assert len(pin) == 6 # Default is 6 digits
def test_generate_pin_custom_length(self):
pin = generate_pin(length=9)
assert pin.isdigit()
assert len(pin) == 9
def test_generate_channel_key(self):
key = generate_channel_key()
# Format: XXXX-XXXX-XXXX-XXXX-XXXX-XXXX-XXXX-XXXX (39 chars)
assert len(key) == 39
assert key.count("-") == 7
class TestValidation:
"""Test validation functions."""
def test_validate_passphrase_valid(self):
result = validate_passphrase(TEST_PASSPHRASE)
assert result.is_valid
def test_validate_passphrase_too_short(self):
result = validate_passphrase("one two")
assert not result.is_valid
def test_validate_pin_valid(self):
result = validate_pin(TEST_PIN)
assert result.is_valid
def test_validate_pin_too_short(self):
result = validate_pin("123")
assert not result.is_valid
def test_validate_pin_non_numeric(self):
result = validate_pin("abc123")
assert not result.is_valid
def test_validate_message_valid(self):
result = validate_message("Hello world")
assert result.is_valid
def test_validate_message_empty(self):
result = validate_message("")
assert not result.is_valid
def test_validate_image_valid(self, carrier_bytes):
result = validate_image(carrier_bytes)
assert result.is_valid
def test_validate_image_invalid(self):
result = validate_image(b"not an image")
assert not result.is_valid
class TestLSBMode:
"""Test LSB (Least Significant Bit) encoding/decoding."""
def test_encode_decode_roundtrip(self, carrier_bytes, ref_bytes):
"""Basic encode/decode roundtrip."""
result = encode(
message=TEST_MESSAGE,
reference_photo=ref_bytes,
carrier_image=carrier_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
embed_mode="lsb",
)
assert result.stego_image
assert len(result.stego_image) > 0
decoded = decode(
stego_image=result.stego_image,
reference_photo=ref_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
embed_mode="lsb",
)
assert decoded.message == TEST_MESSAGE
def test_decode_text_helper(self, carrier_bytes, ref_bytes):
"""Test decode_text convenience function."""
result = encode(
message=TEST_MESSAGE,
reference_photo=ref_bytes,
carrier_image=carrier_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
embed_mode="lsb",
)
text = decode_text(
stego_image=result.stego_image,
reference_photo=ref_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
embed_mode="lsb",
)
assert text == TEST_MESSAGE
def test_wrong_passphrase_fails(self, carrier_bytes, ref_bytes):
"""Decoding with wrong passphrase should fail."""
result = encode(
message=TEST_MESSAGE,
reference_photo=ref_bytes,
carrier_image=carrier_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
embed_mode="lsb",
)
with pytest.raises(Exception):
decode(
stego_image=result.stego_image,
reference_photo=ref_bytes,
passphrase="wrong passphrase words here now",
pin=TEST_PIN,
embed_mode="lsb",
)
def test_wrong_pin_fails(self, carrier_bytes, ref_bytes):
"""Decoding with wrong PIN should fail."""
result = encode(
message=TEST_MESSAGE,
reference_photo=ref_bytes,
carrier_image=carrier_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
embed_mode="lsb",
)
with pytest.raises(Exception):
decode(
stego_image=result.stego_image,
reference_photo=ref_bytes,
passphrase=TEST_PASSPHRASE,
pin="999999999",
embed_mode="lsb",
)
def test_wrong_reference_fails(self, carrier_bytes, ref_bytes, small_image):
"""Decoding with wrong reference should fail."""
result = encode(
message=TEST_MESSAGE,
reference_photo=ref_bytes,
carrier_image=carrier_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
embed_mode="lsb",
)
with pytest.raises(Exception):
decode(
stego_image=result.stego_image,
reference_photo=small_image, # Wrong reference
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
embed_mode="lsb",
)
class TestDCTMode:
"""Test DCT (Discrete Cosine Transform) encoding/decoding."""
@pytest.fixture(autouse=True)
def check_dct_support(self):
"""Skip DCT tests if not supported."""
if not has_dct_support():
pytest.skip("DCT support not available")
def test_encode_decode_roundtrip(self, carrier_bytes, ref_bytes):
"""Basic DCT encode/decode roundtrip."""
result = encode(
message=TEST_MESSAGE,
reference_photo=ref_bytes,
carrier_image=carrier_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
embed_mode="dct",
)
assert result.stego_image
decoded = decode(
stego_image=result.stego_image,
reference_photo=ref_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
embed_mode="dct",
)
assert decoded.message == TEST_MESSAGE
def test_dct_jpeg_output(self, carrier_bytes, ref_bytes):
"""Test DCT mode with JPEG output."""
result = encode(
message=TEST_MESSAGE,
reference_photo=ref_bytes,
carrier_image=carrier_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
embed_mode="dct",
dct_output_format="jpeg",
)
assert result.stego_image
# Verify it's JPEG by checking magic bytes
assert result.stego_image[:2] == b"\xff\xd8"
class TestChannelKey:
"""Test channel key functionality."""
def test_encode_with_channel_key(self, carrier_bytes, ref_bytes):
"""Encode with channel key."""
channel_key = generate_channel_key()
result = encode(
message=TEST_MESSAGE,
reference_photo=ref_bytes,
carrier_image=carrier_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
channel_key=channel_key,
embed_mode="lsb",
)
assert result.stego_image
# Decode with same channel key
decoded = decode(
stego_image=result.stego_image,
reference_photo=ref_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
channel_key=channel_key,
embed_mode="lsb",
)
assert decoded.message == TEST_MESSAGE
def test_wrong_channel_key_fails(self, carrier_bytes, ref_bytes):
"""Decoding with wrong channel key should fail."""
channel_key = generate_channel_key()
wrong_key = generate_channel_key()
result = encode(
message=TEST_MESSAGE,
reference_photo=ref_bytes,
carrier_image=carrier_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
channel_key=channel_key,
embed_mode="lsb",
)
with pytest.raises(Exception):
decode(
stego_image=result.stego_image,
reference_photo=ref_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
channel_key=wrong_key,
embed_mode="lsb",
)
class TestCompression:
"""Test message compression."""
def test_long_message_compresses(self, carrier_bytes, ref_bytes):
"""Long messages should be compressed."""
long_message = "A" * 1000
result = encode(
message=long_message,
reference_photo=ref_bytes,
carrier_image=carrier_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
embed_mode="lsb",
)
assert result.stego_image
decoded = decode(
stego_image=result.stego_image,
reference_photo=ref_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
embed_mode="lsb",
)
assert decoded.message == long_message
class TestEdgeCases:
"""Test edge cases and error handling."""
def test_unicode_message(self, carrier_bytes, ref_bytes):
"""Test encoding Unicode messages."""
unicode_msg = "Hello 🦖 Stegasoo! 日本語 émojis"
result = encode(
message=unicode_msg,
reference_photo=ref_bytes,
carrier_image=carrier_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
embed_mode="lsb",
)
assert result.stego_image
decoded = decode(
stego_image=result.stego_image,
reference_photo=ref_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
embed_mode="lsb",
)
assert decoded.message == unicode_msg
def test_minimum_passphrase(self, carrier_bytes, ref_bytes):
"""Test with minimum valid passphrase."""
min_passphrase = "one two three four" # 4 words minimum
result = encode(
message=TEST_MESSAGE,
reference_photo=ref_bytes,
carrier_image=carrier_bytes,
passphrase=min_passphrase,
pin=TEST_PIN,
embed_mode="lsb",
)
assert result.stego_image
def test_special_characters_in_message(self, carrier_bytes, ref_bytes):
"""Test special characters in message."""
special_msg = "Line1\nLine2\tTab\r\nCRLF"
result = encode(
message=special_msg,
reference_photo=ref_bytes,
carrier_image=carrier_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
embed_mode="lsb",
)
assert result.stego_image
decoded = decode(
stego_image=result.stego_image,
reference_photo=ref_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
embed_mode="lsb",
)
assert decoded.message == special_msg
# =============================================================================
# VIDEO STEGANOGRAPHY TESTS (v4.4.0)
# =============================================================================
@pytest.fixture
def test_video_bytes():
"""Create a minimal test video using ffmpeg.
Creates a 2-second test video with solid color frames.
Returns None if ffmpeg is not available.
"""
import shutil
import subprocess
import tempfile
if not shutil.which("ffmpeg"):
return None
with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
output_path = f.name
try:
# Create a simple 2-second video with colored frames
# Using lavfi (libavfilter) to generate test pattern
result = subprocess.run(
[
"ffmpeg",
"-y",
"-f",
"lavfi",
"-i",
"color=c=blue:s=320x240:d=2:r=10",
"-c:v",
"libx264",
"-pix_fmt",
"yuv420p",
"-g",
"5", # GOP size - creates I-frames every 5 frames
output_path,
],
capture_output=True,
timeout=30,
)
if result.returncode != 0:
return None
with open(output_path, "rb") as f:
video_data = f.read()
return video_data
except Exception:
return None
finally:
import os
try:
os.unlink(output_path)
except OSError:
pass
class TestVideoSupport:
"""Test video steganography support detection."""
def test_video_support_flag_exists(self):
"""HAS_VIDEO_SUPPORT flag should exist."""
assert hasattr(stegasoo, "HAS_VIDEO_SUPPORT")
assert isinstance(stegasoo.HAS_VIDEO_SUPPORT, bool)
def test_video_constants_exist(self):
"""Video-related constants should exist."""
assert hasattr(stegasoo, "EMBED_MODE_VIDEO_LSB")
assert hasattr(stegasoo, "EMBED_MODE_VIDEO_AUTO")
@pytest.mark.skipif(
not stegasoo.HAS_VIDEO_SUPPORT,
reason="Video support not available (ffmpeg or dependencies missing)",
)
class TestVideoFormatDetection:
"""Test video format detection."""
def test_detect_video_format_mp4(self, test_video_bytes):
"""Should detect MP4 format from magic bytes."""
if test_video_bytes is None:
pytest.skip("Could not create test video")
from soosef.stegasoo import detect_video_format
fmt = detect_video_format(test_video_bytes)
assert fmt in ("mp4", "mov")
def test_detect_video_format_unknown(self):
"""Should return 'unknown' for non-video data."""
from soosef.stegasoo import detect_video_format
fmt = detect_video_format(b"not a video")
assert fmt == "unknown"
@pytest.mark.skipif(
not stegasoo.HAS_VIDEO_SUPPORT,
reason="Video support not available (ffmpeg or dependencies missing)",
)
class TestVideoInfo:
"""Test video metadata extraction."""
def test_get_video_info(self, test_video_bytes):
"""Should extract video metadata."""
if test_video_bytes is None:
pytest.skip("Could not create test video")
from soosef.stegasoo import get_video_info
info = get_video_info(test_video_bytes)
assert info.width == 320
assert info.height == 240
assert info.fps > 0
assert info.duration_seconds > 0
assert info.total_frames > 0
assert info.format in ("mp4", "mov")
def test_validate_video(self, test_video_bytes):
"""Should validate video data."""
if test_video_bytes is None:
pytest.skip("Could not create test video")
from soosef.stegasoo import validate_video
result = validate_video(test_video_bytes, check_duration=False)
assert result.is_valid
assert result.details.get("format") in ("mp4", "mov")
@pytest.mark.skipif(
not stegasoo.HAS_VIDEO_SUPPORT,
reason="Video support not available (ffmpeg or dependencies missing)",
)
class TestVideoCapacity:
"""Test video capacity calculation."""
def test_calculate_video_capacity(self, test_video_bytes):
"""Should calculate steganographic capacity."""
if test_video_bytes is None:
pytest.skip("Could not create test video")
from soosef.stegasoo import calculate_video_capacity
capacity_info = calculate_video_capacity(test_video_bytes)
assert capacity_info.total_frames > 0
assert capacity_info.i_frames > 0
assert capacity_info.usable_capacity_bytes > 0
assert capacity_info.embed_mode == "video_lsb"
assert capacity_info.resolution == (320, 240)
@pytest.mark.skipif(
not stegasoo.HAS_VIDEO_SUPPORT,
reason="Video support not available (ffmpeg or dependencies missing)",
)
class TestVideoEncodeDecode:
"""Test video steganography round-trip."""
def test_video_roundtrip(self, test_video_bytes, ref_bytes):
"""Test encoding and decoding a message in video."""
if test_video_bytes is None:
pytest.skip("Could not create test video")
from soosef.stegasoo import decode_video, encode_video
message = "Secret video message!"
# Encode
stego_video, stats = encode_video(
message=message,
reference_photo=ref_bytes,
carrier_video=test_video_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
)
assert stego_video
assert len(stego_video) > 0
assert stats.frames_modified > 0
assert stats.codec == "ffv1" # Should use lossless codec
# Decode
result = decode_video(
stego_video=stego_video,
reference_photo=ref_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
)
assert result.is_text
assert result.message == message
def test_video_wrong_passphrase_fails(self, test_video_bytes, ref_bytes):
"""Decoding with wrong passphrase should fail."""
if test_video_bytes is None:
pytest.skip("Could not create test video")
from soosef.stegasoo import decode_video, encode_video
message = "Secret video message!"
stego_video, _ = encode_video(
message=message,
reference_photo=ref_bytes,
carrier_video=test_video_bytes,
passphrase=TEST_PASSPHRASE,
pin=TEST_PIN,
)
with pytest.raises(Exception):
decode_video(
stego_video=stego_video,
reference_photo=ref_bytes,
passphrase="wrong passphrase words here",
pin=TEST_PIN,
)