Merge "Add unit tests for guestagent.models.py Delete guestagent.utils.py since it is no longer invoked anywhere."

reddwarf/guestagent/utils.py

@@ -1,63 +0,0 @@
-# Copyright (c) 2011 OpenStack, LLC.
-# All Rights Reserved.
-#
-#    Licensed under the Apache License, Version 2.0 (the "License"); you may
-#    not use this file except in compliance with the License. You may obtain
-#    a copy of the License at
-#
-#         http://www.apache.org/licenses/LICENSE-2.0
-#
-#    Unless required by applicable law or agreed to in writing, software
-#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-#    License for the specific language governing permissions and limitations
-#    under the License.
-
-"""
-Set of utilities for the Guest Manager
-"""
-
-import fcntl
-import socket
-import struct
-
-
-from nova import context
-from nova import flags
-from reddwarf.db import db_api as dbapi
-# from nova.db import api as dbapi
-
-
-flags.DEFINE_string('guest_ethernet_device', "eth0",
-                    'Default Ethernet device for the guest agent')
-FLAGS = flags.FLAGS
-
-
-instance_id = None
-
-
-def get_instance_id():
-    """Return the instance id for this guest"""
-    global instance_id
-    if not instance_id:
-        # TODO(rnirmal): Better way to get the instance id
-        address = get_ipv4_address()
-        instance = dbapi.instance_get_by_fixed_ip(context.get_admin_context(),
-                                                  address)
-        instance_id = instance.id
-    return instance_id
-
-
-def get_ipv4_address():
-    """ Get the ip address provided an ethernet device"""
-    # Create an IPV4 (AF_INET) datagram socket
-    soc = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-
-    # fcntl is a system fuction that takes in the socket file descriptor,
-    # 0x8915 = SIOCGIFADDR which is an os call passed to ioctl which returns
-    # the list of interface addresses.
-    # struct.pack, packs the ethernet device string into a binary buffer
-    return socket.inet_ntoa(fcntl.ioctl(soc.fileno(), 0x8915,
-                                        struct.pack('256s',
-                                        FLAGS.guest_ethernet_device[:15])
-                                        )[20:24])

reddwarf/tests/guestagent/test_models.py

@@ -0,0 +1,59 @@
+#    Copyright 2012 OpenStack LLC
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License
+
+import testtools
+from mock import Mock, MagicMock
+from reddwarf.guestagent import models
+from reddwarf.common import utils
+from reddwarf.db import sqlalchemy
+from reddwarf.db import models as dbmodels
+from proboscis import test
+
+from datetime import datetime
+
+
+@test(groups=["dbaas.guestagent.dbaas"])
+class AgentHeartBeatTest(testtools.TestCase):
+    def setUp(self):
+        super(AgentHeartBeatTest, self).setUp()
+
+    def tearDown(self):
+        super(AgentHeartBeatTest, self).tearDown()
+
+    def test_create(self):
+        utils.generate_uuid = Mock()
+        sqlalchemy.api.save = MagicMock(
+            return_value=dbmodels.DatabaseModelBase)
+        dbmodels.DatabaseModelBase.is_valid = Mock(return_value=True)
+        models.AgentHeartBeat.create()
+        self.assertEqual(1, utils.generate_uuid.call_count)
+        self.assertEqual(3,
+                         dbmodels.DatabaseModelBase.is_valid.call_count)
+
+    def test_save(self):
+        utils.utcnow = Mock()
+        dbmodels.DatabaseModelBase = Mock
+        dbmodels.get_db_api = MagicMock(
+            return_value=dbmodels.DatabaseModelBase)
+        sqlalchemy.api.save = Mock()
+        dbmodels.DatabaseModelBase.is_valid = Mock(return_value=True)
+        self.heartBeat = models.AgentHeartBeat()
+        self.heartBeat.save()
+        self.assertEqual(1, utils.utcnow.call_count)
+
+    def test_is_active(self):
+        models.AGENT_HEARTBEAT = 10000000000
+        mock = models.AgentHeartBeat()
+        models.AgentHeartBeat.__setitem__(mock, 'updated_at', datetime.now())
+        self.assertTrue(models.AgentHeartBeat.is_active(mock))

run_tests.py

@@ -126,5 +126,6 @@ if __name__=="__main__":
     from reddwarf.tests.guestagent import test_manager
     from reddwarf.tests.guestagent import test_service
     from reddwarf.tests.guestagent import test_query
+    from reddwarf.tests.guestagent import test_models
 
     proboscis.TestProgram().run_and_exit()