diff --git a/backend/bot/app/factories/build_services.py b/backend/bot/app/factories/build_services.py index 3966d2c..c8e0e05 100644 --- a/backend/bot/app/factories/build_services.py +++ b/backend/bot/app/factories/build_services.py @@ -11,6 +11,7 @@ from bot.services.email_auth_service import EmailAuthService from bot.services.lknpd_service import LknpdService from bot.services.notification_service import NotificationService from bot.services.panel_api_service import PanelApiService +from bot.services.panel_dry_run_api_service import PanelDryRunApiService from bot.services.panel_webhook_service import PanelWebhookService from bot.services.promo_code_service import PromoCodeService from bot.services.referral_service import ReferralService @@ -26,7 +27,11 @@ def build_core_services( i18n: JsonI18n, bot_username_for_default_return: str, ): - panel_service = PanelApiService(settings) + panel_service = ( + PanelDryRunApiService(settings) + if bool(getattr(settings, "panel_dry_run_enabled", False)) + else PanelApiService(settings) + ) subscription_service = SubscriptionService(settings, panel_service, bot, i18n) referral_service = ReferralService(settings, subscription_service, bot, i18n) promo_code_service = PromoCodeService(settings, subscription_service, bot, i18n) diff --git a/backend/bot/services/panel_dry_run_api_service.py b/backend/bot/services/panel_dry_run_api_service.py new file mode 100644 index 0000000..fa47339 --- /dev/null +++ b/backend/bot/services/panel_dry_run_api_service.py @@ -0,0 +1,482 @@ +import json +import logging +import re +import uuid +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, Dict, List, Optional + +from config.settings import Settings + +from .panel_api_service import PanelApiService + +logger = logging.getLogger(__name__) + +_USER_ACTION_RE = re.compile( + r"^/users/(?P[^/]+)/actions/(?Penable|disable|reset-traffic)$" +) +_INTERNAL_SQUAD_BULK_RE = re.compile( + r"^/internal-squads/(?P[^/]+)/bulk-actions/" + r"(?Padd-users|remove-users)$" +) +_LIVE_POST_ENDPOINTS = frozenset({"/system/tools/happ/encrypt"}) +_KNOWN_TRAFFIC_STRATEGIES = frozenset({"NO_RESET", "DAY", "WEEK", "MONTH"}) + + +@dataclass +class _DryRunValidation: + errors: List[str] = field(default_factory=list) + + @property + def ok(self) -> bool: + return not self.errors + + def add(self, message: str) -> None: + self.errors.append(message) + + +class PanelDryRunApiService(PanelApiService): + """Panel API client that reads live data but never mutates Remnawave users.""" + + def __init__(self, settings: Settings): + super().__init__(settings) + self._synthetic_users: Dict[str, Dict[str, Any]] = {} + + async def _request( + self, method: str, endpoint: str, log_full_response: bool = False, **kwargs + ) -> Optional[Dict[str, Any]]: + method_upper = method.upper() + normalized_endpoint = self._normalize_endpoint(endpoint) + if not self._should_intercept(method_upper, normalized_endpoint): + return await super()._request( + method_upper, + endpoint, + log_full_response=log_full_response, + **kwargs, + ) + + validation = await self._validate_dry_run_request( + method_upper, + normalized_endpoint, + kwargs.get("json"), + ) + if not validation.ok: + self._log_dry_run( + "BLOCKED", + method_upper, + normalized_endpoint, + kwargs.get("json"), + errors=validation.errors, + ) + return { + "error": True, + "status_code": 400, + "errorCode": "DRY_RUN_VALIDATION_FAILED", + "message": "Panel dry-run validation failed.", + "details": {"errors": validation.errors}, + } + + response = await self._dry_run_response( + method_upper, + normalized_endpoint, + kwargs.get("json"), + ) + self._log_dry_run("OK", method_upper, normalized_endpoint, kwargs.get("json")) + return {"response": response, "dryRun": True} + + @staticmethod + def _normalize_endpoint(endpoint: str) -> str: + return f"/{str(endpoint or '').lstrip('/')}" + + @staticmethod + def _payload_preview(payload: Any) -> str: + try: + text = json.dumps(payload, ensure_ascii=False, default=str, sort_keys=True) + except Exception: + text = str(payload) + if len(text) > 1200: + return f"{text[:1200]}..." + return text + + def _log_dry_run( + self, + status: str, + method: str, + endpoint: str, + payload: Any, + *, + errors: Optional[List[str]] = None, + ) -> None: + logger.info( + "[PANEL DRY-RUN %s] would %s %s payload=%s%s", + status, + method, + endpoint, + self._payload_preview(payload), + f" errors={errors}" if errors else "", + ) + + @staticmethod + def _should_intercept(method: str, endpoint: str) -> bool: + if method in PanelApiService._SAFE_METHODS: + return False + if method == "POST" and endpoint in _LIVE_POST_ENDPOINTS: + return False + return True + + async def _validate_dry_run_request( + self, + method: str, + endpoint: str, + payload: Any, + ) -> _DryRunValidation: + validation = _DryRunValidation() + data = payload if isinstance(payload, dict) else {} + if payload is not None and not isinstance(payload, dict): + validation.add("JSON payload must be an object.") + return validation + + if method == "POST" and endpoint == "/users": + await self._validate_create_user_payload(data, validation) + return validation + if method == "PATCH" and endpoint == "/users": + await self._validate_update_user_payload(data, validation) + return validation + if method == "POST" and (match := _USER_ACTION_RE.match(endpoint)): + user_uuid = match.group("user_uuid") + self._validate_non_empty_string(user_uuid, "user uuid", validation) + await self._validate_remote_user(user_uuid, validation) + return validation + if method == "DELETE" and endpoint.startswith("/users/"): + user_uuid = endpoint.removeprefix("/users/").strip() + self._validate_non_empty_string(user_uuid, "user uuid", validation) + await self._validate_remote_user(user_uuid, validation) + return validation + if method == "POST" and endpoint == "/hwid/devices/delete": + user_uuid = self._validate_non_empty_string( + data.get("userUuid"), + "userUuid", + validation, + ) + self._validate_non_empty_string(data.get("hwid"), "hwid", validation) + await self._validate_remote_user(user_uuid, validation) + return validation + if match := _INTERNAL_SQUAD_BULK_RE.match(endpoint): + squad_uuid = match.group("squad_uuid") + self._validate_non_empty_string(squad_uuid, "squad uuid", validation) + user_uuids = self._validate_string_list(data.get("userUuids"), "userUuids", validation) + if not user_uuids: + user_uuids = self._validate_string_list(data.get("users"), "users", validation) + await self._validate_remote_squads([squad_uuid], validation) + for user_uuid in user_uuids: + await self._validate_remote_user(user_uuid, validation) + return validation + + if payload is None: + return validation + self._validate_json_serializable(payload, validation) + return validation + + async def _validate_create_user_payload( + self, + payload: Dict[str, Any], + validation: _DryRunValidation, + ) -> None: + username = self._validate_non_empty_string(payload.get("username"), "username", validation) + if username and ( + not (3 <= len(username) <= 36) or not re.match(r"^[A-Za-z0-9_-]+$", username) + ): + validation.add("username must be 3-36 chars and contain only A-Z, 0-9, _ or -.") + self._validate_user_mutation_payload(payload, validation, require_uuid=False) + await self._validate_remote_squads( + self._validate_string_list( + payload.get("activeInternalSquads"), + "activeInternalSquads", + validation, + required=False, + ), + validation, + ) + if not bool(getattr(self.settings, "PANEL_DRY_RUN_SYNTHETIC_CREATE", True)): + validation.add("PANEL_DRY_RUN_SYNTHETIC_CREATE is disabled.") + if self._remote_validation_enabled and username: + await self._validate_create_uniqueness(payload, validation) + + async def _validate_update_user_payload( + self, + payload: Dict[str, Any], + validation: _DryRunValidation, + ) -> None: + user_uuid = self._validate_non_empty_string(payload.get("uuid"), "uuid", validation) + self._validate_user_mutation_payload(payload, validation, require_uuid=True) + await self._validate_remote_user(user_uuid, validation) + await self._validate_remote_squads( + self._validate_string_list( + payload.get("activeInternalSquads"), + "activeInternalSquads", + validation, + required=False, + ), + validation, + ) + + def _validate_user_mutation_payload( + self, + payload: Dict[str, Any], + validation: _DryRunValidation, + *, + require_uuid: bool, + ) -> None: + if require_uuid: + self._validate_non_empty_string(payload.get("uuid"), "uuid", validation) + if "expireAt" in payload: + self._validate_datetime(payload.get("expireAt"), "expireAt", validation) + if "trafficLimitBytes" in payload: + self._validate_non_negative_int( + payload.get("trafficLimitBytes"), + "trafficLimitBytes", + validation, + ) + if "trafficLimitStrategy" in payload: + strategy = self._validate_non_empty_string( + payload.get("trafficLimitStrategy"), + "trafficLimitStrategy", + validation, + ) + if strategy and strategy.upper() not in _KNOWN_TRAFFIC_STRATEGIES: + validation.add(f"trafficLimitStrategy {strategy!r} is not supported.") + if "hwidDeviceLimit" in payload: + self._validate_non_negative_int( + payload.get("hwidDeviceLimit"), + "hwidDeviceLimit", + validation, + ) + if "telegramId" in payload: + self._validate_positive_int(payload.get("telegramId"), "telegramId", validation) + if "email" in payload and payload.get("email") is not None: + self._validate_non_empty_string(payload.get("email"), "email", validation) + if "externalSquadUuid" in payload and payload.get("externalSquadUuid") is not None: + self._validate_non_empty_string( + payload.get("externalSquadUuid"), + "externalSquadUuid", + validation, + ) + self._validate_json_serializable(payload, validation) + + @property + def _remote_validation_enabled(self) -> bool: + return bool(getattr(self.settings, "PANEL_DRY_RUN_VALIDATE_REMOTE", True)) + + async def _validate_remote_user( + self, + user_uuid: Optional[str], + validation: _DryRunValidation, + ) -> Optional[Dict[str, Any]]: + if not user_uuid or not self._remote_validation_enabled: + return self._synthetic_users.get(str(user_uuid or "")) + user = self._synthetic_users.get(str(user_uuid)) + if user: + return user + try: + user = await super().get_user_by_uuid(str(user_uuid), log_response=False) + except Exception as exc: + validation.add(f"failed to validate panel user {user_uuid}: {type(exc).__name__}") + return None + if not user: + validation.add(f"panel user {user_uuid} was not found.") + return user + + async def _validate_remote_squads( + self, + squad_uuids: List[str], + validation: _DryRunValidation, + ) -> None: + if not squad_uuids or not self._remote_validation_enabled: + return + try: + squads = await super().get_internal_squads() + except Exception as exc: + validation.add(f"failed to validate panel squads: {type(exc).__name__}") + return + if squads is None: + validation.add("failed to validate panel squads: empty panel response.") + return + known = { + str(squad.get("uuid") or squad.get("id") or "").strip() + for squad in squads + if isinstance(squad, dict) + } + missing = sorted({squad_uuid for squad_uuid in squad_uuids if squad_uuid not in known}) + if missing: + validation.add(f"panel squads were not found: {', '.join(missing)}.") + + async def _validate_create_uniqueness( + self, + payload: Dict[str, Any], + validation: _DryRunValidation, + ) -> None: + checks = ( + ("username", "username", payload.get("username")), + ("telegramId", "telegram_id", payload.get("telegramId")), + ("email", "email", payload.get("email")), + ) + for label, argument_name, value in checks: + if value in (None, ""): + continue + try: + users = await super().get_users_by_filter(**{argument_name: value}) + except Exception as exc: + validation.add(f"failed to validate unique {label}: {type(exc).__name__}") + continue + if users: + validation.add(f"panel user with {label} {value!r} already exists.") + + @staticmethod + def _validate_non_empty_string( + value: Any, + name: str, + validation: _DryRunValidation, + ) -> Optional[str]: + if not isinstance(value, str) or not value.strip(): + validation.add(f"{name} must be a non-empty string.") + return None + return value.strip() + + @staticmethod + def _validate_string_list( + value: Any, + name: str, + validation: _DryRunValidation, + *, + required: bool = True, + ) -> List[str]: + if value is None: + if required: + validation.add(f"{name} must be a list of strings.") + return [] + if not isinstance(value, list): + validation.add(f"{name} must be a list of strings.") + return [] + result = [] + for item in value: + if not isinstance(item, str) or not item.strip(): + validation.add(f"{name} contains an empty or non-string value.") + continue + result.append(item.strip()) + return result + + @staticmethod + def _validate_non_negative_int( + value: Any, + name: str, + validation: _DryRunValidation, + ) -> None: + try: + parsed = int(value) + except (TypeError, ValueError): + validation.add(f"{name} must be an integer.") + return + if parsed < 0: + validation.add(f"{name} must be >= 0.") + + @staticmethod + def _validate_positive_int(value: Any, name: str, validation: _DryRunValidation) -> None: + try: + parsed = int(value) + except (TypeError, ValueError): + validation.add(f"{name} must be an integer.") + return + if parsed <= 0: + validation.add(f"{name} must be > 0.") + + @staticmethod + def _validate_datetime(value: Any, name: str, validation: _DryRunValidation) -> None: + if not isinstance(value, str) or not value.strip(): + validation.add(f"{name} must be an ISO datetime string.") + return + try: + datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + validation.add(f"{name} must be a valid ISO datetime string.") + + @staticmethod + def _validate_json_serializable(value: Any, validation: _DryRunValidation) -> None: + try: + json.dumps(value, default=str) + except (TypeError, ValueError): + validation.add("payload must be JSON serializable.") + + async def _dry_run_response( + self, + method: str, + endpoint: str, + payload: Any, + ) -> Dict[str, Any]: + data = payload if isinstance(payload, dict) else {} + if method == "POST" and endpoint == "/users": + return self._dry_run_create_user_response(data) + if method == "PATCH" and endpoint == "/users": + return await self._dry_run_patch_user_response(data) + if method == "POST" and (match := _USER_ACTION_RE.match(endpoint)): + return self._dry_run_user_action_response( + match.group("user_uuid"), + match.group("action"), + ) + if method == "DELETE" and endpoint.startswith("/users/"): + return {"uuid": endpoint.removeprefix("/users/"), "deleted": True, "dryRun": True} + if method == "POST" and endpoint == "/hwid/devices/delete": + return {"userUuid": data.get("userUuid"), "hwid": data.get("hwid"), "dryRun": True} + if match := _INTERNAL_SQUAD_BULK_RE.match(endpoint): + return { + "squadUuid": match.group("squad_uuid"), + "action": match.group("action"), + "users": data.get("userUuids") or data.get("users") or [], + "dryRun": True, + } + return {"dryRun": True} + + def _dry_run_create_user_response(self, payload: Dict[str, Any]) -> Dict[str, Any]: + identity = ":".join( + str(payload.get(key) or "") for key in ("username", "telegramId", "email") + ) + user_uuid = str(uuid.uuid5(uuid.NAMESPACE_URL, f"remnawave-minishop:dry-run:{identity}")) + short_uuid = user_uuid.split("-")[0] + response = { + **payload, + "uuid": user_uuid, + "shortUuid": short_uuid, + "subscriptionUuid": short_uuid, + "subscriptionUrl": self._subscription_url(short_uuid), + "dryRun": True, + } + self._synthetic_users[user_uuid] = response + return response + + async def _dry_run_patch_user_response(self, payload: Dict[str, Any]) -> Dict[str, Any]: + user_uuid = str(payload.get("uuid") or "") + existing = self._synthetic_users.get(user_uuid) + if not existing and self._remote_validation_enabled: + try: + existing = await super().get_user_by_uuid(user_uuid, log_response=False) + except Exception: + existing = None + response = {**(existing or {"uuid": user_uuid}), **payload, "dryRun": True} + if user_uuid in self._synthetic_users: + self._synthetic_users[user_uuid] = response + return response + + @staticmethod + def _dry_run_user_action_response(user_uuid: str, action: str) -> Dict[str, Any]: + response: Dict[str, Any] = {"uuid": user_uuid, "action": action, "dryRun": True} + if action == "enable": + response["status"] = "ACTIVE" + elif action == "disable": + response["status"] = "DISABLED" + elif action == "reset-traffic": + response["userTraffic"] = {"usedTrafficBytes": 0} + return response + + def _subscription_url(self, short_uuid: str) -> Optional[str]: + if not self.settings.PANEL_API_URL: + return None + return f"{self.settings.PANEL_API_URL.rstrip('/')}/sub/{short_uuid}" diff --git a/backend/config/settings.py b/backend/config/settings.py index b256771..a349829 100644 --- a/backend/config/settings.py +++ b/backend/config/settings.py @@ -287,6 +287,30 @@ class Settings(BaseSettings): description="Allow legacy referral links like ref_ to continue working. Defaults to True when unset.", # noqa: E501 ) + APP_RUNTIME_MODE: str = Field( + default="production", + description="Runtime profile: production, development, staging or test.", + ) + PANEL_WRITE_MODE: str = Field( + default="auto", + description=( + "Panel write behavior: auto uses dry-run in development/test runtimes, " + "live always writes to Remnawave, dry_run validates and logs mutations only." + ), + ) + PANEL_DRY_RUN_VALIDATE_REMOTE: bool = Field( + default=True, + description=( + "When panel dry-run is enabled, validate referenced users and squads " + "via live GET requests." + ), + ) + PANEL_DRY_RUN_SYNTHETIC_CREATE: bool = Field( + default=True, + description=( + "When panel dry-run is enabled, return synthetic users for create-user attempts." + ), + ) PANEL_API_URL: Optional[str] = None PANEL_API_KEY: Optional[str] = None USER_TRAFFIC_LIMIT_GB: Optional[float] = Field(default=0.0) @@ -560,6 +584,17 @@ class Settings(BaseSettings): ids = self.ADMIN_IDS return ids[0] if ids else None + @computed_field + @property + def panel_dry_run_enabled(self) -> bool: + mode = str(self.PANEL_WRITE_MODE or "auto").strip().lower().replace("-", "_") + if mode == "dry_run": + return True + if mode == "live": + return False + runtime = str(self.APP_RUNTIME_MODE or "production").strip().lower() + return runtime in {"dev", "development", "local", "test", "testing"} + @computed_field @property def trial_traffic_limit_bytes(self) -> int: @@ -1025,6 +1060,28 @@ class Settings(BaseSettings): return None return v + @field_validator("APP_RUNTIME_MODE", mode="before") + @classmethod + def normalize_app_runtime_mode(cls, v): + value = str(v or "production").strip().lower().replace("-", "_") + if not value: + return "production" + aliases = { + "prod": "production", + "dev": "development", + "local_dev": "development", + "testing": "test", + } + return aliases.get(value, value) + + @field_validator("PANEL_WRITE_MODE", mode="before") + @classmethod + def validate_panel_write_mode(cls, v): + value = str(v or "auto").strip().lower().replace("-", "_") + if value not in {"auto", "live", "dry_run"}: + raise ValueError("PANEL_WRITE_MODE must be one of: auto, live, dry_run") + return value + # Notification types LOG_NEW_USERS: bool = Field( default=True, description="Send notifications for new user registrations" @@ -1066,6 +1123,11 @@ def get_settings() -> Settings: logging.warning( "CRITICAL: PANEL_API_URL is not set. Panel integration will not work." ) + if _settings_instance.panel_dry_run_enabled: + logging.warning( + "PANEL_WRITE_MODE dry-run is enabled: Remnawave write requests will be " + "validated and logged without changing panel users." + ) if not os.getenv("WEBAPP_SESSION_SECRET"): logging.warning( "WEBAPP_SESSION_SECRET is not set. A generated secret will be used for this process only." # noqa: E501 diff --git a/tests/test_build_services_wiring.py b/tests/test_build_services_wiring.py index 80bebeb..9b47b82 100644 --- a/tests/test_build_services_wiring.py +++ b/tests/test_build_services_wiring.py @@ -22,6 +22,7 @@ from unittest.mock import MagicMock from bot.app.factories.build_services import build_core_services from bot.payment_providers.yookassa import YooKassaService +from bot.services.panel_dry_run_api_service import PanelDryRunApiService from bot.services.panel_webhook_service import PanelWebhookService from bot.services.subscription_service import SubscriptionService from config.settings import Settings @@ -125,6 +126,23 @@ class BuildServicesWiringTests(unittest.TestCase): self.assertIsInstance(panel_webhook, PanelWebhookService) self.assertIs(getattr(panel_webhook, "subscription_service", None), subscription) + def test_development_runtime_wires_panel_dry_run_service(self): + with tempfile.TemporaryDirectory() as tmpdir: + settings = _make_settings(tmpdir, APP_RUNTIME_MODE="development") + services = build_core_services( + settings=settings, + bot=MagicMock(), + async_session_factory=MagicMock(), + i18n=MagicMock(), + bot_username_for_default_return="testbot", + ) + + self.assertIsInstance(services["panel_service"], PanelDryRunApiService) + self.assertIs( + services["subscription_service"].panel_service, + services["panel_service"], + ) + def test_factory_returns_every_documented_service(self): """Guards against silently dropping a service from the bundle. The web layer reads these keys off ``request.app`` directly — a missing diff --git a/tests/test_panel_dry_run_api_service.py b/tests/test_panel_dry_run_api_service.py new file mode 100644 index 0000000..daceaaf --- /dev/null +++ b/tests/test_panel_dry_run_api_service.py @@ -0,0 +1,100 @@ +import unittest +from types import SimpleNamespace +from unittest.mock import AsyncMock + +from bot.services.panel_dry_run_api_service import PanelDryRunApiService + + +def _settings(**overrides): + values = { + "PANEL_API_URL": "https://panel.example.test/api", + "PANEL_API_KEY": "panel-key", + "PANEL_DRY_RUN_VALIDATE_REMOTE": False, + "PANEL_DRY_RUN_SYNTHETIC_CREATE": True, + "PANEL_USER_CACHE_TTL_SECONDS": 0, + "PANEL_DEVICES_CACHE_TTL_SECONDS": 0, + "PANEL_ALL_USERS_CACHE_TTL_SECONDS": 0, + "PANEL_ALL_USERS_PAGE_SIZE": 1000, + "REDIS_KEY_PREFIX": "tests", + "USER_HWID_DEVICE_LIMIT": None, + } + values.update(overrides) + return SimpleNamespace(**values) + + +class PanelDryRunApiServiceTests(unittest.IsolatedAsyncioTestCase): + async def test_update_user_details_returns_synthetic_success_without_http_write(self): + service = PanelDryRunApiService(_settings()) + service._request_once = AsyncMock() + + result = await service.update_user_details_on_panel( + "user-uuid", + {"trafficLimitBytes": 1024, "trafficLimitStrategy": "NO_RESET"}, + ) + + self.assertEqual(result["uuid"], "user-uuid") + self.assertTrue(result["dryRun"]) + self.assertEqual(result["trafficLimitBytes"], 1024) + service._request_once.assert_not_awaited() + + async def test_update_user_details_rejects_invalid_payload(self): + service = PanelDryRunApiService(_settings()) + service._request_once = AsyncMock() + + result = await service.update_user_details_on_panel( + "user-uuid", + {"trafficLimitBytes": -1, "trafficLimitStrategy": "NO_RESET"}, + ) + + self.assertIsNone(result) + service._request_once.assert_not_awaited() + + async def test_remote_validation_blocks_missing_panel_user(self): + service = PanelDryRunApiService(_settings(PANEL_DRY_RUN_VALIDATE_REMOTE=True)) + service._request_once = AsyncMock( + return_value={"error": True, "status_code": 404, "details": {"errorCode": "A062"}} + ) + + result = await service.update_user_details_on_panel( + "missing-user", + {"trafficLimitBytes": 1024, "trafficLimitStrategy": "NO_RESET"}, + ) + + self.assertIsNone(result) + service._request_once.assert_awaited_once() + + async def test_create_panel_user_returns_synthetic_user(self): + service = PanelDryRunApiService(_settings()) + service._request_once = AsyncMock() + + result = await service.create_panel_user( + username_on_panel="tg_42", + telegram_id=42, + default_traffic_limit_bytes=2048, + default_traffic_limit_strategy="NO_RESET", + ) + + panel_user = result["response"] + self.assertTrue(panel_user["dryRun"]) + self.assertEqual(panel_user["username"], "tg_42") + self.assertIn("uuid", panel_user) + self.assertTrue( + panel_user["subscriptionUrl"].startswith("https://panel.example.test/api/sub/") + ) + self.assertTrue(panel_user["subscriptionUrl"].endswith(panel_user["shortUuid"])) + service._request_once.assert_not_awaited() + + async def test_happ_encrypt_post_stays_live(self): + service = PanelDryRunApiService(_settings()) + service._request_once = AsyncMock( + return_value={"response": {"encryptedLink": "happ://crypt4/test"}} + ) + + result = await service.encrypt_happ_link("https://panel.example.test/sub/abc") + + self.assertEqual(result, "happ://crypt4/test") + service._request_once.assert_awaited_once() + + +if __name__ == "__main__": # pragma: no cover + unittest.main() diff --git a/tests/test_settings.py b/tests/test_settings.py index 19fa416..57c8d17 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -42,6 +42,51 @@ class SettingsTests(unittest.TestCase): self.assertEqual(settings.WEBAPP_TITLE, "/minishop") + def test_panel_write_mode_defaults_to_live_in_production(self): + settings = Settings( + _env_file=None, + BOT_TOKEN="token", + POSTGRES_USER="app_user", + POSTGRES_PASSWORD="app_password", + ) + + self.assertEqual(settings.APP_RUNTIME_MODE, "production") + self.assertEqual(settings.PANEL_WRITE_MODE, "auto") + self.assertFalse(settings.panel_dry_run_enabled) + + def test_development_runtime_enables_panel_dry_run_in_auto_mode(self): + settings = Settings( + _env_file=None, + BOT_TOKEN="token", + POSTGRES_USER="app_user", + POSTGRES_PASSWORD="app_password", + APP_RUNTIME_MODE="development", + ) + + self.assertTrue(settings.panel_dry_run_enabled) + + def test_panel_write_mode_live_overrides_development_runtime(self): + settings = Settings( + _env_file=None, + BOT_TOKEN="token", + POSTGRES_USER="app_user", + POSTGRES_PASSWORD="app_password", + APP_RUNTIME_MODE="development", + PANEL_WRITE_MODE="live", + ) + + self.assertFalse(settings.panel_dry_run_enabled) + + def test_panel_write_mode_rejects_unknown_value(self): + with self.assertRaises(ValidationError): + Settings( + _env_file=None, + BOT_TOKEN="token", + POSTGRES_USER="app_user", + POSTGRES_PASSWORD="app_password", + PANEL_WRITE_MODE="danger", + ) + def test_legacy_subscription_prices_have_defaults(self): settings = Settings( _env_file=None,