Merge pull request #36 from Snawoot/improve_coverage

Improve coverage

postfix_mta_sts_resolver/responder.py

@@ -106,47 +106,43 @@ async def stop(self):
         await self._cache.teardown()
 
     async def sender(self, queue, writer):
+        def cleanup_queue():
+            while not queue.empty():
+                task = queue.get_nowait()
+                try:
+                    task.cancel()
+                except Exception:  # pragma: no cover
+                    pass
+
         try:
             while True:
                 fut = await queue.get()
-
                 # Check for shutdown
                 if fut is None:
-                    writer.close()
                     return
-
                 self._logger.debug("Got new future from queue")
-                try:
-                    data = await fut
-                except asyncio.CancelledError:
-                    writer.close()
-                    return
-                except Exception as exc:
-                    self._logger.exception("Unhandled exception from future: %s", exc)
-                    writer.close()
-                    return
+                data = await fut
                 self._logger.debug("Future await complete: data=%s", repr(data))
                 writer.write(data)
                 self._logger.debug("Wrote: %s", repr(data))
                 await writer.drain()
         except asyncio.CancelledError:
-            try:
-                fut.cancel()
-            except Exception:
-                pass
-            while not queue.empty():
-                task = queue.get_nowait()
-                task.cancel()
+            cleanup_queue()
+        except Exception as exc: # pragma: no cover
+            self._logger.exception("Exception in sender coro: %s", exc)
+            cleanup_queue()
+        finally:
+            writer.close()
 
     # pylint: disable=too-many-locals,too-many-branches,too-many-statements
     async def process_request(self, raw_req):
         # Update local cache
         async def cache_set(domain, entry):
             try:
                 await self._cache.set(domain, entry)
-            except asyncio.CancelledError:  # pylint: disable=try-except-raise
+            except asyncio.CancelledError:  # pragma: no cover pylint: disable=try-except-raise
                 raise
-            except Exception as exc:
+            except Exception as exc: # pragma: no cover
                 self._logger.exception("Cache set failed: %s", str(exc))
 
         have_policy = True
@@ -173,9 +169,9 @@ async def cache_set(domain, entry):
         # Lookup for cached policy
         try:
             cached = await self._cache.get(domain)
-        except asyncio.CancelledError:  # pylint: disable=try-except-raise
+        except asyncio.CancelledError:  # pragma: no cover pylint: disable=try-except-raise
             raise
-        except Exception as exc:
+        except Exception as exc:  # pragma: no cover
             self._logger.exception("Cache get failed: %s", str(exc))
             cached = None
 
@@ -239,7 +235,7 @@ class EndOfStream(Exception):
         async def finalize():
             try:
                 await queue.put(None)
-            except asyncio.CancelledError:
+            except asyncio.CancelledError:  # pragma: no cover
                 sender.cancel()
                 raise
             await sender
@@ -267,7 +263,7 @@ async def finalize():
         except (EndOfStream, ConnectionError, TimeoutError):
             self._logger.debug("Client disconnected")
             await finalize()
-        except OSError as exc:
+        except OSError as exc:  # pragma: no cover
             if exc.errno == 107:
                 self._logger.debug("Client disconnected")
                 await finalize()
@@ -277,11 +273,6 @@ async def finalize():
         except asyncio.CancelledError:
             sender.cancel()
             raise
-        except Exception as exc:
+        except Exception as exc:  # pragma: no cover
             self._logger.exception("Unhandled exception: %s", exc)
             await finalize()
-        finally:
-            try:
-                writer.close()
-            except Exception:
-                pass

postfix_mta_sts_resolver/sqlite_cache.py

@@ -80,7 +80,7 @@ def __init__(self, filename, *,
         self._threads = threads
         self._timeout = timeout
         sqlitelogger = logging.getLogger("aiosqlite")
-        if not sqlitelogger.hasHandlers():
+        if not sqlitelogger.hasHandlers():  # pragma: no cover
             sqlitelogger.addHandler(logging.NullHandler())
         self._pool = None
 

tests/test_resolver.py

@@ -87,7 +87,7 @@ async def test_resolve_dns_timeout(event_loop):
 @pytest.mark.asyncio
 @pytest.mark.timeout(5)
 async def test_proxy(event_loop):
-    with set_env(https_proxy='http://127.0.0.2:8888'):
+    with set_env(https_proxy='http://127.0.0.2:1380'):
         resolver = Resolver(loop=event_loop)
     status, (ver, pol) = await resolver.resolve("good.loc")
     assert status is FR.VALID

tests/test_responder_expiration.py

@@ -0,0 +1,69 @@
+import asyncio
+import tempfile
+import os
+import contextlib
+
+import pytest
+import pynetstring
+
+from postfix_mta_sts_resolver.responder import STSSocketmapResponder
+import postfix_mta_sts_resolver.utils as utils
+import postfix_mta_sts_resolver.base_cache as base_cache
+
+@contextlib.contextmanager
+def set_env(**environ):
+    old_environ = dict(os.environ)
+    os.environ.update(environ)
+    try:
+        yield
+    finally:
+        os.environ.clear()
+        os.environ.update(old_environ)
+
+@pytest.mark.asyncio
+@pytest.mark.timeout(10)
+async def test_responder_expiration(event_loop):
+    async def query(host, port, domain):
+        reader, writer = await asyncio.open_connection(host, port)
+        decoder = pynetstring.Decoder()
+        writer.write(pynetstring.encode(b'test ' + domain.encode('ascii')))
+        try:
+            while True:
+                data = await reader.read(4096)
+                assert data
+                res = decoder.feed(data)
+                if res:
+                    return res[0]
+        finally:
+            writer.close()
+    with tempfile.NamedTemporaryFile() as cachedb:
+        cfg = {}
+        cfg["port"] = 18461
+        cfg["cache_grace"] = 0
+        cfg["shutdown_timeout"] = 1
+        cfg["cache"] = {
+            "type": "sqlite",
+            "options": {
+                "filename": cachedb.name,
+            },
+        }
+        cfg = utils.populate_cfg_defaults(cfg)
+        cache = utils.create_cache(cfg['cache']['type'],
+                                   cfg['cache']['options'])
+        await cache.setup()
+        pol_body = {
+            "version": "STSv1",
+            "mode": "enforce",
+            "mx": [ "mail.loc" ],
+            "max_age": 1,
+        }
+        await cache.set("no-record.loc", base_cache.CacheEntry(0, "0", pol_body))
+        await cache.teardown()
+
+        resp = STSSocketmapResponder(cfg, event_loop)
+        await resp.start()
+        try:
+            result = await query(cfg['host'], cfg['port'], 'no-record.loc')
+            assert result == b'NOTFOUND '
+        finally:
+            await resp.stop()

tests/test_responder_strict.py

@@ -16,6 +16,7 @@ async def responder(event_loop):
     import postfix_mta_sts_resolver.utils as utils
     cfg = utils.populate_cfg_defaults({"default_zone": {"strict_testing": True}})
     cfg["zones"]["test2"] = cfg["default_zone"]
+    cfg["port"] = 28461
     resp = STSSocketmapResponder(cfg, event_loop)
     await resp.start()
     result = resp, cfg['host'], cfg['port']

tests/test_responder_volatile.py

@@ -15,6 +15,7 @@
 async def responder(event_loop):
     import postfix_mta_sts_resolver.utils as utils
     cfg = utils.populate_cfg_defaults(None)
+    cfg["port"] = 38461
     cfg["shutdown_timeout"] = 1
     cfg["cache_grace"] = 0
     cfg["zones"]["test2"] = cfg["default_zone"]
@@ -33,6 +34,32 @@ async def test_hanging_stop(responder):
     assert await reader.read() == b''
     writer.close()
 
+@pytest.mark.asyncio
+@pytest.mark.timeout(5)
+async def test_inprogress_stop(responder):
+    resp, host, port = responder
+    reader, writer = await asyncio.open_connection(host, port)
+    writer.write(pynetstring.encode(b'test blackhole.loc'))
+    await writer.drain()
+    await asyncio.sleep(0.2)
+    await resp.stop()
+    assert await reader.read() == b''
+    writer.close()
+
+@pytest.mark.asyncio
+@pytest.mark.timeout(5)
+async def test_extended_stop(responder):
+    resp, host, port = responder
+    reader, writer = await asyncio.open_connection(host, port)
+    writer.write(pynetstring.encode(b'test blackhole.loc'))
+    writer.write(pynetstring.encode(b'test blackhole.loc'))
+    writer.write(pynetstring.encode(b'test blackhole.loc'))
+    await writer.drain()
+    await asyncio.sleep(0.2)
+    await resp.stop()
+    assert await reader.read() == b''
+    writer.close()
+
 @pytest.mark.asyncio
 @pytest.mark.timeout(7)
 async def test_grace_expired(responder):

tests/tinyproxy.conf

@@ -20,7 +20,7 @@ Group nogroup
 # that should you choose to run on a port lower than 1024 you will need
 # to start tinyproxy using root.
 #
-Port 8888
+Port 1380
 
 #
 # Listen: If you have multiple interfaces this allows you to bind to

tox.ini

@@ -18,4 +18,4 @@ basepython = python3.7
 commands =
     pip install -e ".[dev,sqlite,redis]"
     pytest --cov . --cov-append --cov-report= .
-    coverage report --fail-under=90 --include="postfix_mta_sts_resolver/*" --show-missing
+    coverage report --fail-under=97 --include="postfix_mta_sts_resolver/*" --show-missing