#!/usr/bin/env python3 """Original offline Dropbox SDK witness, with synthetic metadata only. Install dropbox==12.2.3, stone==3.5.5 and requests==2.34.2 in a local venv. No Dropbox account, payload, credential, permission or ERP write is tested. The proposed staged projection is an in-memory example, not a durable database. """ from __future__ import annotations import copy import hashlib import importlib.metadata import json import platform import socket from dataclasses import dataclass CONNECTION_ATTEMPTS = [] REAL_HTTP_ATTEMPTS = [] CHECKS = [] TRACE = [] def deny_connect(*args, **kwargs): CONNECTION_ATTEMPTS.append("blocked") raise AssertionError("Network connection forbidden in this offline witness") socket.socket.connect = deny_connect socket.socket.connect_ex = deny_connect socket.create_connection = deny_connect import requests import dropbox from dropbox.exceptions import ApiError from dropbox.files import FileMetadata def deny_http(*args, **kwargs): REAL_HTTP_ATTEMPTS.append("blocked") raise AssertionError("Real requests transport forbidden") requests.sessions.Session.request = deny_http def check(label, condition): if not condition: raise AssertionError(label) CHECKS.append(label) def fingerprint(value): return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest() @dataclass(frozen=True) class Scope: account: str = "compte-fictif-ADV" namespace: str = "10001" path: str = "/affaire-208" recursive: bool = True include_deleted: bool = True SCOPE = Scope() class PreparedSession(requests.Session): """Consume prepared replies through the SDK's actual request/decoder path.""" def __init__(self, case, replies): super().__init__() self.case = case self.replies = list(replies) def post(self, url, headers=None, data=None, **kwargs): if not self.replies: raise AssertionError("Unexpected prepared-transport invocation") route, response = self.replies.pop(0) assert url == "https://api.dropboxapi.com/2/" + route args = json.loads(data) TRACE.append({"case": self.case, "route": route, "args": args, "path_root": json.loads(headers["Dropbox-API-Path-Root"]), "prepared_status": response[0] if isinstance(response, tuple) else "timeout"}) if isinstance(response, Exception): raise response status, payload = response reply = requests.Response() reply.status_code = status reply.headers = {"content-type": "application/json", "x-dropbox-request-id": "requete-fictive"} reply._content = json.dumps(payload).encode() reply.url = url return reply def client(session): return dropbox.Dropbox( oauth2_access_token="jeton-fictif-inutilisable", session=session, max_retries_on_error=0, max_retries_on_rate_limit=0, headers={"Dropbox-API-Path-Root": json.dumps({".tag": "namespace_id", "namespace_id": SCOPE.namespace})}, ) def file_entry(letter): return {".tag": "file", "name": "plan-" + letter + ".pdf", "path_lower": SCOPE.path + "/plan-" + letter + ".pdf", "path_display": SCOPE.path + "/plan-" + letter + ".pdf", "id": "id:piece-fictive-" + letter, "client_modified": "2026-10-01T10:00:00Z", "server_modified": "2026-10-01T10:00:00Z", "rev": "00000000" + ("1" if letter == "a" else "2"), "size": 20} def page(entries, cursor, has_more): return 200, {"entries": entries, "cursor": cursor, "has_more": has_more} def error(tag, **values): return 409, {"error_summary": tag + "/fictif", "error": {".tag": tag, **values}} def staged_collection(dbx, expected_scope, requested_scope): """Return a complete candidate; never mutate the active projection or ERP.""" if requested_scope != expected_scope: raise ValueError("Collection scope changed: review required") candidate = {} result = dbx.files_list_folder(path=expected_scope.path, recursive=expected_scope.recursive, include_deleted=expected_scope.include_deleted) while True: for entry in result.entries: # Narrow witness: project only file IDs/revisions and explicit deletions. # Folder/implicit-parent/read_only propagation is deliberately NOT implemented. if isinstance(entry, FileMetadata): candidate[entry.path_lower] = {"id": entry.id, "rev": entry.rev} elif isinstance(entry, dropbox.files.DeletedMetadata): candidate.pop(entry.path_lower, None) else: raise ValueError("This narrow file-only witness does not project folders") if not result.has_more: return {"files": candidate, "cursor": result.cursor, "scope": expected_scope.__dict__} result = dbx.files_list_folder_continue(result.cursor) def run(): for package, version in [("dropbox", "12.2.3"), ("stone", "3.5.5"), ("requests", "2.34.2")]: check(package + " pinned version", importlib.metadata.version(package) == version) old = {"files": {SCOPE.path + "/plan-a.pdf": {"id": "id:piece-fictive-a", "rev": "ancienne-a"}, SCOPE.path + "/plan-b.pdf": {"id": "id:piece-fictive-b", "rev": "ancienne-b"}}, "cursor": "curseur-fictif-ancien", "scope": SCOPE.__dict__} erp_references = {"affaire": "AFF-208", "pieces": ["id:piece-fictive-a", "id:piece-fictive-b"]} old_hash, erp_hash = fingerprint(old), fingerprint(erp_references) session = PreparedSession("reset_then_rebuild", [ ("files/list_folder/continue", error("reset")), ("files/list_folder", page([file_entry("a")], "curseur-fictif-page-1", True)), ("files/list_folder/continue", page([file_entry("b")], "curseur-fictif-page-2", False)), ("files/list_folder/continue", page([], "curseur-fictif-vide-3", False)), ]) dbx = client(session) try: dbx.files_list_folder_continue(old["cursor"]) raise AssertionError("Expected supplied reset error") except ApiError as exc: check("SDK decodes reset union", exc.error.is_reset()) check("reset is not path", not exc.error.is_path()) check("SDK does not silently enumerate", len(TRACE) == 1) check("reset leaves active projection intact", fingerprint(old) == old_hash) candidate = staged_collection(dbx, SCOPE, SCOPE) first = next(t for t in TRACE if t["route"] == "files/list_folder") check("fresh enumeration uses expected folder", first["args"]["path"] == SCOPE.path) check("recursive selection is explicit", first["args"]["recursive"] is True) check("include_deleted selection is explicit", first["args"]["include_deleted"] is True) check("namespace header is constructed", first["path_root"] == {".tag": "namespace_id", "namespace_id": "10001"}) check("next request uses returned first cursor", TRACE[2]["args"]["cursor"] == "curseur-fictif-page-1") check("second file is retained after complete scan", len(candidate["files"]) == 2) check("terminal cursor belongs to complete candidate", candidate["cursor"] == "curseur-fictif-page-2") check("candidate building never changes active state", fingerprint(old) == old_hash) check("candidate building never changes ERP references", fingerprint(erp_references) == erp_hash) active = copy.deepcopy(candidate) # Proposed activation, in memory only. check("activation retains both projected file IDs", sorted(x["id"] for x in active["files"].values()) == sorted(erp_references["pieces"])) empty = dbx.files_list_folder_continue(active["cursor"]) check("SDK decodes empty successful entries", empty.entries == []) check("empty reply is terminal at this observation", empty.has_more is False) active["cursor"] = empty.cursor check("latest cursor is saved despite empty page", active["cursor"] == "curseur-fictif-vide-3") check("empty reply does not erase projected files", len(active["files"]) == 2) check("all reset-rebuild prepared replies consumed", not session.replies) interrupted = PreparedSession("interrupted_rebuild", [ ("files/list_folder", page([file_entry("a")], "curseur-fictif-partiel", True)), ("files/list_folder/continue", requests.exceptions.ReadTimeout("timeout prepare")), ]) active_hash = fingerprint(active) try: staged_collection(client(interrupted), SCOPE, SCOPE) raise AssertionError("Expected prepared timeout") except requests.exceptions.ReadTimeout: check("interrupted scan keeps previous active projection", fingerprint(active) == active_hash) check("interrupted scan keeps previous active cursor", active["cursor"] == "curseur-fictif-vide-3") check("interrupted scan leaves ERP references intact", fingerprint(erp_references) == erp_hash) check("interrupted continuation used returned cursor", TRACE[-1]["args"]["cursor"] == "curseur-fictif-partiel") check("zero automatic retry requested", not interrupted.replies) path_session = PreparedSession("path_not_found", [ ("files/list_folder/continue", error("path", path={".tag": "not_found"}))]) try: client(path_session).files_list_folder_continue("curseur-fictif-chemin") raise AssertionError("Expected supplied path error") except ApiError as exc: check("SDK decodes path union", exc.error.is_path()) check("nested lookup error is not_found", exc.error.get_path().is_not_found()) check("path not_found is not reset", not exc.error.is_reset()) check("path error performs no fresh enumeration", len([t for t in TRACE if t["case"] == "path_not_found"]) == 1) check("path error does not mutate active state", fingerprint(active) == active_hash) other_session = PreparedSession("unrecognized_error", [ ("files/list_folder/continue", error("future_tag_fictif"))]) try: client(other_session).files_list_folder_continue("curseur-fictif-autre") raise AssertionError("Expected supplied unknown union") except ApiError as exc: check("unknown union is decoded as other", exc.error.is_other()) check("unknown union is not silently reset", not exc.error.is_reset()) check("unknown error performs no enumeration", len([t for t in TRACE if t["case"] == "unrecognized_error"]) == 1) never_sent = PreparedSession("changed_scope", []) changed = Scope(namespace="20002") try: staged_collection(client(never_sent), SCOPE, changed) raise AssertionError("Expected local scope guard") except ValueError: check("changed scope causes zero prepared requests", not [t for t in TRACE if t["case"] == "changed_scope"]) check("scope guard keeps current active state", fingerprint(active) == active_hash) folders = PreparedSession("unsupported_folder_projection", [ ("files/list_folder", page([{".tag": "folder", "id": "id:dossier-fictif", "name": "etudes", "path_lower": SCOPE.path + "/etudes", "path_display": SCOPE.path + "/etudes"}], "curseur-fictif-dossier", False))]) try: staged_collection(client(folders), SCOPE, SCOPE) raise AssertionError("The file-only witness must refuse folder projection") except ValueError as exc: check("folder projection is explicitly refused", "does not project folders" in str(exc)) check("folder refusal keeps active projection intact", fingerprint(active) == active_hash) check("folder witness actually consumed the SDK reply", not folders.replies) # Counterexample: replacing the live cache after only page one drops B. unsafe_page_one = {SCOPE.path + "/plan-a.pdf": {"id": "id:piece-fictive-a", "rev": "000000001"}} absent_on_incomplete_page = set(old["files"]) - set(unsafe_page_one) check("partial-page replacement wrongly omits B", absent_on_incomplete_page == {SCOPE.path + "/plan-b.pdf"}) check("completed candidate still includes B", SCOPE.path + "/plan-b.pdf" in candidate["files"]) check("ERP links survive every prepared case", fingerprint(erp_references) == erp_hash) check("no real requests transport invocation", REAL_HTTP_ATTEMPTS == []) check("no network connection attempt", CONNECTION_ATTEMPTS == []) return {"checks_passed": len(CHECKS), "checks": CHECKS, "prepared_transport_invocations": len(TRACE), "trace": TRACE, "real_http_requests": len(REAL_HTTP_ATTEMPTS), "connection_attempts": len(CONNECTION_ATTEMPTS), "python": platform.python_version(), "packages": {p: importlib.metadata.version(p) for p in ["dropbox", "stone", "requests"]}, "active_projection": active, "erp_references_unchanged": erp_references, "scope": "Synthetic metadata and prepared replies. File-only, in-memory proposed projection; no folder/read_only projection, durable transaction, native snapshot, permissions, payload or ERP write certified."} if __name__ == "__main__": print(json.dumps(run(), ensure_ascii=False, indent=2))