From 2b975bd4e6380d73efc1eb33f8f27923fd5de982 Mon Sep 17 00:00:00 2001 From: Ivan Date: Sun, 3 May 2026 00:14:55 -0500 Subject: [PATCH] feat(legacy_migrator): implement automatic upstream folder migration and add related tests --- meshchatx/src/backend/legacy_migrator.py | 47 +++++++ tests/backend/test_legacy_migrator.py | 150 +++++++++++++++++++++++ 2 files changed, 197 insertions(+) diff --git a/meshchatx/src/backend/legacy_migrator.py b/meshchatx/src/backend/legacy_migrator.py index 77cdf6d..6247b5e 100644 --- a/meshchatx/src/backend/legacy_migrator.py +++ b/meshchatx/src/backend/legacy_migrator.py @@ -1,5 +1,6 @@ # SPDX-License-Identifier: 0BSD +import logging import os import shutil @@ -7,6 +8,8 @@ import RNS LEGACY_DIR = ".reticulum-meshchat" CURRENT_DIR = ".reticulum-meshchatx" +UPSTREAM_DIR = "reticulum-meshchat" +UPSTREAM_X_DIR = "reticulum-meshchatx" def _basename_norm(p: str) -> str: @@ -27,6 +30,17 @@ def paired_new_from_legacy(legacy_storage_path: str) -> str | None: return os.path.join(parent, CURRENT_DIR) +def _is_meshchatx_storage_basename(base: str) -> bool: + return base in (CURRENT_DIR, UPSTREAM_X_DIR) + + +def paired_upstream_plain_from_meshchatx(meshchatx_path: str) -> str | None: + if not _is_meshchatx_storage_basename(_basename_norm(meshchatx_path)): + return None + parent = os.path.dirname(os.path.normpath(meshchatx_path)) + return os.path.join(parent, UPSTREAM_DIR) + + def storage_has_meshchat_data(storage_dir: str) -> bool: if not storage_dir or not os.path.isdir(storage_dir): return False @@ -57,7 +71,40 @@ def resolve_startup_storage(request_dir: str) -> tuple[str, dict]: if skip: return planned, empty_ctx + skip_upstream_auto = os.environ.get( + "MESHCHAT_SKIP_UPSTREAM_FOLDER_MIGRATION", + "", + ).strip().lower() in ("1", "true", "yes") + base = _basename_norm(planned) + if ( + not skip_upstream_auto + and _is_meshchatx_storage_basename(base) + and not storage_has_meshchat_data(planned) + ): + upstream_plain = paired_upstream_plain_from_meshchatx(planned) + if upstream_plain and storage_has_meshchat_data(upstream_plain): + try: + migrate_legacy_to_target(upstream_plain, planned) + logging.getLogger(__name__).info( + "Auto-copied upstream storage %s -> %s", + upstream_plain, + planned, + ) + return planned, { + **empty_ctx, + "did_auto_upstream_folder_copy": True, + "upstream_copy_source": upstream_plain, + "upstream_copy_target": planned, + } + except OSError: + logging.getLogger(__name__).warning( + "Upstream folder auto-migration failed (%s -> %s)", + upstream_plain, + planned, + exc_info=True, + ) + if base == CURRENT_DIR and not storage_has_meshchat_data(planned): legacy = paired_legacy_from_new(planned) if legacy and storage_has_meshchat_data(legacy): diff --git a/tests/backend/test_legacy_migrator.py b/tests/backend/test_legacy_migrator.py index c43d376..3db40d2 100644 --- a/tests/backend/test_legacy_migrator.py +++ b/tests/backend/test_legacy_migrator.py @@ -6,6 +6,7 @@ import sqlite3 import stat import tempfile from pathlib import Path +from unittest.mock import patch import pytest import RNS @@ -15,12 +16,15 @@ from hypothesis import strategies as st from meshchatx.src.backend.legacy_migrator import ( CURRENT_DIR, LEGACY_DIR, + UPSTREAM_DIR, + UPSTREAM_X_DIR, assert_migration_context_paths, copy_legacy_storage_tree, fresh_storage_at_target, migrate_legacy_to_target, paired_legacy_from_new, paired_new_from_legacy, + paired_upstream_plain_from_meshchatx, resolve_startup_storage, storage_has_meshchat_data, ) @@ -453,3 +457,149 @@ def test_resolve_identities_only_no_root_identity(monkeypatch): assert ctx["show_choice"] is True finally: shutil.rmtree(root, ignore_errors=True) + + +def test_paired_upstream_plain_from_meshchatx_paths(): + p_dot = os.path.join("/tmp", "o", CURRENT_DIR) + assert paired_upstream_plain_from_meshchatx(p_dot) == os.path.join( + "/tmp", "o", UPSTREAM_DIR + ) + p_plain = os.path.join("/tmp", "o", UPSTREAM_X_DIR) + assert paired_upstream_plain_from_meshchatx(p_plain) == os.path.join( + "/tmp", "o", UPSTREAM_DIR + ) + assert paired_upstream_plain_from_meshchatx("/tmp/storage") is None + + +def test_auto_upstream_plain_folder_to_dot_meshchatx(monkeypatch): + root = tempfile.mkdtemp() + try: + new_home = os.path.join(root, CURRENT_DIR) + plain = os.path.join(root, UPSTREAM_DIR) + os.makedirs(plain, exist_ok=True) + with open(os.path.join(plain, "identity"), "wb") as f: + f.write(RNS.Identity(create_keys=True).get_private_key()) + os.makedirs(new_home, exist_ok=True) + monkeypatch.delenv("MESHCHAT_SKIP_LEGACY_MIGRATION_UI", raising=False) + monkeypatch.delenv("MESHCHAT_SKIP_UPSTREAM_FOLDER_MIGRATION", raising=False) + eff, ctx = resolve_startup_storage(new_home) + assert eff == os.path.abspath(new_home) + assert ctx.get("did_auto_upstream_folder_copy") is True + assert ctx.get("show_choice") is not True + assert storage_has_meshchat_data(new_home) + assert os.path.isfile(os.path.join(new_home, "identity")) + finally: + shutil.rmtree(root, ignore_errors=True) + + +def test_auto_upstream_plain_folder_to_plain_meshchatx(monkeypatch): + root = tempfile.mkdtemp() + try: + new_home = os.path.join(root, UPSTREAM_X_DIR) + plain = os.path.join(root, UPSTREAM_DIR) + os.makedirs(plain, exist_ok=True) + with open(os.path.join(plain, "identity"), "wb") as f: + f.write(RNS.Identity(create_keys=True).get_private_key()) + os.makedirs(new_home, exist_ok=True) + monkeypatch.delenv("MESHCHAT_SKIP_LEGACY_MIGRATION_UI", raising=False) + monkeypatch.delenv("MESHCHAT_SKIP_UPSTREAM_FOLDER_MIGRATION", raising=False) + eff, ctx = resolve_startup_storage(new_home) + assert eff == os.path.abspath(new_home) + assert ctx.get("did_auto_upstream_folder_copy") is True + assert storage_has_meshchat_data(new_home) + finally: + shutil.rmtree(root, ignore_errors=True) + + +def test_auto_upstream_skipped_by_env(monkeypatch): + root = tempfile.mkdtemp() + try: + new_home = os.path.join(root, CURRENT_DIR) + plain = os.path.join(root, UPSTREAM_DIR) + os.makedirs(plain, exist_ok=True) + with open(os.path.join(plain, "identity"), "wb") as f: + f.write(RNS.Identity(create_keys=True).get_private_key()) + os.makedirs(new_home, exist_ok=True) + leg = os.path.join(root, LEGACY_DIR) + os.makedirs(leg, exist_ok=True) + with open(os.path.join(leg, "identity"), "wb") as f: + f.write(RNS.Identity(create_keys=True).get_private_key()) + monkeypatch.delenv("MESHCHAT_SKIP_LEGACY_MIGRATION_UI", raising=False) + monkeypatch.setenv("MESHCHAT_SKIP_UPSTREAM_FOLDER_MIGRATION", "1") + eff, ctx = resolve_startup_storage(new_home) + assert eff == leg + assert ctx.get("did_auto_upstream_folder_copy") is not True + assert ctx["show_choice"] is True + finally: + shutil.rmtree(root, ignore_errors=True) + + +def test_auto_upstream_skipped_when_target_already_has_data(monkeypatch): + root = tempfile.mkdtemp() + try: + new_home = os.path.join(root, CURRENT_DIR) + plain = os.path.join(root, UPSTREAM_DIR) + os.makedirs(plain, exist_ok=True) + with open(os.path.join(plain, "identity"), "wb") as f: + f.write(RNS.Identity(create_keys=True).get_private_key()) + os.makedirs(new_home, exist_ok=True) + with open(os.path.join(new_home, "identity"), "wb") as f: + f.write(RNS.Identity(create_keys=True).get_private_key()) + monkeypatch.delenv("MESHCHAT_SKIP_LEGACY_MIGRATION_UI", raising=False) + monkeypatch.delenv("MESHCHAT_SKIP_UPSTREAM_FOLDER_MIGRATION", raising=False) + eff, ctx = resolve_startup_storage(new_home) + assert eff == os.path.abspath(new_home) + assert ctx.get("did_auto_upstream_folder_copy") is not True + assert ctx.get("show_choice") is not True + finally: + shutil.rmtree(root, ignore_errors=True) + + +def test_auto_upstream_prefers_plain_over_dot_legacy_redirect(monkeypatch): + root = tempfile.mkdtemp() + try: + new_home = os.path.join(root, CURRENT_DIR) + plain = os.path.join(root, UPSTREAM_DIR) + os.makedirs(plain, exist_ok=True) + with open(os.path.join(plain, "identity"), "wb") as f: + f.write(RNS.Identity(create_keys=True).get_private_key()) + leg = os.path.join(root, LEGACY_DIR) + os.makedirs(leg, exist_ok=True) + with open(os.path.join(leg, "identity"), "wb") as f: + f.write(RNS.Identity(create_keys=True).get_private_key()) + os.makedirs(new_home, exist_ok=True) + monkeypatch.delenv("MESHCHAT_SKIP_LEGACY_MIGRATION_UI", raising=False) + monkeypatch.delenv("MESHCHAT_SKIP_UPSTREAM_FOLDER_MIGRATION", raising=False) + eff, ctx = resolve_startup_storage(new_home) + assert eff == os.path.abspath(new_home) + assert ctx.get("did_auto_upstream_folder_copy") is True + assert ctx.get("show_choice") is not True + finally: + shutil.rmtree(root, ignore_errors=True) + + +def test_auto_upstream_oserror_falls_through_to_dot_legacy(monkeypatch): + root = tempfile.mkdtemp() + try: + new_home = os.path.join(root, CURRENT_DIR) + plain = os.path.join(root, UPSTREAM_DIR) + os.makedirs(plain, exist_ok=True) + with open(os.path.join(plain, "identity"), "wb") as f: + f.write(RNS.Identity(create_keys=True).get_private_key()) + leg = os.path.join(root, LEGACY_DIR) + os.makedirs(leg, exist_ok=True) + with open(os.path.join(leg, "identity"), "wb") as f: + f.write(RNS.Identity(create_keys=True).get_private_key()) + os.makedirs(new_home, exist_ok=True) + monkeypatch.delenv("MESHCHAT_SKIP_LEGACY_MIGRATION_UI", raising=False) + monkeypatch.delenv("MESHCHAT_SKIP_UPSTREAM_FOLDER_MIGRATION", raising=False) + with patch( + "meshchatx.src.backend.legacy_migrator.migrate_legacy_to_target", + side_effect=OSError("simulated copy failure"), + ): + eff, ctx = resolve_startup_storage(new_home) + assert eff == leg + assert ctx["show_choice"] is True + assert ctx.get("did_auto_upstream_folder_copy") is not True + finally: + shutil.rmtree(root, ignore_errors=True)