feat: import Remnashop payment settings

This commit is contained in:
3252a8
2026-06-02 11:21:26 +03:00
parent 2ccead9b49
commit 9508eece5f
8 changed files with 926 additions and 8 deletions
+595 -1
View File
@@ -17,6 +17,7 @@ import asyncio
import json import json
import logging import logging
import re import re
import shlex
import sys import sys
from collections import defaultdict from collections import defaultdict
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
@@ -53,13 +54,47 @@ from db.models import ( # noqa: E402
User, User,
) )
try: # cryptography is already used by the app for payment webhook validation.
from cryptography.fernet import Fernet
except Exception: # pragma: no cover - defensive fallback for minimal tooling.
Fernet = None # type: ignore[assignment]
SOURCE = "remnashop" SOURCE = "remnashop"
REMNASHOP_ENCRYPTED_PREFIX = "enc_"
GIB = 1024**3 GIB = 1024**3
UUID_RE = re.compile( UUID_RE = re.compile(
r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-" r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-"
r"[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b" r"[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b"
) )
SAFE_SCHEMA_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") SAFE_SCHEMA_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
REMNASHOP_PAYMENT_WEBHOOK_PATH = "/api/v1/payments/{gateway}"
REMNASHOP_PANEL_WEBHOOK_PATH = "/api/v1/remnawave"
SUPPORTED_REMNASHOP_PROVIDER_TYPES = {
"TELEGRAM_STARS",
"YOOKASSA",
"HELEKET",
"CRYPTOPAY",
"FREEKASSA",
"PLATEGA",
"WATA",
}
UNSUPPORTED_REMNASHOP_PROVIDER_TYPES = {
"YOOMONEY",
"CRYPTOMUS",
"MULENPAY",
"PAYMASTER",
"ROBOKASSA",
"URLPAY",
}
PAYMENT_WEBHOOK_PATHS = {
"yookassa": "/webhook/yookassa",
"wata": "/webhook/wata",
"cryptopay": "/webhook/cryptopay",
"heleket": "/webhook/heleket",
"freekassa": "/webhook/freekassa",
"platega": "/webhook/platega",
}
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -160,6 +195,365 @@ def _jsonish(value: Any) -> dict[str, Any]:
return {} return {}
def _strip_env_value(value: str) -> str:
lexer = shlex.shlex(value, posix=True)
lexer.whitespace_split = True
lexer.commenters = "#"
try:
tokens = list(lexer)
except ValueError:
return value.strip().strip("\"'")
return " ".join(tokens).strip()
def parse_remnashop_env_text(text_value: str) -> dict[str, str]:
env: dict[str, str] = {}
for raw_line in str(text_value or "").splitlines():
line = raw_line.strip()
if not line or line.startswith("#"):
continue
if line.startswith("export "):
line = line[len("export ") :].strip()
if "=" not in line:
continue
key, value = line.split("=", 1)
key = key.strip()
if not key or not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", key):
continue
env[key] = _strip_env_value(value)
return env
def read_remnashop_env_file(path: Optional[str]) -> dict[str, str]:
if not path:
return {}
return parse_remnashop_env_text(Path(path).read_text(encoding="utf-8"))
def _clean_url(value: Any) -> Optional[str]:
text_value = str(value or "").strip().rstrip("/")
return text_value or None
def _remnashop_panel_api_url(value: Any) -> Optional[str]:
host = _clean_url(value)
if not host:
return None
if "://" not in host:
if "." in host:
host = f"https://{host}"
else:
host = f"http://{host}:3000"
if not host.rstrip("/").endswith("/api"):
host = f"{host.rstrip('/')}/api"
return host
def _source_public_base_from_env(env: dict[str, str]) -> Optional[str]:
domain = _clean_url(env.get("APP_DOMAIN"))
if not domain:
return None
if "://" not in domain:
domain = f"https://{domain}"
return domain
def _support_link_from_username(value: Any) -> Optional[str]:
username = str(value or "").strip().lstrip("@")
if not username:
return None
return f"https://t.me/{username}"
def _mini_app_url_from_env(value: Any) -> Optional[str]:
raw = str(value or "").strip()
if not raw:
return None
if raw.lower() in {"true", "false", "0", "1"}:
return None
return raw if raw.startswith("https://") else None
def _add_override(overrides: dict[str, Any], key: str, value: Any) -> None:
if value is None:
return
if isinstance(value, str) and not value.strip():
return
overrides[key] = value
def remnashop_env_overrides(env: dict[str, str]) -> dict[str, Any]:
overrides: dict[str, Any] = {}
_add_override(overrides, "PANEL_API_URL", _remnashop_panel_api_url(env.get("REMNAWAVE_HOST")))
_add_override(overrides, "PANEL_API_KEY", env.get("REMNAWAVE_TOKEN"))
_add_override(overrides, "PANEL_WEBHOOK_SECRET", env.get("REMNAWAVE_WEBHOOK_SECRET"))
_add_override(
overrides,
"SUPPORT_LINK",
_support_link_from_username(env.get("BOT_SUPPORT_USERNAME")),
)
_add_override(overrides, "DEFAULT_LANGUAGE", env.get("APP_DEFAULT_LOCALE"))
_add_override(
overrides,
"SUBSCRIPTION_MINI_APP_URL",
_mini_app_url_from_env(env.get("BOT_MINI_APP")),
)
return overrides
def remnashop_source_urls_from_env(env: dict[str, str]) -> dict[str, str]:
base = _source_public_base_from_env(env)
if not base:
return {}
return {
"telegram": f"{base}/api/v1/telegram",
"remnawave_panel": f"{base}{REMNASHOP_PANEL_WEBHOOK_PATH}",
"payments": f"{base}/api/v1/payments/<gateway>",
}
def _normalize_gateway_type(value: Any) -> str:
if hasattr(value, "value"):
value = value.value
text_value = str(value or "").strip().upper()
if "." in text_value:
text_value = text_value.rsplit(".", 1)[-1]
return re.sub(r"[^A-Z0-9_]+", "_", text_value).strip("_")
def _normalize_currency(value: Any) -> Optional[str]:
text_value = str(value or "").strip().upper()
if "." in text_value:
text_value = text_value.rsplit(".", 1)[-1]
aliases = {"RUR": "RUB", "STARS": "XTR", "STAR": "XTR"}
normalized = aliases.get(text_value, text_value)
return normalized or None
def _truthy(value: Any) -> bool:
if isinstance(value, bool):
return value
if isinstance(value, (int, float)):
return bool(value)
return str(value or "").strip().lower() in {"1", "true", "yes", "on", "active"}
def _is_encrypted_remnashop_value(value: Any) -> bool:
return isinstance(value, str) and value.startswith(REMNASHOP_ENCRYPTED_PREFIX)
def remnashop_decrypt_value(value: Any, crypt_key: Optional[str]) -> tuple[Any, bool]:
if not _is_encrypted_remnashop_value(value):
return value, False
if not crypt_key or Fernet is None:
return None, True
try:
token = str(value).removeprefix(REMNASHOP_ENCRYPTED_PREFIX).encode()
return Fernet(crypt_key.encode()).decrypt(token).decode(), False
except Exception:
return None, True
def remnashop_decrypt_recursive(
value: Any,
crypt_key: Optional[str],
*,
skipped_paths: Optional[list[str]] = None,
path: str = "",
) -> Any:
if isinstance(value, dict):
return {
key: remnashop_decrypt_recursive(
item,
crypt_key,
skipped_paths=skipped_paths,
path=f"{path}.{key}" if path else str(key),
)
for key, item in value.items()
}
if isinstance(value, list):
return [
remnashop_decrypt_recursive(
item,
crypt_key,
skipped_paths=skipped_paths,
path=f"{path}[{index}]",
)
for index, item in enumerate(value)
]
decrypted, skipped = remnashop_decrypt_value(value, crypt_key)
if skipped and skipped_paths is not None:
skipped_paths.append(path or "<value>")
return decrypted
def _provider_mapping_result(
gateway_type: str,
provider_ids: Iterable[str],
overrides: dict[str, Any],
warnings: Optional[list[str]] = None,
) -> dict[str, Any]:
return {
"source_type": gateway_type,
"provider_ids": [provider for provider in provider_ids if provider],
"overrides": overrides,
"warnings": warnings or [],
"supported": True,
}
def remnashop_payment_gateway_overrides(
row: dict[str, Any],
*,
crypt_key: Optional[str] = None,
) -> dict[str, Any]:
gateway_type = _normalize_gateway_type(row.get("type"))
if gateway_type not in SUPPORTED_REMNASHOP_PROVIDER_TYPES:
return {
"source_type": gateway_type,
"provider_ids": [],
"overrides": {},
"warnings": [],
"supported": False,
}
skipped_secret_paths: list[str] = []
settings = remnashop_decrypt_recursive(
_jsonish(row.get("settings")),
crypt_key,
skipped_paths=skipped_secret_paths,
)
active = _truthy(row.get("is_active"))
currency = _normalize_currency(row.get("currency"))
overrides: dict[str, Any] = {}
warnings = [
(
f"Skipped encrypted Remnashop {gateway_type} setting '{path}': "
"APP_CRYPT_KEY is missing or invalid"
)
for path in skipped_secret_paths
]
if gateway_type == "TELEGRAM_STARS":
_add_override(overrides, "STARS_ENABLED", active)
return _provider_mapping_result(gateway_type, ["stars"], overrides, warnings)
if gateway_type == "YOOKASSA":
_add_override(overrides, "YOOKASSA_ENABLED", active)
_add_override(overrides, "YOOKASSA_SHOP_ID", settings.get("shop_id"))
_add_override(overrides, "YOOKASSA_SECRET_KEY", settings.get("api_key"))
_add_override(overrides, "YOOKASSA_DEFAULT_RECEIPT_EMAIL", settings.get("customer"))
_add_override(overrides, "YOOKASSA_VAT_CODE", settings.get("vat_code"))
if currency and currency != "RUB":
warnings.append(
f"YooKassa supports RUB only in this shop; source currency was {currency}"
)
return _provider_mapping_result(gateway_type, ["yookassa"], overrides, warnings)
if gateway_type == "WATA":
_add_override(overrides, "WATA_ENABLED", active)
_add_override(overrides, "WATA_API_TOKEN", settings.get("api_key"))
return _provider_mapping_result(gateway_type, ["wata"], overrides, warnings)
if gateway_type == "CRYPTOPAY":
_add_override(overrides, "CRYPTOPAY_ENABLED", active)
_add_override(overrides, "CRYPTOPAY_TOKEN", settings.get("api_key"))
if currency:
_add_override(overrides, "CRYPTOPAY_ASSET", currency)
return _provider_mapping_result(gateway_type, ["cryptopay"], overrides, warnings)
if gateway_type == "HELEKET":
_add_override(overrides, "HELEKET_ENABLED", active)
_add_override(overrides, "HELEKET_MERCHANT_ID", settings.get("merchant_id"))
_add_override(overrides, "HELEKET_API_KEY", settings.get("api_key"))
if currency:
_add_override(overrides, "HELEKET_CURRENCY", currency)
_add_override(overrides, "HELEKET_SUPPORTED_CURRENCIES", currency)
return _provider_mapping_result(gateway_type, ["heleket"], overrides, warnings)
if gateway_type == "FREEKASSA":
_add_override(overrides, "FREEKASSA_ENABLED", active)
_add_override(overrides, "FREEKASSA_MERCHANT_ID", settings.get("shop_id"))
_add_override(overrides, "FREEKASSA_API_KEY", settings.get("api_key"))
_add_override(overrides, "FREEKASSA_SECOND_SECRET", settings.get("secret_word_2"))
_add_override(overrides, "FREEKASSA_PAYMENT_METHOD_ID", settings.get("payment_system_id"))
_add_override(overrides, "FREEKASSA_PAYMENT_IP", settings.get("customer_ip"))
if settings.get("customer_email"):
warnings.append(
"FreeKassa customer_email was captured by Remnashop but is not a "
"Minishop provider setting"
)
return _provider_mapping_result(gateway_type, ["freekassa"], overrides, warnings)
if gateway_type == "PLATEGA":
_add_override(overrides, "PLATEGA_ENABLED", active)
_add_override(overrides, "PLATEGA_SBP_ENABLED", active)
_add_override(overrides, "PLATEGA_MERCHANT_ID", settings.get("merchant_id"))
_add_override(overrides, "PLATEGA_SECRET", settings.get("api_key"))
_add_override(overrides, "PLATEGA_PAYMENT_METHOD", settings.get("payment_method"))
_add_override(overrides, "PLATEGA_SBP_METHOD", settings.get("payment_method"))
if currency:
_add_override(overrides, "PLATEGA_SUPPORTED_CURRENCIES", currency)
return _provider_mapping_result(gateway_type, ["platega_sbp"], overrides, warnings)
return _provider_mapping_result(gateway_type, [], overrides, warnings)
def _target_webhook_url(base_url: Optional[str], path: str) -> Optional[str]:
base = _clean_url(base_url)
if not base:
return None
return f"{base}{path if path.startswith('/') else '/' + path}"
def remnashop_post_migration_actions(
*,
target_webhook_base_url: Optional[str],
imported_provider_ids: Iterable[str],
source_env: Optional[dict[str, str]] = None,
) -> dict[str, Any]:
provider_ids = list(dict.fromkeys(imported_provider_ids))
payment_actions = []
seen_paths: set[str] = set()
for provider_id in provider_ids:
path = PAYMENT_WEBHOOK_PATHS.get(provider_id)
if not path or path in seen_paths:
continue
seen_paths.add(path)
payment_actions.append(
{
"provider": provider_id,
"new_url": _target_webhook_url(target_webhook_base_url, path),
"where": {
"yookassa": "YooKassa merchant cabinet -> HTTP notifications URL",
"wata": "WATA merchant dashboard -> webhook/callback URL",
"cryptopay": "CryptoBot/Crypto Pay app -> webhook URL",
"heleket": "Heleket merchant dashboard -> payment webhook/callback URL",
"freekassa": "FreeKassa shop settings -> notification/result URL",
"platega": "Platega merchant/project settings -> webhook URL",
}.get(provider_id, "Payment provider dashboard -> webhook/callback URL"),
}
)
return {
"webhook_base_url_configured": bool(_clean_url(target_webhook_base_url)),
"source_urls": remnashop_source_urls_from_env(source_env or {}),
"remnawave_panel": {
"new_url": _target_webhook_url(target_webhook_base_url, "/webhook/panel"),
"where": "Remnawave Panel -> WEBHOOK_URL",
"secret": (
"Set the Remnawave webhook secret to the value stored in "
"PANEL_WEBHOOK_SECRET."
),
},
"payment_providers": payment_actions,
"telegram": {
"new_url": _target_webhook_url(target_webhook_base_url, "/tg/webhook"),
"where": "Telegram webhook is set automatically by Minishop on startup.",
},
}
def _listish(value: Any) -> list[Any]: def _listish(value: Any) -> list[Any]:
if value is None: if value is None:
return [] return []
@@ -307,6 +701,9 @@ class RemnashopImporter:
created_by_admin_id: int, created_by_admin_id: int,
tariff_map: dict[str, str], tariff_map: dict[str, str],
write_admin_compat_overrides: bool, write_admin_compat_overrides: bool,
source_env: Optional[dict[str, str]] = None,
source_crypt_key: Optional[str] = None,
target_webhook_base_url: Optional[str] = None,
) -> None: ) -> None:
self.source = source self.source = source
self.target = target self.target = target
@@ -317,8 +714,12 @@ class RemnashopImporter:
self.created_by_admin_id = created_by_admin_id self.created_by_admin_id = created_by_admin_id
self.tariff_map = tariff_map self.tariff_map = tariff_map
self.write_admin_compat_overrides = write_admin_compat_overrides self.write_admin_compat_overrides = write_admin_compat_overrides
self.source_env = source_env or {}
self.source_crypt_key = source_crypt_key or self.source_env.get("APP_CRYPT_KEY")
self.target_webhook_base_url = target_webhook_base_url
self.tables: set[str] = set() self.tables: set[str] = set()
self.user_map: dict[int, int] = {} self.user_map: dict[int, int] = {}
self.imported_payment_provider_ids: list[str] = []
self.summary: dict[str, Any] = { self.summary: dict[str, Any] = {
"source": SOURCE, "source": SOURCE,
"dry_run": dry_run, "dry_run": dry_run,
@@ -328,6 +729,7 @@ class RemnashopImporter:
"subscriptions": _counter(), "subscriptions": _counter(),
"payments": _counter(), "payments": _counter(),
"promocodes": _counter(), "promocodes": _counter(),
"payment_provider_settings": _counter(),
"settings": _counter(), "settings": _counter(),
"warnings": [], "warnings": [],
} }
@@ -349,6 +751,12 @@ class RemnashopImporter:
if self._should_run("settings"): if self._should_run("settings"):
await self.import_settings() await self.import_settings()
self.summary["post_migration_actions"] = remnashop_post_migration_actions(
target_webhook_base_url=self.target_webhook_base_url,
imported_provider_ids=self.imported_payment_provider_ids,
source_env=self.source_env,
)
if self.write_admin_compat_overrides: if self.write_admin_compat_overrides:
await self._write_admin_overrides() await self._write_admin_overrides()
@@ -486,7 +894,19 @@ class RemnashopImporter:
result = await self.target.execute(stmt) result = await self.target.execute(stmt)
return result.scalar_one_or_none() return result.scalar_one_or_none()
async def _upsert_setting_override(self, key: str, value: Any) -> None: async def _upsert_setting_override(self, key: str, value: Any) -> bool:
from bot.app.web.admin_settings_manifest import coerce_value, get_field_by_key
field = get_field_by_key(key)
if field is None:
self.summary["warnings"].append(f"Skipped unknown admin setting override: {key}")
return False
try:
value = coerce_value(field, value)
except ValueError as exc:
self.summary["warnings"].append(f"Skipped invalid admin setting override {key}: {exc}")
return False
now = datetime.now(timezone.utc) now = datetime.now(timezone.utc)
encoded = json.dumps(value, ensure_ascii=False, separators=(",", ":")) encoded = json.dumps(value, ensure_ascii=False, separators=(",", ":"))
stmt = ( stmt = (
@@ -507,6 +927,142 @@ class RemnashopImporter:
) )
) )
await self.target.execute(stmt) await self.target.execute(stmt)
return True
async def _write_setting_overrides(
self,
overrides: dict[str, Any],
*,
summary_key: str,
) -> list[str]:
written: list[str] = []
for key, value in overrides.items():
if await self._upsert_setting_override(key, value):
written.append(key)
self.summary[summary_key]["overrides_written"] += 1
else:
self.summary[summary_key]["overrides_skipped"] += 1
return written
async def import_env_settings(self) -> list[str]:
if not self.source_env:
self.summary["settings"]["source_env_missing"] += 1
return []
overrides = remnashop_env_overrides(self.source_env)
if not overrides:
self.summary["settings"]["source_env_no_supported_values"] += 1
return []
written = await self._write_setting_overrides(overrides, summary_key="settings")
if written:
self.summary["settings"]["source_env_overrides_written"] += 1
await self._upsert_mapping(
entity_type="settings_env",
source_id="remnashop.env",
target_table="app_setting_overrides",
target_id=",".join(written) if written else "none",
metadata={
"override_keys": written,
"source_keys_used": sorted(
key
for key in (
"REMNAWAVE_HOST",
"REMNAWAVE_TOKEN",
"REMNAWAVE_WEBHOOK_SECRET",
"BOT_SUPPORT_USERNAME",
"APP_DEFAULT_LOCALE",
"BOT_MINI_APP",
)
if self.source_env.get(key)
),
"has_app_crypt_key": bool(self.source_env.get("APP_CRYPT_KEY")),
"source_urls": remnashop_source_urls_from_env(self.source_env),
},
)
return written
async def import_payment_provider_settings(self) -> None:
if "payment_gateways" not in self.tables:
self.summary["payment_provider_settings"]["missing_source_table"] += 1
return
rows = await self._fetch_rows("payment_gateways", order_by="order_index, id")
if not rows:
self.summary["payment_provider_settings"]["empty_source_table"] += 1
return
active_provider_ids: list[str] = []
for index, row in enumerate(rows):
source_id = row.get("id") or row.get("type") or f"row:{index}"
mapping = remnashop_payment_gateway_overrides(
row,
crypt_key=self.source_crypt_key,
)
gateway_type = mapping["source_type"]
self.summary["payment_provider_settings"]["seen"] += 1
for warning in mapping["warnings"]:
self.summary["warnings"].append(warning)
if not mapping["supported"]:
self.summary["payment_provider_settings"]["unsupported"] += 1
display_type = gateway_type or str(row.get("type") or "unknown")
self.summary["warnings"].append(
f"Remnashop payment provider {display_type} is not supported by "
"Minishop; configure it manually if it is still needed."
)
await self._upsert_mapping(
entity_type="payment_provider_settings",
source_id=source_id,
target_table="manual_configuration_required",
target_id=display_type,
metadata={
"source_type": display_type,
"active": _truthy(row.get("is_active")),
"currency": _normalize_currency(row.get("currency")),
"supported": False,
},
)
continue
written = await self._write_setting_overrides(
mapping["overrides"],
summary_key="payment_provider_settings",
)
if written:
self.summary["payment_provider_settings"]["providers_mapped"] += 1
else:
self.summary["payment_provider_settings"]["providers_without_overrides"] += 1
if _truthy(row.get("is_active")):
for provider_id in mapping["provider_ids"]:
if provider_id and provider_id not in active_provider_ids:
active_provider_ids.append(provider_id)
if provider_id and provider_id not in self.imported_payment_provider_ids:
self.imported_payment_provider_ids.append(provider_id)
await self._upsert_mapping(
entity_type="payment_provider_settings",
source_id=source_id,
target_table="app_setting_overrides",
target_id=",".join(written) if written else "none",
metadata={
"source_type": gateway_type,
"provider_ids": mapping["provider_ids"],
"active": _truthy(row.get("is_active")),
"currency": _normalize_currency(row.get("currency")),
"override_keys": written,
"source_settings_keys": sorted(_jsonish(row.get("settings")).keys()),
"warnings_count": len(mapping["warnings"]),
"supported": True,
},
)
if active_provider_ids:
order_value = ",".join(active_provider_ids)
if await self._upsert_setting_override("PAYMENT_METHODS_ORDER", order_value):
self.summary["payment_provider_settings"]["payment_order_written"] += 1
async def _upsert_legacy_referral_code(self, *, code: str, user_id: int) -> None: async def _upsert_legacy_referral_code(self, *, code: str, user_id: int) -> None:
if len(code) > 128: if len(code) > 128:
@@ -1029,7 +1585,29 @@ class RemnashopImporter:
} }
for plan in plans[:100] for plan in plans[:100]
], ],
"source_env": {
"provided": bool(self.source_env),
"supported_keys_present": sorted(
key
for key in (
"REMNAWAVE_HOST",
"REMNAWAVE_TOKEN",
"REMNAWAVE_WEBHOOK_SECRET",
"BOT_SUPPORT_USERNAME",
"APP_DEFAULT_LOCALE",
"BOT_MINI_APP",
"APP_DOMAIN",
"APP_CRYPT_KEY",
)
if self.source_env.get(key)
),
"source_urls": remnashop_source_urls_from_env(self.source_env),
},
} }
env_override_keys = await self.import_env_settings()
await self.import_payment_provider_settings()
notes["env_override_keys"] = env_override_keys
notes["payment_provider_ids"] = list(dict.fromkeys(self.imported_payment_provider_ids))
await self._upsert_mapping( await self._upsert_mapping(
entity_type="settings", entity_type="settings",
source_id="singleton", source_id="singleton",
@@ -1080,6 +1658,17 @@ def build_arg_parser() -> argparse.ArgumentParser:
parser.add_argument("--source-type", choices=[SOURCE], default=SOURCE) parser.add_argument("--source-type", choices=[SOURCE], default=SOURCE)
parser.add_argument("--source-dsn", required=True) parser.add_argument("--source-dsn", required=True)
parser.add_argument("--source-schema", default="public") parser.add_argument("--source-schema", default="public")
parser.add_argument(
"--source-env-file",
help=(
"Path to the source Remnashop .env. Used for APP_CRYPT_KEY, Remnawave "
"API settings and selected safe compatibility values."
),
)
parser.add_argument(
"--source-crypt-key",
help="Explicit Remnashop APP_CRYPT_KEY. Overrides the value from --source-env-file.",
)
parser.add_argument("--target-dsn") parser.add_argument("--target-dsn")
parser.add_argument( parser.add_argument(
"--only", "--only",
@@ -1116,6 +1705,8 @@ async def _prepare_target_schema(engine: Any) -> None:
async def run_import(args: argparse.Namespace) -> dict[str, Any]: async def run_import(args: argparse.Namespace) -> dict[str, Any]:
settings = Settings() settings = Settings()
source_env = read_remnashop_env_file(args.source_env_file)
source_crypt_key = args.source_crypt_key or source_env.get("APP_CRYPT_KEY")
source_engine = create_async_engine(normalize_async_postgres_dsn(args.source_dsn)) source_engine = create_async_engine(normalize_async_postgres_dsn(args.source_dsn))
target_engine = create_async_engine( target_engine = create_async_engine(
normalize_async_postgres_dsn(args.target_dsn or settings.DATABASE_URL) normalize_async_postgres_dsn(args.target_dsn or settings.DATABASE_URL)
@@ -1141,6 +1732,9 @@ async def run_import(args: argparse.Namespace) -> dict[str, Any]:
created_by_admin_id=args.created_by_admin_id, created_by_admin_id=args.created_by_admin_id,
tariff_map=parse_tariff_map(args.tariff_map_json), tariff_map=parse_tariff_map(args.tariff_map_json),
write_admin_compat_overrides=not args.no_admin_compat_overrides, write_admin_compat_overrides=not args.no_admin_compat_overrides,
source_env=source_env,
source_crypt_key=source_crypt_key,
target_webhook_base_url=settings.WEBHOOK_BASE_URL,
) )
summary = await importer.run() summary = await importer.run()
if args.dry_run: if args.dry_run:
+5 -1
View File
@@ -42,7 +42,11 @@ sh install.sh
``` ```
Миграция Remnashop в wizard сначала запускает `dry-run`, показывает JSON-сводку Миграция Remnashop в wizard сначала запускает `dry-run`, показывает JSON-сводку
и только после отдельного подтверждения применяет изменения в целевую БД. и только после отдельного подтверждения применяет изменения в целевую БД. Если
указать старый Remnashop `.env`, wizard передаст importer-у `APP_CRYPT_KEY`,
Remnawave API settings и поддерживаемые payment provider settings из таблицы
`payment_gateways`. После применения wizard печатает новые webhook URL для
Remnawave Panel и платежных провайдеров.
Миграция со старого `remnawave-tg-shop` работает как upgrade совместимой БД: Миграция со старого `remnawave-tg-shop` работает как upgrade совместимой БД:
либо копирует старый Docker volume, либо делает `pg_dump` по source DSN, либо копирует старый Docker volume, либо делает `pg_dump` по source DSN,
восстанавливает дамп в целевую compose-БД и запускает сервис `migrate`. восстанавливает дамп в целевую compose-БД и запускает сервис `migrate`.
+1 -1
View File
@@ -5,4 +5,4 @@
| Источник | Поддерживаемый случай | Документы | | Источник | Поддерживаемый случай | Документы |
| --- | --- | --- | | --- | --- | --- |
| [remnawave-tg-shop](https://github.com/kavore/remnawave-tg-shop/) | Полный перенос всех данных | [Инструкция](remnawave-tg-shop.md) | | [remnawave-tg-shop](https://github.com/kavore/remnawave-tg-shop/) | Полный перенос всех данных | [Инструкция](remnawave-tg-shop.md) |
| [Remnashop](https://github.com/snoups/remnashop/) | Автоматический импорт пользователей, подписок, платежей, рефералов и промокодов | [Инструкция](remnashop.md) | | [Remnashop](https://github.com/snoups/remnashop/) | Автоматический импорт пользователей, подписок, платежей, рефералов, промокодов и поддерживаемых платежных настроек | [Инструкция](remnashop.md) |
+50 -4
View File
@@ -24,16 +24,61 @@ sh install.sh
Данные, которые не имеют прямого аналога, сохраняются в служебных таблицах миграции или Данные, которые не имеют прямого аналога, сохраняются в служебных таблицах миграции или
message logs как заметки, чтобы администратор мог проверить их после переноса. message logs как заметки, чтобы администратор мог проверить их после переноса.
## Настройки и платежные провайдеры
Если указать старый Remnashop `.env`, importer дополнительно переносит часть
настроек в админские overrides:
- `REMNAWAVE_HOST` -> `PANEL_API_URL`;
- `REMNAWAVE_TOKEN` -> `PANEL_API_KEY`;
- `REMNAWAVE_WEBHOOK_SECRET` -> `PANEL_WEBHOOK_SECRET`;
- `BOT_SUPPORT_USERNAME` -> `SUPPORT_LINK`;
- `APP_DEFAULT_LOCALE` -> `DEFAULT_LANGUAGE`;
- `BOT_MINI_APP` -> `SUBSCRIPTION_MINI_APP_URL`, если там уже HTTPS URL.
Платежные провайдеры берутся из таблицы Remnashop `payment_gateways`.
Поддерживаются и автоматически маппятся: Telegram Stars, YooKassa, WATA,
CryptoPay, Heleket, FreeKassa и Platega. Для них importer переносит флаги
включения, API-ключи/merchant IDs и доступные provider-specific параметры в
раздел настроек админки.
Провайдеры YooMoney, Cryptomus, MulenPay, PayMaster, RoboKassa и UrlPay сейчас
не имеют прямого аналога в Minishop. Если они были в Remnashop, importer
оставит предупреждение в JSON-сводке и notes миграции, а настроить их нужно
вручную или через будущий отдельный provider.
Remnashop может хранить секреты в формате `enc_...`. Для расшифровки нужен
старый `APP_CRYPT_KEY`; проще всего указать путь к старому `.env` в wizard или
передать `--source-env-file`. Если ключ не передан или неверный, зашифрованные
значения будут пропущены с предупреждением, остальные данные продолжат
импортироваться.
После успешного применения wizard печатает список новых адресов webhook. Их
нужно указать во внешних сервисах вместо старых Remnashop URL:
- Remnawave Panel -> `WEBHOOK_URL`: `WEBHOOK_BASE_URL` + `/webhook/panel`;
- YooKassa HTTP notifications URL: `WEBHOOK_BASE_URL` + `/webhook/yookassa`;
- WATA webhook/callback URL: `WEBHOOK_BASE_URL` + `/webhook/wata`;
- CryptoBot/Crypto Pay webhook URL: `WEBHOOK_BASE_URL` + `/webhook/cryptopay`;
- Heleket payment webhook/callback URL: `WEBHOOK_BASE_URL` + `/webhook/heleket`;
- FreeKassa notification/result URL: `WEBHOOK_BASE_URL` + `/webhook/freekassa`;
- Platega webhook URL: `WEBHOOK_BASE_URL` + `/webhook/platega`;
- Telegram webhook `WEBHOOK_BASE_URL` + `/tg/webhook` выставляется ботом
автоматически при старте.
## Flow wizard ## Flow wizard
1. Wizard скачивает compose-профиль и `backend/scripts/import_legacy.py` через 1. Wizard скачивает compose-профиль и `backend/scripts/import_legacy.py` через
`raw.githubusercontent.com`, без клонирования репозитория. `raw.githubusercontent.com`, без клонирования репозитория.
2. Вы указываете source PostgreSQL DSN Remnashop и schema, обычно `public`. 2. Вы указываете source PostgreSQL DSN Remnashop и schema, обычно `public`.
3. Вы выбираете целевую БД: текущую compose-БД или ручной target DSN. 3. Опционально указываете путь к старому Remnashop `.env` для `APP_CRYPT_KEY`,
4. При необходимости указываете JSON map тарифов Remnashop в локальные Remnawave API settings и переносимых payment/provider settings.
4. Вы выбираете целевую БД: текущую compose-БД или ручной target DSN.
5. При необходимости указываете JSON map тарифов Remnashop в локальные
`tariff_key`, например `{"basic": "standard_month"}`. `tariff_key`, например `{"basic": "standard_month"}`.
5. Wizard запускает `dry-run` и показывает JSON-сводку. 6. Wizard запускает `dry-run` и показывает JSON-сводку.
6. После подтверждения `y` importer применяет изменения и перезапускает 7. После подтверждения `y` importer применяет изменения, печатает список новых
webhook URL для Remnawave Panel и платежных провайдеров, затем перезапускает
`backend`/`worker`, чтобы настройки совместимости перечитались. `backend`/`worker`, чтобы настройки совместимости перечитались.
Если source DB находится на том же Docker host, помните, что DSN выполняется Если source DB находится на том же Docker host, помните, что DSN выполняется
@@ -51,6 +96,7 @@ docker compose run --rm backend \
--source-type remnashop \ --source-type remnashop \
--source-dsn 'postgresql://old_user:old_password@old_host:5432/remnashop' \ --source-dsn 'postgresql://old_user:old_password@old_host:5432/remnashop' \
--source-schema public \ --source-schema public \
--source-env-file /path/to/remnashop/.env \
--dry-run --dry-run
``` ```
+63 -1
View File
@@ -47,6 +47,7 @@ COMPOSE_STYLE=""
PROMPT_VALUE="" PROMPT_VALUE=""
CHOICE_VALUE="" CHOICE_VALUE=""
LEGACY_SOURCE="" LEGACY_SOURCE=""
SOURCE_ENV_PATH=""
COMPOSE_PROJECT_NAME_VALUE="" COMPOSE_PROJECT_NAME_VALUE=""
IMAGE_TAG_VALUE="" IMAGE_TAG_VALUE=""
@@ -128,6 +129,7 @@ Environment overrides:
MINISHOP_INSTALL_REF default ref ($DEFAULT_REF) MINISHOP_INSTALL_REF default ref ($DEFAULT_REF)
MINISHOP_IMAGE_TAG default image tag ($DEFAULT_IMAGE_TAG) MINISHOP_IMAGE_TAG default image tag ($DEFAULT_IMAGE_TAG)
REMNASHOP_SOURCE_DSN default source DSN for migration REMNASHOP_SOURCE_DSN default source DSN for migration
REMNASHOP_SOURCE_ENV_FILE default source Remnashop .env path for migration
LEGACY_TGSHOP_SOURCE_DSN default remnawave-tg-shop source DSN for dump/restore LEGACY_TGSHOP_SOURCE_DSN default remnawave-tg-shop source DSN for dump/restore
The wizard is interactive by design. It never overwrites files without The wizard is interactive by design. It never overwrites files without
@@ -838,10 +840,52 @@ local_target_dsn() {
printf 'postgresql://%s:%s@postgres:5432/%s' "$POSTGRES_USER_VALUE" "$POSTGRES_PASSWORD_VALUE" "$POSTGRES_DB_VALUE" printf 'postgresql://%s:%s@postgres:5432/%s' "$POSTGRES_USER_VALUE" "$POSTGRES_PASSWORD_VALUE" "$POSTGRES_DB_VALUE"
} }
target_webhook_base_url() {
public_url=$(env_get WEBHOOK_PUBLIC_URL "")
if [ -n "$public_url" ]; then
printf '%s' "$public_url" | sed 's:/*$::'
return 0
fi
host=$(env_get WEBHOOK_HOST "")
if [ -n "$host" ]; then
printf 'https://%s' "$host" | sed 's:/*$::'
return 0
fi
printf ''
}
remnashop_webhook_checklist() {
section "Update external webhooks"
base_url=$(target_webhook_base_url)
if [ -z "$base_url" ]; then
warn "Could not determine webhook base URL from .env. Set WEBHOOK_HOST or WEBHOOK_PUBLIC_URL, then use WEBHOOK_BASE_URL + paths below."
base_url="WEBHOOK_BASE_URL"
fi
info "Set these URLs in external dashboards after the migration:"
printf ' Remnawave Panel -> WEBHOOK_URL: %s/webhook/panel\n' "$base_url"
panel_secret=$(env_get PANEL_WEBHOOK_SECRET "")
if [ -n "$panel_secret" ]; then
printf ' Remnawave Panel -> webhook secret: %s\n' "$(mask_secret "$panel_secret")"
else
warn "PANEL_WEBHOOK_SECRET is empty; set it in Minishop and in Remnawave Panel."
fi
printf ' YooKassa merchant cabinet -> HTTP notifications URL: %s/webhook/yookassa\n' "$base_url"
printf ' WATA merchant dashboard -> webhook/callback URL: %s/webhook/wata\n' "$base_url"
printf ' CryptoBot/Crypto Pay app -> webhook URL: %s/webhook/cryptopay\n' "$base_url"
printf ' Heleket merchant dashboard -> payment webhook/callback URL: %s/webhook/heleket\n' "$base_url"
printf ' FreeKassa shop settings -> notification/result URL: %s/webhook/freekassa\n' "$base_url"
printf ' Platega merchant/project settings -> webhook URL: %s/webhook/platega\n' "$base_url"
printf ' Telegram webhook: %s/tg/webhook (configured automatically on bot startup)\n' "$base_url"
}
run_import_command() { run_import_command() {
dry="$1" dry="$1"
set -- run --rm \ set -- run --rm \
-v "$IMPORTER_PATH:/app/backend/scripts/import_legacy.py:ro" -v "$IMPORTER_PATH:/app/backend/scripts/import_legacy.py:ro"
if [ -n "$SOURCE_ENV_PATH" ]; then
set -- "$@" -v "$SOURCE_ENV_PATH:/tmp/remnashop.env:ro"
fi
if [ -n "$TARIFF_MAP_PATH" ]; then if [ -n "$TARIFF_MAP_PATH" ]; then
set -- "$@" -v "$TARIFF_MAP_PATH:/tmp/tariff-map.json:ro" set -- "$@" -v "$TARIFF_MAP_PATH:/tmp/tariff-map.json:ro"
fi fi
@@ -850,6 +894,9 @@ run_import_command() {
--source-dsn "$SOURCE_DSN" \ --source-dsn "$SOURCE_DSN" \
--source-schema "$SOURCE_SCHEMA" \ --source-schema "$SOURCE_SCHEMA" \
--target-dsn "$TARGET_DSN" --target-dsn "$TARGET_DSN"
if [ -n "$SOURCE_ENV_PATH" ]; then
set -- "$@" --source-env-file /tmp/remnashop.env
fi
if [ -n "$TARIFF_MAP_PATH" ]; then if [ -n "$TARIFF_MAP_PATH" ]; then
set -- "$@" --tariff-map-json /tmp/tariff-map.json set -- "$@" --tariff-map-json /tmp/tariff-map.json
fi fi
@@ -861,7 +908,7 @@ run_import_command() {
choose_legacy_source() { choose_legacy_source() {
choose "Source bot" "1" "1|2|3" \ choose "Source bot" "1" "1|2|3" \
"1. Remnashop - import users, subscriptions, payments, referrals and promo codes." \ "1. Remnashop - import users, subscriptions, payments, provider settings and promo codes." \
"2. Old remnawave-tg-shop - upgrade an old compatible database/volume." \ "2. Old remnawave-tg-shop - upgrade an old compatible database/volume." \
"3. Skip migration" "3. Skip migration"
case "$CHOICE_VALUE" in case "$CHOICE_VALUE" in
@@ -895,6 +942,20 @@ run_remnashop_migration() {
SOURCE_DSN="$PROMPT_VALUE" SOURCE_DSN="$PROMPT_VALUE"
prompt_value "Source schema" "public" 1 0 "" prompt_value "Source schema" "public" 1 0 ""
SOURCE_SCHEMA="$PROMPT_VALUE" SOURCE_SCHEMA="$PROMPT_VALUE"
prompt_value "Optional source Remnashop .env path (empty to skip)" "${REMNASHOP_SOURCE_ENV_FILE:-}" 0 0 ""
SOURCE_ENV_PATH="$PROMPT_VALUE"
if [ -n "$SOURCE_ENV_PATH" ]; then
source_env_dir=$(dirname "$SOURCE_ENV_PATH")
if [ ! -d "$source_env_dir" ]; then
fail "Source .env directory not found: $source_env_dir"
return 1
fi
SOURCE_ENV_PATH=$(cd "$source_env_dir" && pwd)/$(basename "$SOURCE_ENV_PATH")
if [ ! -f "$SOURCE_ENV_PATH" ]; then
fail "Source Remnashop .env not found: $SOURCE_ENV_PATH"
return 1
fi
fi
choose "Target database" "1" "1|2" \ choose "Target database" "1" "1|2" \
"1. This Docker Compose stack database (recommended)" \ "1. This Docker Compose stack database (recommended)" \
@@ -936,6 +997,7 @@ run_remnashop_migration() {
section "Apply import" section "Apply import"
run_import_command 0 || return 1 run_import_command 0 || return 1
remnashop_webhook_checklist
if confirm "Restart backend and worker so setting overrides are reloaded?" 1; then if confirm "Restart backend and worker so setting overrides are reloaded?" 1; then
(cd "$TARGET_DIR" && run_compose restart backend worker) || true (cd "$TARGET_DIR" && run_compose restart backend worker) || true
fi fi
+18
View File
@@ -36,6 +36,8 @@ def test_shell_installer_downloads_raw_files_and_runs_import_in_container():
assert "raw.githubusercontent.com" in script assert "raw.githubusercontent.com" in script
assert "git clone" not in script assert "git clone" not in script
assert "backend python backend/scripts/import_legacy.py" in script assert "backend python backend/scripts/import_legacy.py" in script
assert "Optional source Remnashop .env path" in script
assert "--source-env-file /tmp/remnashop.env" in script
assert "--dry-run" in script assert "--dry-run" in script
assert "Install new stack and run migration" in script assert "Install new stack and run migration" in script
assert "Run migration only" in script assert "Run migration only" in script
@@ -61,3 +63,19 @@ def test_shell_installer_only_prepares_data_mount_not_runtime_content():
assert "webapp-logo" not in script assert "webapp-logo" not in script
assert "webapp-emoji" not in script assert "webapp-emoji" not in script
assert "locales-overrides.json" not in script assert "locales-overrides.json" not in script
def test_shell_installer_prints_remnashop_webhook_checklist():
script = INSTALL_SCRIPT.read_text(encoding="utf-8")
assert "remnashop_webhook_checklist" in script
assert "Remnawave Panel -> WEBHOOK_URL" in script
assert "PANEL_WEBHOOK_SECRET" in script
assert "/webhook/panel" in script
assert "/webhook/yookassa" in script
assert "/webhook/wata" in script
assert "/webhook/cryptopay" in script
assert "/webhook/heleket" in script
assert "/webhook/freekassa" in script
assert "/webhook/platega" in script
assert "/tg/webhook" in script
+26
View File
@@ -8,6 +8,7 @@ from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1] REPO_ROOT = Path(__file__).resolve().parents[1]
DOC_PATH = REPO_ROOT / "docs" / "migrations" / "remnawave-tg-shop.md" DOC_PATH = REPO_ROOT / "docs" / "migrations" / "remnawave-tg-shop.md"
REMNASHOP_DOC_PATH = REPO_ROOT / "docs" / "migrations" / "remnashop.md"
INSTALL_SCRIPT_PATH = REPO_ROOT / "scripts" / "install.sh" INSTALL_SCRIPT_PATH = REPO_ROOT / "scripts" / "install.sh"
REMOVED_SCRIPT_PATH = REPO_ROOT / "scripts" / "migrate_to_minishop.sh" REMOVED_SCRIPT_PATH = REPO_ROOT / "scripts" / "migrate_to_minishop.sh"
COMPOSE_FILES = ( COMPOSE_FILES = (
@@ -185,6 +186,31 @@ class DocComposeFileReferencesTests(unittest.TestCase):
self.assertTrue((REPO_ROOT / "backend" / "db" / "migrator.py").is_file()) self.assertTrue((REPO_ROOT / "backend" / "db" / "migrator.py").is_file())
class RemnashopMigrationDocumentationFactsTests(unittest.TestCase):
def setUp(self) -> None:
self.doc = _read(REMNASHOP_DOC_PATH)
def test_doc_mentions_env_payment_gateways_and_encrypted_secrets(self):
self.assertIn("--source-env-file", self.doc)
self.assertIn("APP_CRYPT_KEY", self.doc)
self.assertIn("payment_gateways", self.doc)
self.assertIn("enc_", self.doc)
def test_doc_lists_new_webhook_paths_after_migration(self):
for path in (
"/webhook/panel",
"/webhook/yookassa",
"/webhook/wata",
"/webhook/cryptopay",
"/webhook/heleket",
"/webhook/freekassa",
"/webhook/platega",
"/tg/webhook",
):
with self.subTest(path=path):
self.assertIn(path, self.doc)
class MigrationFootprintRegexTests(unittest.TestCase): class MigrationFootprintRegexTests(unittest.TestCase):
def test_every_compose_volume_documented(self): def test_every_compose_volume_documented(self):
doc = _read(DOC_PATH) doc = _read(DOC_PATH)
+168
View File
@@ -1,7 +1,12 @@
from datetime import datetime, timezone from datetime import datetime, timezone
import pytest
from scripts.import_legacy import ( from scripts.import_legacy import (
parse_remnashop_env_text,
remnashop_env_overrides,
remnashop_months_from_plan_snapshot, remnashop_months_from_plan_snapshot,
remnashop_payment_gateway_overrides,
remnashop_post_migration_actions,
remnashop_pricing_amount, remnashop_pricing_amount,
remnashop_pricing_currency, remnashop_pricing_currency,
remnashop_sale_mode, remnashop_sale_mode,
@@ -42,3 +47,166 @@ def test_remnashop_plan_months_prefers_snapshot_then_dates():
) )
== 3 == 3
) )
def test_remnashop_env_parser_and_overrides_map_safe_values():
env = parse_remnashop_env_text(
"""
# old Remnashop
export REMNAWAVE_HOST=panel.example.com
REMNAWAVE_TOKEN='panel-token#kept'
REMNAWAVE_WEBHOOK_SECRET="panel secret"
BOT_SUPPORT_USERNAME=@support_bot # comment
APP_DEFAULT_LOCALE=en
BOT_MINI_APP=https://app.example.com/
APP_DOMAIN=old.example.com
"""
)
assert env["REMNAWAVE_TOKEN"] == "panel-token#kept"
assert env["BOT_SUPPORT_USERNAME"] == "@support_bot"
overrides = remnashop_env_overrides(env)
assert overrides == {
"PANEL_API_URL": "https://panel.example.com/api",
"PANEL_API_KEY": "panel-token#kept",
"PANEL_WEBHOOK_SECRET": "panel secret",
"SUPPORT_LINK": "https://t.me/support_bot",
"DEFAULT_LANGUAGE": "en",
"SUBSCRIPTION_MINI_APP_URL": "https://app.example.com/",
}
def test_remnashop_yookassa_gateway_maps_to_current_provider_settings():
result = remnashop_payment_gateway_overrides(
{
"type": "YOOKASSA",
"currency": "RUB",
"is_active": True,
"settings": {
"shop_id": "shop-1",
"api_key": "secret",
"customer": "receipt@example.com",
"vat_code": 1,
},
}
)
assert result["supported"] is True
assert result["provider_ids"] == ["yookassa"]
assert result["overrides"] == {
"YOOKASSA_ENABLED": True,
"YOOKASSA_SHOP_ID": "shop-1",
"YOOKASSA_SECRET_KEY": "secret",
"YOOKASSA_DEFAULT_RECEIPT_EMAIL": "receipt@example.com",
"YOOKASSA_VAT_CODE": 1,
}
def test_remnashop_free_kassa_and_platega_gateways_map_available_settings():
freekassa = remnashop_payment_gateway_overrides(
{
"type": "FREEKASSA",
"is_active": True,
"settings": {
"shop_id": "merchant",
"api_key": "api",
"secret_word_2": "notify-secret",
"payment_system_id": 42,
"customer_ip": "203.0.113.10",
"customer_email": "payer@example.com",
},
}
)
assert freekassa["provider_ids"] == ["freekassa"]
assert freekassa["overrides"]["FREEKASSA_SECOND_SECRET"] == "notify-secret"
assert freekassa["overrides"]["FREEKASSA_PAYMENT_METHOD_ID"] == 42
assert any("customer_email" in warning for warning in freekassa["warnings"])
platega = remnashop_payment_gateway_overrides(
{
"type": "PLATEGA",
"currency": "RUB",
"is_active": True,
"settings": {
"merchant_id": "merchant",
"api_key": "secret",
"payment_method": 2,
},
}
)
assert platega["provider_ids"] == ["platega_sbp"]
assert platega["overrides"]["PLATEGA_SBP_ENABLED"] is True
assert platega["overrides"]["PLATEGA_SBP_METHOD"] == 2
def test_remnashop_unsupported_gateway_is_reported_without_overrides():
result = remnashop_payment_gateway_overrides(
{
"type": "ROBOKASSA",
"is_active": True,
"settings": {"merchant_login": "shop"},
}
)
assert result["supported"] is False
assert result["provider_ids"] == []
assert result["overrides"] == {}
def test_remnashop_encrypted_gateway_settings_need_app_crypt_key():
result = remnashop_payment_gateway_overrides(
{
"type": "WATA",
"is_active": True,
"settings": {"api_key": "enc_not-a-fernet-token"},
}
)
assert result["overrides"] == {"WATA_ENABLED": True}
assert any("APP_CRYPT_KEY" in warning for warning in result["warnings"])
def test_remnashop_encrypted_gateway_settings_decrypt_with_app_crypt_key():
cryptography = pytest.importorskip("cryptography.fernet")
key = cryptography.Fernet.generate_key().decode()
token = cryptography.Fernet(key.encode()).encrypt(b"wata-token").decode()
result = remnashop_payment_gateway_overrides(
{
"type": "WATA",
"is_active": True,
"settings": {"api_key": f"enc_{token}"},
},
crypt_key=key,
)
assert result["overrides"] == {
"WATA_ENABLED": True,
"WATA_API_TOKEN": "wata-token",
}
assert result["warnings"] == []
def test_remnashop_post_migration_actions_include_new_webhook_urls():
actions = remnashop_post_migration_actions(
target_webhook_base_url="https://webhooks.example.com/",
imported_provider_ids=["yookassa", "wata", "yookassa"],
source_env={"APP_DOMAIN": "old.example.com"},
)
assert actions["remnawave_panel"]["new_url"] == "https://webhooks.example.com/webhook/panel"
assert actions["telegram"]["new_url"] == "https://webhooks.example.com/tg/webhook"
assert actions["source_urls"]["payments"] == "https://old.example.com/api/v1/payments/<gateway>"
assert actions["payment_providers"] == [
{
"provider": "yookassa",
"new_url": "https://webhooks.example.com/webhook/yookassa",
"where": "YooKassa merchant cabinet -> HTTP notifications URL",
},
{
"provider": "wata",
"new_url": "https://webhooks.example.com/webhook/wata",
"where": "WATA merchant dashboard -> webhook/callback URL",
},
]