Logstash rework

tests/extras/test_logstash.py

@@ -1,7 +1,115 @@
-from yellowbox.extras.logstash import LogstashService
+import json
+import socket
+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()

yellowbox/extras/http_server.py

@@ -15,6 +15,7 @@
 
 from yellowbox.service import YellowService
 from yellowbox.retry import RetrySpec
+from  yellowbox.utils import docker_host_name
 
 __all__ = ['HttpService', 'RouterHTTPRequestHandler']
 SideEffectResponse = Union[bytes, str, int]
@@ -109,11 +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):
     """
@@ -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