Increase unit test coverage for changes in PR #1357 (#1358)

Add unit test for enum usage and others

deploy-agent/tests/unit/deploy/client/test_base_client.py

@@ -13,6 +13,8 @@
 # limitations under the License.
 
 import unittest
+from abc import ABCMeta, abstractmethod
+from future.utils import with_metaclass
 from tests import TestCase
 
 from deployd.client.base_client import BaseClient
@@ -28,6 +30,16 @@ class MyClient(BaseClient):
         with self.assertRaises(TypeError):
             myclient = MyClient()
 
+    def test_abc_equivalent_to_old(self):
+        """
+        Make sure that new changes to base client extend the original class
+        """
+        class OldBaseClient(with_metaclass(ABCMeta, object)):
+            @abstractmethod
+            def send_reports(self, env_reports=None):
+                pass
+        self.assertLessEqual(set(dir(OldBaseClient)), set(dir(BaseClient)))
+
 
 if __name__ == '__main__':
     unittest.main()

deploy-agent/tests/unit/deploy/client/test_client.py

@@ -0,0 +1,33 @@
+import mock
+import unittest
+from tests import TestCase
+
+from deployd.client.client import Client 
+from deployd.client.base_client import BaseClient
+from deployd.common.config import Config
+
+
+class TestClient(TestCase):
+    def test_extends_base_client(self):
+        self.assertTrue(issubclass(Client, BaseClient))
+
+    def test_no_config_provided(self):
+        with self.assertRaises(AttributeError):
+            Client()
+
+    def test_read_host_info(self):
+        client = Client(config=Config())
+        client._ec2_tags = {}
+        client._availability_zone = "us-east-1"
+        return_value: bool = client._read_host_info()
+        self.assertIsNotNone(client._hostname)
+        self.assertIsNotNone(client._ip)
+        self.assertTrue(return_value)
+
+    def test_read_host_info_no_ec2_tags_provided(self):
+        client = Client(config=Config())
+        with self.assertRaises(AttributeError):
+            client._read_host_info()
+
+if __name__ == '__main__':
+    unittest.main()

deploy-agent/tests/unit/deploy/client/test_serverless_client.py

@@ -103,6 +103,14 @@ def test_unknow_status_cause_retry(self):
 
         response = self.client.send_reports(env_status)
         self.assertEqual(response.deployGoal.deployStage, 'PRE_DOWNLOAD')
+
+    def test_create_response_last_deploy_stage(self):
+        report = self._new_report()
+        report.deployId = self.client._deploy_id
+        report.status = "SUCCEEDED"
+        report.deployStage = "SERVING_BUILD" 
+        response: PingResponse = self.client._create_response(report)
+        self.assertEqual(response.opCode, "NOOP")
 
 
 if __name__ == '__main__':

deploy-agent/tests/unit/deploy/common/test_config.py

@@ -20,7 +20,7 @@
 import tests
 
 from deployd.common.config import Config
-from deployd.common.types import DeployStatus, OpCode, DeployStage
+from deployd.common.types import DeployStatus, DeployType, OpCode, DeployStage
 from deployd.types.ping_response import PingResponse
 
 
@@ -58,6 +58,42 @@ def test_get_target(self):
         self.assertEqual(os.environ['COMPUTE_ENV_TYPE'], 'DEFAULT')
         self.assertEqual(self.config.get_target(), '/tmp/pinboard')
 
+    def test_init(self):
+        Config()
+
+        with self.assertRaises(TypeError):
+            Config(filenames=[
+                os.path.join(self.dirname, "test_file1.conf"),
+                os.path.join(self.dirname, "test_file2.conf"),
+                ])
+
+        open(os.path.join(self.dirname, "test_file1.conf"), "w")
+        Config(filenames=os.path.join(self.dirname, "test_file1.conf"))
+
+        os.remove(os.path.join(self.dirname, "test_file1.conf"))
+        with self.assertRaises(SystemExit), mock.patch('builtins.print') as print_mock:
+            Config(filenames=os.path.join(self.dirname, "test_file1.conf"))
+            print_mock.assert_called_once_with(f"Cannot find config files: {os.path.join(self.dirname, 'test_file1.conf')}")
+
+    def test_get_config_filename(self):
+        filename = os.path.join(self.dirname, "test_file1.conf")
+        open(os.path.join(self.dirname, "test_file1.conf"), "w")
+
+        config = Config(filenames=filename) 
+        self.assertEqual(config.get_config_filename(), filename)
+
+    def test_get_deploy_type_from_op_code(self):
+        config = Config()
+        self.assertEqual(config._get_deploy_type_from_opcode(opCode="NOOP"), DeployType.REGULAR)
+        self.assertEqual(config._get_deploy_type_from_opcode(opCode="DEPLOY"), DeployType.REGULAR)
+        self.assertEqual(config._get_deploy_type_from_opcode(opCode="UPDATE"), DeployType.REGULAR)
+        self.assertEqual(config._get_deploy_type_from_opcode(opCode="RESTART"), DeployType.RESTART)
+        self.assertEqual(config._get_deploy_type_from_opcode(opCode="DELETE"), DeployType.REGULAR)
+        self.assertEqual(config._get_deploy_type_from_opcode(opCode="TERMINATE"), DeployType.STOP)
+        self.assertEqual(config._get_deploy_type_from_opcode(opCode="WAIT"), DeployType.REGULAR)
+        self.assertEqual(config._get_deploy_type_from_opcode(opCode="ROLLBACK"), DeployType.ROLLBACK)
+        self.assertEqual(config._get_deploy_type_from_opcode(opCode="STOP"), DeployType.STOP)
+
 
 if __name__ == '__main__':
     unittest.main()

deploy-agent/tests/unit/deploy/types/test_ping_report.py

@@ -0,0 +1,23 @@
+from deployd.types.ping_report import PingReport 
+import unittest
+
+
+class TestPingReport(unittest.TestCase):
+    def test_init_enums(self):
+        report = PingReport(jsonValue={
+            "deployStage": 0,
+            "status": 0,
+            })
+        self.assertEqual(report.deployStage, "UNKNOWN") 
+        self.assertEqual(report.status, "SUCCEEDED") 
+
+        report = PingReport(jsonValue={
+            "deployStage": 10,
+            "status": 7,
+            })
+        self.assertEqual(report.deployStage, "STOPPED") 
+        self.assertEqual(report.status, "TOO_MANY_RETRY") 
+
+
+if __name__ == "__main__":
+    unittest.main()

deploy-agent/tests/unit/deploy/types/test_ping_request.py

@@ -0,0 +1,27 @@
+from deployd.types.ping_request import PingRequest 
+from deployd.types.ping_report import PingReport 
+import unittest
+
+
+class TestPingRequest(unittest.TestCase):
+    def test_to_json_enums(self):
+        reports = [
+            PingReport({
+                "deployStage": 7,
+                "status": 4,
+            }),
+            PingReport({
+                "deployStage": "POST_RESTART",
+                "status": "SCRIPT_FAILED",
+            }),
+        ]
+        request = PingRequest(reports=reports)
+        json_request = request.to_json()
+        self.assertEqual(json_request["reports"][0]["deployStage"], "POST_RESTART")
+        self.assertEqual(json_request["reports"][0]["agentStatus"], "SCRIPT_FAILED")
+        self.assertEqual(json_request["reports"][1]["deployStage"], "POST_RESTART")
+        self.assertEqual(json_request["reports"][1]["agentStatus"], "SCRIPT_FAILED")
+
+
+if __name__ == "__main__":
+    unittest.main()