Spaces:
Running
Running
File size: 4,143 Bytes
fbcb300 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | """Unit tests for migrate_sessions_to_folders (REQ-058–REQ-062)."""
import sys
import os
import pytest
import datetime
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src'))
from db import StateKV
from functions import migrate_sessions_to_folders, KV
def make_kv(tmp_path):
db_path = os.path.join(str(tmp_path), 'test.db')
return StateKV(db_path=db_path)
def seed_session(kv, session_id='sess_001', cwd='/home/user/proj', agent='kiro', obs_count=3):
"""Seed a legacy session with observations into the old schema."""
ts = datetime.datetime.utcnow().isoformat() + 'Z'
session = {
'id': session_id,
'project': cwd,
'cwd': cwd,
'agentId': agent,
'startedAt': ts,
'updatedAt': ts,
'status': 'completed',
}
kv.set(KV.sessions, session_id, session)
obs_ids = []
for i in range(obs_count):
obs_id = f'obs_{i}'
obs = {
'id': obs_id,
'sessionId': session_id,
'timestamp': ts,
'type': 'file_edit',
'title': f'Edit {i}',
'narrative': f'Edited file {i}',
'concepts': ['python'],
'files': [f'src/file_{i}.py'],
'importance': 5,
}
kv.set(KV.observations(session_id), obs_id, obs)
obs_ids.append(obs_id)
return obs_ids
class TestMigrateDryRun:
def test_dry_run_writes_nothing(self, tmp_path):
kv = make_kv(tmp_path)
seed_session(kv)
result = migrate_sessions_to_folders(kv, dry_run=True)
assert result['migrated_sessions'] > 0
assert result['migrated_observations'] > 0
# No folder obs should have been written
fp = 'home/user/proj'
obs = kv.list(KV.folder_obs(fp, 'kiro'))
assert len(obs) == 0
def test_dry_run_returns_counts(self, tmp_path):
kv = make_kv(tmp_path)
seed_session(kv, obs_count=5)
result = migrate_sessions_to_folders(kv, dry_run=True)
assert result['migrated_observations'] == 5
assert result['dry_run'] is True
class TestMigrateActual:
def test_migrates_observations(self, tmp_path):
kv = make_kv(tmp_path)
seed_session(kv, obs_count=3)
migrate_sessions_to_folders(kv, dry_run=False)
fp = 'home/user/proj'
obs = kv.list(KV.folder_obs(fp, 'kiro'))
assert len(obs) == 3
def test_skips_raw_observations(self, tmp_path):
kv = make_kv(tmp_path)
seed_session(kv, obs_count=2)
# Add a raw obs
kv.set(KV.observations('sess_001'), 'obs_0:raw', {
'id': 'obs_0:raw', 'sessionId': 'sess_001',
'timestamp': datetime.datetime.utcnow().isoformat() + 'Z',
})
result = migrate_sessions_to_folders(kv, dry_run=False)
# Raw obs should not be counted
assert result['migrated_observations'] == 2
def test_nondestructive(self, tmp_path):
kv = make_kv(tmp_path)
seed_session(kv)
migrate_sessions_to_folders(kv, dry_run=False)
# Old session data still there
session = kv.get(KV.sessions, 'sess_001')
assert session is not None
def test_unknown_fallback(self, tmp_path):
kv = make_kv(tmp_path)
# Session with no cwd or project
ts = datetime.datetime.utcnow().isoformat() + 'Z'
kv.set(KV.sessions, 'sess_no_path', {
'id': 'sess_no_path', 'startedAt': ts, 'updatedAt': ts, 'status': 'completed',
})
kv.set(KV.observations('sess_no_path'), 'obs_x', {
'id': 'obs_x', 'sessionId': 'sess_no_path',
'timestamp': ts, 'type': 'other', 'title': 'x', 'narrative': 'x',
})
result = migrate_sessions_to_folders(kv, dry_run=False)
# Should succeed with 'unknown' fallback
assert result['migrated_sessions'] >= 1
def test_returns_error_list(self, tmp_path):
kv = make_kv(tmp_path)
seed_session(kv)
result = migrate_sessions_to_folders(kv, dry_run=False)
assert 'errors' in result
assert isinstance(result['errors'], list)
|