add unit test in label-studio django project to azure storage

Question

I am trying to add a new unit test to label-studio project

I am not able to successfully create the storage upload below

- name: stage
  request:
    data:
      container: pytest-azure-images
      project: '{project_pk}'
      title: Testing Azure SPI storage (bucket from conftest.py)
      use_blob_urls: true
    method: POST
    url: '{django_live_url}/api/storages/azure_spi'
  response:
    save:
      json:
        storage_pk: id
    status_code: 201

since I get last_sync_count: 0 in the results. How is the bucket from "conftest.py" being uploaded?

- name: stage
  request:
    method: POST
    url: '{django_live_url}/api/storages/azure_spi/{storage_pk}/sync'
  response:
    json:
      last_sync_count: 0
    status_code: 200

Here is my branch and PR if you can help: https://github.com/HumanSignal/label-studio/pull/5926/files

Answer 1

I was able to fix this. The issue was that the storage account class was not properly patched, and the three files were not being returned by the mock test. This solved the issue:

class DummyAzureContainer:
         def __init__(self, container_name, **kwargs):
             self.name = container_name

         def list_blobs(self, name_starts_with):
             return [File('abc'), File('def'), File('ghi')]

         def get_blob_client(self, key):
             return DummyAzureBlob(self.name, key)

         def get_container_properties(self, **kwargs):
             return SimpleNamespace(
                 name='test-container',
                 last_modified='2022-01-01 01:01:01',
                 etag='test-etag',
                 lease='test-lease',
                 public_access='public',
                 has_immutability_policy=True,
                 has_legal_hold=True,
                 immutable_storage_with_versioning_enabled=True,
                 metadata={'key': 'value'},
                 encryption_scope='test-scope',
                 deleted=False,
                 version='1.0.0',
             )

         def download_blob(self, key):
             return DummyAzureBlob(self.name, key)

     class DummyBlobServiceClient:
         def __init__(self, *args, **kwargs):
             pass

         def get_container_client(self, container_name):
             return DummyAzureContainer(container_name)

     class DummyClientSecretCredential:
         def __init__(self, *args, **kwargs):
             pass

         def get_token(self):
             return 'token'

     with mock.patch.object(models, 'ClientSecretCredential' , DummyClientSecretCredential):
         with mock.patch.object(models, 'BlobServiceClient' , DummyBlobServiceClient):
             with mock.patch.object(models, 'generate_blob_sas', return_value='token'):
                 yield

Answer 2

Follow these steps to add a unit test to the Label Studio project for uploading file to Azure Storage:

Update the settings.py of your Label Studio project to include Azure Storage settings.

# settings.py
AZURE_ACCOUNT_NAME = 'your_account_name'
AZURE_ACCOUNT_KEY = 'your_account_key'
AZURE_CONTAINER = 'azure-images'

• Connect Label Studio to Azure Blob Storageusing this reference

• I used this link to write unit tests in Django application and run the tests with django.test.

• Create a unit test file test_azure_storage.py under the tests directory.

# tests/test_azure_storage.py

import pytest
from django.conf import settings
from rest_framework.test import APIClient

@pytest.fixture
def api_client():
    return APIClient()

@pytest.mark.django_db
def test_azure_storage_upload(api_client):
    # Create a storage
    response = api_client.post(
        f'{settings.DJANGO_LIVE_URL}/api/storages/azure_spi',
        data={
            'container': settings.AZURE_CONTAINER,
            'project': '{project_pk}',
            'title': 'Testing Azure SPI storage (bucket from conftest.py)',
            'use_blob_urls': True,
        }
    )
    assert response.status_code == 201
    storage_pk = response.json().get('id')

    # Sync the storage
    response = api_client.post(
        f'{settings.DJANGO_LIVE_URL}/api/storages/azure_spi/{storage_pk}/sync'
    )
    assert response.status_code == 200
    assert response.json().get('last_sync_count') > 0

    def test_upload_blob(self):
        container_client = self.blob_service_client.get_container_client(self.container_name)
        blob_client = container_client.get_blob_client("test_blob.txt")
        blob_client.upload_blob("This is a test blob")