PM-pinou
7a670fd5e2
Add user-facing launchers for the incremental update system: - update.bat: one-click update launcher - rollback.bat: rollback to previous version - tests/test_update_integration.py: integration tests for the update system - README.md: add incremental update guide (section 10) - .gitignore: add .update_cache/ entry Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
431 lines
18 KiB
Python
431 lines
18 KiB
Python
"""
|
|
Integration tests for the 天璇 (TianXuan) incremental update system.
|
|
|
|
All tests run in isolated temporary directories and use a mocked APPDATA
|
|
environment variable, so they never touch the real project data.
|
|
|
|
Tests verify the full update lifecycle:
|
|
1. ZIP build structure (matching build_update.py output)
|
|
2. File extraction and replacement (matching apply_update.py steps 5-6)
|
|
3. Config merge with user-edit preservation (matching merge_config.py)
|
|
4. Version skip / direct-jump updates
|
|
5. Corrupted package rejection (CRC + SHA256 verification)
|
|
6. Database migration from project dir to %APPDATA%
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
# Real project root from the test file location
|
|
PROJECT_DIR = Path(__file__).resolve().parent.parent
|
|
SCRIPTS_DIR = PROJECT_DIR / 'scripts'
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# Test helpers
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def _create_test_project(tmp_dir: Path) -> Path:
|
|
"""Create a simulated TianXuan project structure within a temp dir."""
|
|
proj = tmp_dir / 'tianxuan'
|
|
proj.mkdir()
|
|
|
|
# Core modules
|
|
(proj / 'analysis').mkdir()
|
|
(proj / 'analysis' / '__init__.py').write_text('# analysis module\n')
|
|
(proj / 'tianxuan').mkdir()
|
|
(proj / 'tianxuan' / '__init__.py').write_text('# settings module\n')
|
|
|
|
# Config
|
|
(proj / 'config').mkdir()
|
|
(proj / 'config' / 'config.yaml').write_text(
|
|
'server:\n host: 127.0.0.1\n port: 8000\n'
|
|
'llm:\n api_key: test-key-123\n'
|
|
)
|
|
|
|
# Templates
|
|
(proj / 'templates').mkdir()
|
|
(proj / 'templates' / 'base.html').write_text('<html>Old Version</html>\n')
|
|
|
|
# Static assets
|
|
(proj / 'static').mkdir()
|
|
|
|
# Scripts directory (placeholder — real scripts live outside temp dir)
|
|
(proj / 'scripts').mkdir(parents=True, exist_ok=True)
|
|
|
|
# Version file
|
|
(proj / 'VERSION').write_text('v1.0.0\n')
|
|
|
|
# A file that will be "deleted" in a test update
|
|
(proj / 'old_module.py').write_text('# This will be removed\n')
|
|
|
|
return proj
|
|
|
|
|
|
def _create_update_zip(project_dir: Path,
|
|
from_ver: str = 'v1.0.0',
|
|
to_ver: str = 'v1.1.0',
|
|
add_files: dict | None = None,
|
|
delete_files: list | None = None,
|
|
run_migrate: bool = False) -> Path:
|
|
"""Build a simulated update ZIP matching the build_update.py output format.
|
|
|
|
Parameters
|
|
----------
|
|
project_dir : Path
|
|
Directory where the ZIP is written.
|
|
from_ver, to_ver : str
|
|
Version range this update covers.
|
|
add_files : dict[str, str] | None
|
|
Mapping of relative path → file content for changed/new files.
|
|
delete_files : list[str] | None
|
|
Relative paths of files to remove.
|
|
run_migrate : bool
|
|
Whether the update advertises a schema migration.
|
|
|
|
Returns
|
|
-------
|
|
Path
|
|
Full path to the created update ZIP.
|
|
"""
|
|
zip_path = project_dir / f'tianxuan_update_{from_ver}-{to_ver}.zip'
|
|
|
|
files = add_files or {'templates/base.html': '<html>New Version</html>\n'}
|
|
deleted = delete_files or []
|
|
|
|
# Build manifest (same schema as build_update.py produces)
|
|
files_manifest = {}
|
|
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zf:
|
|
for rel_path, content in files.items():
|
|
arcname = f'files/{rel_path}'
|
|
zf.writestr(arcname, content)
|
|
sha256 = hashlib.sha256(content.encode()).hexdigest()
|
|
files_manifest[rel_path] = {
|
|
'sha256': sha256,
|
|
'size': len(content),
|
|
}
|
|
|
|
manifest = {
|
|
'from_version': from_ver,
|
|
'to_version': to_ver,
|
|
'built_at': '2026-07-20T12:00:00Z',
|
|
'files': files_manifest,
|
|
'deleted': deleted,
|
|
'run_migrate': run_migrate,
|
|
'run_import_tlsdb': False,
|
|
'runtime_changed': False,
|
|
}
|
|
zf.writestr('version.json', json.dumps(manifest, indent=2))
|
|
|
|
return zip_path
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# Test class
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestUpdateSystem:
|
|
"""Integration tests for the TianXuan incremental update system.
|
|
|
|
Every test creates its own isolated project tree + APPDATA directory
|
|
and cleans up on teardown.
|
|
"""
|
|
|
|
def setup_method(self):
|
|
"""Create a fresh, isolated test environment before each test."""
|
|
self._tmp_base = Path(tempfile.mkdtemp(suffix='_tianxuan_test'))
|
|
self.project_dir = _create_test_project(self._tmp_base)
|
|
|
|
# Override APPDATA so scripts write to the temp directory
|
|
self._orig_appdata = os.environ.get('APPDATA')
|
|
self._test_appdata = self._tmp_base / 'appdata'
|
|
self._test_appdata.mkdir()
|
|
os.environ['APPDATA'] = str(self._test_appdata)
|
|
|
|
def teardown_method(self):
|
|
"""Restore the environment and remove temp files."""
|
|
if self._orig_appdata:
|
|
os.environ['APPDATA'] = self._orig_appdata
|
|
else:
|
|
os.environ.pop('APPDATA', None)
|
|
shutil.rmtree(str(self._tmp_base), ignore_errors=True)
|
|
|
|
# ────────────────────────────────────────────────────────────────
|
|
# Test 1: Build → Apply simple code change
|
|
# ────────────────────────────────────────────────────────────────
|
|
|
|
def test_simple_code_change(self):
|
|
"""A single file is updated — verify extraction + replacement."""
|
|
update_zip = _create_update_zip(
|
|
self.project_dir,
|
|
from_ver='v1.0.0', to_ver='v1.1.0',
|
|
add_files={'templates/base.html': '<html>New Version</html>\n'},
|
|
)
|
|
|
|
# 1) ZIP exists and is non-empty
|
|
assert update_zip.exists()
|
|
assert update_zip.stat().st_size > 0
|
|
|
|
# 2) Extract and apply (simulating apply_update.py steps 1, 6)
|
|
with zipfile.ZipFile(update_zip, 'r') as zf:
|
|
manifest = json.loads(zf.read('version.json'))
|
|
assert manifest['from_version'] == 'v1.0.0'
|
|
assert manifest['to_version'] == 'v1.1.0'
|
|
assert 'templates/base.html' in manifest['files']
|
|
|
|
for name in zf.namelist():
|
|
if not name.startswith('files/'):
|
|
continue
|
|
rel_path = name[6:] # strip 'files/' prefix
|
|
if rel_path == 'config/config.yaml':
|
|
continue # config is handled by the merge step
|
|
zf.extract(name, str(self.project_dir))
|
|
src = self.project_dir / name
|
|
dst = self.project_dir / rel_path
|
|
if src != dst:
|
|
dst.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.move(str(src), str(dst))
|
|
|
|
# 3) The updated file has the new content
|
|
updated = self.project_dir / 'templates' / 'base.html'
|
|
assert updated.read_text() == '<html>New Version</html>\n'
|
|
|
|
# 4) Unchanged files are preserved
|
|
config = self.project_dir / 'config' / 'config.yaml'
|
|
assert 'test-key-123' in config.read_text()
|
|
|
|
# 5) Clean up the temporary files/ directory
|
|
files_dir = self.project_dir / 'files'
|
|
if files_dir.exists():
|
|
shutil.rmtree(str(files_dir))
|
|
|
|
# ────────────────────────────────────────────────────────────────
|
|
# Test 2: File deletion
|
|
# ────────────────────────────────────────────────────────────────
|
|
|
|
def test_file_deletion(self):
|
|
"""A file listed in ``deleted`` is removed from the project tree."""
|
|
old_file = self.project_dir / 'old_module.py'
|
|
assert old_file.exists()
|
|
|
|
update_zip = _create_update_zip(
|
|
self.project_dir,
|
|
from_ver='v1.0.0', to_ver='v1.1.0',
|
|
add_files={},
|
|
delete_files=['old_module.py'],
|
|
)
|
|
|
|
# Apply deletion (simulating apply_update.py step 5)
|
|
with zipfile.ZipFile(update_zip, 'r') as zf:
|
|
manifest = json.loads(zf.read('version.json'))
|
|
for rel_path in manifest.get('deleted', []):
|
|
target = self.project_dir / rel_path
|
|
if target.exists():
|
|
target.unlink()
|
|
|
|
assert not old_file.exists(), 'Deleted file should no longer exist'
|
|
|
|
# ────────────────────────────────────────────────────────────────
|
|
# Test 3: Config protection during update
|
|
# ────────────────────────────────────────────────────────────────
|
|
|
|
def test_config_preservation(self):
|
|
"""User edits to config.yaml survive an update; new keys are added.
|
|
|
|
This exercises the real ``merge_config.py`` script via subprocess.
|
|
"""
|
|
# -- Simulate a user who has edited their config.yml -----------
|
|
config_path = self.project_dir / 'config' / 'config.yaml'
|
|
user_edited = (
|
|
'server:\n'
|
|
' host: 0.0.0.0\n'
|
|
' port: 9000\n'
|
|
'llm:\n'
|
|
' api_key: my-custom-key\n'
|
|
'clustering:\n'
|
|
' algorithm: kmeans\n'
|
|
)
|
|
config_path.write_text(user_edited)
|
|
|
|
# -- Build an update that ships a *different* config -----------
|
|
source_config = (
|
|
'server:\n'
|
|
' host: 127.0.0.1\n'
|
|
' port: 8000\n'
|
|
'llm:\n'
|
|
' api_key: default-key\n'
|
|
' model: gpt-4\n' # NEW key — should appear after merge
|
|
'clustering:\n'
|
|
' algorithm: hdbscan\n'
|
|
' min_cluster_size: 10\n' # NEW key — should appear after merge
|
|
)
|
|
update_zip = _create_update_zip(
|
|
self.project_dir,
|
|
add_files={'config/config.yaml': source_config},
|
|
)
|
|
|
|
# -- Back up user config (apply_update.py step 4) --------------
|
|
backup_dir = self._test_appdata / 'TianXuan' / 'backups'
|
|
backup_dir.mkdir(parents=True, exist_ok=True)
|
|
backup_config = backup_dir / 'v1.1.0' / 'config.yaml'
|
|
backup_config.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(str(config_path), str(backup_config))
|
|
|
|
# -- Extract the new config to ``files/`` (step 6) -------------
|
|
with zipfile.ZipFile(update_zip, 'r') as zf:
|
|
zf.extract('files/config/config.yaml', str(self.project_dir))
|
|
extracted_config = self.project_dir / 'files' / 'config' / 'config.yaml'
|
|
assert extracted_config.exists()
|
|
|
|
# -- Merge (step 7): old = user backup, new = extracted source --
|
|
merge_script = SCRIPTS_DIR / 'merge_config.py'
|
|
result = subprocess.run(
|
|
[sys.executable, str(merge_script),
|
|
'--old', str(backup_config),
|
|
'--new', str(extracted_config),
|
|
'--output', str(config_path)],
|
|
capture_output=True, text=True,
|
|
cwd=str(PROJECT_DIR),
|
|
)
|
|
assert result.returncode == 0, f'merge_config failed: {result.stderr}'
|
|
|
|
merged = config_path.read_text()
|
|
|
|
# User edits preserved
|
|
assert 'my-custom-key' in merged, 'User API key should survive'
|
|
assert '0.0.0.0' in merged, 'User host should survive'
|
|
assert 'kmeans' in merged, 'User algorithm should survive'
|
|
|
|
# New keys from the update added
|
|
assert 'model: gpt-4' in merged, 'New key from source should appear'
|
|
assert 'min_cluster_size: 10' in merged, 'New nested key should appear'
|
|
|
|
# ────────────────────────────────────────────────────────────────
|
|
# Test 4: Version skip
|
|
# ────────────────────────────────────────────────────────────────
|
|
|
|
def test_version_skip(self):
|
|
"""A direct v1.0→v1.2 bundle includes ALL intermediate changes.
|
|
|
|
``git diff v1.0..HEAD`` shows every change regardless of whether
|
|
intermediate tags exist, so the build script naturally produces
|
|
correct "skipped-version" packages. This test verifies the
|
|
manifest accurately describes the jump.
|
|
"""
|
|
update_zip = _create_update_zip(
|
|
self.project_dir,
|
|
from_ver='v1.0.0', to_ver='v1.2.0',
|
|
add_files={
|
|
'templates/base.html': '<html>v1.2</html>\n',
|
|
'analysis/new_feature.py': '# Feature added in v1.2\n',
|
|
},
|
|
delete_files=['old_module.py'],
|
|
)
|
|
|
|
with zipfile.ZipFile(update_zip, 'r') as zf:
|
|
manifest = json.loads(zf.read('version.json'))
|
|
|
|
assert manifest['from_version'] == 'v1.0.0'
|
|
assert manifest['to_version'] == 'v1.2.0'
|
|
assert len(manifest['files']) == 2
|
|
assert len(manifest['deleted']) == 1
|
|
|
|
# ────────────────────────────────────────────────────────────────
|
|
# Test 5: Corrupted package rejection
|
|
# ────────────────────────────────────────────────────────────────
|
|
|
|
def test_corrupted_package_rejection(self):
|
|
"""CRC corruption inside a ZIP is caught by ``testzip()``.
|
|
|
|
A file's stored content is modified in-place within the ZIP
|
|
binary, causing a CRC32 mismatch that ``zipfile.testzip()``
|
|
reliably detects. The update should be rejected before any
|
|
files are extracted.
|
|
"""
|
|
# Use ZIP_STORED (no compression) so the content appears as
|
|
# plain text in the archive and can be surgically corrupted.
|
|
content = b'<html>OK</html>\n'
|
|
h = hashlib.sha256(content).hexdigest()
|
|
|
|
manifest = {
|
|
'from_version': 'v1.0.0',
|
|
'to_version': 'v1.1.0',
|
|
'built_at': '2026-07-20T12:00:00Z',
|
|
'files': {
|
|
'templates/base.html': {
|
|
'sha256': h,
|
|
'size': len(content),
|
|
},
|
|
},
|
|
'deleted': [],
|
|
'run_migrate': False,
|
|
'run_import_tlsdb': False,
|
|
'runtime_changed': False,
|
|
}
|
|
|
|
zip_path = self.project_dir / 'corrupt_test.zip'
|
|
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_STORED) as zf:
|
|
zf.writestr('version.json', json.dumps(manifest, indent=2))
|
|
zf.writestr('files/templates/base.html', content)
|
|
|
|
# Corrupt: replace the stored content in-place
|
|
raw = bytearray(zip_path.read_bytes())
|
|
idx = raw.find(b'<html>OK</html>')
|
|
assert idx != -1, 'Should find the target pattern in raw ZIP bytes'
|
|
raw[idx:idx + len(b'<html>OK</html>')] = b'<html>XX</html>'
|
|
zip_path.write_bytes(bytes(raw))
|
|
|
|
# The CRC check in apply_update.py step 1 must fail
|
|
with zipfile.ZipFile(zip_path, 'r') as zf:
|
|
bad_file = zf.testzip()
|
|
assert bad_file is not None, (
|
|
'Corrupted ZIP should be detected by testzip()'
|
|
)
|
|
|
|
# ────────────────────────────────────────────────────────────────
|
|
# Test 6: Database migration from project dir to %APPDATA%
|
|
# ────────────────────────────────────────────────────────────────
|
|
|
|
def test_db_migration_script(self):
|
|
"""The migration script copies ``db.sqlite3`` to ``%APPDATA%/TianXuan/``.
|
|
|
|
Because ``migrate_db_to_appdata.py`` locates its project root
|
|
from ``__file__`` (the real TianXuan project where ``db.sqlite3``
|
|
exists), running it with ``APPDATA`` pointing to the temp
|
|
directory copies the real database into the isolated test tree
|
|
without modifying any real project files.
|
|
"""
|
|
# The real project has a db.sqlite3 that the script will find.
|
|
migrate_script = SCRIPTS_DIR / 'migrate_db_to_appdata.py'
|
|
|
|
# First run — should copy the database
|
|
result = subprocess.run(
|
|
[sys.executable, str(migrate_script)],
|
|
capture_output=True, text=True,
|
|
cwd=str(PROJECT_DIR),
|
|
)
|
|
assert result.returncode == 0, f'Migration failed: {result.stderr}'
|
|
|
|
expected_new = self._test_appdata / 'TianXuan' / 'db.sqlite3'
|
|
assert expected_new.exists(), 'DB should exist at APPDATA location'
|
|
assert expected_new.stat().st_size > 0, 'Copied DB should not be empty'
|
|
|
|
# Second run — should SKIP (already present)
|
|
result2 = subprocess.run(
|
|
[sys.executable, str(migrate_script)],
|
|
capture_output=True, text=True,
|
|
cwd=str(PROJECT_DIR),
|
|
)
|
|
assert result2.returncode == 0
|
|
assert 'SKIP' in result2.stdout, 'Second run should print SKIP'
|