# SPDX-License-Identifier: AGPL-3.0-only # Copyright (c) 2026 sol pbc """Tests for observe/transfer.py - day archive export, import, and send.""" import inspect import json import tarfile from pathlib import Path from unittest.mock import patch import pytest class TestSegmentDeconfliction: """Tests for segment deconfliction via find_available_segment.""" def test_find_available_segment_returns_original_if_free(self, tmp_path): """Test find_available_segment returns original if available.""" from solstone.observe.utils import find_available_segment # No existing segments result = find_available_segment(tmp_path, "120000_300") assert result == "120000_300" def test_find_available_segment_finds_alternative(self, tmp_path): """Test find_available_segment finds alternative when original taken.""" from solstone.observe.utils import find_available_segment # Create existing segment (tmp_path / "120000_300").mkdir() result = find_available_segment(tmp_path, "120000_300") assert result is not None assert result != "120000_300" # Should be a valid segment key format assert "_" in result def test_find_available_segment_returns_none_when_exhausted(self, tmp_path): """Test find_available_segment returns None when all slots taken.""" from solstone.observe.utils import find_available_segment # Create many segments around the target for delta in range(-50, 51): for dur_delta in range(-50, 51): total_seconds = 12 * 3600 + delta if 0 <= total_seconds < 86400: h = total_seconds // 3600 m = (total_seconds % 3600) // 60 s = total_seconds % 60 dur = 300 + dur_delta if dur > 0: (tmp_path / f"{h:02d}{m:02d}{s:02d}_{dur}").mkdir(exist_ok=True) # With so many slots filled, should eventually fail result = find_available_segment(tmp_path, "120000_300", max_attempts=10) # May or may not find one depending on random walk, but shouldn't crash assert result is None or "_" in result class TestComputeSha256: """Tests for SHA256 computation utilities.""" def test_compute_file_sha256(self, tmp_path): """Test compute_file_sha256 returns correct hash.""" from solstone.observe.utils import compute_file_sha256 test_file = tmp_path / "test.txt" test_file.write_bytes(b"hello world") # Known SHA256 of "hello world" expected = "b94d27b9934d3e08a52e52d7da7dabfac484efe37a5380ee9088f7ace2efcde9" assert compute_file_sha256(test_file) == expected def test_compute_bytes_sha256(self): """Test compute_bytes_sha256 returns correct hash.""" from solstone.observe.utils import compute_bytes_sha256 # Known SHA256 of "hello world" expected = "b94d27b9934d3e08a52e52d7da7dabfac484efe37a5380ee9088f7ace2efcde9" assert compute_bytes_sha256(b"hello world") == expected class TestTransferExport: """Tests for archive creation (export).""" def test_create_archive_basic(self, tmp_path, monkeypatch): """Test create_archive creates valid archive.""" from solstone.observe.transfer import create_archive # Set up mock journal with day/stream/segment structure journal_path = tmp_path / "journal" day_dir = journal_path / "chronicle" / "20250101" segment_dir = day_dir / "default" / "120000_300" segment_dir.mkdir(parents=True) # Add test files to segment (segment_dir / "audio.flac").write_bytes(b"fake audio data") (segment_dir / "audio.jsonl").write_text('{"raw": "audio.flac"}\n') monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) # Clear cache import solstone.think.utils as think_utils think_utils._journal_path_cache = None output_path = tmp_path / "test.tgz" result = create_archive("20250101", output_path) assert result == output_path assert output_path.exists() # Verify archive contents with tarfile.open(output_path, "r:gz") as tar: names = tar.getnames() assert "manifest.json" in names assert "default/120000_300/audio.flac" in names assert "default/120000_300/audio.jsonl" in names # Verify manifest manifest_file = tar.extractfile("manifest.json") manifest = json.load(manifest_file) assert manifest["version"] == 1 assert manifest["day"] == "20250101" assert "default/120000_300" in manifest["segments"] def test_create_archive_no_segments_error(self, tmp_path, monkeypatch): """Test create_archive raises error for empty day.""" from solstone.observe.transfer import create_archive journal_path = tmp_path / "journal" day_dir = journal_path / "chronicle" / "20250101" day_dir.mkdir(parents=True) monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None with pytest.raises(ValueError, match="No segments found"): create_archive("20250101") def test_create_archive_no_day_error(self, tmp_path, monkeypatch): """Test create_archive raises error for missing day.""" from solstone.observe.transfer import create_archive journal_path = tmp_path / "journal" journal_path.mkdir(parents=True) monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None with pytest.raises(ValueError, match="does not exist"): create_archive("20250101") class TestTransferImport: """Tests for archive import.""" def _create_test_archive(self, tmp_path, segments: dict) -> Path: """Helper to create test archive.""" archive_path = tmp_path / "test.tgz" manifest = { "version": 1, "day": "20250101", "created_at": 1704067200000, "host": "test-host", "segments": {}, } with tarfile.open(archive_path, "w:gz") as tar: for segment, files in segments.items(): manifest["segments"][segment] = {"files": []} for filename, content in files.items(): # Add to manifest from solstone.observe.utils import compute_bytes_sha256 manifest["segments"][segment]["files"].append( { "name": filename, "sha256": compute_bytes_sha256(content), "size": len(content), } ) # Add file to archive import io info = tarfile.TarInfo(name=f"{segment}/{filename}") info.size = len(content) tar.addfile(info, io.BytesIO(content)) # Add manifest import io manifest_json = json.dumps(manifest).encode() info = tarfile.TarInfo(name="manifest.json") info.size = len(manifest_json) tar.addfile(info, io.BytesIO(manifest_json)) return archive_path def test_validate_archive_all_new(self, tmp_path, monkeypatch): """Test validate_archive with no existing segments.""" from solstone.observe.transfer import validate_archive # Create archive archive_path = self._create_test_archive( tmp_path, { "120000_300": {"audio.flac": b"audio data"}, "130000_300": {"audio.flac": b"more audio"}, }, ) # Set up empty journal journal_path = tmp_path / "journal" journal_path.mkdir() monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None result = validate_archive(archive_path) assert result["skip"] == [] assert len(result["import_as"]) == 2 assert result["import_as"]["120000_300"] == "120000_300" assert result["import_as"]["130000_300"] == "130000_300" def test_validate_archive_skip_matching(self, tmp_path, monkeypatch): """Test validate_archive skips segments with matching hashes.""" from solstone.observe.transfer import validate_archive # Create archive content = b"audio data" archive_path = self._create_test_archive( tmp_path, {"120000_300": {"audio.flac": content}}, ) # Set up journal with matching segment journal_path = tmp_path / "journal" segment_dir = journal_path / "chronicle" / "20250101" / "120000_300" segment_dir.mkdir(parents=True) (segment_dir / "audio.flac").write_bytes(content) monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None result = validate_archive(archive_path) assert "120000_300" in result["skip"] assert "120000_300" not in result["import_as"] def test_validate_archive_deconflict_different(self, tmp_path, monkeypatch): """Test validate_archive deconflicts segments with different content.""" from solstone.observe.transfer import validate_archive # Create archive archive_path = self._create_test_archive( tmp_path, {"120000_300": {"audio.flac": b"new audio data"}}, ) # Set up journal with different content in same segment journal_path = tmp_path / "journal" segment_dir = journal_path / "chronicle" / "20250101" / "120000_300" segment_dir.mkdir(parents=True) (segment_dir / "audio.flac").write_bytes(b"existing different data") monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None result = validate_archive(archive_path) assert "120000_300" in result["deconflicted"] assert result["import_as"]["120000_300"] != "120000_300" def test_import_archive_basic(self, tmp_path, monkeypatch): """Test import_archive extracts segments correctly.""" from solstone.observe.transfer import import_archive # Create archive audio_content = b"fake audio data" jsonl_content = b'{"raw": "audio.flac"}\n' archive_path = self._create_test_archive( tmp_path, { "120000_300": { "audio.flac": audio_content, "audio.jsonl": jsonl_content, } }, ) # Set up empty journal journal_path = tmp_path / "journal" journal_path.mkdir() monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None # Mock subprocess to avoid running real indexer with patch("subprocess.run"): result = import_archive(archive_path) assert result["status"] == "imported" assert "120000_300" in result["imported"] # Verify files were extracted segment_dir = journal_path / "chronicle" / "20250101" / "120000_300" assert segment_dir.exists() assert (segment_dir / "audio.flac").read_bytes() == audio_content assert (segment_dir / "audio.jsonl").read_bytes() == jsonl_content def test_import_archive_preserves_content_and_mtime(self, tmp_path, monkeypatch): """Test import_archive preserves file bytes and tar member mtime.""" import io from solstone.observe.transfer import import_archive from solstone.observe.utils import compute_bytes_sha256 archive_path = tmp_path / "test.tgz" content = b"known audio bytes" mtime = 1700000000 manifest = { "version": 1, "day": "20250101", "created_at": 1704067200000, "host": "test-host", "segments": { "120000_300": { "files": [ { "name": "audio.flac", "sha256": compute_bytes_sha256(content), "size": len(content), } ] } }, } with tarfile.open(archive_path, "w:gz") as tar: file_info = tarfile.TarInfo(name="120000_300/audio.flac") file_info.size = len(content) file_info.mtime = mtime tar.addfile(file_info, io.BytesIO(content)) manifest_json = json.dumps(manifest).encode() manifest_info = tarfile.TarInfo(name="manifest.json") manifest_info.size = len(manifest_json) tar.addfile(manifest_info, io.BytesIO(manifest_json)) journal_path = tmp_path / "journal" journal_path.mkdir() monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None result = import_archive(archive_path) target_path = ( journal_path / "chronicle" / "20250101" / "120000_300" / "audio.flac" ) assert result["status"] == "imported" assert target_path.read_bytes() == content assert int(target_path.stat().st_mtime) == mtime def test_import_archive_installs_zero_byte_member(self, tmp_path, monkeypatch): """Test import_archive installs empty regular-file members.""" from solstone.observe.transfer import import_archive archive_path = self._create_test_archive( tmp_path, {"120000_300": {"audio.flac": b""}}, ) journal_path = tmp_path / "journal" journal_path.mkdir() monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None result = import_archive(archive_path) target_path = ( journal_path / "chronicle" / "20250101" / "120000_300" / "audio.flac" ) assert target_path.exists() assert target_path.stat().st_size == 0 assert "120000_300" in result["imported"] def test_import_archive_deconflict_installs_into_new_segment( self, tmp_path, monkeypatch ): """Test deconflicted imports install into the generated segment.""" from solstone.observe.transfer import import_archive original_content = b"existing different data" new_content = b"new content" archive_path = self._create_test_archive( tmp_path, {"120000_300": {"audio.flac": new_content}}, ) journal_path = tmp_path / "journal" original_path = ( journal_path / "chronicle" / "20250101" / "120000_300" / "audio.flac" ) original_path.parent.mkdir(parents=True) original_path.write_bytes(original_content) monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None with patch("subprocess.run"): result = import_archive(archive_path) day_dir = journal_path / "chronicle" / "20250101" deconflicted_paths = [ path for path in day_dir.iterdir() if path.is_dir() and path.name != "120000_300" and (path / "audio.flac").read_bytes() == new_content ] assert result["deconflicted"] assert deconflicted_paths assert original_path.read_bytes() == original_content def test_import_archive_mid_extract_failure_cleans_temp_files( self, tmp_path, monkeypatch ): """Test import_archive removes temp files when a later promote fails.""" import solstone.observe.transfer as transfer archive_path = self._create_test_archive( tmp_path, { "120000_300": {"audio.flac": b"first content"}, "130000_300": {"audio.flac": b"second content"}, }, ) journal_path = tmp_path / "journal" journal_path.mkdir() monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None real_install_file = transfer.install_file calls = 0 def failing_second_install(temp_path, target_path, *, mode=None): nonlocal calls calls += 1 if calls == 1: real_install_file(temp_path, target_path, mode=mode) return raise RuntimeError("simulated install failure") with patch( "solstone.observe.transfer.install_file", side_effect=failing_second_install, ): with pytest.raises(RuntimeError, match="simulated install failure"): transfer.import_archive(archive_path) day_dir = journal_path / "chronicle" / "20250101" first_path = day_dir / "120000_300" / "audio.flac" assert first_path.read_bytes() == b"first content" assert list(day_dir.rglob("*.tmp")) == [] def test_import_archive_rejects_member_sha256_mismatch(self, tmp_path, monkeypatch): """Test manifest/member hash mismatch aborts before promoting that file.""" import io from solstone.observe.transfer import import_archive from solstone.observe.utils import compute_bytes_sha256 archive_path = tmp_path / "sha-mismatch.tgz" synced_content = b"already synced" good_content = b"good import" bad_manifest_content = b"manifest bytes" bad_member_content = b"tampered bytes" manifest = { "version": 1, "day": "20250101", "created_at": 1704067200000, "host": "test-host", "segments": { "120000_300": { "files": [ { "name": "audio.flac", "sha256": compute_bytes_sha256(synced_content), "size": len(synced_content), } ] }, "125000_300": { "files": [ { "name": "audio.flac", "sha256": compute_bytes_sha256(good_content), "size": len(good_content), } ] }, "130000_300": { "files": [ { "name": "audio.flac", "sha256": compute_bytes_sha256(bad_manifest_content), "size": len(bad_manifest_content), } ] }, }, } with tarfile.open(archive_path, "w:gz") as tar: for segment, content in ( ("120000_300", synced_content), ("125000_300", good_content), ("130000_300", bad_member_content), ): info = tarfile.TarInfo(name=f"{segment}/audio.flac") info.size = len(content) tar.addfile(info, io.BytesIO(content)) manifest_json = json.dumps(manifest).encode() manifest_info = tarfile.TarInfo(name="manifest.json") manifest_info.size = len(manifest_json) tar.addfile(manifest_info, io.BytesIO(manifest_json)) journal_path = tmp_path / "journal" synced_path = ( journal_path / "chronicle" / "20250101" / "120000_300" / "audio.flac" ) synced_path.parent.mkdir(parents=True) synced_path.write_bytes(synced_content) monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None bad_target = ( journal_path / "chronicle" / "20250101" / "130000_300" / "audio.flac" ) with pytest.raises(ValueError) as excinfo: import_archive(archive_path) message = str(excinfo.value) assert "Archive content mismatch" in message assert str(bad_target) in message assert not bad_target.exists() assert ( journal_path / "chronicle" / "20250101" / "125000_300" / "audio.flac" ).read_bytes() == good_content assert synced_path.read_bytes() == synced_content assert list((journal_path / "chronicle").rglob("*.tmp")) == [] def test_import_archive_routes_member_writes_through_install_file(self): """Test import_archive has no raw durable member write path.""" from solstone.observe import transfer src = inspect.getsource(transfer.import_archive) assert "install_file(" in src assert "open(target_path" not in src assert ".write(source" not in src assert "write_bytes" not in src assert "write_text" not in src def test_import_archive_dry_run(self, tmp_path, monkeypatch): """Test import_archive dry run doesn't modify filesystem.""" from solstone.observe.transfer import import_archive archive_path = self._create_test_archive( tmp_path, {"120000_300": {"audio.flac": b"audio data"}}, ) journal_path = tmp_path / "journal" journal_path.mkdir() monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None result = import_archive(archive_path, dry_run=True) assert result["status"] == "dry_run" # Directory should not be created assert not (journal_path / "chronicle" / "20250101").exists() def test_import_archive_nothing_to_import(self, tmp_path, monkeypatch): """Test import_archive when all segments already synced.""" from solstone.observe.transfer import import_archive content = b"audio data" archive_path = self._create_test_archive( tmp_path, {"120000_300": {"audio.flac": content}}, ) # Set up journal with matching content journal_path = tmp_path / "journal" segment_dir = journal_path / "chronicle" / "20250101" / "120000_300" segment_dir.mkdir(parents=True) (segment_dir / "audio.flac").write_bytes(content) monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None result = import_archive(archive_path) assert result["status"] == "nothing_to_import" def test_import_archive_rejects_dotdot_member(self, tmp_path, monkeypatch): """AC#1a: dot-dot member names are rejected before install.""" import io from solstone.observe.transfer import import_archive archive_path = tmp_path / "dotdot-member.tgz" manifest = { "version": 1, "day": "20250101", "created_at": 1704067200000, "host": "test-host", "segments": {"120000_300": {"files": []}}, } with tarfile.open(archive_path, "w:gz") as tar: file_info = tarfile.TarInfo(name="120000_300/../../../evil") content = b"evil" file_info.size = len(content) tar.addfile(file_info, io.BytesIO(content)) manifest_json = json.dumps(manifest).encode() manifest_info = tarfile.TarInfo(name="manifest.json") manifest_info.size = len(manifest_json) tar.addfile(manifest_info, io.BytesIO(manifest_json)) journal_path = tmp_path / "journal" journal_path.mkdir() monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None with pytest.raises(ValueError, match="unsafe member filename"): import_archive(archive_path) assert not (journal_path / "evil").exists() assert not ( journal_path / "chronicle" / "20250101" / "120000_300" / "evil" ).exists() def test_import_archive_rejects_absolute_member(self, tmp_path, monkeypatch): """AC#1b: absolute member filenames are rejected before install.""" import io from solstone.observe.transfer import import_archive archive_path = tmp_path / "absolute-member.tgz" sentinel = tmp_path / "abs_member_sentinel" manifest = { "version": 1, "day": "20250101", "created_at": 1704067200000, "host": "test-host", "segments": {"120000_300": {"files": []}}, } with tarfile.open(archive_path, "w:gz") as tar: file_info = tarfile.TarInfo(name=f"120000_300/{sentinel.as_posix()}") content = b"evil" file_info.size = len(content) tar.addfile(file_info, io.BytesIO(content)) manifest_json = json.dumps(manifest).encode() manifest_info = tarfile.TarInfo(name="manifest.json") manifest_info.size = len(manifest_json) tar.addfile(manifest_info, io.BytesIO(manifest_json)) journal_path = tmp_path / "journal" journal_path.mkdir() monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None with pytest.raises(ValueError, match="unsafe member filename"): import_archive(archive_path) assert not sentinel.exists() def test_import_archive_prepass_blocks_partial_import(self, tmp_path, monkeypatch): """AC#2: a later unsafe member aborts before any earlier extraction.""" import io from solstone.observe.transfer import import_archive from solstone.observe.utils import compute_bytes_sha256 archive_path = tmp_path / "partial-import.tgz" good_content = b"good audio" sentinel = tmp_path / "partial_abs_member_sentinel" manifest = { "version": 1, "day": "20250101", "created_at": 1704067200000, "host": "test-host", "segments": { "120000_300": { "files": [ { "name": "audio.flac", "sha256": compute_bytes_sha256(good_content), "size": len(good_content), } ] }, "130000_300": {"files": []}, }, } with tarfile.open(archive_path, "w:gz") as tar: good_info = tarfile.TarInfo(name="120000_300/audio.flac") good_info.size = len(good_content) tar.addfile(good_info, io.BytesIO(good_content)) bad_info = tarfile.TarInfo(name=f"130000_300/{sentinel.as_posix()}") bad_content = b"evil" bad_info.size = len(bad_content) tar.addfile(bad_info, io.BytesIO(bad_content)) manifest_json = json.dumps(manifest).encode() manifest_info = tarfile.TarInfo(name="manifest.json") manifest_info.size = len(manifest_json) tar.addfile(manifest_info, io.BytesIO(manifest_json)) journal_path = tmp_path / "journal" journal_path.mkdir() monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None with pytest.raises(ValueError, match="unsafe member filename"): import_archive(archive_path) day_dir = journal_path / "chronicle" / "20250101" assert not (day_dir / "120000_300" / "audio.flac").exists() assert not day_dir.exists() assert not sentinel.exists() assert list((journal_path / "chronicle").rglob("*.tmp")) == [] def test_import_archive_rejects_dotdot_arc_key(self, tmp_path, monkeypatch): """AC#3a: traversal segment keys are rejected before target mkdir.""" import io from solstone.observe.transfer import import_archive archive_path = tmp_path / "dotdot-arc.tgz" manifest = { "version": 1, "day": "20250101", "created_at": 1704067200000, "host": "test-host", "segments": {"../../../escape_arc": {"files": []}}, } with tarfile.open(archive_path, "w:gz") as tar: manifest_json = json.dumps(manifest).encode() manifest_info = tarfile.TarInfo(name="manifest.json") manifest_info.size = len(manifest_json) tar.addfile(manifest_info, io.BytesIO(manifest_json)) journal_path = tmp_path / "journal" journal_path.mkdir() monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None with pytest.raises(ValueError, match="unsafe segment key"): import_archive(archive_path) assert not (tmp_path / "escape_arc").exists() def test_import_archive_rejects_absolute_arc_key(self, tmp_path, monkeypatch): """AC#3b: absolute segment keys are rejected before target mkdir.""" import io from solstone.observe.transfer import import_archive archive_path = tmp_path / "absolute-arc.tgz" sentinel_dir = tmp_path / "abs_arc_sentinel" manifest = { "version": 1, "day": "20250101", "created_at": 1704067200000, "host": "test-host", "segments": {sentinel_dir.as_posix(): {"files": []}}, } with tarfile.open(archive_path, "w:gz") as tar: manifest_json = json.dumps(manifest).encode() manifest_info = tarfile.TarInfo(name="manifest.json") manifest_info.size = len(manifest_json) tar.addfile(manifest_info, io.BytesIO(manifest_json)) journal_path = tmp_path / "journal" journal_path.mkdir() monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None with pytest.raises(ValueError, match="unsafe segment key"): import_archive(archive_path) assert not sentinel_dir.exists() def test_import_archive_rejects_invalid_day(self, tmp_path, monkeypatch): """AC#4: invalid manifest day still fails through day_path validation.""" import io from solstone.observe.transfer import import_archive archive_path = tmp_path / "bad-day.tgz" manifest = { "version": 1, "day": "bad", "created_at": 1704067200000, "host": "test-host", "segments": {"120000_300": {"files": []}}, } with tarfile.open(archive_path, "w:gz") as tar: manifest_json = json.dumps(manifest).encode() manifest_info = tarfile.TarInfo(name="manifest.json") manifest_info.size = len(manifest_json) tar.addfile(manifest_info, io.BytesIO(manifest_json)) journal_path = tmp_path / "journal" journal_path.mkdir() monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None with pytest.raises(ValueError, match="day must be in YYYYMMDD format"): import_archive(archive_path) def test_import_archive_rejects_empty_member_filename(self, tmp_path, monkeypatch): """AC#5: a member exactly matching the segment prefix is rejected.""" import io from solstone.observe.transfer import import_archive archive_path = tmp_path / "empty-filename.tgz" manifest = { "version": 1, "day": "20250101", "created_at": 1704067200000, "host": "test-host", "segments": {"120000_300": {"files": []}}, } with tarfile.open(archive_path, "w:gz") as tar: file_info = tarfile.TarInfo(name="120000_300/") content = b"evil" file_info.size = len(content) tar.addfile(file_info, io.BytesIO(content)) manifest_json = json.dumps(manifest).encode() manifest_info = tarfile.TarInfo(name="manifest.json") manifest_info.size = len(manifest_json) tar.addfile(manifest_info, io.BytesIO(manifest_json)) journal_path = tmp_path / "journal" journal_path.mkdir() monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None with pytest.raises(ValueError, match="empty filename.*120000_300/"): import_archive(archive_path) assert not (journal_path / "chronicle" / "20250101" / "120000_300").exists() def test_create_archive_import_archive_round_trip_deconflicts( self, tmp_path, monkeypatch ): """AC#6: a real exported archive imports and deconflicts collisions.""" from solstone.observe.transfer import create_archive, import_archive source_journal = tmp_path / "source-journal" source_day = source_journal / "chronicle" / "20250101" first_source = source_day / "120000_300" / "audio.flac" second_source = source_day / "130000_300" / "screen.jsonl" first_content = b"source collision content" second_content = b'{"frame": "source noncollision"}\n' first_source.parent.mkdir(parents=True) second_source.parent.mkdir(parents=True) first_source.write_bytes(first_content) second_source.write_bytes(second_content) monkeypatch.setenv("SOLSTONE_JOURNAL", str(source_journal)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None archive_path = tmp_path / "roundtrip.tgz" create_archive("20250101", archive_path) default_stream = think_utils.DEFAULT_STREAM target_journal = tmp_path / "target-journal" colliding_path = ( target_journal / "chronicle" / "20250101" / default_stream / "120000_300" / "audio.flac" ) existing_content = b"target existing content" colliding_path.parent.mkdir(parents=True) colliding_path.write_bytes(existing_content) monkeypatch.setenv("SOLSTONE_JOURNAL", str(target_journal)) think_utils._journal_path_cache = None result = import_archive(archive_path) target_day = target_journal / "chronicle" / "20250101" / default_stream non_colliding_path = target_day / "130000_300" / "screen.jsonl" deconflicted_paths = [ path for path in target_day.iterdir() if path.is_dir() and path.name != "120000_300" and (path / "audio.flac").exists() and (path / "audio.flac").read_bytes() == first_content ] assert result["status"] == "imported" assert non_colliding_path.read_bytes() == second_content assert colliding_path.read_bytes() == existing_content assert deconflicted_paths class TestManifestValidation: """Tests for manifest reading and validation.""" def test_read_manifest_missing(self, tmp_path): """Test error when manifest is missing from archive.""" from solstone.observe.transfer import _read_manifest # Create archive without manifest archive_path = tmp_path / "test.tgz" with tarfile.open(archive_path, "w:gz") as tar: import io info = tarfile.TarInfo(name="some_file.txt") info.size = 4 tar.addfile(info, io.BytesIO(b"test")) with pytest.raises(ValueError, match="manifest.json not found"): _read_manifest(archive_path) def test_read_manifest_wrong_version(self, tmp_path): """Test error when manifest has wrong version.""" from solstone.observe.transfer import _read_manifest archive_path = tmp_path / "test.tgz" with tarfile.open(archive_path, "w:gz") as tar: import io manifest = json.dumps({"version": 999, "day": "20250101", "segments": {}}) info = tarfile.TarInfo(name="manifest.json") info.size = len(manifest) tar.addfile(info, io.BytesIO(manifest.encode())) with pytest.raises(ValueError, match="Unsupported manifest version"): _read_manifest(archive_path) def test_read_manifest_missing_fields(self, tmp_path): """Test error when manifest has missing required fields.""" from solstone.observe.transfer import _read_manifest archive_path = tmp_path / "test.tgz" with tarfile.open(archive_path, "w:gz") as tar: import io manifest = json.dumps({"version": 1}) # Missing day and segments info = tarfile.TarInfo(name="manifest.json") info.size = len(manifest) tar.addfile(info, io.BytesIO(manifest.encode())) with pytest.raises(ValueError, match="missing required fields"): _read_manifest(archive_path) class TestTransferSend: """Tests for transfer send functionality.""" def _setup_journal(self, tmp_path, *, include_stream_json: bool = False) -> Path: journal = tmp_path / "journal" day_dir = journal / "chronicle" / "20250103" / "default" / "120000_300" day_dir.mkdir(parents=True) (day_dir / "audio.flac").write_bytes(b"audio data") (day_dir / "transcript.jsonl").write_text('{"text": "hello"}\n') if include_stream_json: (day_dir / "stream.json").write_text('{"stream": "default"}\n') return journal def _set_journal_override(self, monkeypatch, journal: Path) -> None: monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) import solstone.think.utils as think_utils think_utils._journal_path_cache = None def test_parse_day_spec_single(self, tmp_path): from solstone.observe.transfer import _parse_day_spec journal_root = tmp_path / "journal" journal_root.mkdir() assert _parse_day_spec("20250103", journal_root) == ["20250103"] def test_parse_day_spec_range(self, tmp_path): from solstone.observe.transfer import _parse_day_spec journal_root = tmp_path / "journal" journal_root.mkdir() assert _parse_day_spec("20250101-20250103", journal_root) == [ "20250101", "20250102", "20250103", ] def test_parse_day_spec_all_days(self, tmp_path): from solstone.observe.transfer import _parse_day_spec journal_root = tmp_path / "journal" journal_root.mkdir() (journal_root / "chronicle" / "20250101").mkdir(parents=True) (journal_root / "chronicle" / "20250103").mkdir(parents=True) (journal_root / "config").mkdir() (journal_root / "streams").mkdir() assert _parse_day_spec(None, journal_root) == ["20250101", "20250103"] def test_parse_day_spec_invalid(self, tmp_path): from solstone.observe.transfer import _parse_day_spec journal_root = tmp_path / "journal" journal_root.mkdir() with pytest.raises(ValueError, match="Invalid day format"): _parse_day_spec("invalid", journal_root) def test_normalize_url(self): from solstone.observe.transfer import _normalize_url assert _normalize_url("example.com") == "https://example.com" assert _normalize_url("example.com/") == "https://example.com" assert _normalize_url("https://example.com/") == "https://example.com" assert _normalize_url("http://example.com/api/") == "http://example.com/api"