diff --git a/.github/workflows/unused-imports.yml b/.github/workflows/unused-imports.yml
index 83098ddf6..e716a9c01 100644
--- a/.github/workflows/unused-imports.yml
+++ b/.github/workflows/unused-imports.yml
@@ -1,29 +1,20 @@
name: Check for unused imports using Pylint
-on:
- pull_request_target:
- types: [opened, reopened, ready_for_review, locked]
+on: push
+ # pull_request_target:
+ # types: [opened, reopened, ready_for_review, locked]
jobs:
- build:
+ Check-Unused-Imports:
runs-on: ubuntu-latest
- strategy:
- matrix:
- python-version: ["3.11"]
steps:
- - uses: actions/checkout@v3
- - name: Set up Python ${{ matrix.python-version }}
- uses: actions/setup-python@v3
- with:
- python-version: ${{ matrix.python-version }}
- - name: Install dependencies
- run: |
- python -m pip install --upgrade pip
- pip install pylint
- - name: Analysing the code with pylint
- run: |
- tmp=$(git ls-files *.py | xargs pylint | grep W0611 | wc -l)
- if [[ $tmp -gt 0 ]]; then
- echo "Found $tmp occurrence of unused Python import statements!"
- exit 1
- fi
- exit 0
+ - uses: actions/checkout@v3
+ - name: Set up Python
+ uses: actions/setup-python@v3
+ with:
+ python-version: 3.11
+ - name: Install dependencies
+ run: |
+ python -m pip install --upgrade pip
+ pip install pylint
+ - name: Analysing the code with pylint
+ run: make unused-imports
diff --git a/Makefile b/Makefile
index 0868025ae..432de7547 100644
--- a/Makefile
+++ b/Makefile
@@ -1,125 +1,125 @@
TMPL_DIR := templates-cfg
OP_TMPL_DIR := templates-op
BUILD_DIR := build
DATA_DIR := data
SHIM_DIR := src/shim
LIBS := -lzmq
CFLAGS :=
BUILD_ARCH := $(shell dpkg-architecture -q DEB_BUILD_ARCH)
-
J2LINT := $(shell command -v j2lint 2> /dev/null)
+PYLINT_FILES := $(shell git ls-files *.py)
config_xml_src = $(wildcard interface-definitions/*.xml.in)
config_xml_obj = $(config_xml_src:.xml.in=.xml)
op_xml_src = $(wildcard op-mode-definitions/*.xml.in)
op_xml_obj = $(op_xml_src:.xml.in=.xml)
%.xml: %.xml.in
@echo Generating $(BUILD_DIR)/$@ from $<
mkdir -p $(BUILD_DIR)/$(dir $@)
$(CURDIR)/scripts/transclude-template $< > $(BUILD_DIR)/$@
.PHONY: interface_definitions
.ONESHELL:
interface_definitions: $(config_xml_obj)
mkdir -p $(TMPL_DIR)
$(CURDIR)/scripts/override-default $(BUILD_DIR)/interface-definitions
find $(BUILD_DIR)/interface-definitions -type f -name "*.xml" | xargs -I {} $(CURDIR)/scripts/build-command-templates {} $(CURDIR)/schema/interface_definition.rng $(TMPL_DIR) || exit 1
$(CURDIR)/python/vyos/xml_ref/generate_cache.py --xml-dir $(BUILD_DIR)/interface-definitions || exit 1
# XXX: delete top level node.def's that now live in other packages
# IPSec VPN EAP-RADIUS does not support source-address
rm -rf $(TMPL_DIR)/vpn/ipsec/remote-access/radius/source-address
# T2472 - EIGRP support
rm -rf $(TMPL_DIR)/protocols/eigrp
# T2773 - EIGRP support for VRF
rm -rf $(TMPL_DIR)/vrf/name/node.tag/protocols/eigrp
# XXX: test if there are empty node.def files - this is not allowed as these
# could mask help strings or mandatory priority statements
find $(TMPL_DIR) -name node.def -type f -empty -exec false {} + || sh -c 'echo "There are empty node.def files! Check your interface definitions." && exit 1'
ifeq ($(BUILD_ARCH),arm64)
# There is currently no telegraf support in VyOS for ARM64, remove CLI definitions
rm -rf $(TMPL_DIR)/service/monitoring/telegraf
endif
.PHONY: op_mode_definitions
.ONESHELL:
op_mode_definitions: $(op_xml_obj)
mkdir -p $(OP_TMPL_DIR)
find $(BUILD_DIR)/op-mode-definitions/ -type f -name "*.xml" | xargs -I {} $(CURDIR)/scripts/build-command-op-templates {} $(CURDIR)/schema/op-mode-definition.rng $(OP_TMPL_DIR) || exit 1
# XXX: delete top level op mode node.def's that now live in other packages
rm -f $(OP_TMPL_DIR)/add/node.def
rm -f $(OP_TMPL_DIR)/clear/interfaces/node.def
rm -f $(OP_TMPL_DIR)/clear/node.def
rm -f $(OP_TMPL_DIR)/delete/node.def
# XXX: ping, traceroute and mtr must be able to recursivly call themselves as the
# options are provided from the scripts themselves
ln -s ../node.tag $(OP_TMPL_DIR)/ping/node.tag/node.tag/
ln -s ../node.tag $(OP_TMPL_DIR)/traceroute/node.tag/node.tag/
ln -s ../node.tag $(OP_TMPL_DIR)/mtr/node.tag/node.tag/
# XXX: test if there are empty node.def files - this is not allowed as these
# could mask help strings or mandatory priority statements
find $(OP_TMPL_DIR) -name node.def -type f -empty -exec false {} + || sh -c 'echo "There are empty node.def files! Check your interface definitions." && exit 1'
.PHONY: vyshim
vyshim:
$(MAKE) -C $(SHIM_DIR)
.PHONY: all
all: clean interface_definitions op_mode_definitions check test j2lint vyshim
.PHONY: check
.ONESHELL:
check:
@echo "Checking which CLI scripts are not enabled to work with vyos-configd..."
@for file in `ls src/conf_mode -I__pycache__`
do
if ! grep -q $$file data/configd-include.json; then
echo "* $$file"
fi
done
.PHONY: clean
clean:
rm -rf $(BUILD_DIR)
rm -rf $(TMPL_DIR)
rm -rf $(OP_TMPL_DIR)
$(MAKE) -C $(SHIM_DIR) clean
.PHONY: test
test:
set -e; python3 -m compileall -q -x '/vmware-tools/scripts/, /ppp/' .
PYTHONPATH=python/ python3 -m "nose" --with-xunit src --with-coverage --cover-erase --cover-xml --cover-package src/conf_mode,src/op_mode,src/completion,src/helpers,src/validators,src/tests --verbose
.PHONY: j2lint
j2lint:
ifndef J2LINT
$(error "j2lint binary not found, consider installing: pip install git+https://github.com/aristanetworks/j2lint.git@341b5d5db86")
endif
$(J2LINT) data/
.PHONY: sonar
sonar:
sonar-scanner -X -Dsonar.login=${SONAR_TOKEN}
.PHONY: unused-imports
unused-imports:
- git ls-files *.py | xargs pylint | grep W0611
+ @pylint --disable=all --enable=W0611 $(PYLINT_FILES)
deb:
dpkg-buildpackage -uc -us -tc -b
.PHONY: schema
schema:
trang -I rnc -O rng schema/interface_definition.rnc schema/interface_definition.rng
trang -I rnc -O rng schema/op-mode-definition.rnc schema/op-mode-definition.rng
diff --git a/src/services/api/graphql/generate/composite_function.py b/src/services/api/graphql/generate/composite_function.py
index bc9d80fbb..d6626fd1f 100644
--- a/src/services/api/graphql/generate/composite_function.py
+++ b/src/services/api/graphql/generate/composite_function.py
@@ -1,11 +1,7 @@
# typing information for composite functions: those that invoke several
# elementary requests, and return the result as a single dict
-import typing
-
def system_status():
pass
queries = {'system_status': system_status}
-
mutations = {}
-
diff --git a/src/services/api/graphql/graphql/auth_token_mutation.py b/src/services/api/graphql/graphql/auth_token_mutation.py
index 603a13758..a53fa4d60 100644
--- a/src/services/api/graphql/graphql/auth_token_mutation.py
+++ b/src/services/api/graphql/graphql/auth_token_mutation.py
@@ -1,61 +1,61 @@
-# Copyright 2022 VyOS maintainers and contributors <maintainers@vyos.io>
+# Copyright 2022-2024 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this library. If not, see <http://www.gnu.org/licenses/>.
-import jwt
import datetime
-from typing import Any, Dict
-from ariadne import ObjectType, UnionType
+from typing import Any
+from typing import Dict
+from ariadne import ObjectType
from graphql import GraphQLResolveInfo
from .. libs.token_auth import generate_token
from .. session.session import get_user_info
from .. import state
auth_token_mutation = ObjectType("Mutation")
@auth_token_mutation.field('AuthToken')
def auth_token_resolver(obj: Any, info: GraphQLResolveInfo, data: Dict):
# non-nullable fields
user = data['username']
passwd = data['password']
secret = state.settings['secret']
exp_interval = int(state.settings['app'].state.vyos_token_exp)
expiration = (datetime.datetime.now(tz=datetime.timezone.utc) +
datetime.timedelta(seconds=exp_interval))
res = generate_token(user, passwd, secret, expiration)
try:
res |= get_user_info(user)
except ValueError:
# non-existent user already caught
pass
if 'token' in res:
data['result'] = res
return {
"success": True,
"data": data
}
if 'errors' in res:
return {
"success": False,
"errors": res['errors']
}
return {
"success": False,
"errors": ['token generation failed']
}
diff --git a/src/services/api/graphql/graphql/directives.py b/src/services/api/graphql/graphql/directives.py
index a7919854a..3927aee58 100644
--- a/src/services/api/graphql/graphql/directives.py
+++ b/src/services/api/graphql/graphql/directives.py
@@ -1,87 +1,87 @@
-# Copyright 2021 VyOS maintainers and contributors <maintainers@vyos.io>
+# Copyright 2021-2024 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this library. If not, see <http://www.gnu.org/licenses/>.
-from ariadne import SchemaDirectiveVisitor, ObjectType
+from ariadne import SchemaDirectiveVisitor
from . queries import *
from . mutations import *
def non(arg):
pass
class VyosDirective(SchemaDirectiveVisitor):
def visit_field_definition(self, field, object_type, make_resolver=non):
name = f'{field.type}'
# field.type contains the return value of the mutation; trim value
# to produce canonical name
name = name.replace('Result', '', 1)
func = make_resolver(name)
field.resolve = func
return field
class ConfigSessionQueryDirective(VyosDirective):
"""
Class providing implementation of 'configsessionquery' directive in schema.
"""
def visit_field_definition(self, field, object_type):
super().visit_field_definition(field, object_type,
make_resolver=make_config_session_query_resolver)
class ConfigSessionMutationDirective(VyosDirective):
"""
Class providing implementation of 'configsessionmutation' directive in schema.
"""
def visit_field_definition(self, field, object_type):
super().visit_field_definition(field, object_type,
make_resolver=make_config_session_mutation_resolver)
class GenOpQueryDirective(VyosDirective):
"""
Class providing implementation of 'genopquery' directive in schema.
"""
def visit_field_definition(self, field, object_type):
super().visit_field_definition(field, object_type,
make_resolver=make_gen_op_query_resolver)
class GenOpMutationDirective(VyosDirective):
"""
Class providing implementation of 'genopmutation' directive in schema.
"""
def visit_field_definition(self, field, object_type):
super().visit_field_definition(field, object_type,
make_resolver=make_gen_op_mutation_resolver)
class CompositeQueryDirective(VyosDirective):
"""
Class providing implementation of 'system_status' directive in schema.
"""
def visit_field_definition(self, field, object_type):
super().visit_field_definition(field, object_type,
make_resolver=make_composite_query_resolver)
class CompositeMutationDirective(VyosDirective):
"""
Class providing implementation of 'system_status' directive in schema.
"""
def visit_field_definition(self, field, object_type):
super().visit_field_definition(field, object_type,
make_resolver=make_composite_mutation_resolver)
directives_dict = {"configsessionquery": ConfigSessionQueryDirective,
"configsessionmutation": ConfigSessionMutationDirective,
"genopquery": GenOpQueryDirective,
"genopmutation": GenOpMutationDirective,
"compositequery": CompositeQueryDirective,
"compositemutation": CompositeMutationDirective}
diff --git a/src/services/api/graphql/graphql/mutations.py b/src/services/api/graphql/graphql/mutations.py
index 8254e22b1..d115a8e94 100644
--- a/src/services/api/graphql/graphql/mutations.py
+++ b/src/services/api/graphql/graphql/mutations.py
@@ -1,137 +1,139 @@
-# Copyright 2021-2022 VyOS maintainers and contributors <maintainers@vyos.io>
+# Copyright 2021-2024 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this library. If not, see <http://www.gnu.org/licenses/>.
from importlib import import_module
-from typing import Any, Dict, Optional
from ariadne import ObjectType, convert_camel_case_to_snake
-from graphql import GraphQLResolveInfo
from makefun import with_signature
+# used below by func_sig
+from typing import Any, Dict, Optional # pylint: disable=W0611
+from graphql import GraphQLResolveInfo # pylint: disable=W0611
+
from .. import state
from .. libs import key_auth
from api.graphql.session.session import Session
from api.graphql.session.errors.op_mode_errors import op_mode_err_msg, op_mode_err_code
from vyos.opmode import Error as OpModeError
mutation = ObjectType("Mutation")
def make_mutation_resolver(mutation_name, class_name, session_func):
"""Dynamically generate a resolver for the mutation named in the
schema by 'mutation_name'.
Dynamic generation is provided using the package 'makefun' (via the
decorator 'with_signature'), which provides signature-preserving
function wrappers; it provides several improvements over, say,
functools.wraps.
:raise Exception:
raising ConfigErrors, or internal errors
"""
func_base_name = convert_camel_case_to_snake(class_name)
resolver_name = f'resolve_{func_base_name}'
func_sig = '(obj: Any, info: GraphQLResolveInfo, data: Optional[Dict]=None)'
@mutation.field(mutation_name)
@with_signature(func_sig, func_name=resolver_name)
async def func_impl(*args, **kwargs):
try:
auth_type = state.settings['app'].state.vyos_auth_type
if auth_type == 'key':
data = kwargs['data']
key = data['key']
auth = key_auth.auth_required(key)
if auth is None:
return {
"success": False,
"errors": ['invalid API key']
}
# We are finished with the 'key' entry, and may remove so as to
# pass the rest of data (if any) to function.
del data['key']
elif auth_type == 'token':
data = kwargs['data']
if data is None:
data = {}
info = kwargs['info']
user = info.context.get('user')
if user is None:
error = info.context.get('error')
if error is not None:
return {
"success": False,
"errors": [error]
}
return {
"success": False,
"errors": ['not authenticated']
}
else:
# AtrributeError will have already been raised if no
# vyos_auth_type; validation and defaultValue ensure it is
# one of the previous cases, so this is never reached.
pass
session = state.settings['app'].state.vyos_session
# one may override the session functions with a local subclass
try:
mod = import_module(f'api.graphql.session.override.{func_base_name}')
klass = getattr(mod, class_name)
except ImportError:
# otherwise, dynamically generate subclass to invoke subclass
# name based functions
klass = type(class_name, (Session,), {})
k = klass(session, data)
method = getattr(k, session_func)
result = method()
data['result'] = result
return {
"success": True,
"data": data
}
except OpModeError as e:
typename = type(e).__name__
msg = str(e)
return {
"success": False,
"errore": ['op_mode_error'],
"op_mode_error": {"name": f"{typename}",
"message": msg if msg else op_mode_err_msg.get(typename, "Unknown"),
"vyos_code": op_mode_err_code.get(typename, 9999)}
}
except Exception as error:
return {
"success": False,
"errors": [repr(error)]
}
return func_impl
def make_config_session_mutation_resolver(mutation_name):
return make_mutation_resolver(mutation_name, mutation_name,
convert_camel_case_to_snake(mutation_name))
def make_gen_op_mutation_resolver(mutation_name):
return make_mutation_resolver(mutation_name, mutation_name, 'gen_op_mutation')
def make_composite_mutation_resolver(mutation_name):
return make_mutation_resolver(mutation_name, mutation_name,
convert_camel_case_to_snake(mutation_name))
diff --git a/src/services/api/graphql/graphql/queries.py b/src/services/api/graphql/graphql/queries.py
index daccc19b2..717098259 100644
--- a/src/services/api/graphql/graphql/queries.py
+++ b/src/services/api/graphql/graphql/queries.py
@@ -1,137 +1,139 @@
-# Copyright 2021-2022 VyOS maintainers and contributors <maintainers@vyos.io>
+# Copyright 2021-2024 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this library. If not, see <http://www.gnu.org/licenses/>.
from importlib import import_module
-from typing import Any, Dict, Optional
from ariadne import ObjectType, convert_camel_case_to_snake
-from graphql import GraphQLResolveInfo
from makefun import with_signature
+# used below by func_sig
+from typing import Any, Dict, Optional # pylint: disable=W0611
+from graphql import GraphQLResolveInfo # pylint: disable=W0611
+
from .. import state
from .. libs import key_auth
from api.graphql.session.session import Session
from api.graphql.session.errors.op_mode_errors import op_mode_err_msg, op_mode_err_code
from vyos.opmode import Error as OpModeError
query = ObjectType("Query")
def make_query_resolver(query_name, class_name, session_func):
"""Dynamically generate a resolver for the query named in the
schema by 'query_name'.
Dynamic generation is provided using the package 'makefun' (via the
decorator 'with_signature'), which provides signature-preserving
function wrappers; it provides several improvements over, say,
functools.wraps.
:raise Exception:
raising ConfigErrors, or internal errors
"""
func_base_name = convert_camel_case_to_snake(class_name)
resolver_name = f'resolve_{func_base_name}'
func_sig = '(obj: Any, info: GraphQLResolveInfo, data: Optional[Dict]=None)'
@query.field(query_name)
@with_signature(func_sig, func_name=resolver_name)
async def func_impl(*args, **kwargs):
try:
auth_type = state.settings['app'].state.vyos_auth_type
if auth_type == 'key':
data = kwargs['data']
key = data['key']
auth = key_auth.auth_required(key)
if auth is None:
return {
"success": False,
"errors": ['invalid API key']
}
# We are finished with the 'key' entry, and may remove so as to
# pass the rest of data (if any) to function.
del data['key']
elif auth_type == 'token':
data = kwargs['data']
if data is None:
data = {}
info = kwargs['info']
user = info.context.get('user')
if user is None:
error = info.context.get('error')
if error is not None:
return {
"success": False,
"errors": [error]
}
return {
"success": False,
"errors": ['not authenticated']
}
else:
# AtrributeError will have already been raised if no
# vyos_auth_type; validation and defaultValue ensure it is
# one of the previous cases, so this is never reached.
pass
session = state.settings['app'].state.vyos_session
# one may override the session functions with a local subclass
try:
mod = import_module(f'api.graphql.session.override.{func_base_name}')
klass = getattr(mod, class_name)
except ImportError:
# otherwise, dynamically generate subclass to invoke subclass
# name based functions
klass = type(class_name, (Session,), {})
k = klass(session, data)
method = getattr(k, session_func)
result = method()
data['result'] = result
return {
"success": True,
"data": data
}
except OpModeError as e:
typename = type(e).__name__
msg = str(e)
return {
"success": False,
"errors": ['op_mode_error'],
"op_mode_error": {"name": f"{typename}",
"message": msg if msg else op_mode_err_msg.get(typename, "Unknown"),
"vyos_code": op_mode_err_code.get(typename, 9999)}
}
except Exception as error:
return {
"success": False,
"errors": [repr(error)]
}
return func_impl
def make_config_session_query_resolver(query_name):
return make_query_resolver(query_name, query_name,
convert_camel_case_to_snake(query_name))
def make_gen_op_query_resolver(query_name):
return make_query_resolver(query_name, query_name, 'gen_op_query')
def make_composite_query_resolver(query_name):
return make_query_resolver(query_name, query_name,
convert_camel_case_to_snake(query_name))
diff --git a/src/services/api/graphql/libs/op_mode.py b/src/services/api/graphql/libs/op_mode.py
index 5022f7d4e..86e38eae6 100644
--- a/src/services/api/graphql/libs/op_mode.py
+++ b/src/services/api/graphql/libs/op_mode.py
@@ -1,101 +1,103 @@
-# Copyright 2022 VyOS maintainers and contributors <maintainers@vyos.io>
+# Copyright 2022-2024 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this library. If not, see <http://www.gnu.org/licenses/>.
import os
import re
import typing
-from typing import Union, Tuple, Optional
+
+from typing import Union
+from typing import Optional
from humps import decamelize
from vyos.defaults import directories
from vyos.utils.system import load_as_module
from vyos.opmode import _normalize_field_names
from vyos.opmode import _is_literal_type, _get_literal_values
def load_op_mode_as_module(name: str):
path = os.path.join(directories['op_mode'], name)
name = os.path.splitext(name)[0].replace('-', '_')
return load_as_module(name, path)
def is_show_function_name(name):
if re.match(r"^show", name):
return True
return False
def _nth_split(delim: str, n: int, s: str):
groups = s.split(delim)
l = len(groups)
if n > l-1 or n < 1:
return (s, '')
return (delim.join(groups[:n]), delim.join(groups[n:]))
def _nth_rsplit(delim: str, n: int, s: str):
groups = s.split(delim)
l = len(groups)
if n > l-1 or n < 1:
return (s, '')
return (delim.join(groups[:l-n]), delim.join(groups[l-n:]))
# Since we have mangled possible hyphens in the file name while constructing
# the snake case of the query/mutation name, we will need to recover the
# file name by searching with mangling:
def _filter_on_mangled(test):
def func(elem):
mangle = os.path.splitext(elem)[0].replace('-', '_')
return test == mangle
return func
# Find longest name in concatenated string that matches the basename of an
# op-mode script. Should one prefer to concatenate in the reverse order
# (script_name + '_' + function_name), use _nth_rsplit.
def split_compound_op_mode_name(name: str, files: list):
for i in range(1, name.count('_') + 1):
pair = _nth_split('_', i, name)
f = list(filter(_filter_on_mangled(pair[1]), files))
if f:
pair = (pair[0], f[0])
return pair
return (name, '')
def snake_to_pascal_case(name: str) -> str:
res = ''.join(map(str.title, name.split('_')))
return res
def map_type_name(type_name: type, enums: Optional[dict] = None, optional: bool = False) -> str:
if type_name == str:
return 'String!' if not optional else 'String = null'
if type_name == int:
return 'Int!' if not optional else 'Int = null'
if type_name == bool:
return 'Boolean = false'
if typing.get_origin(type_name) == list:
if not optional:
return f'[{map_type_name(typing.get_args(type_name)[0], enums=enums)}]!'
return f'[{map_type_name(typing.get_args(type_name)[0], enums=enums)}]'
if _is_literal_type(type_name):
mapped = enums.get(_get_literal_values(type_name), '')
if not mapped:
raise ValueError(typing.get_args(type_name))
return f'{mapped}!' if not optional else mapped
# typing.Optional is typing.Union[_, NoneType]
if (typing.get_origin(type_name) is typing.Union and
typing.get_args(type_name)[1] == type(None)):
return f'{map_type_name(typing.get_args(type_name)[0], enums=enums, optional=True)}'
# scalar 'Generic' is defined in schema.graphql
return 'Generic'
def normalize_output(result: Union[dict, list]) -> Union[dict, list]:
return _normalize_field_names(decamelize(result))
diff --git a/src/services/api/graphql/session/composite/system_status.py b/src/services/api/graphql/session/composite/system_status.py
index d809f32e3..516a4eff6 100755
--- a/src/services/api/graphql/session/composite/system_status.py
+++ b/src/services/api/graphql/session/composite/system_status.py
@@ -1,38 +1,29 @@
#!/usr/bin/env python3
#
-# Copyright (C) 2022 VyOS maintainers and contributors
+# Copyright (C) 2022-2024 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
-#
-#
-
-import os
-import sys
-import json
-import importlib.util
-
-from vyos.defaults import directories
from api.graphql.libs.op_mode import load_op_mode_as_module
def get_system_version() -> dict:
show_version = load_op_mode_as_module('version.py')
return show_version.show(raw=True, funny=False)
def get_system_uptime() -> dict:
show_uptime = load_op_mode_as_module('uptime.py')
return show_uptime._get_raw_data()
def get_system_ram_usage() -> dict:
show_ram = load_op_mode_as_module('memory.py')
return show_ram.show(raw=True)
diff --git a/src/services/api/graphql/session/session.py b/src/services/api/graphql/session/session.py
index 3c5a062b6..6ae44b9bf 100644
--- a/src/services/api/graphql/session/session.py
+++ b/src/services/api/graphql/session/session.py
@@ -1,212 +1,211 @@
-# Copyright 2021-2022 VyOS maintainers and contributors <maintainers@vyos.io>
+# Copyright 2021-2024 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this library. If not, see <http://www.gnu.org/licenses/>.
import os
import json
from ariadne import convert_camel_case_to_snake
from vyos.config import Config
from vyos.configtree import ConfigTree
from vyos.defaults import directories
-from vyos.template import render
from vyos.opmode import Error as OpModeError
from api.graphql.libs.op_mode import load_op_mode_as_module, split_compound_op_mode_name
from api.graphql.libs.op_mode import normalize_output
op_mode_include_file = os.path.join(directories['data'], 'op-mode-standardized.json')
def get_config_dict(path=[], effective=False, key_mangling=None,
get_first_key=False, no_multi_convert=False,
no_tag_node_value_mangle=False):
config = Config()
return config.get_config_dict(path=path, effective=effective,
key_mangling=key_mangling,
get_first_key=get_first_key,
no_multi_convert=no_multi_convert,
no_tag_node_value_mangle=no_tag_node_value_mangle)
def get_user_info(user):
user_info = {}
info = get_config_dict(['system', 'login', 'user', user],
get_first_key=True)
if not info:
raise ValueError("No such user")
user_info['user'] = user
user_info['full_name'] = info.get('full-name', '')
return user_info
class Session:
"""
Wrapper for calling configsession functions based on GraphQL requests.
Non-nullable fields in the respective schema allow avoiding a key check
in 'data'.
"""
def __init__(self, session, data):
self._session = session
self._data = data
self._name = convert_camel_case_to_snake(type(self).__name__)
try:
with open(op_mode_include_file) as f:
self._op_mode_list = json.loads(f.read())
except Exception:
self._op_mode_list = None
def show_config(self):
session = self._session
data = self._data
out = ''
try:
out = session.show_config(data['path'])
if data.get('config_format', '') == 'json':
config_tree = ConfigTree(out)
out = json.loads(config_tree.to_json())
except Exception as error:
raise error
return out
def save_config_file(self):
session = self._session
data = self._data
if 'file_name' not in data or not data['file_name']:
data['file_name'] = '/config/config.boot'
try:
session.save_config(data['file_name'])
except Exception as error:
raise error
def load_config_file(self):
session = self._session
data = self._data
try:
session.load_config(data['file_name'])
session.commit()
except Exception as error:
raise error
def show(self):
session = self._session
data = self._data
out = ''
try:
out = session.show(data['path'])
except Exception as error:
raise error
return out
def add_system_image(self):
session = self._session
data = self._data
try:
res = session.install_image(data['location'])
except Exception as error:
raise error
return res
def delete_system_image(self):
session = self._session
data = self._data
try:
res = session.remove_image(data['name'])
except Exception as error:
raise error
return res
def show_user_info(self):
session = self._session
data = self._data
user_info = {}
user = data['user']
try:
user_info = get_user_info(user)
except Exception as error:
raise error
return user_info
def system_status(self):
import api.graphql.session.composite.system_status as system_status
session = self._session
data = self._data
status = {}
status['host_name'] = session.show(['host', 'name']).strip()
status['version'] = system_status.get_system_version()
status['uptime'] = system_status.get_system_uptime()
status['ram'] = system_status.get_system_ram_usage()
return status
def gen_op_query(self):
session = self._session
data = self._data
name = self._name
op_mode_list = self._op_mode_list
# handle the case that the op-mode file contains underscores:
if op_mode_list is None:
raise FileNotFoundError(f"No op-mode file list at '{op_mode_include_file}'")
(func_name, scriptname) = split_compound_op_mode_name(name, op_mode_list)
if scriptname == '':
raise FileNotFoundError(f"No op-mode file named in string '{name}'")
mod = load_op_mode_as_module(f'{scriptname}')
func = getattr(mod, func_name)
try:
res = func(True, **data)
except OpModeError as e:
raise e
res = normalize_output(res)
return res
def gen_op_mutation(self):
session = self._session
data = self._data
name = self._name
op_mode_list = self._op_mode_list
# handle the case that the op-mode file name contains underscores:
if op_mode_list is None:
raise FileNotFoundError(f"No op-mode file list at '{op_mode_include_file}'")
(func_name, scriptname) = split_compound_op_mode_name(name, op_mode_list)
if scriptname == '':
raise FileNotFoundError(f"No op-mode file named in string '{name}'")
mod = load_op_mode_as_module(f'{scriptname}')
func = getattr(mod, func_name)
try:
res = func(**data)
except OpModeError as e:
raise e
return res