I've writting some unit tests for some auxiliary classes that are currently working. The class is in charge of loading some files that are inside a blob storage, splitting their inner lines to read them and process.
My unit test is trying to assert that giving some list of files the process is doing whatever is expected to do. This is working in the test explorer, it's also working in the console when executing pytest and also the coverage 100% method. When commiting the changes into the repo, I got an error in the build inside the azure pipelines process and I tried all I know without success.
The class:
Hanimport logging
import re
from abc import abstractmethod
from common.blob_storage import BlobStorageHandler
class BlobProcessor:
def __init__(self, logger: logging.Logger) -> None:
self.logger = logger
async def process_files(
self,
blob_storage_handler: BlobStorageHandler,
dir_path: str,
filename_filter: str,
) -> None:
# List blobs in the directory
blobs = blob_storage_handler.container_client.list_blobs(
name_starts_with=dir_path
)
# Loop through matching blobs and read their content and publish a message for each line
for blob in blobs:
if re.search(filename_filter, blob.name.split("/")[-1]):
self.logger.info(f"{blob.name}: Processing file")
# Read file contents and split by newlines
blob_content_raw = blob_storage_handler.get_blob_data(blob.name)
blob_content_decoded = blob_content_raw.decode("utf-8-sig")
lines = [line.strip() for line in blob_content_decoded.split("\n") if line]
# Process and send lines
await self.process_lines(lines)
self.logger.info(f"{blob.name}: Successfully processed {len(lines)} messages")
@abstractmethod
async def process_lines(self, lines: list[str]) -> None:
pass
This is the unit test done :
from unittest import IsolatedAsyncioTestCase
from unittest.mock import Mock
from common.blob_storage import BlobStorageHandler
from common.blob_processor import BlobProcessor
class TestBlobProcessor(IsolatedAsyncioTestCase):
async def test_process_files_with_matching_blob(self):
# Arrange
mock_logger = Mock()
mock_blob_storage_handler = Mock(spec=BlobStorageHandler)
mock_blob_storage_handler.container_client = Mock()
mock_blob = Mock()
mock_blob.name = "test_file.txt"
mock_blob_storage_handler.container_client.list_blobs.return_value = [mock_blob]
mock_blob_storage_handler.get_blob_data.return_value = b"Line 1\nLine 2\nLine 3\n"
data_processor = BlobProcessor(mock_logger)
# Act
await data_processor.process_files(mock_blob_storage_handler, "test_dir", ".*\\.txt")
# Assert
mock_logger.info.assert_any_call("test_file.txt: Processing file")
mock_logger.info.assert_any_call("test_file.txt: Successfully processed 3 messages")
The error that I get on azure pipelines is this:
E TypeError: BlobProcessor.init() missing 3 required positional arguments: 'blob_storage_handler', 'dir_path', and 'filename_filter'
Also the test is being debugged perfectly, I got the for each iterated correctly and I got the exact lines readed that I put in the test parameter...
The problem was solved by doing a re-base of the original branch into my current developing branch, basically, the pipeline in the build activity, prevents a future merge.
So, the merge into the original branch is doing versus the published origin code. In that code, another developer approved a pull request with 3 adittional parameters in the dataprocessor class, so my code was working, but only at my local environment.
In resume: the pipeline does the build using the last version of the published code, if your branch is outdated it may trigger a problem like this.