Fix mikrotik worker v1.6.x

.gitignore

@@ -107,4 +107,7 @@ venv.bak/
 *.db
 
 # Logs
-*.log
+*.log
+
+# Database files
+*.sqlite

Dockerfile

@@ -13,7 +13,9 @@ RUN poetry install --only main
 COPY . .
 
 RUN mkdir /data && chown nobody /data
+RUN mkdir /logs && chown nobody /logs
 VOLUME ["/data"]
+VOLUME ["/logs"]
 
 USER nobody
 EXPOSE 8000

env.sh

@@ -1,5 +1,6 @@
 export APP_DB_URL="postgresql://whohacks:S3cret@localhost:5432/whohacks"
 export OAUTH_OPENID="http://sso.hsp.sh/auth/realms/hsp/.well-known/openid-configuration"
 export OAUTH_CLIENT_ID="fake-development-client-id"
+export LOGLEVEL="DEBUG" # DEBUG | INFO | WARNING | ERROR
 env | grep APP_ > .env
 env | grep OUATH_ > .env

helpers/logger.py

@@ -0,0 +1,76 @@
+import logging
+import os
+
+loggers = {}
+
+logging_dir = str(os.environ.get("LOG_DIR", "logs"))
+is_log_to_file = bool(os.environ.get("LOG_TO_FILE", "false").lower() == "true")
+
+is_logging_to_file_configured = False
+
+
+def configure_logging_to_file():
+    config_value = os.environ.get("LOG_TO_FILE")
+    if is_log_to_file:
+        print(f"Logging to file is enabled" f"(LOG_TO_FILE is set to '{config_value}')")
+        if logging_dir:
+            print(f"Logging files directory: '{logging_dir}'")
+            if not os.path.exists(logging_dir):
+                os.makedirs(logging_dir, exist_ok=True)
+    else:
+        print(f"Logging to file is disabled" f"(LOG_TO_FILE is set to '{config_value}')")
+
+
+
+def get_loglevel(loglevel: str):
+    match loglevel:
+        case "DEBUG":
+            loglevel = logging.DEBUG
+        case "INFO":
+            loglevel = logging.INFO
+        case "WARNING":
+            loglevel = logging.WARNING
+        case "ERROR":
+            loglevel = logging.ERROR
+        case "CRITICAL":
+            loglevel = logging.CRITICAL
+        case _:
+            loglevel = logging.INFO
+
+    return loglevel
+
+
+def init_logger(name: str) -> logging.Logger:
+    global is_logging_to_file_configured
+    if not is_logging_to_file_configured:
+        configure_logging_to_file()
+        is_logging_to_file_configured = True
+
+    if loggers.get(name):
+        return loggers[name]
+
+    logger = logging.getLogger(name)
+    loglevel = get_loglevel(os.environ.get("LOGLEVEL"))
+
+    formatter = logging.Formatter(
+        fmt=f"({name}) %(asctime)s %(module)s %(levelname)s: %(message)s",
+        datefmt="%m/%d/%Y %I:%M:%S %p",
+    )
+
+    for handler in logger.handlers:
+        logger.removeHandler(handler)
+
+    stream_handler = logging.StreamHandler()
+    stream_handler.setFormatter(formatter)
+    logger.addHandler(stream_handler)
+
+    if is_log_to_file:
+        file_handler = logging.FileHandler(f"{logging_dir}/{name}.log")
+        file_handler.setFormatter(formatter)
+        logger.addHandler(file_handler)
+
+    logger.setLevel(loglevel)
+
+    loggers[name] = logger
+
+    return logger

tests/test_db.py

@@ -0,0 +1,36 @@
+import logging
+from copy import copy
+from datetime import datetime, timedelta
+from unittest import TestCase
+
+from helpers.logger import init_logger
+from whois.data.db.database import Database
+from whois.data.repository.device_repository import DeviceRepository
+from whois.entity.device import Device
+
+
+class TestMikrotik(TestCase):
+
+    def setUp(self):
+        self.logger = init_logger(__name__)
+        self.logger.addHandler(logging.FileHandler(f"{__name__}.log"))
+
+        self.db = Database("sqlite:///whohacks.test.sqlite")
+        self.db.drop()
+        self.db.create_db()
+        self.device_repository = DeviceRepository(self.db)
+
+    def test_create_device(self):
+        device = Device("mock_mac", "device_host", datetime.now(), owner=0, flags=0)
+        self.device_repository.insert(device)
+
+    def test_update_device(self):
+        device1 = Device("mac1", "device1", datetime.now(), owner=0, flags=0)
+        self.device_repository.insert(device1)
+        device2 = copy(device1)
+        new_last_seen = datetime.now() - timedelta(days=1)
+        device2.last_seen = new_last_seen
+        self.device_repository.update(device2)
+
+        result_device = self.device_repository.get_by_mac_address("mac1")
+        assert result_device == device2

tests/test_helpers.py

@@ -37,7 +37,7 @@ def test_iterable(self):
         All functions should return iterable
         :return:
         """
-        assert hasattr(helpers.filter_anon_names(users), "__len__")
+        assert hasattr(helpers.filter_anon_users(users), "__len__")
         assert hasattr(helpers.filter_hidden(users), "__len__")
         assert hasattr(helpers.owners_from_devices(devices), "__len__")
         assert hasattr(helpers.unclaimed_devices(devices), "__len__")

tests/test_integration.py

@@ -1,6 +1,7 @@
 import logging
 from unittest import TestCase
 
+from helpers.logger import init_logger
 from whois.app import WhohacksApp
 from whois.data.db.database import Database
 from whois.settings.testing import app_settings, mikrotik_settings
@@ -9,19 +10,13 @@
 class ApiTestCase(TestCase):
 
     def setUp(self):
-        self.logger = logging.getLogger(__name__)
-        logging.basicConfig(
-            format="%(asctime)s %(module)s %(levelname)s: %(message)s",
-            datefmt="%m/%d/%Y %I:%M:%S %p",
-            level=logging.DEBUG,
-            force=True,
-        )
+        self.logger = init_logger(__name__)
         self.logger.addHandler(logging.FileHandler(f"{__name__}.log"))
 
         self.db = Database("sqlite:///whohacks.test.sqlite")
         self.db.drop()
         self.db.create_db()
-        self.whois = WhohacksApp(app_settings, mikrotik_settings, self.db, self.logger)
+        self.whois = WhohacksApp(app_settings, mikrotik_settings, self.db)
         self.app = self.whois.app.test_client()
         self.app.testing = True