functionality implemented

.gitlab-ci.yml

+before_script:
+  - pip install virtualenv
+  - virtualenv -p "$(which python)" /tmp/test
+
+stages:
+  - 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
+  stage: test
+  <<: *run_script

CHANGELOG.rst

+0.0 (YYYY-MM-DD)
+----------------
+
+* some change

README.rst

+Foris Controller Schnapps module
+==============================
+This is a schnapps module for foris-controller
+
+Requirements
+============
+
+* python3
+* foris-controller
+
+Installation
+============
+
+	``python3 setup.py install``

foris_controller_backends/__init__.py

+__import__("pkg_resources").declare_namespace(__name__)

foris_controller_backends/schnapps/__init__.py

+#
+# foris-controller-schnapps-module
+# Copyright (C) 2019 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
+import json
+import re
+import typing
+
+from foris_controller_backends.cmdline import BaseCmdLine
+
+logger = logging.getLogger(__name__)
+
+
+class SchnappsCmds(BaseCmdLine):
+    def list(self) -> dict:
+        stdout, _ = self._run_command_and_check_retval(["/usr/bin/schnapps", "list", "-j"], 0)
+        return json.loads(stdout.strip())
+
+    def create(self, description: str) -> int:
+        stdout, _ = self._run_command_and_check_retval(
+            ["/usr/bin/schnapps", "create", "-t", "single", description], 0
+        )
+        res = re.match(r"Snapshot number (\d+) created", stdout.decode().strip())
+        return int(res.group(1))
+
+    def delete(self, number: int) -> bool:
+        ret, stdout, _ = self._run_command("/usr/bin/schnapps", "delete", str(number))
+        return ret == 0
+
+    def rollback(self, number: int) -> typing.Tuple[bool, typing.Optional[typing.List[int]]]:
+        snapshots = self.list()["snapshots"]
+        ret, stdout, _ = self._run_command("/usr/bin/schnapps", "rollback", str(number))
+        self._run_command("/usr/bin/maintain-reboot-needed")  # best effort
+        if ret == 0:
+            return True, [e["number"] for e in snapshots if e["number"] > number]
+        else:
+            return False, None

foris_controller_modules/__init__.py

+__import__("pkg_resources").declare_namespace(__name__)

foris_controller_modules/schnapps/__init__.py

+#
+# foris-controller-schnapps-module
+# Copyright (C) 2019 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 SchnappsModule(BaseModule):
+    logger = logging.getLogger(__name__)
+
+    def action_list(self, data: dict) -> dict:
+        return self.handler.list()
+
+    def action_create(self, data: dict) -> dict:
+        res = self.handler.create(**data)
+        self.notify("create", {"number": res})
+        return {"number": res}
+
+    def action_delete(self, data: dict) -> dict:
+        res = self.handler.delete(**data)
+        if res:
+            self.notify("delete", data)
+        return {"result": res}
+
+    def action_rollback(self, data: dict) -> dict:
+        res, shifted = self.handler.rollback(**data)
+        response = {"result": res}
+        if res:
+            self.notify("rollback", {"number": data["number"], "shifted": shifted})
+            response["shifted"] = shifted
+        return response
+
+
+@wrap_required_functions(["list", "create", "delete", "rollback"])
+class Handler(object):
+    pass

foris_controller_modules/schnapps/handlers/__init__.py

+#
+# foris-controller-schnapps-module
+# Copyright (C) 2019 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 MockSchnappsHandler
+from .openwrt import OpenwrtSchnappsHandler
+
+__all__ = ["MockSchnappsHandler", "OpenwrtSchnappsHandler"]

foris_controller_modules/schnapps/handlers/mock.py

+#
+# foris-controller-schnapps-module
+# Copyright (C) 2019 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 datetime
+import logging
+import typing
+import random
+
+from foris_controller.handler_base import BaseMockHandler
+from foris_controller.utils import logger_wrapper
+
+from .. import Handler
+
+logger = logging.getLogger(__name__)
+
+
+class MockSchnappsHandler(Handler, BaseMockHandler):
+    idx = 0
+
+    snapshots: typing.List[dict] = []
+
+    @logger_wrapper(logger)
+    def list(self) -> dict:
+        return {"snapshots": MockSchnappsHandler.snapshots}
+
+    @logger_wrapper(logger)
+    def create(self, description: str) -> int:
+        MockSchnappsHandler.idx += 1
+        MockSchnappsHandler.snapshots.append(
+            {
+                "number": MockSchnappsHandler.idx,
+                "created": datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S +0000"),
+                "type": "single",
+                "size": f"{random.randint(1, 100000) / 100}MiB",
+                "description": description,
+            }
+        )
+        return MockSchnappsHandler.idx
+
+    @logger_wrapper(logger)
+    def delete(self, number: int) -> bool:
+        ids = [e["number"] for e in MockSchnappsHandler.snapshots]
+        if number not in ids:
+            return False
+        MockSchnappsHandler.snapshots = [
+            e for e in MockSchnappsHandler.snapshots if e["number"] != number
+        ]
+        return True
+
+    @logger_wrapper(logger)
+    def rollback(self, number: int) -> typing.Tuple[bool, typing.Optional[typing.List[int]]]:
+        ids = [e["number"] for e in MockSchnappsHandler.snapshots]
+        if number not in ids:
+            return False, None
+        return True, [e["number"] for e in MockSchnappsHandler.snapshots if e["number"] > number]

foris_controller_modules/schnapps/handlers/openwrt.py

+#
+# foris-controller-schnapps-module
+# Copyright (C) 2019 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
+import typing
+
+from foris_controller.handler_base import BaseOpenwrtHandler
+from foris_controller.utils import logger_wrapper
+
+from foris_controller_backends.schnapps import SchnappsCmds
+
+from .. import Handler
+
+logger = logging.getLogger(__name__)
+
+
+class OpenwrtSchnappsHandler(Handler, BaseOpenwrtHandler):
+
+    cmds = SchnappsCmds()
+
+    @logger_wrapper(logger)
+    def get_slices(self):
+        return OpenwrtSchnappsHandler.cmds.get_slices()
+
+    @logger_wrapper(logger)
+    def set_slices(self, value):
+        return OpenwrtSchnappsHandler.uci.set_slices(value)
+
+    @logger_wrapper(logger)
+    def list(self) -> dict:
+        return OpenwrtSchnappsHandler.cmds.list()
+
+    @logger_wrapper(logger)
+    def create(self, description: str) -> int:
+        return OpenwrtSchnappsHandler.cmds.create(description)
+
+    @logger_wrapper(logger)
+    def delete(self, number: int) -> bool:
+        return OpenwrtSchnappsHandler.cmds.delete(number)
+
+    @logger_wrapper(logger)
+    def rollback(self, number: int) -> typing.Tuple[bool, typing.Optional[typing.List[int]]]:
+        return OpenwrtSchnappsHandler.cmds.rollback(number)

foris_controller_modules/schnapps/schema/schnapps.json

+{
+    "definitions": {
+        "number": {"type": "integer", "minimum": 1},
+        "type": {"enum": ["pre", "post", "type", "single"]},
+        "snapshot": {
+            "type": "object",
+            "properties": {
+                "number": {"$ref": "#/definitions/number"},
+                "type": {"$ref": "#/definitions/type"},
+                "created": {"type": "string"},
+                "size": {"type": "string"},
+                "description": {"type": "string"}
+            },
+            "required": ["number", "size", "description", "type", "created"],
+            "additionalProperties": false
+        }
+    },
+    "oneOf": [
+        {
+            "description": "List snapshots request",
+            "properties": {
+                "module": {"enum": ["schnapps"]},
+                "kind": {"enum": ["request"]},
+                "action": {"enum": ["list"]}
+            },
+            "additionalProperties": false
+        },
+        {
+            "description": "List snapshots reply",
+            "properties": {
+                "module": {"enum": ["schnapps"]},
+                "kind": {"enum": ["reply"]},
+                "action": {"enum": ["list"]},
+                "data": {
+                    "type": "object",
+                    "properties": {
+                        "snapshots": {
+                            "type": "array",
+                            "items": {"$ref": "#/definitions/snapshot"}
+                        }
+                    },
+                    "additionalProperties": false,
+                    "required": ["snapshots"]
+                }
+            },
+            "additionalProperties": false,
+            "required": ["data"]
+        },
+        {
+            "description": "Create snapshot request",
+            "properties": {
+                "module": {"enum": ["schnapps"]},
+                "kind": {"enum": ["request"]},
+                "action": {"enum": ["create"]},
+                "data": {
+                    "type": "object",
+                    "properties": {
+                        "description": {"type": "string"}
+                    },
+                    "additionalProperties": false,
+                    "required": ["description"]
+                }
+            },
+            "additionalProperties": false,
+            "required": ["data"]
+        },
+        {
+            "description": "Create snapshot reply",
+            "properties": {
+                "module": {"enum": ["schnapps"]},
+                "kind": {"enum": ["reply"]},
+                "action": {"enum": ["create"]},
+                "data": {
+                    "type": "object",
+                    "properties": {
+                        "number": {"$ref": "#/definitions/number"}
+                    },
+                    "additionalProperties": false,
+                    "required": ["number"]
+                }
+            },
+            "additionalProperties": false,
+            "required": ["data"]
+        },
+        {
+            "description": "Create snapshot notification",
+            "properties": {
+                "module": {"enum": ["schnapps"]},
+                "kind": {"enum": ["notification"]},
+                "action": {"enum": ["create"]},
+                "data": {
+                    "type": "object",
+                    "properties": {
+                        "number": {"$ref": "#/definitions/number"}
+                    },
+                    "additionalProperties": false,
+                    "required": ["number"]
+                }
+            },
+            "additionalProperties": false,
+            "required": ["data"]
+        },
+        {
+            "description": "Delete snapshot request",
+            "properties": {
+                "module": {"enum": ["schnapps"]},
+                "kind": {"enum": ["request"]},
+                "action": {"enum": ["delete"]},
+                "data": {
+                    "type": "object",
+                    "properties": {
+                        "number": {"$ref": "#/definitions/number"}
+                    },
+                    "additionalProperties": false,
+                    "required": ["number"]
+                }
+            },
+            "additionalProperties": false,
+            "required": ["data"]
+        },
+        {
+            "description": "Delete snapshot reply",
+            "properties": {
+                "module": {"enum": ["schnapps"]},
+                "kind": {"enum": ["reply"]},
+                "action": {"enum": ["delete"]},
+                "data": {
+                    "type": "object",
+                    "properties": {
+                        "result": {"type": "boolean"}
+                    },
+                    "additionalProperties": false,
+                    "required": ["result"]
+                }
+            },
+            "additionalProperties": false,
+            "required": ["data"]
+        },
+        {
+            "description": "Delete snapshot notification",
+            "properties": {
+                "module": {"enum": ["schnapps"]},
+                "kind": {"enum": ["notification"]},
+                "action": {"enum": ["delete"]},
+                "data": {
+                    "type": "object",
+                    "properties": {
+                        "number": {"$ref": "#/definitions/number"}
+                    },
+                    "additionalProperties": false,
+                    "required": ["number"]
+                }
+            },
+            "additionalProperties": false,
+            "required": ["data"]
+        },
+        {
+            "description": "Rollback snapshot request",
+            "properties": {
+                "module": {"enum": ["schnapps"]},
+                "kind": {"enum": ["request"]},
+                "action": {"enum": ["rollback"]},
+                "data": {
+                    "type": "object",
+                    "properties": {
+                        "number": {"$ref": "#/definitions/number"}
+                    },
+                    "additionalProperties": false,
+                    "required": ["number"]
+                }
+            },
+            "additionalProperties": false,
+            "required": ["data"]
+        },
+        {
+            "description": "Rollback snapshot reply",
+            "properties": {
+                "module": {"enum": ["schnapps"]},
+                "kind": {"enum": ["reply"]},
+                "action": {"enum": ["rollback"]},
+                "data": {
+                    "oneOf": [
+                        {
+                            "type": "object",
+                            "properties": {
+                                "result": {"enum": [false]}
+                            },
+                            "additionalProperties": false,
+                            "required": ["result"]
+                        },
+                        {
+                            "type": "object",
+                            "properties": {
+                                "result": {"enum": [true]},
+                                "shifted": {"type": "array", "items":{"$ref": "#/definitions/number"}} 
+                            },
+                            "additionalProperties": false,
+                            "required": ["result", "shifted"]
+                        }
+                    ]
+                }
+            },
+            "additionalProperties": false,
+            "required": ["data"]
+        },
+        {
+            "description": "Rollback snapshot notification",
+            "properties": {
+                "module": {"enum": ["schnapps"]},
+                "kind": {"enum": ["notification"]},
+                "action": {"enum": ["rollback"]},
+                "data": {
+                    "type": "object",
+                    "properties": {
+                        "number": {"$ref": "#/definitions/number"},
+                        "shifted": {"type": "array", "items":{"$ref": "#/definitions/number"}} 
+                    },
+                    "additionalProperties": false,
+                    "required": ["number", "shifted"]
+                }
+            },
+            "additionalProperties": false,
+            "required": ["data"]
+        }
+    ]
+}

foris_controller_schnapps_module/__init__.py

+#
+# foris-controller-schnapps-module
+# Copyright (C) 2019 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"

setup.cfg

+[aliases]
+test=pytest
+
+[tool:pytest]
+addopts = --verbose -s
+testpaths = tests
+python_files = test_*.py

setup.py

+#
+# foris-controller-schnapps-module
+# Copyright (C) 2019 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_schnapps_module import __version__
+
+DESCRIPTION = """
+Schnapps module for Foris Controller
+"""
+
+setup(
+    name="foris-controller-schnapps-module",
+    version=__version__,
+    author="CZ.NIC, z.s.p.o. (http://www.nic.cz/)",
+    author_email="my.email@nic.cz",
+    packages=[
+        "foris_controller_schnapps_module",
+        "foris_controller_backends",
+        "foris_controller_backends.schnapps",
+        "foris_controller_modules",
+        "foris_controller_modules.schnapps",
+        "foris_controller_modules.schnapps.handlers",
+    ],
+    package_data={"foris_controller_modules.schnapps": ["schema", "schema/*.json"]},
+    namespace_packages=["foris_controller_modules", "foris_controller_backends"],
+    description=DESCRIPTION,
+    long_description=open("README.rst").read(),
+    install_requires=[
+        "foris-controller @ git+https://gitlab.labs.nic.cz/turris/foris-controller/foris-controller.git"
+    ],
+    setup_requires=["pytest-runner"],
+    tests_require=["pytest", "foris-controller-testtools", "foris-client", "ubus", "paho-mqtt"],
+    dependency_links=[
+        "git+https://gitlab.labs.nic.cz/turris/foris-controller/foris-controller-testtools.git#egg=foris-controller-testtools",
+        "git+https://gitlab.labs.nic.cz/turris/foris-controller/foris-client.git#egg=foris-client",
+    ],
+    include_package_data=True,
+    zip_safe=False,
+)

tests/conftest.py

+#
+# foris-controller-schnapps-module
+# Copyright (C) 2019 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="module")
+def controller_modules():
+    return ["remote", "schnapps"]
+
+
+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")

tests/test_root/usr/bin/maintain-reboot-needed

+#!/bin/sh
+exit 0

tests/test_root/usr/bin/schnapps

+#!/usr/bin/env python3
+
+import datetime
+import argparse
+import pathlib
+import json
+import random
+import sys
+
+PATH = pathlib.Path("/tmp/test-schnapps.json")
+
+
+def load_data():
+    if not PATH.exists():
+        with PATH.open("w") as f:
+            json.dump({"snapshots": []}, f)
+    with PATH.open() as f:
+        return json.load(f)
+
+
+def store_data(data):
+    with PATH.open("w") as f:
+        json.dump(data, f)
+
+
+def handle_list(options):
+    print(json.dumps(load_data()))
+
+    return 0
+
+
+def handle_create(options):
+    now = datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S +0000")
+    data = load_data()
+    new_id = max(e["number"] for e in data["snapshots"] or [{"number": 0}]) + 1
+    data["snapshots"].append(
+        {
+            "number": new_id,
+            "created": now,
+            "type": options.t,
+            "size": f"{random.randint(1, 100000) / 100}MiB",
+            "description": options.descr,
+        }
+    )
+    store_data(data)
+
+    print(f"Snapshot number {new_id} created")
+
+    return 0
+
+
+def handle_delete(options):
+    data = load_data()
+    orig_len = len(data["snapshots"])
+    data["snapshots"] = [e for e in data["snapshots"] if e["number"] != options.id]
+    store_data(data)
+
+    return 1 if orig_len == len(data["snapshots"]) else 0
+
+
+def handle_rollback(options):
+    data = load_data()
+    ids = [e["number"] for e in data["snapshots"]]
+    if options.id not in ids:
+        return 1
+
+    return 0
+
+
+CMD_MAP = {
+    "list": handle_list,
+    "create": handle_create,
+    "delete": handle_delete,
+    "rollback": handle_rollback,
+}
+
+
+def main():
+
+    parser = argparse.ArgumentParser(prog="schnapps")
+
+    subcmds = parser.add_subparsers(dest="cmd")
+    subcmds.required = True
+
+    list_cmd = subcmds.add_parser("list")
+    list_cmd.add_argument("-j", default=False, action="store_true")
+
+    create_cmd = subcmds.add_parser("create")
+    create_cmd.add_argument("-t", choices=["single"], required=True)
+    create_cmd.add_argument("descr")
+
+    delete_cmd = subcmds.add_parser("delete")
+    delete_cmd.add_argument("id", type=int)
+
+    rollback_cmd = subcmds.add_parser("rollback")
+    rollback_cmd.add_argument("id", type=int)
+
+    options = parser.parse_args()
+    sys.exit(CMD_MAP[options.cmd](options))
+
+
+if __name__ == "__main__":
+    main()

tests/test_schnapps.py

+#
+# foris-controller-schnapps-module
+# Copyright (C) 2019 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 json
+import pathlib
+import pytest
+
+from foris_controller_testtools.fixtures import (
+    only_message_buses,
+    backend,
+    infrastructure,
+    start_buses,
+    mosquitto_test,
+    ubusd_test,
+    notify_api,
+)
+
+
+@pytest.fixture(scope="function")
+def init_snapshots():
+    path = pathlib.Path("/tmp/test-schnapps.json")
+    with path.open("w") as f:
+        json.dump(
+            {
+                "snapshots": [
+                    {
+                        "number": 1,
+                        "created": "2019-11-27 14:15:27 +0000",
+                        "type": "single",
+                        "size": "74.16MiB",
+                        "description": "ARGGGHHH",
+                    },
+                    {
+                        "number": 2,
+                        "created": "2019-11-29 17:18:34 +0000",
+                        "type": "single",
+                        "size": "63.28MiB",
+                        "description": "ARRGGHH 2",
+                    },
+                ]
+            },
+            f,
+        )
+
+    yield
+
+    path.unlink()
+
+
+def test_list(infrastructure, start_buses, init_snapshots):
+    res = infrastructure.process_message(
+        {"module": "schnapps", "action": "list", "kind": "request"}
+    )
+    assert "error" not in res
+    assert "data" in res
+    assert "snapshots" in res["data"]
+
+
+def test_create(infrastructure, start_buses, init_snapshots):
+    filters = [("schnapps", "create")]
+
+    notifications = infrastructure.get_notifications(filters=filters)
+    res = infrastructure.process_message(
+        {
+            "module": "schnapps",
+            "action": "create",
+            "kind": "request",
+            "data": {"description": "Testing snapshot 1"},
+        }
+    )
+    assert "number" in res["data"]
+
+    notifications = infrastructure.get_notifications(notifications, filters=filters)
+    assert notifications[-1]["module"] == "schnapps"
+    assert notifications[-1]["action"] == "create"
+    assert notifications[-1]["kind"] == "notification"
+    assert "number" in notifications[-1]["data"]
+
+    res = infrastructure.process_message(
+        {"module": "schnapps", "action": "list", "kind": "request"}
+    )
+    assert res["data"]["snapshots"][-1]["description"] == "Testing snapshot 1"
+    assert res["data"]["snapshots"][-1]["type"] == "single"
+    assert "size" in res["data"]["snapshots"][-1]
+    assert "number" in res["data"]["snapshots"][-1]
+    assert "created" in res["data"]["snapshots"][-1]
+
+
+def test_delete(infrastructure, start_buses, init_snapshots):
+    def create(description: str) -> int:
+        return infrastructure.process_message(
+            {
+                "module": "schnapps",
+                "action": "create",
+                "kind": "request",
+                "data": {"description": description},
+            }
+        )["data"]["number"]
+
+    first = create("firsttt")
+    second = create("seconddd")
+
+    filters = [("schnapps", "delete")]
+    notifications = infrastructure.get_notifications(filters=filters)
+    res = infrastructure.process_message(
+        {"module": "schnapps", "action": "delete", "kind": "request", "data": {"number": first}}
+    )
+    assert res["data"]["result"]
+    notifications = infrastructure.get_notifications(notifications, filters=filters)
+    assert notifications[-1] == {
+        "module": "schnapps",
+        "action": "delete",
+        "kind": "notification",
+        "data": {"number": first},
+    }
+
+    snapshots = infrastructure.process_message(
+        {"module": "schnapps", "action": "list", "kind": "request"}
+    )["data"]["snapshots"]
+
+    ids = [e["number"] for e in snapshots]
+    assert first not in ids
+    assert second in ids
+
+
+def test_rollback(infrastructure, start_buses, init_snapshots):
+    def create(description: str) -> int:
+        return infrastructure.process_message(
+            {
+                "module": "schnapps",
+                "action": "create",
+                "kind": "request",
+                "data": {"description": description},
+            }
+        )["data"]["number"]
+
+    first = create("firstttt")
+    second = create("secondddd")
+
+    filters = [("schnapps", "rollback")]
+    notifications = infrastructure.get_notifications(filters=filters)
+    res = infrastructure.process_message(
+        {"module": "schnapps", "action": "rollback", "kind": "request", "data": {"number": first}}
+    )
+    assert res["data"]["result"]
+    notifications = infrastructure.get_notifications(notifications, filters=filters)
+    assert notifications[-1] == {
+        "module": "schnapps",
+        "action": "rollback",
+        "kind": "notification",
+        "data": {"number": first, "shifted": [second]},
+    }
+
+    snapshots = infrastructure.process_message(
+        {"module": "schnapps", "action": "list", "kind": "request"}
+    )["data"]["snapshots"]
+
+    ids = [e["number"] for e in snapshots]
+    assert first in ids