Source code for dagster_azure.adls2.fake_adls2_resource

import io
import random
from typing import Dict
from unittest import mock

from dagster_azure.blob import FakeBlobServiceClient

from dagster import resource

from .resources import ADLS2Resource
from .utils import ResourceNotFoundError


@resource({"account_name": str})
def fake_adls2_resource(context):
    return FakeADLS2Resource(account_name=context.resource_config["account_name"])


[docs]class FakeADLS2Resource(ADLS2Resource): """Stateful mock of an ADLS2Resource for testing. Wraps a ``mock.MagicMock``. Containers are implemented using an in-memory dict. """ def __init__( self, account_name, credential="fake-creds" ): # pylint: disable=unused-argument,super-init-not-called self._adls2_client = FakeADLS2ServiceClient(account_name) self._blob_client = FakeBlobServiceClient(account_name) self._lease_client_constructor = FakeLeaseClient
class FakeLeaseClient: def __init__(self, client): self.client = client self.id = None # client needs a ref to self to check if a given lease is valid self.client._lease = self def acquire(self, lease_duration=-1): # pylint: disable=unused-argument if self.id is None: self.id = random.randint(0, 2**9) else: raise Exception("Lease already held") def release(self): self.id = None def is_valid(self, lease): if self.id is None: # no lease is held so any operation is valid return True return lease == self.id class FakeADLS2ServiceClient: """Stateful mock of an ADLS2 service client for testing. Wraps a ``mock.MagicMock``. Containers are implemented using an in-memory dict. """ def __init__(self, account_name, credential="fake-creds"): self._account_name = account_name self._credential = mock.MagicMock() self._credential.account_key = credential self._file_systems = {} @property def account_name(self): return self._account_name @property def credential(self): return self._credential @property def file_systems(self): return self._file_systems def get_file_system_client(self, file_system): return self._file_systems.setdefault( file_system, FakeADLS2FilesystemClient(self.account_name, file_system) ) def get_file_client(self, file_system, file_path): return self.get_file_system_client(file_system).get_file_client(file_path) class FakeADLS2FilesystemClient: """Stateful mock of an ADLS2 filesystem client for testing.""" def __init__(self, account_name, file_system_name): self._file_system: Dict[str, FakeADLS2FileClient] = {} self._account_name = account_name self._file_system_name = file_system_name @property def account_name(self): return self._account_name @property def file_system_name(self): return self._file_system_name def keys(self): return self._file_system.keys() def get_file_system_properties(self): return {"account_name": self.account_name, "file_system_name": self.file_system_name} def has_file(self, path): return bool(self._file_system.get(path)) def get_file_client(self, file_path): # pass fileclient a ref to self and its name so the file can delete itself self._file_system.setdefault(file_path, FakeADLS2FileClient(self, file_path)) return self._file_system[file_path] def create_file(self, file): # pass fileclient a ref to self and the file's name so the file can delete itself by # accessing the self._file_system dict self._file_system.setdefault(file, FakeADLS2FileClient(fs_client=self, name=file)) return self._file_system[file] def delete_file(self, file): for k in list(self._file_system.keys()): if k.startswith(file): del self._file_system[k] class FakeADLS2FileClient: """Stateful mock of an ADLS2 file client for testing.""" def __init__(self, name, fs_client): self.name = name self.contents = None self._lease = None self.fs_client = fs_client @property def lease(self): return self._lease if self._lease is None else self._lease.id def get_file_properties(self): if self.contents is None: raise ResourceNotFoundError("File does not exist!") lease_id = None if self._lease is None else self._lease.id return {"lease": lease_id} def upload_data(self, contents, overwrite=False, lease=None): if self._lease is not None: if not self._lease.is_valid(lease): raise Exception("Invalid lease!") if self.contents is not None or overwrite is True: if isinstance(contents, str): self.contents = contents.encode("utf8") elif isinstance(contents, io.BytesIO): self.contents = contents.read() elif isinstance(contents, io.StringIO): self.contents = contents.read().encode("utf8") elif isinstance(contents, bytes): self.contents = contents else: self.contents = contents def download_file(self): if self.contents is None: raise ResourceNotFoundError("File does not exist!") return FakeADLS2FileDownloader(contents=self.contents) def delete_file(self, lease=None): if self._lease is not None: if not self._lease.is_valid(lease): raise Exception("Invalid lease!") self.fs_client.delete_file(self.name) class FakeADLS2FileDownloader: """Mock of an ADLS2 file downloader for testing.""" def __init__(self, contents): self.contents = contents def readall(self): return self.contents def readinto(self, fileobj): fileobj.write(self.contents)