""" 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('Old Version\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': 'New Version\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': 'New Version\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() == 'New Version\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': 'v1.2\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'OK\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'OK') assert idx != -1, 'Should find the target pattern in raw ZIP bytes' raw[idx:idx + len(b'OK')] = b'XX' 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'