# coding: utf-8 #------------------------------------------------------------------------- # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. See License.txt in the project root for # license information. #-------------------------------------------------------------------------- import unittest import azure.mgmt.datamigration import azure.mgmt.network.models from azure.mgmt.datamigration.models import (DataMigrationService, ServiceSku, Project, SqlConnectionInfo, MigrateSqlServerSqlDbTaskProperties, MigrateSqlServerSqlDbTaskInput, MigrateSqlServerSqlDbDatabaseInput, MigrationValidationOptions) from devtools_testutils import AzureMgmtTestCase, ResourceGroupPreparer @unittest.skip("skip test") class MgmtDataMigrationTest(AzureMgmtTestCase): location_name = 'centralus' def setUp(self): super(MgmtDataMigrationTest, self).setUp() self.dms_sdk_client = self.create_mgmt_client(azure.mgmt.datamigration.DataMigrationManagementClient) self.network_sdk_client = self.create_mgmt_client(azure.mgmt.network.NetworkManagementClient) @ResourceGroupPreparer(name_prefix='dms_sdk_test', location=location_name) def test_datamigration(self, resource_group): vnet_name = self.get_resource_name('pysdkdmstestvnet') vsubnet_id = '/subscriptions/{}/resourceGroups/{}/providers/Microsoft.Network/virtualNetworks/{}/subnets/subnet1' service_name = self.get_resource_name('pysdkdmstestservice') sku_name = 'GeneralPurpose_2vCores' project_name = self.get_resource_name('pysdkdmstestproject') task_name = self.get_resource_name('pysdkdmstesttask') # ensure the service name is available name_availability = self.dms_sdk_client.services.check_name_availability( location = self.location_name, name = service_name, type = 'services') self.assertTrue(name_availability.name_available) # DMS requires a real vnet plus subnet if self.is_live: vnet_creation_async = self.network_sdk_client.virtual_networks.create_or_update( resource_group.name, vnet_name, { 'location': self.location_name, 'address_space': { 'address_prefixes': ['10.0.0.0/16'] } } ) vnet_creation_async.wait() self.network_sdk_client.subnets.create_or_update( resource_group.name, vnet_name, 'subnet1', {'address_prefix': '10.0.0.0/24'} ).wait() # create the service params_create_service = DataMigrationService( location = self.location_name, virtual_subnet_id = vsubnet_id.format(self.settings.SUBSCRIPTION_ID, resource_group.name, vnet_name), sku=ServiceSku(name=sku_name)) service_creation_async = self.dms_sdk_client.services.create_or_update( parameters = params_create_service, group_name = resource_group.name, service_name = service_name) service_creation_async.wait() # check the service dms_service = self.dms_sdk_client.services.get(group_name = resource_group.name, service_name = service_name) self.assertEqual(dms_service.provisioning_state, 'Succeeded') self.assertEqual(dms_service.name, service_name) self.assertEqual(dms_service.location, self.location_name) self.assertEqual(dms_service.sku.name, sku_name) # the service name should not be available name_availability = self.dms_sdk_client.services.check_name_availability( location = self.location_name, name = service_name, type = 'services') self.assertFalse(name_availability.name_available) # create a project params_create_project = Project( location = self.location_name, source_platform = 'SQL', target_platform = 'SQLDB') project_creation = self.dms_sdk_client.projects.create_or_update( parameters = params_create_project, group_name = resource_group.name, service_name = service_name, project_name = project_name) # check the project dms_project = self.dms_sdk_client.projects.get( group_name = resource_group.name, service_name = service_name, project_name = project_name) self.assertEqual(dms_project.provisioning_state, 'Succeeded') self.assertEqual(dms_project.name, project_name) self.assertEqual(dms_project.source_platform, 'SQL') self.assertEqual(dms_project.target_platform, 'SQLDB') # create a task database_options = [] database_options.append( MigrateSqlServerSqlDbDatabaseInput( name = 'Test_Source', target_database_name = 'Test_Target', make_source_db_read_only = False, table_map = { "dbo.TestTableForeign": "dbo.TestTableForeign", "dbo.TestTablePrimary": "dbo.TestTablePrimary" })) validation_options = MigrationValidationOptions( enable_schema_validation = False, enable_data_integrity_validation = False, enable_query_analysis_validation = False) task_input = MigrateSqlServerSqlDbTaskInput( source_connection_info = { 'userName': 'testuser', 'password': 'password', 'dataSource': 'testsource.microsoft.com', 'authentication': 'SqlAuthentication', 'encryptConnection': True, 'trustServerCertificate': True }, target_connection_info = { 'userName': 'testuser', 'password': 'password', 'dataSource': 'testtarget.microsoft.com', 'authentication': 'SqlAuthentication', 'encryptConnection': True, 'trustServerCertificate': True }, selected_databases = database_options, validation_options = validation_options) migration_properties = MigrateSqlServerSqlDbTaskProperties(input = task_input) task_creation = self.dms_sdk_client.tasks.create_or_update( group_name = resource_group.name, service_name = service_name, project_name = project_name, task_name = task_name, properties = migration_properties) # check the task dms_task = self.dms_sdk_client.tasks.get( group_name = resource_group.name, service_name = service_name, project_name = project_name, task_name = task_name) self.assertEqual(dms_task.name, task_name) self.assertEqual(dms_task.properties.input.selected_databases[0].name, 'Test_Source') self.assertEqual(dms_task.properties.input.source_connection_info.data_source, 'testsource.microsoft.com') self.assertEqual(dms_task.properties.input.target_connection_info.data_source, 'testtarget.microsoft.com') self.assertFalse(dms_task.properties.input.validation_options.enable_schema_validation) self.assertEqual(dms_task.properties.task_type, 'Migrate.SqlServer.SqlDb') # delete the task self.dms_sdk_client.tasks.delete( group_name = resource_group.name, service_name = service_name, project_name = project_name, task_name = task_name, delete_running_tasks = True) # delete the project self.dms_sdk_client.projects.delete( group_name = resource_group.name, service_name = service_name, project_name = project_name) # delete the service service_deletion_async = self.dms_sdk_client.services.delete( group_name = resource_group.name, service_name = service_name) service_deletion_async.wait() # ensure the service name is available again name_availability = self.dms_sdk_client.services.check_name_availability( location = self.location_name, name = service_name, type = 'services') self.assertTrue(name_availability.name_available) #------------------------------------------------------------------------------ if __name__ == '__main__': unittest.main()