Merge pull request #1337 from tirkarthi/fix-assert-1

Refactor deprecated unittest aliases for Python 3.11 compatibility.

Author: micafer

test/integration/TestREST.py

@@ -452,7 +452,7 @@ def test_94_get_outputs(self):
                          msg="ERROR getting TOSCA outputs:" + resp.text)
         res = json.loads(resp.text)
         server_url = str(res['outputs']['server_url'][0])
-        self.assertRegexpMatches(
+        self.assertRegex(
             server_url, '\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', msg="Unexpected outputs: " + resp.text)
 
     def test_95_add_tosca(self):

test/unit/connectors/Azure.py

@@ -217,20 +217,20 @@ def test_20_launch(self, save_data, credentials, network_client, compute_client,
         self.assertTrue(res[0][0])
         self.assertTrue(res[1][0])
         self.assertTrue(res[2][0])
-        self.assertEquals(nclient.network_interfaces.begin_delete.call_count, 1)
+        self.assertEqual(nclient.network_interfaces.begin_delete.call_count, 1)
         self.assertIn("nic_name", nclient.network_interfaces.begin_delete.call_args_list[0][0][1])
 
         json_vm_req = cclient.virtual_machines.begin_create_or_update.call_args_list[0][0][2]
-        self.assertEquals(json_vm_req['storage_profile']['data_disks'][0]['disk_size_gb'], 1)
-        self.assertEquals(json_vm_req['storage_profile']['data_disks'][1]['managed_disk']['id'], "did")
+        self.assertEqual(json_vm_req['storage_profile']['data_disks'][0]['disk_size_gb'], 1)
+        self.assertEqual(json_vm_req['storage_profile']['data_disks'][1]['managed_disk']['id'], "did")
         image_res = {'sku': '16.04.0-LTS', 'publisher': 'Canonical', 'version': 'latest', 'offer': 'UbuntuServer'}
-        self.assertEquals(json_vm_req['storage_profile']['image_reference'], image_res)
-        self.assertEquals(json_vm_req['hardware_profile']['vm_size'], 'Standard_A1')
-        self.assertEquals(json_vm_req['os_profile']['admin_username'], 'user')
-        self.assertEquals(json_vm_req['os_profile']['admin_password'], 'pass')
-        self.assertEquals(json_vm_req['os_profile']['admin_password'], 'pass')
-        self.assertEquals(nclient.subnets.begin_create_or_update.call_args_list[0][0][3]['address_prefix'],
-                          '10.0.1.0/24')
+        self.assertEqual(json_vm_req['storage_profile']['image_reference'], image_res)
+        self.assertEqual(json_vm_req['hardware_profile']['vm_size'], 'Standard_A1')
+        self.assertEqual(json_vm_req['os_profile']['admin_username'], 'user')
+        self.assertEqual(json_vm_req['os_profile']['admin_password'], 'pass')
+        self.assertEqual(json_vm_req['os_profile']['admin_password'], 'pass')
+        self.assertEqual(nclient.subnets.begin_create_or_update.call_args_list[0][0][3]['address_prefix'],
+                         '10.0.1.0/24')
 
         radl_data = """
             network net1 (outbound = 'yes')
@@ -252,7 +252,7 @@ def test_20_launch(self, save_data, credentials, network_client, compute_client,
         radl.check()
         with self.assertRaises(Exception) as ex:
             azure_cloud.launch(inf, radl, radl, 1, auth)
-        self.assertEquals(str(ex.exception), "Incorrect image url: it must be snapshot or disk.")
+        self.assertEqual(str(ex.exception), "Incorrect image url: it must be snapshot or disk.")
 
         radl_data = """
             network net1 (outbound = 'yes')
@@ -274,9 +274,9 @@ def test_20_launch(self, save_data, credentials, network_client, compute_client,
         radl.check()
         res = azure_cloud.launch(inf, radl, radl, 1, auth)
         json_vm_req = cclient.virtual_machines.begin_create_or_update.call_args_list[5][0][2]
-        self.assertEquals(json_vm_req['storage_profile']['os_disk']['os_type'], 'linux')
-        self.assertEquals(nclient.subnets.begin_create_or_update.call_args_list[2][0][3]['address_prefix'],
-                          '192.168.1.0/24')
+        self.assertEqual(json_vm_req['storage_profile']['os_disk']['os_type'], 'linux')
+        self.assertEqual(nclient.subnets.begin_create_or_update.call_args_list[2][0][3]['address_prefix'],
+                         '192.168.1.0/24')
 
         radl_data = """
             network net1 (outbound = 'yes')
@@ -304,10 +304,10 @@ def test_20_launch(self, save_data, credentials, network_client, compute_client,
         subnet_create.address_prefix = "10.0.1.0/24"
         nclient.subnets.get.return_value = subnet_create
         res = azure_cloud.launch(inf, radl, radl, 1, auth)
-        self.assertEquals(nclient.subnets.get.call_args_list[3][0][1], 'vnet')
-        self.assertEquals(nclient.subnets.get.call_args_list[3][0][2], 'subnet1')
-        self.assertEquals(nclient.subnets.begin_create_or_update.call_count, 3)
-        self.assertEquals(nclient.virtual_networks.begin_create_or_update.call_count, 3)
+        self.assertEqual(nclient.subnets.get.call_args_list[3][0][1], 'vnet')
+        self.assertEqual(nclient.subnets.get.call_args_list[3][0][2], 'subnet1')
+        self.assertEqual(nclient.subnets.begin_create_or_update.call_count, 3)
+        self.assertEqual(nclient.virtual_networks.begin_create_or_update.call_count, 3)
 
     @patch('IM.connectors.Azure.NetworkManagementClient')
     @patch('IM.connectors.Azure.ComputeManagementClient')
@@ -393,11 +393,11 @@ def test_30_updateVMInfo(self, credentials, dns_client, compute_client, network_
         success, vm = azure_cloud.updateVMInfo(vm, auth)
 
         self.assertTrue(success, msg="ERROR: updating VM info.")
-        self.assertEquals(dclient.zones.create_or_update.call_args_list,
-                          [call('rg0', 'domain.com', {'location': 'global'})])
-        self.assertEquals(dclient.record_sets.create_or_update.call_args_list,
-                          [call('rg0', 'domain.com', 'test', 'A',
-                                {'arecords': [{'ipv4_address': '13.0.0.1'}], 'ttl': 300})])
+        self.assertEqual(dclient.zones.create_or_update.call_args_list,
+                         [call('rg0', 'domain.com', {'location': 'global'})])
+        self.assertEqual(dclient.record_sets.create_or_update.call_args_list,
+                         [call('rg0', 'domain.com', 'test', 'A',
+                               {'arecords': [{'ipv4_address': '13.0.0.1'}], 'ttl': 300})])
         self.assertNotIn("ERROR", self.log.getvalue(), msg="ERROR found in log: %s" % self.log.getvalue())
 
     @patch('IM.connectors.Azure.ComputeManagementClient')
@@ -564,7 +564,7 @@ def test_60_finalize(self, credentials, resource_client, compute_client):
         self.assertTrue(success, msg="ERROR: finalizing VM info.")
         self.assertNotIn("ERROR", self.log.getvalue(), msg="ERROR found in log: %s" % self.log.getvalue())
 
-        self.assertEquals(cclient.virtual_machines.begin_delete.call_count, 2)
+        self.assertEqual(cclient.virtual_machines.begin_delete.call_count, 2)
         self.assertEqual(cclient.virtual_machines.begin_delete.call_args_list[0][0], ('rg0', 'vm0'))
         self.assertEqual(rclient.resource_groups.begin_delete.call_count, 1)
         self.assertEqual(rclient.resource_groups.begin_delete.call_args_list[0][0], ('rg0',))

test/unit/connectors/CloudStack.py

@@ -206,7 +206,7 @@ def test_30_updateVMInfo(self, get_driver):
         success, vm = osc_cloud.updateVMInfo(vm, auth)
 
         self.assertTrue(success, msg="ERROR: updating VM info.")
-        self.assertEquals(vm.info.systems[0].getValue("net_interface.0.ip"), "8.8.8.8")
+        self.assertEqual(vm.info.systems[0].getValue("net_interface.0.ip"), "8.8.8.8")
         self.assertNotIn("ERROR", self.log.getvalue(), msg="ERROR found in log: %s" % self.log.getvalue())
 
     @patch('libcloud.compute.drivers.cloudstack.CloudStackNodeDriver')

test/unit/connectors/Docker.py

@@ -222,7 +222,7 @@ def test_30_updateVMInfo(self, requests):
         success, vm = docker_cloud.updateVMInfo(vm, auth)
 
         self.assertTrue(success, msg="ERROR: updating VM info.")
-        self.assertEquals(vm.info.systems[0].getValue("net_interface.1.ip"), "10.0.0.1")
+        self.assertEqual(vm.info.systems[0].getValue("net_interface.1.ip"), "10.0.0.1")
         self.assertNotIn("ERROR", self.log.getvalue(), msg="ERROR found in log: %s" % self.log.getvalue())
 
         self.activate_swarm()
@@ -232,7 +232,7 @@ def test_30_updateVMInfo(self, requests):
         docker_cloud._swarm = None
 
         self.assertTrue(success, msg="ERROR: updating VM info.")
-        self.assertEquals(vm.info.systems[0].getValue("net_interface.1.ip"), "10.0.0.1")
+        self.assertEqual(vm.info.systems[0].getValue("net_interface.1.ip"), "10.0.0.1")
         self.assertNotIn("ERROR", self.log.getvalue(), msg="ERROR found in log: %s" % self.log.getvalue())
 
     @patch('requests.request')

test/unit/connectors/EC2.py

@@ -177,10 +177,10 @@ def test_20_launch(self, save_data, blockdevicemapping, VPCConnection, get_regio
         success, _ = res[0]
         self.assertTrue(success, msg="ERROR: launching a VM.")
         self.assertNotIn("ERROR", self.log.getvalue(), msg="ERROR found in log: %s" % self.log.getvalue())
-        self.assertEquals(len(conn.create_security_group.call_args_list), 3)
-        self.assertEquals(conn.create_security_group.call_args_list[0][0][0], "im-%s" % inf.id)
-        self.assertEquals(conn.create_security_group.call_args_list[1][0][0], "sgname")
-        self.assertEquals(conn.create_security_group.call_args_list[2][0][0], "im-%s-net2" % inf.id)
+        self.assertEqual(len(conn.create_security_group.call_args_list), 3)
+        self.assertEqual(conn.create_security_group.call_args_list[0][0][0], "im-%s" % inf.id)
+        self.assertEqual(conn.create_security_group.call_args_list[1][0][0], "sgname")
+        self.assertEqual(conn.create_security_group.call_args_list[2][0][0], "im-%s-net2" % inf.id)
 
         # Check the case that we do not use VPC
         radl_data = """
@@ -225,10 +225,10 @@ def test_20_launch(self, save_data, blockdevicemapping, VPCConnection, get_regio
         self.assertTrue(success, msg="ERROR: launching a VM.")
         # check the instance_type selected is correct
         self.assertIn(".micro", conn.run_instances.call_args_list[1][1]["instance_type"])
-        self.assertEquals(conn.create_vpc.call_args_list[0][0][0], "10.0.128.0/22")
-        self.assertEquals(conn.create_subnet.call_args_list[0][0], ('vpc-id', '10.0.128.0/24'))
-        self.assertEquals(conn.create_subnet.call_args_list[1][0], ('vpc-id', '10.0.129.0/24'))
-        self.assertEquals(conn.create_subnet.call_args_list[2][0], ('vpc-id', '10.0.130.0/24'))
+        self.assertEqual(conn.create_vpc.call_args_list[0][0][0], "10.0.128.0/22")
+        self.assertEqual(conn.create_subnet.call_args_list[0][0], ('vpc-id', '10.0.128.0/24'))
+        self.assertEqual(conn.create_subnet.call_args_list[1][0], ('vpc-id', '10.0.129.0/24'))
+        self.assertEqual(conn.create_subnet.call_args_list[2][0], ('vpc-id', '10.0.130.0/24'))
 
         self.assertNotIn("ERROR", self.log.getvalue(), msg="ERROR found in log: %s" % self.log.getvalue())
 
@@ -400,11 +400,11 @@ def test_30_updateVMInfo(self, record_sets, connect_to_region, get_connection):
         self.assertTrue(success, msg="ERROR: updating VM info.")
         self.assertNotIn("ERROR", self.log.getvalue(), msg="ERROR found in log: %s" % self.log.getvalue())
 
-        self.assertEquals(dns_conn.create_zone.call_count, 1)
-        self.assertEquals(dns_conn.create_zone.call_args_list[0][0][0], "domain.com.")
-        self.assertEquals(changes.add_change.call_args_list, [call('CREATE', 'test.domain.com.', 'A')])
-        self.assertEquals(change.add_value.call_args_list, [call('158.42.1.1')])
-        self.assertEquals(conn.create_route.call_args_list, [call('routet-id', '10.0.10.0/24', instance_id='int-id')])
+        self.assertEqual(dns_conn.create_zone.call_count, 1)
+        self.assertEqual(dns_conn.create_zone.call_args_list[0][0][0], "domain.com.")
+        self.assertEqual(changes.add_change.call_args_list, [call('CREATE', 'test.domain.com.', 'A')])
+        self.assertEqual(change.add_value.call_args_list, [call('158.42.1.1')])
+        self.assertEqual(conn.create_route.call_args_list, [call('routet-id', '10.0.10.0/24', instance_id='int-id')])
 
     @patch('IM.connectors.EC2.EC2CloudConnector.get_connection')
     def test_30_updateVMInfo_spot(self, get_connection):
@@ -690,17 +690,17 @@ def test_60_finalize(self, record_sets, connect_to_region, sleep, get_connection
         self.assertTrue(success, msg="ERROR: finalizing VM info.")
         self.assertNotIn("ERROR", self.log.getvalue(), msg="ERROR found in log: %s" % self.log.getvalue())
 
-        self.assertEquals(dns_conn.delete_hosted_zone.call_count, 1)
-        self.assertEquals(dns_conn.delete_hosted_zone.call_args_list[0][0][0], zone.id)
-        self.assertEquals(changes.add_change.call_args_list, [call('DELETE', 'test.domain.com.', 'A')])
-        self.assertEquals(change.add_value.call_args_list, [call('158.42.1.1')])
-        self.assertEquals(sg.delete.call_args_list, [call()])
-        self.assertEquals(sg1.delete.call_args_list, [])
-        self.assertEquals(sg2.delete.call_args_list, [call()])
-        self.assertEquals(conn.delete_subnet.call_args_list, [call('subnet-id')])
-        self.assertEquals(conn.delete_vpc.call_args_list, [call('vpc-id')])
-        self.assertEquals(conn.delete_internet_gateway.call_args_list, [call('ig-id')])
-        self.assertEquals(conn.detach_internet_gateway.call_args_list, [call('ig-id', 'vpc-id')])
+        self.assertEqual(dns_conn.delete_hosted_zone.call_count, 1)
+        self.assertEqual(dns_conn.delete_hosted_zone.call_args_list[0][0][0], zone.id)
+        self.assertEqual(changes.add_change.call_args_list, [call('DELETE', 'test.domain.com.', 'A')])
+        self.assertEqual(change.add_value.call_args_list, [call('158.42.1.1')])
+        self.assertEqual(sg.delete.call_args_list, [call()])
+        self.assertEqual(sg1.delete.call_args_list, [])
+        self.assertEqual(sg2.delete.call_args_list, [call()])
+        self.assertEqual(conn.delete_subnet.call_args_list, [call('subnet-id')])
+        self.assertEqual(conn.delete_vpc.call_args_list, [call('vpc-id')])
+        self.assertEqual(conn.delete_internet_gateway.call_args_list, [call('ig-id')])
+        self.assertEqual(conn.detach_internet_gateway.call_args_list, [call('ig-id', 'vpc-id')])
 
     @patch('IM.connectors.EC2.EC2CloudConnector.get_connection')
     @patch('time.sleep')

test/unit/connectors/Fogbow.py

@@ -298,10 +298,10 @@ def test_30_updateVMInfo(self, sleep, requests):
 
         self.assertTrue(success, msg="ERROR: updating VM info.")
         self.assertNotIn("ERROR", self.log.getvalue(), msg="ERROR found in log: %s" % self.log.getvalue())
-        self.assertEquals(vm.info.systems[0].getValue("net_interface.1.ip"), "10.0.0.1")
-        self.assertEquals(vm.info.systems[0].getValue("net_interface.0.ip"), "8.8.8.8")
-        self.assertEquals(vm.info.systems[0].getValue("memory.size"), 1073741824)
-        self.assertEquals(vm.info.systems[0].getValue("disk.1.device"), "/dev/sdb")
+        self.assertEqual(vm.info.systems[0].getValue("net_interface.1.ip"), "10.0.0.1")
+        self.assertEqual(vm.info.systems[0].getValue("net_interface.0.ip"), "8.8.8.8")
+        self.assertEqual(vm.info.systems[0].getValue("memory.size"), 1073741824)
+        self.assertEqual(vm.info.systems[0].getValue("disk.1.device"), "/dev/sdb")
 
         data = json.loads(requests.call_args_list[1][1]["data"])
         self.assertEqual(data["computeId"], '1')