Something went wrong. Try again.
personal memory agent
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133# SPDX-License-Identifier: AGPL-3.0-only# Copyright (c) 2026 sol pbc
"""Tests for observe/transfer.py - day archive export, import, and send."""
import inspectimport jsonimport tarfilefrom pathlib import Pathfrom 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"