test_azure_batch_5.py 789 B

1234567891011121314151617
  1. def test_configure_pool_with_latest_vm(self):
  2. with mock.patch(
  3. "airflow.providers.microsoft.azure.hooks."
  4. "batch.AzureBatchHook._get_latest_verified_image_vm_and_sku"
  5. ) as mock_getvm:
  6. hook = AzureBatchHook(azure_batch_conn_id=self.test_cloud_conn_id)
  7. getvm_instance = mock_getvm
  8. getvm_instance.return_value = ['test-image', 'test-sku']
  9. pool = hook.configure_pool(
  10. pool_id='mypool',
  11. vm_size="test_vm_size",
  12. use_latest_image_and_sku=True,
  13. vm_publisher="test.vm.publisher",
  14. vm_offer="test.vm.offer",
  15. sku_starts_with="test-sku",
  16. )
  17. assert isinstance(pool, batch_models.PoolAddParameter)