# TechGuide reproducible lab. Uses only disposable files in a temporary directory.
import os
import tempfile
from pathlib import Path
from typing import Generator
from contextlib import contextmanager

@contextmanager
def atomic_write(filepath: str | Path, mode: str = "w", encoding: str = "utf-8") -> Generator[tempfile._TemporaryFileWrapper, None, None]:
    """
    Context manager providing atomic, crash-resilient file writes.
    
    Ensures that file modifications are fully written and synced to disk
    before atomically replacing the destination path.
    """
    dest_path = Path(filepath).resolve()
    parent_dir = dest_path.parent
    
    # Ensure destination directory exists
    parent_dir.mkdir(parents=True, exist_ok=True)
    
    # Prefix and suffix keep temp files recognizable during debugging
    prefix = f".tmp-{dest_path.name}-"
    
    # Binary vs text mode setup
    is_binary = "b" in mode
    open_kwargs = {} if is_binary else {"encoding": encoding}
    
    # 1. Create temporary file on the exact same filesystem directory
    temp_file = tempfile.NamedTemporaryFile(
        mode=mode,
        dir=parent_dir,
        prefix=prefix,
        delete=False,
        **open_kwargs
    )
    
    temp_path = Path(temp_file.name)
    success = False
    
    try:
        yield temp_file
        
        # 2. Flush Python application runtime buffers
        temp_file.flush()
        
        # 3. Force operating system kernel page cache to physical disk
        os.fsync(temp_file.fileno())
        
        # Close descriptor prior to replacing (essential on Windows)
        temp_file.close()
        
        # 4. Atomically swap destination file with temporary file
        os.replace(temp_path, dest_path)
        
        # Optional: Sync parent directory on POSIX to persist directory entry
        if hasattr(os, "O_DIRECTORY"):
            try:
                dir_fd = os.open(str(parent_dir), os.O_RDONLY | os.O_DIRECTORY)
                try:
                    os.fsync(dir_fd)
                finally:
                    os.close(dir_fd)
            except OSError:
                # Directory fsync may be unsupported on certain filesystems/privileges
                pass
                
        success = True
        
    finally:
        # Guarantee descriptor closure
        if not temp_file.closed:
            temp_file.close()
            
        # Clean up stray temporary file if an exception interrupted the write
        if not success and temp_path.exists():
            try:
                temp_path.unlink()
            except OSError:
                pass

import json
import sqlite3
import platform
from contextlib import closing

def require(condition, message):
    if not condition:
        raise AssertionError(message)

def main():
    print(f"Python {platform.python_version()} | {platform.system()} | SQLite {sqlite3.sqlite_version}")
    with tempfile.TemporaryDirectory(prefix="techguide-lab-") as folder:
        root = Path(folder)
        target = root / "settings.json"
        with atomic_write(target) as stream:
            json.dump({"revision": 1}, stream)
        require(json.loads(target.read_text()) == {"revision": 1}, "replacement failed")
        print("PASS atomic replacement produces complete JSON")

        old = target.read_bytes()
        try:
            with atomic_write(target) as stream:
                stream.write('{"partial":')
                raise ValueError("simulated serialization failure")
        except ValueError:
            pass
        require(target.read_bytes() == old, "old contents lost")
        require(not list(root.glob('.tmp-*')), "temporary files leaked")
        print("PASS failed write preserves old file and removes temporary file")

        binary = root / "payload.bin"
        with atomic_write(binary, mode="wb") as stream:
            stream.write(bytes(range(256)))
        require(binary.read_bytes() == bytes(range(256)), "binary mismatch")
        print("PASS binary replacement preserves all bytes")

        database = root / "source.db"
        backup = root / "snapshot.db"
        with closing(sqlite3.connect(database)) as connection:
            connection.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT UNIQUE)")
            with connection:
                connection.execute("INSERT INTO users(name) VALUES (?)", ("alice",))
            payload = "alice' OR '1'='1"
            rows = connection.execute("SELECT name FROM users WHERE name = ?", (payload,)).fetchall()
            require(rows == [], "query treated data as SQL")
            print("PASS SQL metacharacters remain literal query data")

            try:
                with connection:
                    connection.execute("INSERT INTO users(name) VALUES (?)", ("bob",))
                    connection.execute("INSERT INTO users(name) VALUES (?)", ("alice",))
            except sqlite3.IntegrityError:
                pass
            require(connection.execute("SELECT name FROM users ORDER BY id").fetchall() == [("alice",)], "transaction did not roll back")
            print("PASS uniqueness failure rolls back the preceding insert")

            with closing(sqlite3.connect(backup)) as destination:
                connection.backup(destination, pages=1)

        with closing(sqlite3.connect(backup.as_uri() + "?mode=ro", uri=True)) as restored:
            require(restored.execute("PRAGMA integrity_check").fetchall() == [("ok",)], "integrity failure")
            require(restored.execute("PRAGMA foreign_key_check").fetchall() == [], "foreign key failure")
            require(restored.execute("SELECT name FROM users").fetchall() == [("alice",)], "restored data mismatch")
        print("PASS reopened backup passes integrity and application-data checks")
    print("6 checks passed; disposable fixtures removed")

if __name__ == "__main__":
    main()