init comit

.gitignore

@@ -0,0 +1,172 @@
+
+# Created by https://www.gitignore.io/api/python,pycharm+all
+
+### PyCharm+all ###
+# Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm, CLion, Android Studio and Webstorm
+# Reference: https://intellij-support.jetbrains.com/hc/en-us/articles/206544839
+
+# User-specific stuff:
+.idea/**/workspace.xml
+.idea/**/tasks.xml
+.idea/dictionaries
+
+# Sensitive or high-churn files:
+.idea/**/dataSources/
+.idea/**/dataSources.ids
+.idea/**/dataSources.xml
+.idea/**/dataSources.local.xml
+.idea/**/sqlDataSources.xml
+.idea/**/dynamic.xml
+.idea/**/uiDesigner.xml
+
+# Gradle:
+.idea/**/gradle.xml
+.idea/**/libraries
+
+# CMake
+cmake-build-debug/
+
+# Mongo Explorer plugin:
+.idea/**/mongoSettings.xml
+
+## File-based project format:
+*.iws
+
+## Plugin-specific files:
+
+# IntelliJ
+/out/
+
+# mpeltonen/sbt-idea plugin
+.idea_modules/
+
+# JIRA plugin
+atlassian-ide-plugin.xml
+
+# Cursive Clojure plugin
+.idea/replstate.xml
+
+# Ruby plugin and RubyMine
+/.rakeTasks
+
+# Crashlytics plugin (for Android Studio and IntelliJ)
+com_crashlytics_export_strings.xml
+crashlytics.properties
+crashlytics-build.properties
+fabric.properties
+
+### PyCharm+all Patch ###
+# Ignores the whole .idea folder and all .iml files
+# See https://github.com/joeblau/gitignore.io/issues/186 and https://github.com/joeblau/gitignore.io/issues/360
+
+.idea/
+
+# Reason: https://github.com/joeblau/gitignore.io/issues/186#issuecomment-249601023
+
+*.iml
+modules.xml
+.idea/misc.xml
+*.ipr
+
+### Python ###
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.coverage
+.coverage.*
+.cache
+.pytest_cache/
+nosetests.xml
+coverage.xml
+*.cover
+.hypothesis/
+
+# Translations
+*.mo
+*.pot
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# pyenv
+.python-version
+
+# celery beat schedule file
+celerybeat-schedule.*
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+
+
+# End of https://www.gitignore.io/api/python,pycharm+all

mmgs108/api/http/sap.py

@@ -0,0 +1,14 @@
+from mmgs108.api.http.session import Session
+
+
+class Sap:
+
+    def __init__(self, base_address):
+        self.__address = base_address
+
+    def execute_request(self, request, password):
+        with Session(self.__address, password) as apiSession:
+            apiSession.get("foo")
+
+    def generate_url(self, path):
+        return "http://{}/{}".format(self.__address, path)

mmgs108/api/http/session.py

@@ -0,0 +1,33 @@
+from requests import session
+
+
+class Session:
+    __LOGIN_PATH = "login.cgi"
+    __LOGOUT_PATH = "logout.cgi"
+    __PASSWORD_POST_KEY = "password"
+
+    def __del__(self):
+        self.close()
+
+    def close(self):
+        self.__http_session.get(self.__logout_url)
+        self.__http_session.close()
+
+    def __init__(self, sap, password):
+        self.__api = sap
+        self.__login_url = sap.generate_url(Session.__LOGIN_PATH)
+        self.__logout_url = sap.generate_url(Session.__LOGOUT_PATH)
+        self.__http_session = session()
+        self.__http_session.post(self.__login_url, data={Session.__PASSWORD_POST_KEY: password})
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        return self
+
+    def get(self, path):
+        return self.__http_session.get(self.__api.generate_url(path))
+
+    def post(self, path, data):
+        return self.__http_session.post(self.__api.generate_url(path), data=data)

mmgs108/api/http/vlan_table.py

@@ -0,0 +1,74 @@
+import re
+
+
+class VLanTable:
+    __Q_BASIC = "8021qBasic"
+    __Q_BASIC_HTM = "{}.htm".format(__Q_BASIC)
+    __Q_BASIC_CGI = "{}.cgi".format(__Q_BASIC)
+
+    def __init__(self, session):
+        self.__session = session
+        self.__vlan_table, self.__hash_value = self.__read_table()
+
+    def __read_table(self):
+            response = self.__session.get(VLanTable.__Q_BASIC_HTM)
+            html = response.text
+            table = VLanTable.__html_table_to_dict(html)
+            hash_value = VLanTable.__get_hash_from_html_table(html)
+            return table, hash_value
+
+    def __str__(self):
+            return str(self.__vlan_table)
+
+    def __getitem__(self, item):
+        return self.__vlan_table[item]
+
+    def __setitem__(self, key, value):
+        self.__vlan_table[key] = str(value)
+        self.flush_table()
+
+    def flush_table(self):
+        data = VLanTable._create_post_param(self.__vlan_table, self.__hash_value)
+        self.__session.post(VLanTable.__Q_BASIC_CGI, data)
+        self.__vlan_table, self.__hash_value = self.__read_table()
+
+    def items(self):
+        return self.__vlan_table.items()
+
+    @staticmethod
+    def _create_post_param(table, hash_value):
+        params = {
+            "hash": hash_value,
+            "hiddVlan": "",
+            "status": "Enable",
+        }
+
+        return {**params, **table}
+
+    def to_html_table(self):
+        return VLanTable.__dict_to_html_table(self.__vlan_table)
+
+    @staticmethod
+    def __html_table_to_dict(html):
+        pattern = re.compile("name='(?P<port_name>port[0-9]+)' value='(?P<port_vlan_id>(all|[0-9]+))'")
+        match_iter = pattern.finditer(html)
+        vlan_table = {}
+        for match in match_iter:
+            vlan_table[match["port_name"]] = match["port_vlan_id"]
+
+        return vlan_table
+
+    @staticmethod
+    def __get_hash_from_html_table(html):
+        pattern = re.compile("name='hash' id='hash' value='(?P<hash_value>[0-9]+)'")
+        return pattern.search(html)["hash_value"]
+
+    @staticmethod
+    def __dict_to_html_table(table):
+        header = ""
+        values = ""
+        for key, value in table.items():
+            header = header + "<th>{}</th>\n".format(key)
+            values = values + "<td><input type='text' name='{}' value='{}' size='11' maxlength='4'></td>\n".format(key, value)
+
+        return "<table><tr>{} </tr><tr>{}</tr><table>".format(header, values)

mmgs108/ui/kivy/app.py

@@ -0,0 +1,72 @@
+from kivy.app import App
+from mmgs108.api.http.session import Session
+from mmgs108.api.http.vlan_table import VLanTable
+
+from mmgs108.api.http.sap import Sap
+from kivy.uix.boxlayout import BoxLayout
+from kivy.uix.textinput import TextInput
+from kivy.uix.label import Label
+from kivy.uix.scrollview import ScrollView
+from kivy.uix.gridlayout import GridLayout
+from kivy.core.window import Window
+
+
+class Gs108App(App):
+
+    def __init__(self, ips, **kwargs):
+        super().__init__(**kwargs)
+        self.__switches = {}
+        self.__ips = ips
+
+    def stop(self, *largs):
+        for key, switch in self.__switches.items():
+            switch["session"].close()
+
+    def build(self):
+
+        for ip in self.__ips:
+            sap = Sap(ip)
+            session = Session(sap, "password")
+            table = VLanTable(session)
+            self.__switches[ip] = {
+                "sap": sap,
+                "session": session,
+                "table": table
+            }
+
+        grid = GridLayout(cols=1, spacing=10, size_hint_y=None)
+        grid.bind(minimum_height=grid.setter('height'))
+        for name, switch in self.__switches.items():
+            grid.add_widget(self.build_switch(name, switch["table"]))
+
+        root = ScrollView(size_hint=(1, None), size=(Window.width, Window.height))
+        root.add_widget(grid)
+        return root
+
+    def build_vtable(self, table):
+        layout = BoxLayout()
+        for key, value in table.items():
+            tmp_layout = BoxLayout(orientation='vertical')
+            tmp_layout.add_widget(Label(text='{}:'.format(key), size_hint=(1, None), height=30))
+            vlan_input = VLanIdInput(table, key, text='{}'.format(value), multiline=False, size_hint=(1, None), height=30)
+            tmp_layout.add_widget(vlan_input)
+            layout.add_widget(tmp_layout)
+
+        return layout
+
+    def build_switch(self, switch, table):
+        layout = GridLayout(cols=1, spacing=10, size_hint_y=None)
+        layout.add_widget(Label(text='{}:'.format(switch), size_hint=(1, None), height=30))
+        layout.add_widget(self.build_vtable(table))
+        return layout
+
+
+class VLanIdInput(TextInput):
+
+    def __init__(self, table, port, **kwargs):
+        super().__init__(**kwargs)
+        self.__table = table
+        self.__port = port
+
+    def on_text_validate(self):
+        self.__table[self.__port] = self.text

tests/api/http/test_sap.py

@@ -0,0 +1,16 @@
+import unittest
+from mmgs108.api.http.sap import Sap
+from mmgs108.ui.kivy.app import Gs108App
+
+
+class TestSap(unittest.TestCase):
+    def test_generate_url(self):
+        sap = Sap("foo")
+        self.assertEqual("http://foo/bar", sap.generate_url("bar"))
+
+    def test_runServer(self):
+        Gs108App(["10.0.0.254"]).run()
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/api/http/test_vlan_table.py

@@ -0,0 +1,25 @@
+import unittest
+import random
+from mmgs108.api.http.sap import Sap
+from mmgs108.api.http.session import Session
+from mmgs108.api.http.vlan_table import VLanTable
+
+
+class TestVLanTable(unittest.TestCase):
+
+    def test_get_vlan_table(self):
+        sap = Sap("10.0.0.254")
+        vlan_id = random.randrange(1, 4095)
+        port_key = "port{}".format(random.randrange(2, 8))
+        with Session(sap, "password") as session1:
+            table = VLanTable(session1)
+            table[port_key] = vlan_id
+            self.assertEqual(str(vlan_id), table[port_key])
+
+        with Session(sap, "password") as session2:
+            table = VLanTable(session2)
+            self.assertEqual(str(vlan_id), table[port_key])
+
+
+if __name__ == '__main__':
+    unittest.main()