feat: Update Remnawave SDK to version 2.4.4 with new subscription page management features

- Bump version to 2.4.4 and update description in pyproject.toml
- Refactor RemnawaveSDK to include SubscriptionPageConfigController
- Introduce new subscription page management endpoints:
  - Get, create, update, delete, reorder, and clone subscription page configs
- Remove deprecated NodesUsageHistoryController and UsersStatsController
- Add new bandwidth stats models and endpoints for legacy and new stats
- Enhance tests for bandwidth stats and subscription page management
- Ensure backward compatibility with legacy endpoints while introducing new stats models
This commit is contained in:
Artem 2025-12-25 22:30:53 +01:00
parent 3eaad58131
commit ba1a221593
No known key found for this signature in database
GPG key ID: 833485276B7902CE
16 changed files with 847 additions and 152 deletions

View file

@ -21,19 +21,6 @@
A Python SDK client for interacting with the **[Remnawave API](https://remna.st)**.
This library simplifies working with the API by providing convenient controllers, Pydantic models for requests and responses, and fast serialization with `orjson`.
**🎉 Version 2.0.0** brings full compatibility with the latest Remnawave backend API, including new endpoints, improved response wrappers, and enhanced type safety.
## ✨ Key Features
- **Full v2.0.0 API compatibility**: Updated for latest Remnawave backend features
- **New controllers**: ConfigProfiles, InternalSquads, InfraBilling, NodesUsageHistory
- **Enhanced models**: OpenAPI-compliant response wrappers with improved field mappings
- **Controller-based design**: Split functionality into separate controllers for flexibility. Use only what you need!
- **Pydantic models**: Strongly-typed requests and responses for better reliability.
- **Fast serialization**: Powered by `orjson` for efficient JSON handling.
- **Modular usage**: Import individual controllers or the full SDK as needed.
- **Backward compatibility**: Legacy aliases maintained for smooth migration.
## 📦 Installation
### New Package (Recommended)
@ -63,7 +50,8 @@ pip install git+https://github.com/remnawave/python-sdk.git@development
| Contract Version | Remnawave Panel Version |
| ---------------- | ----------------------- |
| 2.3.2 | >=2.3.0 |
| 2.4.4 | >=2.4.0 |
| 2.3.2 | >=2.3.0, <2.4.0 |
| 2.3.0 | >=2.3.0, <2.3.2 |
| 2.2.6 | ==2.2.6 |
| 2.2.3 | >=2.2.13 |
@ -127,14 +115,6 @@ if __name__ == "__main__":
---
## 🧪 Running Tests
To run the test suite, use Poetry:
```bash
poetry run pytest
```
## ❤️ About
This SDK was originally developed by [@kesevone](https://github.com/kesevone) for integration with Remnawave's API.

View file

@ -1,7 +1,7 @@
[project]
name = "remnawave"
version = "2.3.2rc3"
description = "A Python SDK for interacting with the Remnawave API v2.3.2."
version = "2.4.4"
description = "A Python SDK for interacting with the Remnawave API v2.4.4."
authors = [
{name = "Artem",email = "dev@forestsnet.com"}
]

View file

@ -17,8 +17,6 @@ from remnawave.controllers import (
InternalSquadsController,
KeygenController,
NodesController,
NodesUsageHistoryController,
NodesUserUsageHistoryController,
SubscriptionController,
SubscriptionsController,
SubscriptionsSettingsController,
@ -26,7 +24,6 @@ from remnawave.controllers import (
SystemController,
UsersBulkActionsController,
UsersController,
UsersStatsController,
WebhookUtility,
XrayConfigController,
SubscriptionRequestHistoryController,
@ -34,7 +31,7 @@ from remnawave.controllers import (
ExternalSquadsController,
SnippetsController,
RemnawaveSettingsController,
# WebhookUtility is not a controller, but it's included in the controllers module for convenience
SubscriptionPageConfigController,
)
@ -84,8 +81,6 @@ class RemnawaveSDK:
self.internal_squads = InternalSquadsController(self._client)
self.keygen = KeygenController(self._client)
self.nodes = NodesController(self._client)
self.nodes_usage_history = NodesUsageHistoryController(self._client)
self.nodes_user_usage_history = NodesUserUsageHistoryController(self._client)
self.subscription = SubscriptionController(self._client)
self.subscriptions = SubscriptionsController(self._client)
self.subscriptions_settings = SubscriptionsSettingsController(self._client)
@ -94,13 +89,13 @@ class RemnawaveSDK:
self.system = SystemController(self._client)
self.users = UsersController(self._client)
self.users_bulk_actions = UsersBulkActionsController(self._client)
self.users_stats = UsersStatsController(self._client)
self.webhook_utility = WebhookUtility()
self.xray_config = XrayConfigController(self._client)
self.passkeys = PasskeysController(self._client)
self.external_squads = ExternalSquadsController(self._client)
self.snippets = SnippetsController(self._client)
self.remnawave_settings = RemnawaveSettingsController(self._client)
self.subscription_page_config = SubscriptionPageConfigController(self._client)
def _validate_params(self) -> None:
if self._client is None:

View file

@ -11,7 +11,6 @@ from .infra_billing import InfraBillingController
from .internal_squads import InternalSquadsController
from .keygen import KeygenController
from .nodes import NodesController
from .nodes_usage_history import NodesUsageHistoryController, NodesUserUsageHistoryController
from .subscription import SubscriptionController
from .subscriptions_controller import SubscriptionsController
from .subscriptions_settings import SubscriptionsSettingsController
@ -19,7 +18,6 @@ from .subscriptions_template import SubscriptionsTemplateController
from .system import SystemController
from .users import UsersController
from .users_bulk_actions import UsersBulkActionsController
from .users_stats import UsersStatsController
from .webhooks import WebhookUtility
from .xray_config import XrayConfigController
from .subscriptions_request import SubscriptionRequestHistoryController
@ -27,7 +25,7 @@ from .passkeys import PasskeysController
from .external_squads import ExternalSquadsController
from .snippets import SnippetsController
from .remnawave_settings import RemnawaveSettingsController
from .subscription_page import SubscriptionPageConfigController
__all__ = [
"APITokensManagementController",
@ -43,8 +41,6 @@ __all__ = [
"InternalSquadsController",
"KeygenController",
"NodesController",
"NodesUsageHistoryController",
"NodesUserUsageHistoryController",
"SubscriptionController",
"SubscriptionsController",
"SubscriptionsSettingsController",
@ -52,7 +48,6 @@ __all__ = [
"SystemController",
"UsersController",
"UsersBulkActionsController",
"UsersStatsController",
"WebhookUtility",
"XrayConfigController",
"SubscriptionRequestHistoryController",
@ -60,4 +55,5 @@ __all__ = [
"ExternalSquadsController",
"SnippetsController",
"RemnawaveSettingsController",
"SubscriptionPageConfigController"
]

View file

@ -1,25 +1,102 @@
from typing import Annotated
from rapid_api_client import Query
from rapid_api_client import Path, Query
from remnawave.models import GetNodesUsageByRangeResponseDto, GetNodesRealtimeUsageResponseDto
from remnawave.models.bandwidthstats import (
GetLegacyStatsNodesUsersUsageResponseDto,
GetLegacyStatsUserUsageResponseDto,
GetNodeUserUsageByRangeResponseDto,
GetNodesRealtimeUsageResponseDto,
GetNodesUsageByRangeResponseDto,
GetStatsNodesRealtimeUsageResponseDto,
GetStatsNodesUsageResponseDto,
GetStatsNodeUsersUsageResponseDto,
GetStatsUserUsageResponseDto,
GetUserUsageByRangeResponseDto,
)
from remnawave.rapid import BaseController, get
class BandWidthStatsController(BaseController):
@get("/nodes/usage/range", response_class=GetNodesUsageByRangeResponseDto)
async def get_nodes_usage_by_range(
# ============ Legacy Endpoints (Deprecated) ============
@get("/bandwidth-stats/users/{user_uuid}/legacy-old", response_class=GetUserUsageByRangeResponseDto)
async def get_user_usage_legacy_old(
self,
start: Annotated[str, Query(description="Start date in ISO format")],
end: Annotated[str, Query(description="End date in ISO format")],
) -> GetNodesUsageByRangeResponseDto:
"""Get Nodes Usage By Range"""
user_uuid: Annotated[str, Path(description="UUID of the user", alias="userUuid")],
start: Annotated[str, Query(description="Start date")],
end: Annotated[str, Query(description="End date")],
) -> GetUserUsageByRangeResponseDto:
"""Get User Usage by Range (Legacy - Deprecated)"""
...
@get("/nodes/usage/realtime", response_class=GetNodesRealtimeUsageResponseDto)
async def get_nodes_usage_realtime(
@get("/bandwidth-stats/nodes/{node_uuid}/users/legacy-old", response_class=GetNodeUserUsageByRangeResponseDto)
async def get_node_user_usage_legacy_old(
self,
) -> GetNodesRealtimeUsageResponseDto:
"""Get Nodes Usage Realtime"""
node_uuid: Annotated[str, Path(description="UUID of the node", alias="nodeUuid")],
start: Annotated[str, Query(description="Start date")],
end: Annotated[str, Query(description="End date")],
) -> GetNodeUserUsageByRangeResponseDto:
"""Get Node User Usage by Range and Node UUID (Legacy - Deprecated)"""
...
# ============ New Stats Endpoints ============
@get("/bandwidth-stats/nodes/realtime", response_class=GetStatsNodesRealtimeUsageResponseDto)
async def get_nodes_realtime_usage(
self,
) -> GetStatsNodesRealtimeUsageResponseDto:
"""Get Nodes Realtime Usage"""
...
@get("/bandwidth-stats/nodes/{uuid}/users/legacy", response_class=GetLegacyStatsNodesUsersUsageResponseDto)
async def get_node_users_usage_legacy_stats(
self,
uuid: Annotated[str, Path(description="UUID of the node")],
start: Annotated[str, Query(description="Start date")],
end: Annotated[str, Query(description="End date")],
) -> GetLegacyStatsNodesUsersUsageResponseDto:
"""Get Node Users Usage by Range and Node UUID (Legacy Stats)"""
...
@get("/bandwidth-stats/nodes/{uuid}/users", response_class=GetStatsNodeUsersUsageResponseDto)
async def get_stats_node_users_usage(
self,
uuid: Annotated[str, Path(description="UUID of the node")],
top_users_limit: Annotated[int, Query(description="Limit of top users to return", alias="topUsersLimit")],
start: Annotated[str, Query(description="Start date")],
end: Annotated[str, Query(description="End date")],
) -> GetStatsNodeUsersUsageResponseDto:
"""Get Node Users Usage by Node UUID"""
...
@get("/bandwidth-stats/users/{uuid}", response_class=GetStatsUserUsageResponseDto)
async def get_stats_user_usage(
self,
uuid: Annotated[str, Path(description="UUID of the user")],
top_nodes_limit: Annotated[int, Query(description="Limit of top nodes to return", alias="topNodesLimit")],
start: Annotated[str, Query(description="Start date")],
end: Annotated[str, Query(description="End date")],
) -> GetStatsUserUsageResponseDto:
"""Get User Usage by Range"""
...
@get("/bandwidth-stats/nodes", response_class=GetStatsNodesUsageResponseDto)
async def get_stats_nodes_usage(
self,
top_nodes_limit: Annotated[int, Query(description="Limit of top nodes to return", alias="topNodesLimit")],
start: Annotated[str, Query(description="Start date")],
end: Annotated[str, Query(description="End date")],
) -> GetStatsNodesUsageResponseDto:
"""Get Nodes Usage by Range"""
...
@get("/bandwidth-stats/users/{uuid}/legacy", response_class=GetLegacyStatsUserUsageResponseDto)
async def get_user_usage_legacy_stats(
self,
uuid: Annotated[str, Path(description="UUID of the user")],
start: Annotated[str, Query(description="Start date")],
end: Annotated[str, Query(description="End date")],
) -> GetLegacyStatsUserUsageResponseDto:
"""Get User Usage by Range (Legacy Stats)"""
...

View file

@ -7,26 +7,3 @@ from remnawave.models import (
GetNodesUsageByRangeResponseDto,
)
from remnawave.rapid import BaseController, get
class NodesUsageHistoryController(BaseController):
@get("/nodes/usage/range", response_class=GetNodesUsageByRangeResponseDto)
async def get_nodes_usage_by_range(
self,
start: Annotated[str, Query(description="Start date", format="date-time")],
end: Annotated[str, Query(description="End date", format="date-time")],
) -> GetNodesUsageByRangeResponseDto:
"""Get nodes usage by range"""
...
class NodesUserUsageHistoryController(BaseController):
@get("/nodes/usage/{uuid}/users/range", response_class=GetNodeUserUsageByRangeResponseDto)
async def get_node_user_usage_by_range(
self,
uuid: Annotated[str, Path(description="UUID of the node")],
start: Annotated[str, Query(description="Start date", format="date-time")],
end: Annotated[str, Query(description="End date", format="date-time")],
) -> GetNodeUserUsageByRangeResponseDto:
"""Get nodes user usage by range"""
...

View file

@ -0,0 +1,73 @@
from typing import Annotated
from rapid_api_client.annotations import Path, PydanticBody
from remnawave.models import (
CloneSubscriptionPageConfigRequestDto,
CloneSubscriptionPageConfigResponseDto,
CreateSubscriptionPageConfigRequestDto,
CreateSubscriptionPageConfigResponseDto,
DeleteSubscriptionPageConfigResponseDto,
GetSubscriptionPageConfigResponseDto,
GetSubscriptionPageConfigsResponseDto,
ReorderSubscriptionPageConfigsRequestDto,
ReorderSubscriptionPageConfigsResponseDto,
UpdateSubscriptionPageConfigRequestDto,
UpdateSubscriptionPageConfigResponseDto,
)
from remnawave.rapid import BaseController, delete, get, patch, post
class SubscriptionPageConfigController(BaseController):
@get("/subscription-page-configs", response_class=GetSubscriptionPageConfigsResponseDto)
async def get_all_configs(self) -> GetSubscriptionPageConfigsResponseDto:
"""Get all subscription page configs"""
...
@post("/subscription-page-configs", response_class=CreateSubscriptionPageConfigResponseDto)
async def create_config(
self,
body: Annotated[CreateSubscriptionPageConfigRequestDto, PydanticBody()],
) -> CreateSubscriptionPageConfigResponseDto:
"""Create subscription page config"""
...
@patch("/subscription-page-configs", response_class=UpdateSubscriptionPageConfigResponseDto)
async def update_config(
self,
body: Annotated[UpdateSubscriptionPageConfigRequestDto, PydanticBody()],
) -> UpdateSubscriptionPageConfigResponseDto:
"""Update subscription page config"""
...
@get("/subscription-page-configs/{uuid}", response_class=GetSubscriptionPageConfigResponseDto)
async def get_config_by_uuid(
self,
uuid: Annotated[str, Path(description="Subscription page config UUID")],
) -> GetSubscriptionPageConfigResponseDto:
"""Get subscription page config by uuid"""
...
@delete("/subscription-page-configs/{uuid}", response_class=DeleteSubscriptionPageConfigResponseDto)
async def delete_config(
self,
uuid: Annotated[str, Path(description="Subscription page config UUID")],
) -> DeleteSubscriptionPageConfigResponseDto:
"""Delete subscription page config"""
...
@post("/subscription-page-configs/actions/reorder", response_class=ReorderSubscriptionPageConfigsResponseDto)
async def reorder_configs(
self,
body: Annotated[ReorderSubscriptionPageConfigsRequestDto, PydanticBody()],
) -> ReorderSubscriptionPageConfigsResponseDto:
"""Reorder subscription page configs"""
...
@post("/subscription-page-configs/actions/clone", response_class=CloneSubscriptionPageConfigResponseDto)
async def clone_config(
self,
body: Annotated[CloneSubscriptionPageConfigRequestDto, PydanticBody()],
) -> CloneSubscriptionPageConfigResponseDto:
"""Clone subscription page config"""
...

View file

@ -5,17 +5,3 @@ from rapid_api_client import Path, Query
from remnawave.models import GetUserUsageByRangeResponseDto
from remnawave.rapid import BaseController, get
class UsersStatsController(BaseController):
@get(
"/users/stats/usage/{uuid}/range",
response_class=GetUserUsageByRangeResponseDto,
)
async def get_user_usage_by_range(
self,
uuid: Annotated[str, Path(description="UUID of the user")],
start: Annotated[str, Query(description="Start date in ISO format")],
end: Annotated[str, Query(description="End date in ISO format")],
) -> GetUserUsageByRangeResponseDto:
"""Get User Usage By Range"""
...

View file

@ -31,6 +31,23 @@ from .bandwidthstats import (
NodeUsageResponseDto,
NodesRealtimeUsageResponseDto, # Legacy alias
NodesUsageResponseDto, # Legacy alias
GetLegacyStatsUserUsageResponseDto,
GetLegacyStatsNodesUsersUsageResponseDto,
GetStatsNodesRealtimeUsageResponseDto,
GetStatsNodesUsageResponseDto,
GetStatsNodeUsersUsageResponseDto,
GetStatsUserUsageResponseDto,
# Data Models
LegacyUserUsageItem,
LegacyNodeUserUsageItem,
NodeRealtimeUsageItem,
TopNodeItem,
TopUserItem,
NodeSeriesItem,
StatsNodesUsageData,
StatsNodeUsersUsageData,
StatsUserUsageData,
)
from .config_profiles import (
ConfigProfileDto,
@ -409,6 +426,21 @@ from .remnawave_settings import (
UpdateRemnawaveSettingsResponseDto,
YandexOAuth2Settings,
)
from .subscription_page import (
CloneSubscriptionPageConfigRequestDto,
CloneSubscriptionPageConfigResponseDto,
CreateSubscriptionPageConfigRequestDto,
CreateSubscriptionPageConfigResponseDto,
DeleteSubscriptionPageConfigResponseDto,
GetSubscriptionPageConfigResponseDto,
GetSubscriptionPageConfigsResponseDto,
ReorderSubscriptionPageConfigItem,
ReorderSubscriptionPageConfigsRequestDto,
ReorderSubscriptionPageConfigsResponseDto,
SubscriptionPageConfigDto,
UpdateSubscriptionPageConfigRequestDto,
UpdateSubscriptionPageConfigResponseDto,
)
__all__ = [
# Auth models
@ -567,6 +599,21 @@ __all__ = [
"NodeUsageResponseDto",
"NodesRealtimeUsageResponseDto", # Legacy alias
"NodesUsageResponseDto", # Legacy alias
"GetLegacyStatsUserUsageResponseDto",
"GetLegacyStatsNodesUsersUsageResponseDto",
"GetStatsNodesRealtimeUsageResponseDto",
"GetStatsNodesUsageResponseDto",
"GetStatsNodeUsersUsageResponseDto",
"GetStatsUserUsageResponseDto",
"LegacyUserUsageItem",
"LegacyNodeUserUsageItem",
"NodeRealtimeUsageItem",
"TopNodeItem",
"TopUserItem",
"NodeSeriesItem",
"StatsNodesUsageData",
"StatsNodeUsersUsageData",
"StatsUserUsageData",
# API Tokens models
"CreateApiTokenRequestDto",
"CreateApiTokenResponseDto",
@ -801,4 +848,19 @@ __all__ = [
"UpdateRemnawaveSettingsRequestDto",
"UpdateRemnawaveSettingsResponseDto",
"YandexOAuth2Settings",
# Subscription page config models
"CloneSubscriptionPageConfigRequestDto",
"CloneSubscriptionPageConfigResponseDto",
"CreateSubscriptionPageConfigRequestDto",
"CreateSubscriptionPageConfigResponseDto",
"DeleteSubscriptionPageConfigResponseDto",
"GetSubscriptionPageConfigResponseDto",
"GetSubscriptionPageConfigsResponseDto",
"ReorderSubscriptionPageConfigItem",
"ReorderSubscriptionPageConfigsRequestDto",
"ReorderSubscriptionPageConfigsResponseDto",
"SubscriptionPageConfigDto",
"UpdateSubscriptionPageConfigRequestDto",
"UpdateSubscriptionPageConfigResponseDto",
]

View file

@ -1,11 +1,14 @@
import datetime
from datetime import datetime, date
from typing import List
from uuid import UUID
from pydantic import BaseModel, Field, RootModel
# ============ Legacy Models (Deprecated) ============
class NodeUsageResponseDto(BaseModel):
"""Deprecated: Old node usage model"""
node_uuid: UUID = Field(alias="nodeUuid")
node_name: str = Field(alias="nodeName")
total: int
@ -14,10 +17,11 @@ class NodeUsageResponseDto(BaseModel):
human_readable_total: str = Field(alias="humanReadableTotal")
human_readable_total_download: str = Field(alias="humanReadableTotalDownload")
human_readable_total_upload: str = Field(alias="humanReadableTotalUpload")
date: datetime.date
date: date
class NodesUsageResponseDto(RootModel[List[NodeUsageResponseDto]]):
"""Deprecated: Use GetStatsNodesUsageResponseDto instead"""
def __iter__(self):
return iter(self.root)
@ -25,15 +29,14 @@ class NodesUsageResponseDto(RootModel[List[NodeUsageResponseDto]]):
return self.root[item]
def __bool__(self):
"""Return True if list is not empty"""
return bool(self.root)
def __len__(self):
"""Return length of list"""
return len(self.root)
class GetNodesUsageByRangeResponseDto(RootModel[List[NodeUsageResponseDto]]):
"""Deprecated: Use GetStatsNodesUsageResponseDto instead"""
def __iter__(self):
return iter(self.root)
@ -41,15 +44,14 @@ class GetNodesUsageByRangeResponseDto(RootModel[List[NodeUsageResponseDto]]):
return self.root[item]
def __bool__(self):
"""Return True if list is not empty"""
return bool(self.root)
def __len__(self):
"""Return length of list"""
return len(self.root)
class NodeRealtimeUsageResponseDto(BaseModel):
"""Deprecated: Use NodeRealtimeUsageItem instead"""
node_uuid: UUID = Field(alias="nodeUuid")
node_name: str = Field(alias="nodeName")
country_code: str = Field(alias="countryCode")
@ -62,6 +64,7 @@ class NodeRealtimeUsageResponseDto(BaseModel):
class NodesRealtimeUsageResponseDto(RootModel[List[NodeRealtimeUsageResponseDto]]):
"""Deprecated: Use GetStatsNodesRealtimeUsageResponseDto instead"""
def __iter__(self):
return iter(self.root)
@ -69,15 +72,14 @@ class NodesRealtimeUsageResponseDto(RootModel[List[NodeRealtimeUsageResponseDto]
return self.root[item]
def __bool__(self):
"""Return True if list is not empty"""
return bool(self.root)
def __len__(self):
"""Return length of list"""
return len(self.root)
class GetNodesRealtimeUsageResponseDto(RootModel[List[NodeRealtimeUsageResponseDto]]):
"""Deprecated: Use GetStatsNodesRealtimeUsageResponseDto instead"""
def __iter__(self):
return iter(self.root)
@ -85,15 +87,14 @@ class GetNodesRealtimeUsageResponseDto(RootModel[List[NodeRealtimeUsageResponseD
return self.root[item]
def __bool__(self):
"""Return True if list is not empty"""
return bool(self.root)
def __len__(self):
"""Return length of list"""
return len(self.root)
class UserUsageByRangeItem(BaseModel):
"""Deprecated: Use LegacyUserUsageItem instead"""
user_uuid: UUID = Field(alias="userUuid")
node_uuid: UUID = Field(alias="nodeUuid")
node_name: str = Field(alias="nodeName")
@ -102,6 +103,7 @@ class UserUsageByRangeItem(BaseModel):
class GetUserUsageByRangeResponseDto(RootModel[List[UserUsageByRangeItem]]):
"""Deprecated: Use GetLegacyStatsUserUsageResponseDto instead"""
def __iter__(self):
return iter(self.root)
@ -109,15 +111,14 @@ class GetUserUsageByRangeResponseDto(RootModel[List[UserUsageByRangeItem]]):
return self.root[item]
def __bool__(self):
"""Return True if list is not empty"""
return bool(self.root)
def __len__(self):
"""Return length of list"""
return len(self.root)
class NodeUserUsageItem(BaseModel):
"""Deprecated: Use LegacyNodeUserUsageItem instead"""
user_uuid: UUID = Field(alias="userUuid")
username: str
node_uuid: UUID = Field(alias="nodeUuid")
@ -126,6 +127,7 @@ class NodeUserUsageItem(BaseModel):
class GetNodeUserUsageByRangeResponseDto(RootModel[List[NodeUserUsageItem]]):
"""Deprecated: Use GetLegacyStatsNodesUsersUsageResponseDto instead"""
def __iter__(self):
return iter(self.root)
@ -133,9 +135,160 @@ class GetNodeUserUsageByRangeResponseDto(RootModel[List[NodeUserUsageItem]]):
return self.root[item]
def __bool__(self):
"""Return True if list is not empty"""
return bool(self.root)
def __len__(self):
"""Return length of list"""
return len(self.root)
# ============ New Stats Models ============
# Legacy Stats Models
class LegacyUserUsageItem(BaseModel):
"""Legacy user usage item"""
user_uuid: UUID = Field(alias="userUuid")
node_uuid: UUID = Field(alias="nodeUuid")
node_name: str = Field(alias="nodeName")
country_code: str = Field(alias="countryCode")
total: int
date: str
class GetLegacyStatsUserUsageResponseDto(RootModel[List[LegacyUserUsageItem]]):
"""Response for legacy user usage"""
@property
def response(self) -> List[LegacyUserUsageItem]:
return self.root
def __iter__(self):
return iter(self.root)
def __getitem__(self, item):
return self.root[item]
class LegacyNodeUserUsageItem(BaseModel):
"""Legacy node user usage item"""
user_uuid: UUID = Field(alias="userUuid")
username: str
node_uuid: UUID = Field(alias="nodeUuid")
total: int
date: str
class GetLegacyStatsNodesUsersUsageResponseDto(RootModel[List[LegacyNodeUserUsageItem]]):
"""Response for legacy nodes users usage"""
@property
def response(self) -> List[LegacyNodeUserUsageItem]:
return self.root
def __iter__(self):
return iter(self.root)
def __getitem__(self, item):
return self.root[item]
# Realtime Stats
class NodeRealtimeUsageItem(BaseModel):
"""Node realtime usage item"""
node_uuid: UUID = Field(alias="nodeUuid")
node_name: str = Field(alias="nodeName")
country_code: str = Field(alias="countryCode")
download_bytes: float = Field(alias="downloadBytes")
upload_bytes: float = Field(alias="uploadBytes")
total_bytes: float = Field(alias="totalBytes")
download_speed_bps: float = Field(alias="downloadSpeedBps")
upload_speed_bps: float = Field(alias="uploadSpeedBps")
total_speed_bps: float = Field(alias="totalSpeedBps")
class GetStatsNodesRealtimeUsageResponseDto(RootModel[List[NodeRealtimeUsageItem]]):
"""Response for nodes realtime usage"""
@property
def response(self) -> List[NodeRealtimeUsageItem]:
return self.root
def __iter__(self):
return iter(self.root)
def __getitem__(self, item):
return self.root[item]
# Stats Nodes Usage (with charts)
class TopNodeItem(BaseModel):
"""Top node item"""
uuid: UUID
color: str
name: str
country_code: str = Field(alias="countryCode")
total: int
class NodeSeriesItem(BaseModel):
"""Node series item for charts"""
uuid: UUID
name: str
color: str
country_code: str = Field(alias="countryCode")
total: int
data: List[int]
class StatsNodesUsageData(BaseModel):
"""Stats nodes usage data"""
categories: List[str]
sparkline_data: List[int] = Field(alias="sparklineData")
top_nodes: List[TopNodeItem] = Field(alias="topNodes")
series: List[NodeSeriesItem]
class GetStatsNodesUsageResponseDto(RootModel[StatsNodesUsageData]):
"""Response for stats nodes usage"""
@property
def response(self) -> StatsNodesUsageData:
return self.root
# Stats Node Users Usage (with charts)
class TopUserItem(BaseModel):
"""Top user item"""
color: str
username: str
total: int
class StatsNodeUsersUsageData(BaseModel):
"""Stats node users usage data"""
categories: List[str]
sparkline_data: List[int] = Field(alias="sparklineData")
top_users: List[TopUserItem] = Field(alias="topUsers")
class GetStatsNodeUsersUsageResponseDto(RootModel[StatsNodeUsersUsageData]):
"""Response for stats node users usage"""
@property
def response(self) -> StatsNodeUsersUsageData:
return self.root
# Stats User Usage (with charts)
class StatsUserUsageData(BaseModel):
"""Stats user usage data"""
categories: List[str]
sparkline_data: List[int] = Field(alias="sparklineData")
top_nodes: List[TopNodeItem] = Field(alias="topNodes")
series: List[NodeSeriesItem]
class GetStatsUserUsageResponseDto(RootModel[StatsUserUsageData]):
"""Response for stats user usage"""
@property
def response(self) -> StatsUserUsageData:
return self.root

View file

@ -0,0 +1,102 @@
from typing import Annotated, Any, List, Optional
from uuid import UUID
from pydantic import BaseModel, ConfigDict, Field, StringConstraints
class SubscriptionPageConfigDto(BaseModel):
"""Subscription page config data model"""
model_config = ConfigDict(populate_by_name=True)
uuid: UUID
view_position: int = Field(alias="viewPosition")
name: str
config: Optional[Any] = None
class GetSubscriptionPageConfigsData(BaseModel):
"""Data for getting all subscription page configs"""
total: int
configs: List[SubscriptionPageConfigDto]
class GetSubscriptionPageConfigsResponseDto(GetSubscriptionPageConfigsData):
"""Response with all subscription page configs"""
pass
class GetSubscriptionPageConfigResponseDto(SubscriptionPageConfigDto):
"""Response with single subscription page config"""
pass
class CreateSubscriptionPageConfigRequestDto(BaseModel):
"""Request to create subscription page config"""
name: Annotated[
str,
StringConstraints(min_length=2, max_length=30, pattern=r"^[A-Za-z0-9_\s-]+$")
]
class CreateSubscriptionPageConfigResponseDto(SubscriptionPageConfigDto):
"""Response after creating subscription page config"""
pass
class UpdateSubscriptionPageConfigRequestDto(BaseModel):
"""Request to update subscription page config"""
model_config = ConfigDict(populate_by_name=True)
uuid: UUID
name: Optional[Annotated[
str,
StringConstraints(min_length=2, max_length=30, pattern=r"^[A-Za-z0-9_\s-]+$")
]] = None
config: Optional[Any] = None
class UpdateSubscriptionPageConfigResponseDto(SubscriptionPageConfigDto):
"""Response after updating subscription page config"""
pass
class DeleteSubscriptionPageConfigData(BaseModel):
"""Data for delete response"""
model_config = ConfigDict(populate_by_name=True)
is_deleted: bool = Field(alias="isDeleted")
class DeleteSubscriptionPageConfigResponseDto(DeleteSubscriptionPageConfigData):
"""Response after deleting subscription page config"""
pass
class ReorderSubscriptionPageConfigItem(BaseModel):
"""Item for reordering subscription page configs"""
model_config = ConfigDict(populate_by_name=True)
view_position: int = Field(alias="viewPosition")
uuid: UUID
class ReorderSubscriptionPageConfigsRequestDto(BaseModel):
"""Request to reorder subscription page configs"""
items: List[ReorderSubscriptionPageConfigItem]
class ReorderSubscriptionPageConfigsResponseDto(GetSubscriptionPageConfigsData):
"""Response after reordering subscription page configs"""
pass
class CloneSubscriptionPageConfigRequestDto(BaseModel):
"""Request to clone subscription page config"""
model_config = ConfigDict(populate_by_name=True)
clone_from_uuid: UUID = Field(alias="cloneFromUuid")
class CloneSubscriptionPageConfigResponseDto(SubscriptionPageConfigDto):
"""Response after cloning subscription page config"""
pass

View file

@ -41,7 +41,7 @@ async def remnawave() -> RemnawaveSDK:
assert sdk.system is not None
assert sdk.users is not None
assert sdk.users_bulk_actions is not None
assert sdk.users_stats is not None
assert sdk.subscription_page_config is not None
assert sdk.xray_config is not None
assert sdk.hwid is not None
return sdk

View file

@ -1,14 +1,251 @@
from remnawave.models import GetNodesUsageByRangeResponseDto, GetNodesRealtimeUsageResponseDto
import pytest
from uuid import UUID
from remnawave.models import (
# Legacy models (deprecated)
GetNodesUsageByRangeResponseDto,
GetNodesRealtimeUsageResponseDto,
GetNodeUserUsageByRangeResponseDto,
GetUserUsageByRangeResponseDto,
# New stats models
GetLegacyStatsUserUsageResponseDto,
GetLegacyStatsNodesUsersUsageResponseDto,
GetStatsNodesRealtimeUsageResponseDto,
GetStatsNodesUsageResponseDto,
GetStatsNodeUsersUsageResponseDto,
GetStatsUserUsageResponseDto,
)
from tests.utils import generate_isoformat_range
async def test_bandwidthstats(remnawave):
start, end = generate_isoformat_range()
nodes_usage_by_range = await remnawave.bandwidthstats.get_nodes_usage_by_range(
start=start, end=end
)
assert isinstance(nodes_usage_by_range, GetNodesUsageByRangeResponseDto)
@pytest.mark.asyncio
async def test_legacy_user_usage(remnawave):
"""Test legacy user usage endpoint (deprecated)"""
# Get first user
users = await remnawave.users.get_all_users()
if not users.users:
pytest.skip("No users available for testing")
# Test realtime usage
realtime_usage = await remnawave.bandwidthstats.get_nodes_usage_realtime()
assert isinstance(realtime_usage, GetNodesRealtimeUsageResponseDto)
user_uuid = str(users.users[0].uuid)
start, end = generate_isoformat_range()
user_usage = await remnawave.bandwidthstats.get_user_usage_legacy_old(
user_uuid=user_uuid,
start=start,
end=end
)
assert isinstance(user_usage, GetUserUsageByRangeResponseDto)
assert len(user_usage) >= 0
@pytest.mark.asyncio
async def test_legacy_node_user_usage(remnawave):
"""Test legacy node user usage endpoint (deprecated)"""
# Get first node
nodes = await remnawave.nodes.get_all_nodes()
if not nodes:
pytest.skip("No nodes available for testing")
node_uuid = str(nodes[0].uuid)
start, end = generate_isoformat_range()
node_user_usage = await remnawave.bandwidthstats.get_node_user_usage_legacy_old(
node_uuid=node_uuid,
start=start,
end=end
)
assert isinstance(node_user_usage, GetNodeUserUsageByRangeResponseDto)
assert len(node_user_usage) >= 0
@pytest.mark.asyncio
async def test_stats_nodes_realtime_usage(remnawave):
"""Test new stats nodes realtime usage endpoint"""
realtime_usage = await remnawave.bandwidthstats.get_nodes_realtime_usage()
assert isinstance(realtime_usage, GetStatsNodesRealtimeUsageResponseDto)
assert hasattr(realtime_usage, 'response')
assert isinstance(realtime_usage.response, list)
# Check structure if data exists
if realtime_usage.response:
first_item = realtime_usage.response[0]
assert hasattr(first_item, 'node_uuid')
assert hasattr(first_item, 'node_name')
assert hasattr(first_item, 'download_bytes')
assert hasattr(first_item, 'upload_bytes')
assert hasattr(first_item, 'total_bytes')
@pytest.mark.asyncio
async def test_stats_nodes_usage(remnawave):
"""Test new stats nodes usage endpoint with charts"""
start, end = generate_isoformat_range()
nodes_usage = await remnawave.bandwidthstats.get_stats_nodes_usage(
start=start,
end=end,
top_nodes_limit=5
)
assert isinstance(nodes_usage, GetStatsNodesUsageResponseDto)
assert hasattr(nodes_usage, 'response')
assert hasattr(nodes_usage.response, 'categories')
assert hasattr(nodes_usage.response, 'sparkline_data')
assert hasattr(nodes_usage.response, 'top_nodes')
assert hasattr(nodes_usage.response, 'series')
# Check data types
assert isinstance(nodes_usage.response.categories, list)
assert isinstance(nodes_usage.response.sparkline_data, list)
assert isinstance(nodes_usage.response.top_nodes, list)
assert isinstance(nodes_usage.response.series, list)
@pytest.mark.asyncio
async def test_stats_node_users_usage(remnawave):
"""Test new stats node users usage endpoint"""
# Get first node
nodes = await remnawave.nodes.get_all_nodes()
if not nodes:
pytest.skip("No nodes available for testing")
node_uuid = str(nodes[0].uuid)
start, end = generate_isoformat_range()
node_users_usage = await remnawave.bandwidthstats.get_stats_node_users_usage(
uuid=node_uuid,
start=start,
end=end,
top_users_limit=5
)
assert isinstance(node_users_usage, GetStatsNodeUsersUsageResponseDto)
assert hasattr(node_users_usage, 'response')
assert hasattr(node_users_usage.response, 'categories')
assert hasattr(node_users_usage.response, 'sparkline_data')
assert hasattr(node_users_usage.response, 'top_users')
# Check data types
assert isinstance(node_users_usage.response.categories, list)
assert isinstance(node_users_usage.response.sparkline_data, list)
assert isinstance(node_users_usage.response.top_users, list)
@pytest.mark.asyncio
async def test_stats_user_usage(remnawave):
"""Test new stats user usage endpoint"""
# Get first user
users = await remnawave.users.get_all_users()
if not users.users:
pytest.skip("No users available for testing")
user_uuid = str(users.users[0].uuid)
start, end = generate_isoformat_range()
user_usage = await remnawave.bandwidthstats.get_stats_user_usage(
uuid=user_uuid,
start=start,
end=end,
top_nodes_limit=5
)
assert isinstance(user_usage, GetStatsUserUsageResponseDto)
assert hasattr(user_usage, 'response')
assert hasattr(user_usage.response, 'categories')
assert hasattr(user_usage.response, 'sparkline_data')
assert hasattr(user_usage.response, 'top_nodes')
assert hasattr(user_usage.response, 'series')
# Check data types
assert isinstance(user_usage.response.categories, list)
assert isinstance(user_usage.response.sparkline_data, list)
assert isinstance(user_usage.response.top_nodes, list)
assert isinstance(user_usage.response.series, list)
@pytest.mark.asyncio
async def test_legacy_stats_user_usage(remnawave):
"""Test legacy stats user usage endpoint"""
# Get first user
users = await remnawave.users.get_all_users()
if not users.users:
pytest.skip("No users available for testing")
user_uuid = str(users.users[0].uuid)
start, end = generate_isoformat_range()
legacy_user_usage = await remnawave.bandwidthstats.get_user_usage_legacy_stats(
uuid=user_uuid,
start=start,
end=end
)
assert isinstance(legacy_user_usage, GetLegacyStatsUserUsageResponseDto)
assert hasattr(legacy_user_usage, 'response')
assert isinstance(legacy_user_usage.response, list)
# Check structure if data exists
if legacy_user_usage.response:
first_item = legacy_user_usage.response[0]
assert hasattr(first_item, 'user_uuid')
assert hasattr(first_item, 'node_uuid')
assert hasattr(first_item, 'node_name')
assert hasattr(first_item, 'total')
@pytest.mark.asyncio
async def test_legacy_stats_nodes_users_usage(remnawave):
"""Test legacy stats nodes users usage endpoint"""
# Get first node
nodes = await remnawave.nodes.get_all_nodes()
if not nodes:
pytest.skip("No nodes available for testing")
node_uuid = str(nodes[0].uuid)
start, end = generate_isoformat_range()
legacy_node_users = await remnawave.bandwidthstats.get_node_users_usage_legacy_stats(
uuid=node_uuid,
start=start,
end=end
)
assert isinstance(legacy_node_users, GetLegacyStatsNodesUsersUsageResponseDto)
assert hasattr(legacy_node_users, 'response')
assert isinstance(legacy_node_users.response, list)
# Check structure if data exists
if legacy_node_users.response:
first_item = legacy_node_users.response[0]
assert hasattr(first_item, 'user_uuid')
assert hasattr(first_item, 'username')
assert hasattr(first_item, 'node_uuid')
assert hasattr(first_item, 'total')
@pytest.mark.asyncio
async def test_bandwidth_data_structure(remnawave):
"""Test bandwidth stats data structure validity"""
start, end = generate_isoformat_range()
# Get realtime data
realtime = await remnawave.bandwidthstats.get_nodes_realtime_usage()
if realtime.response:
# Verify each node has required fields
for node in realtime.response:
assert isinstance(node.node_uuid, UUID)
assert isinstance(node.node_name, str)
assert isinstance(node.download_bytes, (int, float))
assert isinstance(node.upload_bytes, (int, float))
assert isinstance(node.total_bytes, (int, float))
assert node.total_bytes >= 0
# Get stats data
stats = await remnawave.bandwidthstats.get_stats_nodes_usage(
start=start,
end=end,
top_nodes_limit=3
)
# Verify stats structure
assert len(stats.response.categories) == len(stats.response.sparkline_data)
assert len(stats.response.top_nodes) <= 3
if stats.response.series:
for series_item in stats.response.series:
assert len(series_item.data) == len(stats.response.categories)

View file

@ -1,25 +0,0 @@
from datetime import datetime, timedelta
import pytest
from remnawave.models import (
GetNodeUserUsageByRangeResponseDto,
GetNodesUsageByRangeResponseDto,
)
from tests.conftest import REMNAWAVE_USER_UUID
@pytest.mark.asyncio
async def test_nodes_usage_history(remnawave) -> None:
# Test get nodes usage by range
start_date = (datetime.now() - timedelta(days=7)).isoformat()
end_date = datetime.now().isoformat()
nodes_usage = await remnawave.nodes_usage_history.get_nodes_usage_by_range(
start=start_date,
end=end_date
)
assert isinstance(nodes_usage, GetNodesUsageByRangeResponseDto)
# Response should be a list now (RootModel)
assert isinstance(nodes_usage.root, list)

96
tests/test_sub_page.py Normal file
View file

@ -0,0 +1,96 @@
import pytest
from remnawave.models import (
CloneSubscriptionPageConfigRequestDto,
CloneSubscriptionPageConfigResponseDto,
CreateSubscriptionPageConfigRequestDto,
CreateSubscriptionPageConfigResponseDto,
DeleteSubscriptionPageConfigResponseDto,
GetSubscriptionPageConfigResponseDto,
GetSubscriptionPageConfigsResponseDto,
ReorderSubscriptionPageConfigItem,
ReorderSubscriptionPageConfigsRequestDto,
ReorderSubscriptionPageConfigsResponseDto,
UpdateSubscriptionPageConfigRequestDto,
UpdateSubscriptionPageConfigResponseDto,
)
def random_string(length=10):
import random
import string
return ''.join(random.choices(string.ascii_letters + string.digits, k=length))
@pytest.mark.asyncio
async def test_get_all_configs(remnawave):
"""Test getting all subscription page configs"""
configs = await remnawave.subscription_page_config.get_all_configs()
assert isinstance(configs, GetSubscriptionPageConfigsResponseDto)
assert configs.total >= 0
assert isinstance(configs.configs, list)
@pytest.mark.asyncio
async def test_subscription_page_config_full_workflow(remnawave):
"""Test full workflow: create, get, update, reorder, clone, delete"""
# Create config
config_name = f"test_config_{random_string()}"
create_request = CreateSubscriptionPageConfigRequestDto(name=config_name)
created_config = await remnawave.subscription_page_config.create_config(create_request)
assert isinstance(created_config, CreateSubscriptionPageConfigResponseDto)
assert created_config.name == config_name
config_uuid = str(created_config.uuid)
# Get config by UUID
config = await remnawave.subscription_page_config.get_config_by_uuid(config_uuid)
assert isinstance(config, GetSubscriptionPageConfigResponseDto)
assert config.name == config_name
# Update config
updated_name = f"updated_{config_name}"
update_request = UpdateSubscriptionPageConfigRequestDto(
uuid=created_config.uuid,
name=updated_name,
config=config.config
)
updated_config = await remnawave.subscription_page_config.update_config(update_request)
assert isinstance(updated_config, UpdateSubscriptionPageConfigResponseDto)
assert updated_config.name == updated_name
# Clone config
clone_request = CloneSubscriptionPageConfigRequestDto(
clone_from_uuid=created_config.uuid
)
cloned_config = await remnawave.subscription_page_config.clone_config(clone_request)
assert isinstance(cloned_config, CloneSubscriptionPageConfigResponseDto)
# Reorder configs
reorder_request = ReorderSubscriptionPageConfigsRequestDto(
items=[
ReorderSubscriptionPageConfigItem(
uuid=created_config.uuid,
view_position=1
),
ReorderSubscriptionPageConfigItem(
uuid=cloned_config.uuid,
view_position=2
)
]
)
reordered = await remnawave.subscription_page_config.reorder_configs(reorder_request)
assert isinstance(reordered, ReorderSubscriptionPageConfigsResponseDto)
# Delete cloned config
delete_response = await remnawave.subscription_page_config.delete_config(
str(cloned_config.uuid)
)
assert isinstance(delete_response, DeleteSubscriptionPageConfigResponseDto)
assert delete_response.is_deleted is True
# Delete original config
delete_response = await remnawave.subscription_page_config.delete_config(config_uuid)
assert isinstance(delete_response, DeleteSubscriptionPageConfigResponseDto)
assert delete_response.is_deleted is True

View file

@ -1,14 +0,0 @@
import pytest
from remnawave.models import GetUserUsageByRangeResponseDto
from tests.conftest import REMNAWAVE_USER_UUID
from tests.utils import generate_isoformat_range
@pytest.mark.asyncio
async def test_users_stats(remnawave):
start, end = generate_isoformat_range()
user_usage_by_range = await remnawave.users_stats.get_user_usage_by_range(
uuid=REMNAWAVE_USER_UUID, start=start, end=end
)
assert isinstance(user_usage_by_range, GetUserUsageByRangeResponseDto)