feat: add validations

Author: ErikKalkoken

pyproject.toml

@@ -4,11 +4,6 @@ build-backend = "flit_core.buildapi"
 
 [project]
 authors = [{ name = "Erik Kalkoken", email = "kalkoken87@gmail.com" }]
-dynamic = ["version", "description"]
-license = "MIT"
-license-files = ["LICENSE"]
-name = "victorialogs-handler"
-readme = "README.md"
 classifiers = [
     "Environment :: Web Environment",
     "Intended Audience :: Developers",
@@ -21,6 +16,12 @@ classifiers = [
     "Programming Language :: Python :: 3.13",
     "Topic :: System :: Logging",
 ]
+dynamic = ["version", "description"]
+license = "MIT"
+license-files = ["LICENSE"]
+name = "victorialogs-handler"
+readme = "README.md"
+requires-python = ">=3.10"
 
 [project.urls]
 Home = "https://github.com/ErikKalkoken/python-victorialogs-handler"

src/vlogs_handler/__init__.py

@@ -3,7 +3,4 @@
 from .handler import VictoriaLogsHandler  # noqa: F401
 
 __title__ = "Victoria Logs Handler"
-__version__ = "0.1.0dev3"
-
-# TO-DOs
-# [ ] Add validations
+__version__ = "0.1.0dev4"

src/vlogs_handler/handler.py

@@ -72,15 +72,27 @@ def __init__(
         url: str = "http://localhost:9428",
     ):
         super().__init__()
-        self._batch_size = batch_size
-        self._queue = queue.Queue(-1)
-        self._request_timeout = request_timeout
-        self._url = url
+
+        if batch_size < 1:
+            raise ValueError(f"batch size must be >= 1 {batch_size}")
+
+        if request_timeout <= 0:
+            raise ValueError(f"request_timeout must be > 0: {request_timeout}")
+
+        if not request.is_url(url):
+            raise ValueError(f"url is not valid: {url}")
+
         name = __package__
         if not name:
-            raise RuntimeError("Must be run as module")
+            raise RuntimeError("Must run as module")
+
         self.addFilter(_create_filter(name))
 
+        self._batch_size = int(batch_size)
+        self._queue = queue.Queue(-1)
+        self._request_timeout = float(request_timeout)
+        self._url = url
+
         # Start background worker
         self._worker_thread = threading.Thread(target=self._worker, daemon=True)
         self._worker_started = False
@@ -202,3 +214,4 @@ def _calc_stream_from_record(record: logging.LogRecord):
         else:
             stream = record.name
     return stream
+    return stream

src/vlogs_handler/request.py

@@ -2,12 +2,22 @@
 
 import logging
 import urllib.error
+import urllib.parse
 import urllib.request
 from typing import Optional
 
 logger = logging.getLogger(__name__)
 
 
+def is_url(url: str) -> bool:
+    """Report whether a string represents a valid URL."""
+    try:
+        result = urllib.parse.urlparse(url)
+        return all([result.scheme, result.netloc])
+    except ValueError:
+        return False
+
+
 def post_ndjson(*, url: str, data: str, timeout: Optional[float] = None) -> bool:
     """Send a POST request with the ndjson protocol
     and report whether it was successful.

tests/test_handler.py

@@ -10,10 +10,30 @@
 MODULE_PATH = "vlogs_handler.handler"
 
 
+class TestVictoriaLogsHandler_Validations(unittest.TestCase):
+    def test_should_validate_batch_size(self):
+        with self.assertRaises(ValueError):
+            VictoriaLogsHandler(batch_size=0)
+
+        with self.assertRaises(ValueError):
+            VictoriaLogsHandler(batch_size=-1)
+
+    def test_should_should_validate_request_timeout(self):
+        with self.assertRaises(ValueError):
+            VictoriaLogsHandler(request_timeout=0)
+
+        with self.assertRaises(ValueError):
+            VictoriaLogsHandler(request_timeout=-1)
+
+    def test_should_should_validate_url(self):
+        with self.assertRaises(ValueError):
+            VictoriaLogsHandler(url="123")
+
+
 @patch(MODULE_PATH + ".request.post_ndjson")
 class TestVictoriaLogsHandler_SingleLog(unittest.TestCase):
     def setUp(self):
-        self.handler = VictoriaLogsHandler(url="http://localhost:30123")
+        self.handler = VictoriaLogsHandler()
         self.logger = logging.getLogger("test_logger")
         self.logger.addHandler(self.handler)
         self.logger.setLevel(logging.INFO)
@@ -94,7 +114,7 @@ def test_should_do_nothing_when_trying_to_start_again(self, _):
 @patch(MODULE_PATH + ".request.post_ndjson")
 class TestVictoriaLogsHandler_MultipleLogs(unittest.TestCase):
     def setUp(self):
-        self.handler = VictoriaLogsHandler(url="http://localhost:30123", batch_size=3)
+        self.handler = VictoriaLogsHandler(batch_size=3)
         self.logger = logging.getLogger("test_logger")
         self.logger.addHandler(self.handler)
         self.logger.setLevel(logging.INFO)

tests/test_request.py

@@ -3,9 +3,10 @@
 from contextlib import contextmanager
 from http.client import HTTPMessage
 from io import BytesIO
+from typing import NamedTuple
 from unittest.mock import MagicMock, patch
 
-from vlogs_handler.request import post_ndjson
+from vlogs_handler import request
 
 MODULE_PATH = "vlogs_handler.request"
 
@@ -20,7 +21,7 @@ def make_http_error(status=404, message="Not Found", url="https://example.com"):
 
 def make_urlopen_fake(exception=None):
     @contextmanager
-    def urlopen_fake(*args, **kwargs):
+    def urlopen_fake(req, timeout):
         if exception:
             raise exception
         yield MagicMock()
@@ -37,7 +38,7 @@ def test_should_submit_successfully(self):
         # when
         with patch(MODULE_PATH + ".urllib.request.urlopen") as m:
             m.side_effect = make_urlopen_fake()
-            got = post_ndjson(url=url, data=data)
+            got = request.post_ndjson(url=url, data=data)
 
         # then
         self.assertTrue(got)
@@ -50,7 +51,7 @@ def test_should_handle_http_exception(self):
         # when
         with patch(MODULE_PATH + ".urllib.request.urlopen") as m:
             m.side_effect = make_urlopen_fake(exception=make_http_error())
-            got = post_ndjson(url=url, data=data)
+            got = request.post_ndjson(url=url, data=data)
 
         # then
         self.assertFalse(got)
@@ -65,7 +66,7 @@ def test_should_handle_url_exception(self):
             m.side_effect = make_urlopen_fake(
                 exception=urllib.error.URLError("Network is unreachable")
             )
-            got = post_ndjson(url=url, data=data)
+            got = request.post_ndjson(url=url, data=data)
 
         # then
         self.assertFalse(got)
@@ -79,7 +80,25 @@ def test_should_handle_general_exception(self):
         # when
         with patch(MODULE_PATH + ".urllib.request.urlopen") as m:
             m.side_effect = make_urlopen_fake(exception=RuntimeError)
-            got = post_ndjson(url=url, data=data, timeout=timeout)
+            got = request.post_ndjson(url=url, data=data, timeout=timeout)
 
         # then
         self.assertFalse(got)
+
+
+class TestIsURL(unittest.TestCase):
+    def test_all(self):
+        class Case(NamedTuple):
+            url: str
+            want: bool
+
+        cases = [
+            Case("http://www.example.com", True),
+            Case("http://localhost:9428", True),
+            Case("http://0.0.0.0:9428", True),
+            Case("www.example.com", False),
+        ]
+
+        for tc in cases:
+            with self.subTest(url=tc.url):
+                self.assertIs(request.is_url(tc.url), tc.want)