mia/bumper
1
0
Fork 0
Code Issues 47 Pull requests 11 Projects Releases Packages Wiki Activity Actions

most plugins have same response schema

Browse source
This commit is contained in:
Robert Resch 2022-03-06 22:20:42 +01:00
parent 3dd905d3d5
commit 1a1dc17648
17 changed files with 293 additions and 470 deletions
Download patch file Download diff file Expand all files Collapse all files

5
bumper/mqttserver.py
View file

@ -152,7 +152,10 @@ class MQTTHelperBot:
assert self._client is not None
try:
topic = f"iot/p2p/{cmdjson['cmdName']}/helperbot/bumper/helperbot/{cmdjson['toId']}/{cmdjson['toType']}/{cmdjson['toRes']}/q/{request_id}/{cmdjson['payloadType']}"
topic = (
f"iot/p2p/{cmdjson['cmdName']}/helperbot/bumper/helperbot/{cmdjson['toId']}/"
f"{cmdjson['toType']}/{cmdjson['toRes']}/q/{request_id}/{cmdjson['payloadType']}"
)
command_dto = CommandDto(cmdjson["payloadType"])
self._commands[request_id] = command_dto

18
bumper/web/images/__init__.py Normal file
View file

@ -0,0 +1,18 @@
import logging
import os
from aiohttp.web_exceptions import HTTPInternalServerError
from aiohttp.web_fileresponse import FileResponse
from aiohttp.web_request import Request
async def get_bot_image(_: Request) -> FileResponse:
"""Return image of bot."""
try:
return FileResponse(
os.path.join(os.path.dirname(__file__), "robotvac_image.jpg")
)
except Exception: # pylint: disable=broad-except
logging.error("Unexpected exception occurred", exc_info=True)
raise HTTPInternalServerError

18
bumper/web/plugins/__init__.py
View file

@ -5,11 +5,15 @@ from abc import abstractmethod
from glob import glob
from os.path import dirname, isfile, join
from types import ModuleType
from typing import Iterable
from typing import Any, Iterable
from aiohttp import web
from aiohttp.web_response import Response
from aiohttp.web_routedef import AbstractRouteDef
from bumper.models import RETURN_API_SUCCESS
from bumper.util import get_current_time_as_millis
class WebserverPlugin:
"""Abstract webserver plugin."""
@ -70,3 +74,15 @@ def add_plugins(app: web.Application) -> None:
continue
_add_routes(app, obj, plugin_module_name)
def get_success_response(data: Any) -> Response:
body = {
"code": RETURN_API_SUCCESS,
"data": data,
"msg": "操作成功",
"success": True,
"time": get_current_time_as_millis(),
}
return web.json_response(body)

18
bumper/web/plugins/api/pim.py
View file

@ -1,15 +1,14 @@
"""Pim plugin module."""
import json
import logging
import os
from typing import Iterable
from aiohttp import web
from aiohttp.web_routedef import AbstractRouteDef
from bumper import bumper_dir
from bumper.models import RETURN_API_SUCCESS, EcoVacsHomeProducts
from ...images import get_bot_image
from .. import WebserverPlugin
@ -28,7 +27,7 @@ class PimPlugin(WebserverPlugin):
web.route(
"*",
"/pim/file/get/{id}",
self.handle_pimFile,
get_bot_image,
),
web.route(
"*",
@ -63,19 +62,6 @@ class PimPlugin(WebserverPlugin):
except Exception as e:
logging.exception(f"{e}")
async def handle_pimFile(self, request):
try:
fileID = request.match_info.get("id", "")
return web.FileResponse(
os.path.join(
bumper_dir, "bumper", "web", "images", "robotvac_image.jpg"
)
)
except Exception as e:
logging.exception(f"{e}")
async def handle_getConfignetAll(self, request):
try:
body = confignetAllResponse

24
bumper/web/plugins/upload/global.py
View file

@ -1,20 +1,14 @@
"""Global plugin module."""
import logging
import os
from typing import Iterable
from aiohttp import web
from aiohttp.web_exceptions import HTTPInternalServerError
from aiohttp.web_request import Request
from aiohttp.web_response import StreamResponse
from aiohttp.web_routedef import AbstractRouteDef
from bumper import bumper_dir
from bumper.web.images import get_bot_image
from .. import WebserverPlugin
# pylint: disable=no-self-use
class GlobalPlugin(WebserverPlugin):
"""Global plugin."""
@ -24,19 +18,7 @@ class GlobalPlugin(WebserverPlugin):
return [
web.route(
"*",
"/global/{year}/{month}/{day}/{fileid}",
self._handle_upload_global_file,
"/global/{year}/{month}/{day}/{id}",
get_bot_image,
),
]
async def _handle_upload_global_file(self, _: Request) -> StreamResponse:
try:
return web.FileResponse(
os.path.join(
bumper_dir, "bumper", "web", "images", "robotvac_image.jpg"
)
)
except Exception: # pylint: disable=broad-except
logging.error("Unexpected exception occurred", exc_info=True)
raise HTTPInternalServerError

2
bumper/web/plugins/v1/private/__init__.py
View file

@ -1 +1,3 @@
"""Private plugins module."""
BASE_URL = "/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/"

47
bumper/web/plugins/v1/private/ad.py
View file

@ -1,14 +1,13 @@
"""Ad plugin module."""
import logging
from typing import Iterable
from aiohttp import web
from aiohttp.web_request import Request
from aiohttp.web_response import Response
from aiohttp.web_routedef import AbstractRouteDef
from bumper.models import RETURN_API_SUCCESS
from bumper.util import get_current_time_as_millis
from ... import WebserverPlugin
from ... import WebserverPlugin, get_success_response
from . import BASE_URL
class AdPlugin(WebserverPlugin):
@ -20,42 +19,16 @@ class AdPlugin(WebserverPlugin):
return [
web.route(
"*",
"/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/ad/getAdByPositionType",
self.handle_getAdByPositionType,
f"{BASE_URL}ad/getAdByPositionType",
_handle,
),
web.route(
"*",
"/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/ad/getBootScreen",
self.handle_getBootScreen,
f"{BASE_URL}ad/getBootScreen",
_handle,
),
]
async def handle_getAdByPositionType(self, request): # EcoVacs Home
try:
body = {
"code": RETURN_API_SUCCESS,
"data": None,
"msg": "操作成功",
"success": True,
"time": get_current_time_as_millis(),
}
return web.json_response(body)
except Exception as e:
logging.exception(f"{e}")
async def handle_getBootScreen(self, request): # EcoVacs Home
try:
body = {
"code": RETURN_API_SUCCESS,
"data": None,
"msg": "操作成功",
"success": True,
"time": get_current_time_as_millis(),
}
return web.json_response(body)
except Exception as e:
logging.exception(f"{e}")
async def _handle(_: Request) -> Response:
return get_success_response(None)

33
bumper/web/plugins/v1/private/campaign.py
View file

@ -1,15 +1,16 @@
"""Campaign plugin module."""
import logging
from datetime import datetime, timedelta
from typing import Iterable
from aiohttp import web
from aiohttp.web_request import Request
from aiohttp.web_response import Response
from aiohttp.web_routedef import AbstractRouteDef
from bumper.models import RETURN_API_SUCCESS
from bumper.util import convert_to_millis, get_current_time_as_millis
from ... import WebserverPlugin
from ... import WebserverPlugin, get_success_response
from . import BASE_URL
class CampaignPlugin(WebserverPlugin):
@ -21,33 +22,21 @@ class CampaignPlugin(WebserverPlugin):
return [
web.route(
"*",
"/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/campaign/homePageAlert",
self.handle_homePageAlert,
name="v1_campaign_homePageAlert",
f"{BASE_URL}campaign/homePageAlert",
_handle_home_page_alert,
),
]
async def handle_homePageAlert(self, request):
try:
nextAlert = convert_to_millis(
(datetime.now() + timedelta(hours=12)).timestamp()
)
body = {
"code": RETURN_API_SUCCESS,
"data": {
async def _handle_home_page_alert(_: Request) -> Response:
nextAlert = convert_to_millis((datetime.now() + timedelta(hours=12)).timestamp())
return get_success_response(
{
"clickSchemeUrl": None,
"clickWebUrl": None,
"hasCampaign": "N",
"imageUrl": None,
"nextAlertTime": nextAlert,
"serverTime": get_current_time_as_millis(),
},
"msg": "操作成功",
"time": get_current_time_as_millis(),
}
return web.json_response(body)
except Exception as e:
logging.exception(f"{e}")
)

116
bumper/web/plugins/v1/private/common.py
View file

@ -10,10 +10,10 @@ from aiohttp.web_request import Request
from aiohttp.web_response import Response
from aiohttp.web_routedef import AbstractRouteDef
from bumper.models import RETURN_API_SUCCESS
from bumper.util import get_current_time_as_millis
from ... import WebserverPlugin
from ... import WebserverPlugin, get_success_response
from . import BASE_URL
class CommonPlugin(WebserverPlugin):
@ -22,55 +22,53 @@ class CommonPlugin(WebserverPlugin):
@property
def routes(self) -> Iterable[AbstractRouteDef]:
"""Plugin routes."""
base_url = "/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/common/"
return [
web.route(
"*",
f"{base_url}checkAPPVersion",
f"{BASE_URL}common/checkAPPVersion",
_handle_check_app_version,
),
web.route(
"*",
f"{base_url}checkVersion",
f"{BASE_URL}common/checkVersion",
_handle_check_version,
),
web.route(
"*",
f"{base_url}uploadDeviceInfo",
f"{BASE_URL}common/uploadDeviceInfo",
_handle_upload_device_info,
),
web.route(
"*",
f"{base_url}getSystemReminder",
f"{BASE_URL}common/getSystemReminder",
_handle_get_system_reminder,
),
web.route(
"*",
f"{base_url}getConfig",
f"{BASE_URL}common/getConfig",
_handle_get_config,
),
web.route(
"*",
f"{base_url}getAreas",
f"{BASE_URL}common/getAreas",
_handle_get_areas,
),
web.route(
"*",
f"{base_url}getAgreementURLBatch",
f"{BASE_URL}common/getAgreementURLBatch",
_handle_get_agreement_url_batch,
),
web.route(
"*",
f"{base_url}getTimestamp",
f"{BASE_URL}common/getTimestamp",
_handle_get_timestamp,
),
]
async def _handle_check_version(_: Request) -> Response:
body = {
"code": RETURN_API_SUCCESS,
"data": {
return get_success_response(
{
"c": None,
"img": None,
"r": 0,
@ -78,18 +76,13 @@ async def _handle_check_version(_: Request) -> Response:
"u": None,
"ut": 0,
"v": None,
},
"msg": "操作成功",
"time": get_current_time_as_millis(),
}
return web.json_response(body)
)
async def _handle_check_app_version(_: Request) -> Response:
body = {
"code": RETURN_API_SUCCESS,
"data": {
return get_success_response(
{
"c": None,
"downPageUrl": None,
"img": None,
@ -99,44 +92,25 @@ async def _handle_check_app_version(_: Request) -> Response:
"u": None,
"ut": 0,
"v": None,
},
"msg": "操作成功",
"success": True,
"time": get_current_time_as_millis(),
}
return web.json_response(body)
)
async def _handle_upload_device_info(_: Request) -> Response:
body = {
"code": RETURN_API_SUCCESS,
"data": None,
"msg": "操作成功",
"success": True,
"time": get_current_time_as_millis(),
}
return web.json_response(body)
return get_success_response(None)
async def _handle_get_system_reminder(_: Request) -> Response:
body = {
"code": RETURN_API_SUCCESS,
"data": {
return get_success_response(
{
"iosGradeTime": {"iodGradeFlag": "N"},
"openNotification": {
"openNotificationContent": None,
"openNotificationFlag": "N",
"openNotificationTitle": None,
},
},
"msg": "操作成功",
"success": True,
"time": get_current_time_as_millis(),
}
return web.json_response(body)
)
async def _handle_get_config(request: Request) -> Response:
@ -145,15 +119,7 @@ async def _handle_get_config(request: Request) -> Response:
for key in request.query["keys"].split(","):
data.append({"key": key, "value": "Y"})
body = {
"code": RETURN_API_SUCCESS,
"data": data,
"msg": "操作成功",
"success": True,
"time": get_current_time_as_millis(),
}
return web.json_response(body)
return get_success_response(data)
except Exception: # pylint: disable=broad-except
logging.error("Unexpected exception occurred", exc_info=True)
@ -162,16 +128,11 @@ async def _handle_get_config(request: Request) -> Response:
async def _handle_get_areas(_: Request) -> Response:
try:
with open(os.path.join(os.path.dirname(__file__), "area.json")) as file:
body = {
"code": RETURN_API_SUCCESS,
"data": json.load(file),
"msg": "操作成功",
"success": True,
"time": get_current_time_as_millis(),
}
return web.json_response(body)
with open(
os.path.join(os.path.dirname(__file__), "common_area.json"),
encoding="utf-8",
) as file:
return get_success_response(json.load(file))
except Exception: # pylint: disable=broad-except
logging.error("Unexpected exception occurred", exc_info=True)
@ -179,9 +140,8 @@ async def _handle_get_areas(_: Request) -> Response:
async def _handle_get_agreement_url_batch(_: Request) -> Response:
body = {
"code": RETURN_API_SUCCESS,
"data": [
return get_success_response(
[
{
"acceptTime": None,
"force": None,
@ -198,23 +158,9 @@ async def _handle_get_agreement_url_batch(_: Request) -> Response:
"url": "https://gl-eu-wap.ecovacs.com/content/agreement?id=20180804040245_4e7c56dfb7ebd3b81b1f2747d0859fac&language=EN",
"version": "1.03",
},
],
"msg": "操作成功",
"success": True,
"time": get_current_time_as_millis(),
}
return web.json_response(body)
]
)
async def _handle_get_timestamp(_: Request) -> Response:
time = get_current_time_as_millis()
body = {
"code": RETURN_API_SUCCESS,
"data": {"timestamp": time},
"msg": "操作成功",
"success": True,
"time": time,
}
return web.json_response(body)
return get_success_response({"timestamp": get_current_time_as_millis()})

0
bumper/web/plugins/v1/private/area.json → bumper/web/plugins/v1/private/common_area.json
View file

47
bumper/web/plugins/v1/private/message.py
View file

@ -1,13 +1,12 @@
import logging
from typing import Iterable
from aiohttp import web
from aiohttp.web_request import Request
from aiohttp.web_response import Response
from aiohttp.web_routedef import AbstractRouteDef
from bumper.models import RETURN_API_SUCCESS
from bumper.util import get_current_time_as_millis
from ... import WebserverPlugin
from ... import WebserverPlugin, get_success_response
from . import BASE_URL
class MessagePlugin(WebserverPlugin):
@ -19,42 +18,20 @@ class MessagePlugin(WebserverPlugin):
return [
web.route(
"*",
"/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/message/hasUnreadMsg",
self.handle_hasUnreadMessage,
f"{BASE_URL}message/hasUnreadMsg",
_handle_has_unread_message,
),
web.route(
"*",
"/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/message/getMsgList",
self.handle_getMsgList,
f"{BASE_URL}message/getMsgList",
_handle_get_msg_list,
),
]
async def handle_hasUnreadMessage(self, request): # EcoVacs Home
try:
body = {
"code": RETURN_API_SUCCESS,
"data": "N",
"msg": "操作成功",
"success": True,
"time": get_current_time_as_millis(),
}
return web.json_response(body)
async def _handle_has_unread_message(_: Request) -> Response:
return get_success_response("N")
except Exception as e:
logging.exception(f"{e}")
async def handle_getMsgList(self, request): # EcoVacs Home
try:
body = {
"code": RETURN_API_SUCCESS,
"data": {"hasNextPage": 0, "items": []},
"msg": "操作成功",
"success": True,
"time": get_current_time_as_millis(),
}
return web.json_response(body)
except Exception as e:
logging.exception(f"{e}")
async def _handle_get_msg_list(_: Request) -> Response:
return get_success_response({"hasNextPage": 0, "items": []})

32
bumper/web/plugins/v1/private/shop.py
View file

@ -1,14 +1,13 @@
"""Shop plugin module."""
import logging
from typing import Iterable
from aiohttp import web
from aiohttp.web_request import Request
from aiohttp.web_response import Response
from aiohttp.web_routedef import AbstractRouteDef
from bumper.models import RETURN_API_SUCCESS
from bumper.util import get_current_time_as_millis
from ... import WebserverPlugin
from ... import WebserverPlugin, get_success_response
from . import BASE_URL
class ShopPlugin(WebserverPlugin):
@ -20,27 +19,18 @@ class ShopPlugin(WebserverPlugin):
return [
web.route(
"*",
"/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/shop/getCnWapShopConfig",
self.handle_getCnWapShopConfig,
f"{BASE_URL}shop/getCnWapShopConfig",
_handle_get_cn_wap_shop_config,
),
]
async def handle_getCnWapShopConfig(self, request): # EcoVacs Home
try:
body = {
"code": RETURN_API_SUCCESS,
"data": {
async def _handle_get_cn_wap_shop_config(_: Request) -> Response:
return get_success_response(
{
"myShopShowFlag": "N",
"myShopUrl": "",
"shopIndexShowFlag": "N",
"shopIndexUrl": "",
},
"msg": "操作成功",
"success": True,
"time": get_current_time_as_millis(),
}
return web.json_response(body)
except Exception as e:
logging.exception(f"{e}")
)

124
bumper/web/plugins/v1/private/user.py
View file

@ -1,14 +1,15 @@
import logging
"""User plugin module."""
from typing import Iterable
from aiohttp import web
from aiohttp.web_request import Request
from aiohttp.web_response import Response
from aiohttp.web_routedef import AbstractRouteDef
from bumper.models import RETURN_API_SUCCESS
from bumper.util import get_current_time_as_millis
from bumper.web import auth_util
from ... import WebserverPlugin
from ... import WebserverPlugin, get_success_response
from . import BASE_URL
class UserPlugin(WebserverPlugin):
@ -20,72 +21,67 @@ class UserPlugin(WebserverPlugin):
return [
web.route(
"*",
"/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/login",
f"{BASE_URL}user/login",
auth_util.login,
),
web.route(
"*",
"/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/checkLogin",
f"{BASE_URL}user/checkLogin",
auth_util.login,
),
web.route(
"*",
"/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/getAuthCode",
f"{BASE_URL}user/getAuthCode",
auth_util.get_authcode,
),
web.route(
"*",
"/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/logout",
f"{BASE_URL}user/logout",
auth_util.logout,
),
web.route(
"*",
"/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/checkAgreement",
self.handle_checkAgreement,
f"{BASE_URL}user/checkAgreement",
_handle_check_agreement,
),
web.route(
"*",
"/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/checkAgreementBatch",
self.handle_checkAgreement,
f"{BASE_URL}user/checkAgreementBatch",
_handle_check_agreement,
),
web.route(
"*",
"/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/getUserAccountInfo",
f"{BASE_URL}user/getUserAccountInfo",
auth_util.get_user_account_info,
),
web.route(
"*",
"/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/getUserMenuInfo",
self.handle_getUserMenuInfo,
f"{BASE_URL}user/getUserMenuInfo",
_handle_get_user_menu_info,
),
web.route(
"*",
"/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/changeArea",
self.handle_changeArea,
f"{BASE_URL}user/changeArea",
_handle_change_area,
),
web.route(
"*",
"/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/queryChangeArea",
self.handle_changeArea,
f"{BASE_URL}user/queryChangeArea",
_handle_change_area,
),
web.route(
"*",
"/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/acceptAgreementBatch",
self.handle_acceptAgreementBatch,
f"{BASE_URL}user/acceptAgreementBatch",
_handle_accept_agreement_batch,
),
# Direct register from app:
# /{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/directRegister
# Register by email
# /registerByEmail
]
async def handle_checkAgreement(self, request):
try:
async def _handle_check_agreement(request: Request) -> Response:
apptype = request.match_info.get("apptype", "")
data = []
if "global_" in apptype:
body = {
"code": RETURN_API_SUCCESS,
"data": [
data = [
{
"force": "N",
"id": "20180804040641_7d746faf18b8cb22a50d145598fe4c90",
@ -100,30 +96,14 @@ class UserPlugin(WebserverPlugin):
"url": "https://bumper.ecovacs.com/content/agreement?id=20180804040245_4e7c56dfb7ebd3b81b1f2747d0859fac&language=EN", # "https://gl-us-wap.ecovacs.com/content/agreement?id=20180804040245_4e7c56dfb7ebd3b81b1f2747d0859fac&language=EN"
"version": "1.01",
},
],
"msg": "操作成功",
"success": True,
"time": get_current_time_as_millis(),
}
else:
body = {
"code": RETURN_API_SUCCESS,
"data": [],
"msg": "操作成功",
"time": get_current_time_as_millis(),
}
]
return web.json_response(body)
return get_success_response(data)
except Exception as e:
logging.exception(f"{e}")
async def handle_getUserMenuInfo(self, request):
try:
apptype = request.match_info.get("apptype", "")
body = {
"code": "0000",
"data": [
async def _handle_get_user_menu_info(_: Request) -> Response:
return get_success_response(
[
{
"menuItems": [
{
@ -176,43 +156,13 @@ class UserPlugin(WebserverPlugin):
],
"menuPositionKey": "D_FOURTH",
},
],
"msg": "操作成功",
"success": True,
"time": get_current_time_as_millis(),
}
]
)
return web.json_response(body)
except Exception as e:
logging.exception(f"{e}")
async def _handle_change_area(_: Request) -> Response:
return get_success_response({"isNeedReLogin": "N"})
async def handle_changeArea(self, request):
try:
body = {
"code": RETURN_API_SUCCESS,
"data": {"isNeedReLogin": "N"},
"msg": "操作成功",
"success": True,
"time": get_current_time_as_millis(),
}
return web.json_response(body)
except Exception as e:
logging.exception(f"{e}")
async def handle_acceptAgreementBatch(self, request):
try:
body = {
"code": RETURN_API_SUCCESS,
"data": None,
"msg": "操作成功",
"success": True,
"time": get_current_time_as_millis(),
}
return web.json_response(body)
except Exception as e:
logging.exception(f"{e}")
async def _handle_accept_agreement_batch(_: Request) -> Response:
return get_success_response(None)

51
bumper/web/plugins/v1/private/user_setting.py Normal file
View file

@ -0,0 +1,51 @@
"""User setting plugin module."""
from typing import Iterable
from aiohttp import web
from aiohttp.web_request import Request
from aiohttp.web_response import Response
from aiohttp.web_routedef import AbstractRouteDef
from ... import WebserverPlugin, get_success_response
from . import BASE_URL
class UserSettingPlugin(WebserverPlugin):
"""User setting plugin."""
@property
def routes(self) -> Iterable[AbstractRouteDef]:
"""Plugin routes."""
return [
web.route(
"*",
f"{BASE_URL}userSetting/getSuggestionSetting",
_handle_get_suggestion_setting,
),
]
async def _handle_get_suggestion_setting(_: Request) -> Response:
return get_success_response(
{
"acceptSuggestion": "Y",
"itemList": [
{
"name": "Aktionen/Angebote/Ereignisse",
"settingKey": "MARKETING",
"val": "Y",
},
{
"name": "Benutzerbefragung",
"settingKey": "QUESTIONNAIRE",
"val": "Y",
},
{
"name": "Produkt-Upgrade/Hilfe für Benutzer",
"settingKey": "INTRODUCTION",
"val": "Y",
},
],
}
)

61
bumper/web/plugins/v1/private/users_setting.py
View file

@ -1,61 +0,0 @@
"""user setting plugin module."""
import logging
from typing import Iterable
from aiohttp import web
from aiohttp.web_routedef import AbstractRouteDef
from bumper.models import RETURN_API_SUCCESS
from bumper.util import get_current_time_as_millis
from ... import WebserverPlugin
class UserSettingPlugin(WebserverPlugin):
"""User setting plugin."""
@property
def routes(self) -> Iterable[AbstractRouteDef]:
"""Plugin routes."""
return [
web.route(
"*",
"/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/userSetting/getSuggestionSetting",
self.handle_getSuggestionSetting,
),
]
async def handle_getSuggestionSetting(self, request):
try:
body = {
"code": RETURN_API_SUCCESS,
"data": {
"acceptSuggestion": "Y",
"itemList": [
{
"name": "Aktionen/Angebote/Ereignisse",
"settingKey": "MARKETING",
"val": "Y",
},
{
"name": "Benutzerbefragung",
"settingKey": "QUESTIONNAIRE",
"val": "Y",
},
{
"name": "Produkt-Upgrade/Hilfe für Benutzer",
"settingKey": "INTRODUCTION",
"val": "Y",
},
],
},
"msg": "操作成功",
"time": get_current_time_as_millis(),
}
return web.json_response(body)
except Exception as e:
logging.exception(f"{e}")

3
tests/web/test_server.py
View file

@ -496,7 +496,8 @@ async def test_getProductIotMap(webserver_client):
jsonresp = json.loads(text)
assert jsonresp["code"] == RETURN_API_SUCCESS
# Test getPimFile
async def test_pim_file(webserver_client):
resp = await webserver_client.get("/api/pim/file/get/123")
assert resp.status == 200