Commits

Jean-Tiare Le Bigot committed d62d074

enhance ConnectionBorg tests

Comments (0)

Files changed (1)

dynamodb_mapper/tests/test_model.py

 class TestConnectionBorg(unittest.TestCase):
     def setUp(self):
         ConnectionBorg()._connections.clear()
+        ConnectionBorg._shared_state = {
+            "_aws_access_key_id": None,
+            "_aws_secret_access_key": None,
+            "_region": None,
+            "_connections": {},
+        }
 
     def tearDown(self):
         ConnectionBorg()._connections.clear()
+        ConnectionBorg._shared_state = {
+            "_aws_access_key_id": None,
+            "_aws_secret_access_key": None,
+            "_region": None,
+            "_connections": {},
+        }
 
     def test_borgness(self):
         aws_access_key_id = "foo"
 
     @mock.patch("dynamodb_mapper.model.boto")
     def test_set_region_valid(self, m_boto):
+        # Make sure internal state is set and shared
         m_regions = [
             mocked_region('us-east-1'),
             mocked_region('eu-west-1'),
 
     @mock.patch("dynamodb_mapper.model.boto")
     def test_set_region_invalid(self, m_boto):
+        # Make sure internal state is set and shared
         m_regions = [
             mocked_region('us-east-1'),
             mocked_region('eu-west-1'),
 
 
     @mock.patch("dynamodb_mapper.model.boto")
-    def test_get_table_nominal(self, m_boto):
+    def test_set_credentials(self, m_boto):
+        # Make sure internal state is set and shared
         m_get_table = m_boto.connect_dynamodb.return_value.get_table
 
         aws_access_key_id = "foo"
         aws_secret_access_key = "bar"
         table_name = "foo"
-        table_name2 = "bar"
 
-        ConnectionBorg().set_credentials(aws_access_key_id, aws_secret_access_key)
+        borg1 = ConnectionBorg()
+        borg2 = ConnectionBorg()
+
+        borg1.set_credentials(aws_access_key_id, aws_secret_access_key)
+
+        self.assertIs(borg1._aws_access_key_id ,aws_access_key_id)
+        self.assertIs(borg2._aws_access_key_id ,aws_access_key_id)
+
+        self.assertIs(borg1._aws_secret_access_key ,aws_secret_access_key)
+        self.assertIs(borg2._aws_secret_access_key ,aws_secret_access_key)
+
+    @mock.patch("dynamodb_mapper.model.boto")
+    def test_get_table_default(self, m_boto):
+        m_get_table = m_boto.connect_dynamodb.return_value.get_table
+
+        aws_access_key_id = "foo"
+        aws_secret_access_key = "bar"
+        table_name = "foo"
+
+        borg1 = ConnectionBorg()
+
+        borg1._aws_access_key_id = aws_access_key_id
+        borg1._aws_secret_access_key = aws_secret_access_key
 
         ConnectionBorg().get_table(table_name)
         m_get_table.assert_called_with(table_name)
 
-        ConnectionBorg().get_table(table_name2)
-        m_get_table.assert_called_with(table_name2)
-
-        # Should be called only once per thread even when getting multiple tables.
-        m_boto.connect_dynamodb.assert_called_once_with(
+        m_boto.connect_dynamodb.assert_called_with(
             aws_access_key_id=aws_access_key_id,
             aws_secret_access_key=aws_secret_access_key,
             region=None,
         )
 
     @mock.patch("dynamodb_mapper.model.boto")
+    def test_get_table_eu_region(self, m_boto):
+        m_get_table = m_boto.connect_dynamodb.return_value.get_table
+        m_region = mocked_region('eu-west-1')
+
+        aws_access_key_id = "foo"
+        aws_secret_access_key = "bar"
+        table_name = "foo"
+
+        borg1 = ConnectionBorg()
+
+        borg1._region = m_region
+        borg1._aws_access_key_id = aws_access_key_id
+        borg1._aws_secret_access_key = aws_secret_access_key
+
+        ConnectionBorg().get_table(table_name)
+        m_get_table.assert_called_with(table_name)
+
+        m_boto.connect_dynamodb.assert_called_with(
+            aws_access_key_id=aws_access_key_id,
+            aws_secret_access_key=aws_secret_access_key,
+            region=m_region,
+        )
+
+    @mock.patch("dynamodb_mapper.model.boto")
     def test_new_batch_list_nominal(self, m_boto):
         m_new_batch_list = m_boto.connect_dynamodb.return_value.new_batch_list
         ConnectionBorg().new_batch_list()