1 files changed, 74 insertions, 0 deletions
diff --git a/designate/tests/unit/test_upgrade_checks.py b/designate/tests/unit/test_upgrade_checks.py
new file mode 100644
index 00000000..31c21ccd
--- /dev/null
+++ b/designate/tests/unit/test_upgrade_checks.py
@@ -0,0 +1,74 @@
+# Copyright 2018 Red Hat Inc.
+#
+# 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.
+
+from migrate.changeset.constraint import UniqueConstraint
+from oslo_upgradecheck import upgradecheck
+from sqlalchemy.schema import MetaData
+from sqlalchemy.schema import Table
+
+from designate.cmd import status
+from designate import tests
+from designate.sqlalchemy import session
+
+
+class TestDuplicateServiceStatus(tests.TestCase):
+    def setUp(self):
+        super(TestDuplicateServiceStatus, self).setUp()
+        self.engine = session.get_engine('storage:sqlalchemy')
+        self.meta = MetaData()
+        self.meta.bind = self.engine
+        self.service_statuses_table = Table('service_statuses', self.meta,
+                                            autoload=True)
+
+    def test_success(self):
+        fake_record = {'id': '1',
+                       'service_name': 'worker',
+                       'hostname': 'localhost',
+                       'status': 'UP',
+                       'stats': '',
+                       'capabilities': '',
+                       }
+        self.service_statuses_table.insert().execute(fake_record)
+        # Different hostname should be fine
+        fake_record['id'] = '2'
+        fake_record['hostname'] = 'otherhost'
+        self.service_statuses_table.insert().execute(fake_record)
+        # Different service_name should be fine
+        fake_record['id'] = '3'
+        fake_record['service_name'] = 'producer'
+        self.service_statuses_table.insert().execute(fake_record)
+        checks = status.Checks()
+        self.assertEqual(upgradecheck.Code.SUCCESS,
+                         checks._duplicate_service_status().code)
+
+    def test_failure(self):
+        # Drop unique constraint so we can test error cases
+        constraint = UniqueConstraint('service_name', 'hostname',
+                                      table=self.service_statuses_table,
+                                      name="unique_service_status")
+        constraint.drop()
+        fake_record = {'id': '1',
+                       'service_name': 'worker',
+                       'hostname': 'localhost',
+                       'status': 'UP',
+                       'stats': '',
+                       'capabilities': '',
+                       }
+        self.service_statuses_table.insert().execute(fake_record)
+        fake_record['id'] = '2'
+        self.service_statuses_table.insert().execute(fake_record)
+
+        checks = status.Checks()
+        self.assertEqual(upgradecheck.Code.FAILURE,
+                         checks._duplicate_service_status().code)