scripts/py_matter_yamltests/matter_yamltests/pseudo_clusters/clusters/system_commands.py

#
#    Copyright (c) 2023 Project CHIP Authors
#
#    Licensed under the Apache License, Version 2.0 (the 'License');
#    you may not use this file except in compliance with the License.
#    You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an 'AS IS' BASIS,
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#    See the License for the specific language governing permissions and
#    limitations under the License.

import sys
import xmlrpc.client

from ..pseudo_cluster import PseudoCluster

DEFAULT_KEY = 'default'
IP = '127.0.0.1'
PORT = 9000

if sys.platform == 'linux':
    IP = '10.10.10.5'


def make_url():
    return 'http://' + IP + ':' + str(PORT) + '/'


def get_register_key(request):
    if request.arguments:
        values = request.arguments['values']
        for item in values:
            name = item['name']
            value = item['value']
            if name == 'registerKey':
                return value

    return DEFAULT_KEY


def get_options(request):
    options = []

    if request.arguments:
        values = request.arguments['values']
        for item in values:
            name = item['name']
            value = item['value']

            if name == 'discriminator':
                options.append('--discriminator')
                options.append(str(value))
            elif name == 'port':
                options.append('--secured-device-port')
                options.append(str(value))
            elif name == 'kvs':
                options.append('--KVS')
                options.append(str(value))
            elif name == 'minCommissioningTimeout':
                options.append('--min_commissioning_timeout')
                options.append(str(value))
            elif name == 'filepath':
                options.append('--filepath')
                options.append(str(value))
            elif name == 'otaDownloadPath':
                options.append('--otaDownloadPath')
                options.append(str(value))
            elif name == 'registerKey':
                pass
            else:
                raise KeyError(f'Unknown key: {name}')

    return options


class SystemCommands(PseudoCluster):
    name = 'SystemCommands'

    async def Start(self, request):
        register_key = get_register_key(request)
        options = get_options(request)

        with xmlrpc.client.ServerProxy(make_url(), allow_none=True) as proxy:
            proxy.start(register_key, options)

    async def Stop(self, request):
        register_key = get_register_key(request)

        with xmlrpc.client.ServerProxy(make_url(), allow_none=True) as proxy:
            proxy.stop(register_key)

    async def Reboot(self, request):
        register_key = get_register_key(request)

        with xmlrpc.client.ServerProxy(make_url(), allow_none=True) as proxy:
            proxy.reboot(register_key)

    async def FactoryReset(self, request):
        register_key = get_register_key(request)

        with xmlrpc.client.ServerProxy(make_url(), allow_none=True) as proxy:
            proxy.factoryReset(register_key)