Feature/tests (#45)

* tests for Containers

* new tests for DockerHelper

* next helpers tests

* new test cases

Author: piontec

docker_enforcer.py

@@ -7,6 +7,7 @@
 from base64 import b64decode
 from logging import StreamHandler
 
+from docker import APIClient
 from flask import Flask, Response, request
 from pygments import highlight
 from pygments.formatters.html import HtmlFormatter
@@ -24,7 +25,8 @@
 from whitelist_rules.whitelist_rules import whitelist_rules
 
 config = Config()
-docker_helper = DockerHelper(config)
+client: APIClient = APIClient(base_url=config.docker_socket, timeout=config.docker_req_timeout_sec)
+docker_helper = DockerHelper(config, client)
 judge = Judge(rules, "container", config, run_whitelists=True, custom_whitelist_rules=whitelist_rules)
 requests_judge = Judge(request_rules, "request", config, run_whitelists=False)
 jurek = Killer(docker_helper, config.mode)

dockerenforcer/docker_helper.py

@@ -38,12 +38,12 @@ def __str__(self, *args, **kwargs) -> str:
 
 
 class DockerHelper:
-    def __init__(self, config: Config) -> None:
+    def __init__(self, config: Config, client: APIClient) -> None:
         super().__init__()
         self._padlock = threading.Lock()
         self._check_in_progress: bool = False
         self._config: Config = config
-        self._client: APIClient = APIClient(base_url=config.docker_socket, timeout=config.docker_req_timeout_sec)
+        self._client: APIClient = client
         self._params_cache: Dict[str, Any] = {}
         self.last_check_containers_run_end_timestamp: datetime.datetime = datetime.datetime.min
         self.last_check_containers_run_start_timestamp: datetime.datetime = datetime.datetime.min
@@ -142,6 +142,7 @@ def get_params(self, container_id: str) -> Optional[Dict[str, Any]]:
             logger.error("Unexpected error when fetching params for container {0}: {1}".format(container_id, e))
             return {}
         logger.debug("[{0}] Params fetched for {1}".format(threading.current_thread().name, container_id))
+
         if not self._config.cache_params:
             return params
 

readme.md

@@ -1,6 +1,6 @@
 # docker enforcer
-[![Build Status](https://travis-ci.org/piontec/docker-enforcer.png?branch=develop)](https://travis-ci.org/piontec/docker-enforcer)
-[![Coverage Status](https://coveralls.io/repos/github/piontec/docker-enforcer/badge.svg?branch=develop)](https://coveralls.io/github/piontec/docker-enforcer?branch=develop)
+[![Build Status](https://travis-ci.org/piontec/docker-enforcer.png?branch=feature/tests)](https://travis-ci.org/piontec/docker-enforcer)
+[![Coverage Status](https://coveralls.io/repos/github/piontec/docker-enforcer/badge.svg?branch=feature/tests)](https://coveralls.io/github/piontec/docker-enforcer?branch=feature/tests)
 
 ## Why?
 Docker enforcer audits containers running on a shared docker host. The aim of docker enforcer is to

test/test_api.py

@@ -15,7 +15,7 @@ class ApiContainerTest(unittest.TestCase):
     mem_rule = {"name": "must have memory limit", "rule": lambda c: c.params['HostConfig']['Memory'] == 0}
     cp_request_rule_regexp = re.compile("^/v1\.[23]\d/containers/test/archive$")
     cp_request_rule = {"name": "cp not allowed", "rule": lambda r, x=cp_request_rule_regexp:
-    r['RequestMethod'] in ['GET', 'HEAD'] and x.match(r['ParsedUri'].path)}
+                       r['RequestMethod'] in ['GET', 'HEAD'] and x.match(r['ParsedUri'].path)}
     test_trigger_flag = False
     test_trigger = {"name": "set local flag", "trigger": lambda v: ApiContainerTest.set_trigger_flag()}
     forbid_privileged_rule = {

test/test_docker_helpers.py

@@ -0,0 +1,93 @@
+import unittest
+from unittest.mock import create_autospec
+
+import docker
+
+from dockerenforcer.config import Config
+from dockerenforcer.docker_helper import Container, CheckSource, DockerHelper
+
+
+class ContainerTests(unittest.TestCase):
+    def test_create_without_name(self):
+        container = Container("123", {"param1": 1}, {"cpu": 7}, 0, CheckSource.Periodic)
+        self.assertEqual("123", str(container))
+
+    def test_create_with_name(self):
+        container = Container("123", {"Name": "container1"}, {"cpu": 7}, 0, CheckSource.Periodic)
+        self.assertEqual("container1", str(container))
+
+
+class DockerHelperTests(unittest.TestCase):
+    def setUp(self):
+        self._config = Config()
+        self._client = create_autospec(docker.APIClient)
+        self._helper = DockerHelper(self._config, self._client)
+        self._cid = "cont_id1"
+        self._cid2 = "cont_id2"
+        self._params = {"Id": self._cid, "param1": "1"}
+        self._params2 = {"Id": self._cid2, "param1": "2"}
+
+    def test_kill_container(self):
+        c = Container(self._cid, params=self._params, metrics={}, position=0, check_source=CheckSource.Periodic)
+        self._helper.kill_container(c)
+        self._client.stop.assert_called_once_with(self._cid)
+
+    def test_get_params_no_cache(self):
+        self._client.inspect_container.return_value = self._params
+        params = self._helper.get_params(self._cid)
+        self._client.inspect_container.assert_called_once_with(self._cid)
+        self.assertDictEqual(params, self._params)
+
+    def test_get_params_fill_cache(self):
+        self._config.cache_params = True
+        self._client.inspect_container.return_value = self._params
+        params = self._helper.get_params(self._cid)
+        self._client.inspect_container.assert_called_once_with(self._cid)
+        self.assertDictEqual(params, self._params)
+        self.assertDictEqual(self._helper._params_cache[self._cid], self._params)
+
+    def test_get_params_from_cache_and_remove(self):
+        self._config.cache_params = True
+        self._helper._params_cache[self._cid] = self._params
+        params = self._helper.get_params(self._cid)
+        self._client.inspect_container.assert_not_called()
+        self.assertDictEqual(params, self._params)
+        self.assertDictEqual(self._helper._params_cache[self._cid], self._params)
+        # now try to remove cached params
+        self._helper.remove_from_cache(self._cid)
+        self.assertFalse(self._cid in self._helper._params_cache)
+
+    def test_purge_cache(self):
+        self._config.cache_params = True
+        self._helper._params_cache[self._cid] = self._params
+        self._helper._params_cache[self._cid2] = self._params2
+        self._client.inspect_container.assert_not_called()
+        self._helper.purge_cache([self._cid])
+        self.assertFalse(self._cid2 in self._helper._params_cache)
+
+    def test_check_containers(self):
+        self._config.disable_metrics = True
+        self._client.containers.return_value = [{'Id': self._cid}, {'Id': self._cid2}]
+        self._client.inspect_container.side_effect = [self._params, self._params2]
+        containers = list(self._helper.check_containers(CheckSource.Periodic))
+        self._client.containers.assert_called_once_with(quiet=True)
+        self._client.inspect_container.side_effect = [self._params, self._params2]
+        self.assertEqual(len(containers), 2)
+        self.assertEqual(containers[0].cid, self._cid)
+        self.assertEqual(containers[0].check_source, CheckSource.Periodic)
+        self.assertDictEqual(containers[0].params, self._params)
+        self.assertEqual(containers[1].cid, self._cid2)
+        self.assertEqual(containers[1].check_source, CheckSource.Periodic)
+        self.assertDictEqual(containers[1].params, self._params2)
+
+    def test_get_events(self):
+        res = [
+            {u'from': u'image/with:tag', u'id': self._cid, u'status': u'start', u'time': 1423339459},
+            {u'from': u'image/with:tag', u'id': self._cid2, u'status': u'start', u'time': 1423339459}
+            ]
+        self._client.events.return_value = res
+        events = list(self._helper.get_events_observable())
+        self._client.events.assert_called_once_with(decode=True)
+        self.assertEqual(len(events), 2)
+        self.assertEqual(events[0]['id'], self._cid)
+        self.assertEqual(events[1]['id'], self._cid2)