Skip to content
Snippets Groups Projects
Verified Commit 33851d5a authored by Martin Matějek's avatar Martin Matějek
Browse files

initial functionality

parent 708a3d35
Branches
Tags
1 merge request!1initial functionality
Pipeline #69882 passed with stage
in 54 seconds
Showing
with 1287 additions and 2 deletions
.eggs/
.mypy/
*.un~
.mypy_cache/
.*~
before_script:
- pip install virtualenv
- virtualenv -p "$(which python)" /tmp/test
.run_script: &run_script
script:
- source /tmp/test/bin/activate
- pip install .
- python setup.py test --addopts="--backend openwrt --backend mock"
test_python3:
image: registry.labs.nic.cz/turris/foris-ci/python3
<<: *run_script
flake8:
image: registry.labs.nic.cz/turris/foris-ci/python3
script:
- python setup.py flake8
0.0 (YYYY-MM-DD)
----------------
* some change
LICENSE 0 → 100644
This diff is collapsed.
Foris Controller haas module
==============================
This is a haas module for foris-controller
Requirements
============
* python3
* foris-controller
Installation
============
```
pip install .
```
__import__("pkg_resources").declare_namespace(__name__)
#
# foris-controller-haas-module
# Copyright (C) 2020 CZ.NIC, z.s.p.o. (http://www.nic.cz/)
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# 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, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
#
import logging
from foris_controller_backends.uci import UciBackend, get_option_named
from foris_controller_backends.services import OpenwrtServices
logger = logging.getLogger(__name__)
class HaasUci:
SERVICE = "haas-proxy"
@classmethod
def get_settings(cls):
with UciBackend() as backend:
haas_data = backend.read("haas")
token = get_option_named(haas_data, "haas", "settings", "token", "")
with OpenwrtServices() as services:
enabled = services.is_enabled(HaasUci.SERVICE)
return {
"token": token,
"enabled": enabled,
}
@classmethod
def update_settings(cls, token, enabled):
with UciBackend() as backend:
backend.set_option("haas", "settings", "token", token)
with OpenwrtServices() as services:
if enabled:
services.enable(HaasUci.SERVICE, fail_on_error=False)
services.restart(HaasUci.SERVICE, fail_on_error=False)
else:
services.disable(HaasUci.SERVICE, fail_on_error=False)
services.stop(HaasUci.SERVICE, fail_on_error=False)
return True
#
# foris-controller-haas-module
# Copyright (C) 2020 CZ.NIC, z.s.p.o. (http://www.nic.cz/)
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# 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, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
#
__version__ = "0.0"
__import__("pkg_resources").declare_namespace(__name__)
#
# foris-controller-haas-module
# Copyright (C) 2020 CZ.NIC, z.s.p.o. (http://www.nic.cz/)
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# 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, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
#
import logging
from foris_controller.module_base import BaseModule
from foris_controller.handler_base import wrap_required_functions
class HaasModule(BaseModule):
logger = logging.getLogger(__name__)
def action_get_settings(self, data):
return self.handler.get_settings()
def action_update_settings(self, data):
return {"result": self.handler.update_settings(data)}
@wrap_required_functions(["get_settings", "update_settings"])
class Handler(object):
pass
#
# foris-controller-haas-module
# Copyright (C) 2020 CZ.NIC, z.s.p.o. (http://www.nic.cz/)
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# 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, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
#
from .mock import MockHaasHandler
from .openwrt import OpenwrtHaasHandler
__all__ = ["MockHaasHandler", "OpenwrtHaasHandler"]
#
# foris-controller-haas-module
# Copyright (C) 2020 CZ.NIC, z.s.p.o. (http://www.nic.cz/)
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# 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, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
#
import logging
from foris_controller.handler_base import BaseMockHandler
from foris_controller.utils import logger_wrapper
from .. import Handler
logger = logging.getLogger(__name__)
class MockHaasHandler(Handler, BaseMockHandler):
settings = {
"enabled": True, # service enabled/disabled
"token": "",
}
@staticmethod
@logger_wrapper(logger)
def get_settings():
return MockHaasHandler.settings
@staticmethod
@logger_wrapper(logger)
def update_settings(data):
MockHaasHandler.settings["token"] = data["token"]
MockHaasHandler.settings["enabled"] = data["enabled"]
return True
#
# foris-controller-haas-module
# Copyright (C) 2020 CZ.NIC, z.s.p.o. (http://www.nic.cz/)
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# 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, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
#
import logging
from foris_controller.handler_base import BaseOpenwrtHandler
from foris_controller.utils import logger_wrapper
from foris_controller_backends.haas import HaasUci
from .. import Handler
logger = logging.getLogger(__name__)
class OpenwrtHaasHandler(Handler, BaseOpenwrtHandler):
uci = HaasUci()
@staticmethod
@logger_wrapper(logger)
def get_settings():
return OpenwrtHaasHandler.uci.get_settings()
@staticmethod
@logger_wrapper(logger)
def update_settings(data):
return OpenwrtHaasHandler.uci.update_settings(**data)
{
"definitions": {
"token": {"type": "string", "pattern": "[a-f0-9]{32}"}
},
"oneOf": [
{
"description": "Get haas settings request",
"properties": {
"module": {"enum": ["haas"]},
"kind": {"enum": ["request"]},
"action": {"enum": ["get_settings"]}
},
"additionalProperties": false
},
{
"description": "Get haas settings reply",
"properties": {
"module": {"enum": ["haas"]},
"kind": {"enum": ["reply"]},
"action": {"enum": ["get_settings"]},
"data": {
"type": "object",
"properties": {
"token": {
"oneOf": [
{"$ref": "#/definitions/token"},
{"enum": [""]}
]
},
"enabled": {"type": "boolean"}
},
"additionalProperties": false,
"required": ["token", "enabled"]
}
},
"additionalProperties": false,
"required": ["data"]
},
{
"description": "Update haas settings request",
"properties": {
"module": {"enum": ["haas"]},
"kind": {"enum": ["request"]},
"action": {"enum": ["update_settings"]},
"data": {
"type": "object",
"properties": {
"token": {"$ref": "#/definitions/token"},
"enabled": {"type": "boolean"}
},
"additionalProperties": false,
"required": ["token", "enabled"]
}
},
"additionalProperties": false,
"required": ["data"]
},
{
"description": "Update haas settings reply",
"properties": {
"module": {"enum": ["haas"]},
"kind": {"enum": ["reply"]},
"action": {"enum": ["update_settings"]},
"data": {
"type": "object",
"properties": {
"result": {"type": "boolean"}
},
"additionalProperties": false,
"required": ["result"]
}
},
"additionalProperties": false,
"required": ["data"]
}
]
}
[aliases]
test=pytest
[tool:pytest]
addopts = --verbose -s
testpaths = tests
python_files = test_*.py
[flake8]
max-line-length = 100
select = C,E,F,W,B,B950
ignore = E203, E231, E501, W503
per-file-ignores =
tests/*: F811, F401
setup.py 0 → 100644
#
# foris-controller-haas-module
# Copyright (C) 2020 CZ.NIC, z.s.p.o. (http://www.nic.cz/)
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# 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, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
#
from setuptools import setup
from foris_controller_haas_module import __version__
DESCRIPTION = """
haas module for Foris Controller
"""
setup(
name="foris-controller-haas-module",
version=__version__,
author="CZ.NIC, z.s.p.o. (http://www.nic.cz/)",
author_email="packaging@turris.cz",
packages=[
"foris_controller_haas_module",
"foris_controller_backends",
"foris_controller_backends.haas",
"foris_controller_modules",
"foris_controller_modules.haas",
"foris_controller_modules.haas.handlers",
],
package_data={"foris_controller_modules.haas": ["schema", "schema/*.json"]},
namespace_packages=["foris_controller_modules", "foris_controller_backends"],
description=DESCRIPTION,
long_description=open("README.md").read(),
long_description_content_type="text/markdown",
install_requires=[
"foris-controller @ git+https://gitlab.nic.cz/turris/foris-controller/foris-controller.git"
],
setup_requires=["pytest-runner", "flake8"],
tests_require=["pytest", "foris-controller-testtools", "foris-client", "ubus", "paho-mqtt"],
dependency_links=[
"git+https://gitlab.nic.cz/turris/foris-controller/foris-controller-testtools.git#egg=foris-controller-testtools",
"git+https://gitlab.nic.cz/turris/foris-controller/foris-client.git#egg=foris-client",
],
include_package_data=True,
zip_safe=False,
)
#
# foris-controller-haas-module
# Copyright (C) 2020 CZ.NIC, z.s.p.o. (http://www.nic.cz/)
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# 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, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
#
import pytest
import os
# load common fixtures
from foris_controller_testtools.fixtures import (
uci_config_default_path,
env_overrides,
cmdline_script_root,
controller_modules,
extra_module_paths,
message_bus,
backend,
)
@pytest.fixture(scope="session")
def uci_config_default_path():
return os.path.join(os.path.dirname(os.path.realpath(__file__)), "uci_configs")
@pytest.fixture(scope="session")
def cmdline_script_root():
return os.path.join(os.path.dirname(os.path.realpath(__file__)), "test_root")
@pytest.fixture(scope="session")
def file_root():
# default src dirctory will be the same as for the scripts (could be override later)
return os.path.join(os.path.dirname(os.path.realpath(__file__)), "test_root")
@pytest.fixture(scope="module")
def controller_modules():
return ["remote", "haas"]
def pytest_addoption(parser):
parser.addoption(
"--backend",
action="append",
default=[],
help=("Set test backend here. available values = (mock, openwrt)"),
)
parser.addoption(
"--message-bus",
action="append",
default=[],
help=("Set test bus here. available values = (unix-socket, ubus, mqtt)"),
)
parser.addoption(
"--debug-output",
action="store_true",
default=False,
help=("Whether show output of foris-controller cmd"),
)
def pytest_generate_tests(metafunc):
if "backend" in metafunc.fixturenames:
backend = metafunc.config.option.backend
if not backend:
backend = ["openwrt"]
metafunc.parametrize("backend_param", backend, scope="module")
if "message_bus" in metafunc.fixturenames:
message_bus = metafunc.config.option.message_bus
if not message_bus:
message_bus = ["mqtt"]
metafunc.parametrize("message_bus_param", message_bus, scope="module")
#
# foris-controller-haas-module
# Copyright (C) 2020 CZ.NIC, z.s.p.o. (http://www.nic.cz/)
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# 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, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
#
import pytest
from foris_controller_testtools.fixtures import (
only_message_buses,
only_backends,
infrastructure,
notify_api,
uci_configs_init,
file_root_init,
UCI_CONFIG_DIR_PATH,
init_script_result,
)
from foris_controller_testtools.utils import get_uci_module, check_service_result, INIT_SCRIPT_TEST_DIR
def test_get_settings(file_root_init, infrastructure):
res = infrastructure.process_message(
{"module": "haas", "action": "get_settings", "kind": "request"}
)
assert "errors" not in res
assert "data" in res
assert res["data"]["enabled"] is True
assert res["data"]["token"] == "" # initial settings with unset token
def test_update_settings(file_root_init, uci_configs_init, infrastructure):
new_token = "81f2cd612ea14da5bbaeaf08e7dc2a39"
data = {"token": new_token, "enabled": False}
res = infrastructure.process_message(
{
"module": "haas",
"action": "update_settings",
"kind": "request",
"data": data,
}
)
assert res == {
"module": "haas",
"action": "update_settings",
"kind": "reply",
"data": {"result": True},
}
res = infrastructure.process_message(
{
"module": "haas",
"action": "get_settings",
"kind": "request",
}
)
assert res["data"]["token"] == new_token
@pytest.mark.only_backends(["openwrt"])
def test_update_settings_uci(file_root_init, uci_configs_init, infrastructure):
new_token = "81f2cd612ea14da5bbaeaf08e7dc2a39"
data = {"token": new_token, "enabled": False}
res = infrastructure.process_message(
{
"module": "haas",
"action": "update_settings",
"kind": "request",
"data": data,
}
)
assert res == {
"module": "haas",
"action": "update_settings",
"kind": "reply",
"data": {"result": True},
}
uci = get_uci_module(infrastructure.name)
with uci.UciBackend(UCI_CONFIG_DIR_PATH) as backend:
data = backend.read()
token = uci.get_option_named(data, "haas", "settings", "token")
assert new_token == token
pass
\ No newline at end of file
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment