Logstash rework

.gitignore

@@ -133,3 +133,4 @@ dmypy.json
 
 # Poetry
 poetry.lock
+.pytype

tests/extras/test_http_server.py

@@ -4,7 +4,7 @@
 import requests
 from pytest import mark
 
-from yellowbox import temp_network, connect
+from yellowbox import connect, temp_network
 from yellowbox.extras.http_server import HttpService, RouterHTTPRequestHandler
 
 
@@ -123,4 +123,4 @@ def square(request_handler: RouterHTTPRequestHandler):
         with square:
             response = requests.request(method, service.local_url + '/square', data=b'12')
             response.raise_for_status()
-            assert response.content.strip() == b'144'
+            assert response.content.strip() == b'144'

tests/extras/test_logstash.py

@@ -1,7 +1,127 @@
-from yellowbox.extras.logstash import LogstashService
+import json
+import socket
+import weakref
+from time import sleep
 
-# rebuilding the logstash service is an ongoing ticket, for now we just want to make sure it runs alright
+import pytest
 
-def test_make(docker_client):
-    with LogstashService.run(docker_client):
-        pass
+from yellowbox.extras.logstash import FakeLogstashService
+
+
+@pytest.fixture
+def logstash():
+    with FakeLogstashService().start() as ls:
+        yield ls
+
+
+def create_socket(logstash):
+    return socket.create_connection((logstash.local_host, logstash.port))
+
+
+def send_record(logstash, **kwargs):
+    s = create_socket(logstash)
+    s.sendall(json.dumps(kwargs).encode("utf-8") + b"\n")
+    s.close()
+    sleep(0.01)  # Wait for LogstashService to process.
+
+
+def send_records(logstash, *records):
+    s = create_socket(logstash)
+    data = "\n".join([json.dumps(record) for record in records]) + "\n"
+    data = data.encode("utf-8")
+    s.sendall(data)
+    s.close()
+    sleep(0.01)
+
+
+def test_sanity(logstash):
+    send_record(logstash, msg="test")
+    assert logstash.records[0] == {"msg": "test"}
+
+
+def test_multiple_records(logstash):
+    send_records(logstash, {"msg": "hello"}, {"msg": "meow"})
+    assert logstash.records == [{"msg": "hello"}, {"msg": "meow"}]
+
+
+def test_multiple_connections(logstash):
+    send_record(logstash, msg="test")
+    send_record(logstash, msg="test2")
+    assert logstash.records == [{"msg": "test"}, {"msg": "test2"}]
+
+
+def test_half_record(logstash):
+    s = create_socket(logstash)
+    s.sendall(b'{"ms')
+    s.sendall(b'g": "t')
+    s.sendall(b'est"}\n')
+    s.close()
+    sleep(0.01)
+    assert logstash.records[0] == {"msg": "test"}
+
+
+def test_bad_record(logstash):
+    s = create_socket(logstash)
+    s.sendall(b"{'sdafasdgsdgs\n")
+    sleep(0.05)
+
+    # Bad socket was closed
+    with pytest.raises(BrokenPipeError):
+        s.sendall(b"asdasd")
+        sleep(0.05)
+        s.sendall(b"asdasd")
+
+    # Server still works
+    send_record(logstash, msg="hello")
+    assert logstash.records == [{"msg": "hello"}]
+
+
+def test_assert_logs(logstash):
+    send_record(logstash, level="INFO")
+
+    with pytest.raises(AssertionError):
+        logstash.assert_logs("ERROR")
+
+    send_record(logstash, level="ERROR")
+
+    # Doesn't throw an error
+    logstash.assert_logs("ERROR")
+
+
+def test_assert_no_logs(logstash):
+    send_record(logstash, level="INFO")
+
+    # Doesn't throw an error
+    logstash.assert_no_logs("ERROR")
+
+    send_record(logstash, level="ERROR", message="hello")
+
+    with pytest.raises(AssertionError):
+        logstash.assert_no_logs("ERROR")
+
+
+def test_filter_records(logstash):
+    send_records(logstash, {"level": "INFO"}, {"level": "WARNING"},
+                 {"level": "ERROR"})
+    assert list(logstash.filter_records("warning")) == [
+        {"level": "WARNING"}, {"level": "ERROR"}]
+
+
+def test_is_alive():
+    logstash = FakeLogstashService()
+    assert not logstash.is_alive()
+    logstash.start()
+    assert logstash.is_alive()
+    logstash.stop()
+    assert not logstash.is_alive()
+
+
+def test_garbage_collection():
+    """Make sure LogstashService gets collected and has no cyclic references
+
+    Important for cleaning OS resources.
+    """
+    ls = FakeLogstashService()
+    r = weakref.ref(ls)
+    del ls
+    assert r() is None

yellowbox/extras/http_server.py

@@ -1,20 +1,21 @@
 from __future__ import annotations
 
-import platform
 from collections import defaultdict
 from contextlib import contextmanager
 from functools import partial
-from http.server import HTTPServer, BaseHTTPRequestHandler
-from threading import Thread, Lock
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from threading import Lock, Thread
 from types import new_class
-from typing import Pattern, Callable, Set, DefaultDict, Union, Optional, Type, cast, NamedTuple, ClassVar
-from urllib.parse import urlparse, ParseResult
+from typing import Callable, ClassVar, DefaultDict, NamedTuple, Optional, Pattern, Set, Type, Union, \
+    cast
+from urllib.parse import ParseResult, urlparse
 
 import requests
-from requests import HTTPError, ConnectionError
+from requests import ConnectionError, HTTPError
 
-from yellowbox.service import YellowService
 from yellowbox.retry import RetrySpec
+from yellowbox.service import YellowService
+from yellowbox.utils import docker_host_name
 
 __all__ = ['HttpService', 'RouterHTTPRequestHandler']
 SideEffectResponse = Union[bytes, str, int]
@@ -109,12 +110,6 @@ def __getattr__(self, item: str):
         raise AttributeError(item)
 
 
-if platform.system() == "Linux":
-    _docker_host_name = '172.17.0.1'
-else:
-    _docker_host_name = 'host.docker.internal'
-
-
 class HttpService(YellowService):
     """
     The HttpService class is used to mock http servers. Although it is a YellowService,
@@ -139,7 +134,8 @@ def __init__(self, host='0.0.0.0', port=0, name='anonymous_yellowbox_HTTPService
         self.router_cls = cast(Type[RouterHTTPRequestHandler],
                                new_class(name + '_RequestHandler', (RouterHTTPRequestHandler,)))
         self.server = HTTPServer((host, port), self.router_cls)
-        self.server_thread = Thread(name=name + '_thread', target=self.server.serve_forever, daemon=True)
+        self.server_thread = Thread(name=name + '_thread', target=self.server.serve_forever,
+                                    daemon=True)
 
     @property
     def server_port(self):
@@ -151,7 +147,7 @@ def local_url(self):
 
     @property
     def container_url(self):
-        return f'http://{_docker_host_name}:{self.server_port}'
+        return f'http://{docker_host_name}:{self.server_port}'
 
     @staticmethod
     def _to_callback(side_effect: SideEffect):
@@ -227,7 +223,7 @@ def start(self, retry_spec: Optional[RetrySpec] = None):
                 lambda: requests.get(self.local_url + '/health').raise_for_status(),
                 (ConnectionError, HTTPError)
             )
-        return self
+        return super(HttpService, self).start(retry_spec=retry_spec)
 
     def stop(self):
         self.server.shutdown()
@@ -239,7 +235,7 @@ def is_alive(self):
     def connect(self, network):
         # since the http service is not docker related, it cannot actually connect to the network. However,
         # other containers, connected to the network or not, can connect to the service with docker's usual host
-        return [_docker_host_name]
+        return [docker_host_name]
 
     def disconnect(self, network):
         pass