Based on our interactions and the provided guidelines, here’s a breakdown of frequent unit test issues:

## 1. Incorrect Mocking Strategy (High Frequency)

This is the most common category of errors encountered.

* **Error Type:** Patching the wrong target.
* **Root Cause:** `unittest.mock.patch` needs to target the name of the object *as it is looked up in the module under test*, not necessarily where it’s defined. If module `A` does `from B import C`, and you are testing module `A`, you need to patch `A.C` (or `module_A_path.C`), not `B.C`.
* **Golden Pattern:**
* Always identify where the object (class, function, method) is being imported and used by the code you are testing.
* Use `patch(‘module_under_test.object_to_mock’, …)` or `patch.object(ClassUsedInModuleUnderTest, ‘method_to_mock’, …)`.
* Refer to `unit_test_guide.mdc` section “Patching Imported Objects/Functions” for examples.

* **Error Type:** Not using `AsyncMock` for `async` methods/functions.
* **Root Cause:** Using `MagicMock` or a regular function for an `async def` callable. When the code under test `await`s the mock, it results in a `TypeError` because `MagicMock` instances are not awaitable by default.
* **Golden Pattern:**
* Always use `unittest.mock.AsyncMock` when mocking asynchronous functions or methods.
* Example: `with patch.object(MyClass, ‘async_method’, new_callable=AsyncMock) as mock_async_method: …`
* Ensure your test function itself is marked with `@pytest.mark.asyncio` if it directly calls or awaits async code.

* **Error Type:** Incorrect `return_value` or `side_effect` for mocks.
* **Root Cause:** The mock doesn’t return the expected type or structure, or a `side_effect` function is overly complex or has incorrect logic. This can also happen when a lambda is used for a side effect and doesn’t handle arguments correctly (e.g., missing `self` if it’s meant to mimic a method).
* **Golden Pattern:**
* Keep mock return values simple. For `AsyncMock`, use `AsyncMock(return_value=…)`.
* If a `side_effect` is needed, ensure it’s a simple function that accurately reflects the behavior needed for the test. Avoid complex logic within `side_effect` functions.
* For methods that return `self` (fluent interfaces), use `mock_obj.return_value = mock_obj`.

* **Error Type:** Overly complex mock setups.
* **Root Cause:** Trying to mock too many implementation details or creating intricate `side_effect` functions when simpler approaches would suffice.
* **Golden Pattern:**
* Focus on mocking the *interface* your code interacts with, not the internal workings of the dependency.
* Prefer `return_value` over `side_effect` when possible.
* Use fixtures (`@pytest.fixture`) to create reusable mock setups for common dependencies.

## 2. Assertion Mismatches (Medium Frequency)

These errors occur when the test’s expectations don’t align with the actual behavior of the code.

* **Error Type:** Outdated assertions or incorrect expected values.
* **Root Cause:** The code under test has changed (e.g., function signature, return value structure, logic), but the test assertions haven’t been updated to reflect these changes. A specific instance was asserting a detailed query string when the code was updated to use a generic `”*”` (as seen in `test_get_chart_by_id_behavior` from `unit_test_guide.mdc`).
* **Golden Pattern:**
* When refactoring code, always review and update corresponding unit tests.
* Make assertions precise but flexible enough to accommodate minor, non-functional changes if appropriate.
* Use descriptive variable names for expected values to improve readability.
* When asserting mock calls, use `mock_object.assert_called_once_with(…)` with the exact arguments and keyword arguments expected. `mock_object.call_args` can be used to inspect actual call arguments for debugging.

* **Error Type:** Asserting non-existent keys or attributes.
* **Root Cause:** The test tries to access a key in a dictionary (e.g., `kwargs[‘permissions’]`) or an attribute of an object that isn’t present in the actual result or call arguments.
* **Golden Pattern:**
* Before asserting the value of a key/attribute, ensure it’s expected to be present in that specific test scenario.
* Use `assert ‘key’ in dictionary` or `hasattr(obj, ‘attribute’)` before asserting the value if its presence is conditional.
* Carefully check the actual arguments passed to mocks (e.g., using `mock_name.call_args`) to verify what is actually being sent.

## 3. Scope and Context Issues (Medium Frequency)

These issues often relate to how the code interacts with its environment or input parameters.

* **Error Type:** Misunderstanding of parameter handling or conditional logic.
* **Root Cause:** The test setup doesn’t correctly reflect the conditions under which certain parameters are passed or certain code paths are taken. For example, initial confusion about whether `permissions` should always be passed to `search_context` or is conditional.
* **Golden Pattern:**
* Thoroughly understand the function/method signature and its conditional logic.
* Create separate tests for different code paths based on input parameters.
* In our “Fixing Unit Test in Zsh” interaction, we saw an issue where `PromptPlugin.build_augmented_prompt` expected a dictionary key named `”question”` but the test was providing `”query”`. This was a direct mismatch in expected input structure. The fix was to align the test’s input with the method’s expectation.

* **Error Type:** Input data structure mismatch.
* **Root Cause:** The test provides input data (e.g., a dictionary for arguments) that doesn’t match the structure expected by the function or method under test. This was the core issue in the `test_prompt_plugin.py` scenario where the key `query` was used instead of `question`.
* **Golden Pattern:**
* Double-check the expected input structure of the function/method being tested, especially if it accepts complex types like dictionaries or Pydantic models.
* Ensure the keys, types, and nesting of test data match the requirements.
* When dealing with Pydantic models or typed dictionaries, ensure your test data conforms to the schema.

## 4. Environment/Import Errors (Less Frequent but Critical)

* **Error Type:** `ImportError` or issues related to test environment setup.
* **Root Cause:** The test environment isn’t configured correctly, Python path issues, or dependencies are missing.
* **Golden Pattern:**
* Ensure your test runner (e.g., `pytest`) is invoked from the correct directory (usually the project root) so that source code modules can be found.
* Use virtual environments to manage dependencies consistently.
* Ensure all necessary packages are installed in the test environment.

## General Golden Patterns for Robust Unit Tests:

* **AAA Pattern (Arrange, Act, Assert):** Structure your tests clearly.
* **Arrange:** Set up all preconditions, including mock configurations and input data.
* **Act:** Execute the single unit of code being tested.
* **Assert:** Verify the outcome (return values, mock calls, state changes).
* **Test One Thing at a Time:** Each test should ideally verify a single behavior or code path. This makes failures easier to diagnose.
* **Descriptive Test Names:** Name your tests clearly to indicate what they are testing (e.g., `test_method_returns_true_when_input_is_valid`).
* **Use Fixtures for Setup:** `@pytest.fixture` is excellent for creating reusable setup code (e.g., mock instances, test data).
* **Keep Tests Independent:** Tests should not depend on the order of execution or the state left by other tests.

When writing code, clarity and intent matter just as much as functionality. A subtle but common error in Python, particularly when handling arrays like those from NumPy, offers a perfect lens through which to explore this principle. Let’s dive into a real-world example, dissect the problem, and uncover why explicitness in programming is a virtue worth embracing.

Root Cause

The issue surfaces in a file called `azure_ai_search_memory.py` at line 165, where the following code resides:

```python
if embeddings and len(embeddings) > 0:
```

At first glance, this looks like a reasonable check: ensure `embeddings` exists and has elements. But Python disagrees, throwing an error. Why?

Why This Error Happens

The error stems from a fundamental mismatch between Python’s expectations and the code’s assumptions. Let’s break it down:

1. Array Boolean Context
- The variable `embeddings` is likely a NumPy array, and the code attempts to evaluate it directly in a boolean context (`if embeddings`).
- Python doesn’t inherently know how to interpret an array as `True` or `False`. Should it mean “all elements are non-zero”? “At least one element exists”? Something else entirely?
- This ambiguity causes Python to raise an exception rather than guess the programmer’s intent.

2. Python’s Boolean Evaluation
- When an object is used in a boolean context (e.g., `if some_object`), Python calls its `__bool__` method (or `__len__` for containers) to determine truthiness.
- NumPy arrays, however, deliberately avoid defining a default boolean conversion. This is a safety feature: it prevents accidental, unclear operations on multi-element data structures.
- The result? A `ValueError` or similar exception, forcing the developer to clarify their logic.

3. Common with NumPy Arrays
- This isn’t a rare edge case — it’s a frequent stumbling block when working with NumPy.
- Unlike Python lists, which evaluate as `False` when empty and `True` when non-empty, NumPy arrays demand explicit handling.
- For example, you might use `embeddings.any()` (are any elements true?) or `embeddings.all()` (are all elements true?), but a bare `if embeddings` won’t cut it.

Solution

To fix this, we need to rewrite the condition to be explicit about what we’re checking. Here’s the transformation:

```python
# Before (problematic)
if embeddings and len(embeddings) > 0:

# After (correct)
if embeddings is not None and len(embeddings) > 0:
```

Why This Works
- `embeddings is not None` explicitly checks if the variable is defined and not `None`, avoiding any attempt to evaluate the array as a boolean.
- `len(embeddings) > 0` then safely checks if the array has elements, assuming it’s a valid object with a length.
- Together, these conditions make the intent crystal clear: “Ensure `embeddings` exists and isn’t empty.”

Takeaways

This small error reveals a broader truth about programming: **explicitness often leads to better code**. Here are the key lessons:

- **Avoid Ambiguity**: Don’t rely on implicit boolean conversions for complex objects like arrays. If you mean “not None,” say `is not None`. If you mean “non-empty,” check the length explicitly.
- **Embrace Python’s Guardrails**: Features like NumPy’s refusal to auto-convert to boolean aren’t bugs — they’re protections. They push us to think harder about what we want.
- **Clarity Over Convenience**: Writing `if nums` might feel concise, but `if nums is not None` leaves no room for misinterpretation.

In the end, this isn’t just about fixing a bug — it’s about adopting a mindset. By being forced to articulate our intentions, we write code that’s not only functional but also self-documenting and robust. And that’s a philosophy worth coding by.

ClickHouse is a high-performance column-oriented database management system (DBMS) developed to handle real-time analytics with low latency. Its design excels in scenarios involving large volumes of read- and write-intensive operations.

ClickHouse Architecture Overview

Massive Parallel Processing (MPP)

• Architecture Model: ClickHouse uses an MPP (Massive Parallel Processing) architecture with a shared-nothing design.
• Decentralized Resource Usage: Each ClickHouse node operates independently, utilizing its own local resources (memory and storage).
• Parallel Processing: Nodes collaborate by processing tasks in parallel, without sharing state directly, which maximizes performance and fault tolerance.

Note:

• Imbalances in how data is distributed across shards can lead to uneven load across nodes, impacting query performance.

Columnar Storage in ClickHouse

• Data is stored and processed by columns instead of rows.
• Operations occur on arrays or chunks of columnar data, enabling vectorized execution.
• Minimizes the cost of data processing by working on batches (vectors), improving efficiency for analytical queries.

Shard, Replica, and Cluster in ClickHouse

1. Shard:

• A shard is a horizontal partition of a table’s data.
• Data is distributed across multiple shards, each located on different nodes.

2. Replica:

• A replica is a copy of a shard.
• Replicas ensure high availability and fault tolerance by allowing operations to continue even if one replica fails.

3. Cluster:

• A cluster in ClickHouse is a group of nodes organized by the MPP model.
• Each cluster consists of multiple shards, with each shard having one or more replicas.

4. KV Store:

Zookeeper is used to keep the shard, replica metadata

Application

ClickHouse plays a critical role in application backend as a data warehouse with three types of clusters:

1. Query Cluster:

• Stores six months’ behavior data, approximately 30 billion records per day, distributed across 100+ nodes.
• Primarily supports query-heavy workloads.

2. Write Cluster:

• Temporarily stores the latest day’s data (<1TB) and updates the query cluster every 30 minutes.
• Reduces query cluster’s load for better performance.

3. Data Durability and Recovery:

• Uses persistent volumes (PV) to ensure data safety.
• If a node or PV fails, a new node and volume are created and populated with replicated data from another node.

High Availability (HA) in ClickHouse

1. Current Setup:

• Single query cluster with replicas for intra-cluster disaster recovery.

2. Future Plan:

• Build duplicate query clusters across different regions and cosmic clusters to ensure high availability and cross-region disaster recovery.
• A multi-cluster setup mitigates risks associated with individual cluster failures.

ClickHouse’s distributed architecture, combined with its columnar data processing, makes it a powerhouse for real-time and analytics-intensive applications.

FAQ

• Q:Differences between sharding, partitioning and indexing
• A: Sharding: Distributes data across multiple nodes (horizontal scaling). Partitioning: Splits data within a single node (logical grouping). Indexing: Speeds up data retrieval by creating pointers to specific rows or values.

Background

Our project began with two primary goals:

1. SQL Copilot: Automating SQL syntax refactoring across database languages and providing explanations for database functions using AI.
2. Data Copilot: Translating natural language into SQL queries (NL2SQL), a problem that has been researched for over two decades but remains challenging in practical applications.

The project stems from our existing data platform, where many product managers and designers lack familiarity with various database query languages. It’s challenging for these non-technical stakeholders to develop data-savvy skills quickly. Enabling them to interact naturally with the platform can bridge this gap and empower decision-making without requiring advanced technical expertise.

Challenges

SQL Copilot

This project focused on setting up a robust infrastructure for AI services. While conceptually straightforward, building a scalable and reliable Retrieval-Augmented Generation (RAG) system required addressing:

• Embedding and Search Integration: Selecting a high-quality embedding model (Azure OpenAI) and a search system that supports vector search (Azure Cognitive Search).
• Document Chunking: Optimizing the division of database documentation for efficient search and retrieval.
• Prompt Design: Ensuring concise and relevant prompts for consistent and accurate outputs.

Data Copilot

The challenges here were more nuanced and multifaceted:

• Ambiguity in Natural Language: Words and phrases have multiple meanings that vary with context, making interpretation non-trivial.
• Database Complexity:
• Abbreviations and Inconsistencies: Column names and values often lack standardized formats, making semantic interpretation difficult.
• Redundancy: Redundant data inflates token costs and increases hallucinations in generated SQL.
• Performance Issues: Balancing retrieval recall and computational efficiency was essential for production-grade performance.

Phases

The evolving speed of LLMs and their surrounding ecosystem is astonishing. The quality of responses continues to have significant room for improvement, making it impractical to define a fixed methodology or final goal upfront. To tackle this, we broke the project into multiple phases, allowing us to adapt to the rapid changes in methodology and technology while ensuring steady progress.

Phase 1 — SQL Copilot

This phase focused on delivering SQL Copilot, leveraging its relative simplicity to establish the foundational AI infrastructure. RAG (Retrieval-Augmented Generation) was implemented as the backbone, enabling efficient query processing and retrieval. This phase ensured the infrastructure could support modern AI applications effectively.

Phase 2 — Data Copilot

Building on the established infrastructure, Phase 2 tackled the more complex challenge of Data Copilot. Here, we explored advanced techniques like semantic parsing and model fine-tuning to address NL2SQL problems. The focus was on adapting methodologies as they evolved, leveraging the modular design from Phase 1 for rapid iteration.

Implementations

Sql Copilot

Overview

There are the below question we need to resolve:

• Selection of embedding and chat service provider — Azure open AI
• Search service that support vector search — Azure AI Search
• Data preparation and chunking — Deciding chunk size
• Prompt representation and format — Deciding instruction, retrieved docs format
• Evaluation — RAGAS

And for the application to be mature, we need to consider(Our language is python and service stack is on Azure):

• Deployment Infra — Azure K8s
• Security — Enable MSI auth on Azure AAD
• Data persistency — Azure CosmosDB
• Web Server Framework — FastAPI
• LLM Orechestration framework — Semantic Kernel
• Observability — Opentelemetry + Azure Monitor

Technical Implementation

SQL Copilot

SQL Copilot was designed as a RAG-based system. Key components include:

1.Embedding and Search System:

• Embedding Model: Azure OpenAI’s embedding model (text-embedding-ada-002).
• Vector Search: Azure AI Search with native support for semantic search and cosine similarity as the distance metric.

2. Document Chunking:
Large documents were split into smaller, contextually relevant pieces using a recursive markdown splitter:

• Step 1: Split by structural elements like headers and paragraphs.
• Step 2: Merge smaller units recursively until a defined chunk size is met.
• Optimization: Empirically determined the optimal chunk size by evaluating retrieval metrics (precision, recall, and relevancy).

3. Prompt Engineering:

• Initially used simple instruction-based templates with retrieved context.
• Later, introduced function calling to improve flexibility. The LLM dynamically determines when to query additional context or external tools, reducing confusion from low-quality retrievals.

4. Evaluation Framework:

• Retrieval Metrics: Precision, recall, and relevancy of retrieved data.
• Generation Metrics: Faithfulness and relevancy of the LLM’s response.
• Ground Truth Comparison: F1 score, semantic similarity, and aspect critique to assess completeness and clarity.

Data Copilot (NL2SQL)

This is a very complex topic, we will discuss in part2.

Evaluation Metrics

The below metrics are more for sql copilot (RAG system), for nl2sql, we will disclose it in part2.

Retrieval Evaluation

• Contextual Precision: Measures how effectively the system ranks relevant context higher than irrelevant data.
• Contextual Recall: Assesses the system’s ability to capture all pertinent information.
• Relevancy: Ensures retrieved context remains focused without noise.

Generation Evaluation

• Answer Relevancy: Assesses whether generated SQL addresses user intent accurately.
• Faithfulness: Checks the factual accuracy of generated SQL against retrieved context.

Ground Truth Comparison

• F1 Score: Evaluates correctness by comparing generated SQL with ground-truth queries.
• Semantic Similarity: Analyzes how closely generated queries align with intent.
• Aspect Critique: Considers relevance, completeness, and clarity.

AI Infrastructure

A robust AI infrastructure is crucial for building scalable and efficient applications. Our focus was on creating a foundation that supports observability, data persistence, and seamless integration with AI services.

Deployment Infrastructure

We deployed our application on Azure Kubernetes Service (AKS) to ensure scalability and fault tolerance.

Technical Benefits:

• Horizontal Scaling: Easily accommodates high request loads by automatically scaling resources.
• Auto-Recovery: Supports resilient application recovery in case of failures.
• Container Orchestration: Simplifies management of containerized services, ensuring efficient resource utilization.

Web Server Framework

We chose FastAPI as our web server framework, leveraging its asynchronous capabilities for handling the orchestration-heavy nature of LLM applications.

Technical Benefits:

• Uvicorn-Based: Operates as a single-threaded server with a queue, ideal for asynchronous tasks like calling external services.
• Efficient Orchestration: Handles LLM workflows that require communication with various APIs and services seamlessly.
• Ease of Development: Provides a developer-friendly environment with built-in support for modern Python features.

Application Framework

We adopted Semantic Kernel as the core application framework for organizing LLM functions and plugins.

Technical Benefits:

• Modularity: Enables clear separation of concerns, making functions and plugins reusable and organized.
• Orchestration: Facilitates efficient coordination of multiple LLM capabilities.
• Scalability: Supports the addition of new functions and plugins without disrupting the existing architecture.

Logging

To enhance observability, we prioritized logging application activities in containerized Linux environments. This approach ensures comprehensive debugging context, correlates Q&A sessions with user feedback, and maintains standardized telemetry conventions.

Technical Selection:

• Framework: OpenTelemetry

Reasons:

• Standardization: Provides consistent telemetry conventions for logs, traces, and spans.
• Comprehensive Observability: Offers detailed views of operations within a trace, helping understand the end-to-end flow of requests.
• Integration with Azure Monitor: Supports seamless integration with Azure Monitor for storage, search, and visualization of telemetry data.
• Agent-Based Approach: Uses an intermediary agent to collect and forward telemetry data, enhancing reliability and efficiency.

Data Persistence

The database design centers on storing chat messages, including user questions, AI responses, and feedback. This supports retrieving past interactions, improving user experience, and serving as a training dataset for AI models.

Technical Selection:

• Database: CosmosDB/NoSQL

Reasons:

• Native Vector Support: Essential for AI tasks involving embeddings.
• Azure AI Search Integration: Facilitates efficient retrieval and search capabilities.
• Document-Based Design: Optimized for read-heavy operations, improving performance.
• Horizontal Scaling and Schema-Free: Ideal for managing large datasets with flexible structures.

This modular infrastructure underpins the AI services, ensuring adaptability for future requirements and rapid iterations.

What is Hybrid Search?

Hybrid search in Azure AI Search merges the results of full-text search and semantic search using the RRF algorithm. The goal is to leverage both keyword-based and meaning-based search techniques for better retrieval performance.

Full-Text Search

Full-text search is a traditional, keyword-based search method. It breaks down content into terms using language-specific analysis, and ranks documents based on keyword relevance using the BM25 scoring algorithm.

Why BM25?

BM25 is effective because it combines term frequency (TF) and inverse document frequency (IDF):

• TF measures how often a term appears in a document.
• IDF evaluates the rarity of a term across the document collection, giving more weight to rare terms.

BM25 excels at lexical matching and provides token importance weights, making it highly effective for keyword-based searches.

Why BM25 has these advantages?

BM25 estimates relevance by combining term frequency (TF) and inverse document frequency (IDF):

• TF: How often does each term in the query (“endangered,” “Bengal,” “tiger”) appear in a document?
• IDF: How rare are the query terms across the entire document collection? Rare terms are given more weight.

BM25’s score will vary based on these:

• The number of occurrences of the search term in each document. -> Lexical matching
• The rarity of the term across the entire collection (IDF). -> gives token importance weights
• The document length and how it affects term frequency. -> Contributes to term frequency counting

Semantic Search

Semantic search takes a different approach by understanding the meaning behind the query rather than relying solely on keyword matches. It converts the query into a vector representation using models like OpenAI’s text-embedding-ada-002, and then performs a vector search to find documents with similar meanings. This method relies on cosine similarity or dot product to measure how closely a document’s vector matches the query.

Unlike full-text search, which focuses on exact lexical matches, semantic search seeks to match the meaning behind the query, complementing the limitations of keyword-based search.

Combining Results

The intuition of semantic search is it complements the exact lexical matching, instead it is search for the closeness of meaning, which full-text search is unable to do. So combing these results could significantly increase the potential matching candidates.

But simply combining the results of both search methods is not enough. Reciprocal Rank Fusion (RRF) is an algorithm used to merge the rankings from both searches. RRF recalculates scores based on the rank position of each document in the result set.

For example, if a document ranks 1st in one result set and 3rd in another, RRF calculates the score for that document as:

• From the first search: 1(1+k)\frac{1}{(1+k)}(1+k)1​
• From the second search: 1(3+k)\frac{1}{(3+k)}(3+k)1​

The scores from both searches are summed to create a unified score, emphasizing documents that appear earlier in the lists.

Final Step: Semantic Re-ranking

After the initial retrieval phase using BM25, semantic search, and RRF, semantic ranking is applied as a secondary step. This re-ranks the top results based on their semantic relevance, ensuring the most meaningful documents appear first.

Service telemetry is an essential aspect of maintaining high availability and stability for any application, but it becomes even more critical for data analytics platforms. This article explores the methodology of implementing service telemetry using OpenTelemetry and Geneva, focusing on the general approach rather than specific products.​

Goal

Becoming modern observability ready, eventually following the 3 pillars of observability

Collection Architecture

Agent way is more friendly to the client as a lot of features irrelevant of sending telemetry itself are refactored in the agent.

More detailed

For Log and Span

For Metrics

Opentelemetry Data Model

For logs

For metrics

Telemetry Flow

Application will have opentelmetry sdk of its language to instrument telemetry through an exporter to an agent, while you could add a processor in between to process the log record. And the agent will be responsible for sending the telemetry to the monitoring service backend

Implementation

For Log

For Span

For Metrics

Log and Span can share the same agent, just the initialization codes are different. Metrics usually need a different agent specifically for aggregating metrics.

Example on how to model the instrumentation

Log And Span: Know your component, span each major stage, and log the info and errors. The key is to know the internal state just by the logs.

Metrics: Produce key metrics indicator that you would like to get alerted or monitor about.

Now you are fully equipped with the modern observability and know the health of your service.

Context:

I have a copilot chat service that needs saving the messages, cosmosDB NoSQL is chosen for this task. The stack is python.

Problem:

Even though cosmosDB theoretically should be very performant on the reads and writes due to the “unlimited” scaling. It is important to test it on our own hands and see the practical relation between the RU and the actual performance

Expectations:

Query and writes should be under second in a database with 500K+ items with 30 concurrent users who should resemble the normal user actions.

Methodology:

Given that a container’s performance could be optimized, we will have two containers for comparison. One is the “bench container” where we configure the basic minimum, and another one is the “optimized container” where we apply the best practices possible. Finally, with the same throughput provisioned and same workload, we will observe the performance for our targeted scenarios.

Setups:

1. Create the resource, database and containers(bench container and optimized container)
2. Ingest enough sample data to containers
3. Run Locust to mimic 30 concurrent users

Details

I will skip #1 as it has so many tutorials on how it could be done.

For #2, I used the stored procedure for fast insertions on the cosmosdb server side.

function generateDataAndInsert(params) {
    var context = getContext();
    var collection = context.getCollection();
    var collectionLink = collection.getSelfLink();

    var numberOfItems = params.numberOfItems;
    var userId = params.userId;
    var results = params.results || [];
    var limit = params.limit;

    function getRandomFloat(min, max) {
        return Math.random() * (max - min) + min;
    }

    function getRandomString(n) {
        var text = "";
        var possible = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789";
        for (var i = 0; i < n; i++) {
            text += possible.charAt(Math.floor(Math.random() * possible.length));
        }
        return text;
    }

    function getRandomInt(min, max) {
        min = Math.ceil(min);
        max = Math.floor(max);
        return Math.floor(Math.random() * (max - min + 1)) + min;
    }

    function getRandomTimestamp() {
        return new Date().toISOString();
    }

    function getRandomEmbeddings() {
        var embeddings = [];
        for (var i = 0; i < 1536; i++) {
            embeddings.push(getRandomFloat(-0.001, 0.01));
        }
        return embeddings;
    }

    function getRandomDateInDays() {
        var date = new Date();
        var randomDays = getRandomInt(0, 365);
        date.setDate(date.getUTCDate() - randomDays);
        return date.toISOString().substring(0, 10);
    }

    tryCreate(numberOfItems, userId, results, limit, callback);

    function tryCreate(count, userId, results, limit, callback) {

        if (count === 0 || limit === 0) {
            context.getResponse().setBody(count);
            return;
        }

        var item = {
            "chat_id": getRandomInt(0, 100000000),
            "user_id": userId,
            "question": getRandomString(20),
            "answers": [{"answer": getRandomString(20), "top_docs": [getRandomString(5), getRandomString(5)], "condensed_question" : getRandomString(20), "condensed_question_embeddings" : getRandomEmbeddings(), "answer_embeddings" : getRandomEmbeddings(), "feedback" : {"rating" : getRandomInt(0, 6), "content" : getRandomString(20), "timestamp" : getRandomTimestamp()}}],
            "created_day": getRandomDateInDays(),
            "chat_created_day": getRandomDateInDays(),
        };
        var isAccepted = collection.createDocument(collectionLink, item, {}, callback);
        if (!isAccepted) getContext().getResponse().setBody(count);
    }

    function callback(err, documentCreated) {
        if (err) getContext().getResponse().setBody(count);
        
        results.push(documentCreated.id);
        numberOfItems--;
        limit--;
        tryCreate(numberOfItems, userId, results, limit, callback);

    }
}

After you create it on the container, you could execute it on your machine. Note the stored procedure could only run within each logic partition

# The count is how many items you eventually want to create, the limit is
# to control how many to be processed at a time, since there is fixed 5 seconds
# timeout when calling stored procedure, you don't want to get timeout
def execute_stored_procedure(container, count, user_id, limit):
    """Execute the stored procedure and handle the response."""
    response = container.scripts.execute_stored_procedure(
        sproc="generateDataAndInsert",
        params=[{"numberOfItems": count, "userId": user_id, "limit": limit}],
        partition_key=user_id,
        enable_script_logging=True,
    )

# Using try catch to resume the remaining if exception occurs
count = 10000
while count > 0:
    try:
        count = execute_stored_procedure(container, count, user_id, 5)
        print(f"Remaining count: {count} for user_id: {user_id}")
    except Exception as e:
        print("Exception occurred, sleeping for 2 seconds...")
        print(f"Error: {e}")
        sleep(2)

For #3, I used locust to mimic a user behavior. A few take aways:

• Implement a class inheriting User from locust
• Create methods for the scenarios you want to test
• Adding @task(n) where n is the ratio to balance among scenarios, mimicking your real user usage
• Customize your fire events if you need to, in my case, I want to observe the actual time spent in container, excluding the time spent on network, so I extracted the db operation time from the response header and fire it to my locust host.

import json
import os
import random
from datetime import datetime, timezone
from typing import Callable

from azure.core.credentials import TokenCredential
from azure.cosmos import CosmosClient
from azure.identity import (
    CertificateCredential,
    DefaultAzureCredential,
    get_bearer_token_provider,
)
from azure.keyvault.secrets import SecretClient
from datasets import load_dataset
from dotenv import load_dotenv
from locust import User, between, events, tag, task
from semantic_kernel.connectors.ai.open_ai import AzureTextEmbedding

class Mongouser(User):
    wait_time = between(2, 5)

    def on_start(self):
        credential = CredentialManager(ConfigLoader()).get_credential()
        client = CosmosClient(
            url="https://nezha-wukong-nosql-local-wus2.documents.azure.com",
            credential=credential,
        )

        # Select your database and container
        database_name = "WukongDB"
        container_name = "OptimizedContainer"
        database = client.get_database_client(database_name)
        self.container = database.get_container_client(container_name)
        self.user_id = random.randint(0, 50)
        self.chat_id = []
        self.item_ids = []
        self.dataset = load_dataset("squad", split="train[:5000]").train_test_split(
            test_size=0.1
        )["train"]
        ad_token_provider = CredentialManager(ConfigLoader()).get_bearer_token_provider(
            credential, "https://cognitiveservices.azure.com/.default"
        )
        AOAI_API_BASE_WUS = ConfigLoader().load("AOAI_API_BASE_WUS")
        self.embedding_model = AzureTextEmbedding(
            deployment_name="text-embedding",
            ad_token_provider=ad_token_provider,
            endpoint=AOAI_API_BASE_WUS,
        )
        with open("./src/wukong/data_providers/cosmosdb/data.json") as f:
            self.documents = json.load(f)

    def success_fire(self, request_type, name):
        events.request.fire(
            request_type=request_type,
            name=name,
            response_time=float(
                self.container.client_connection.last_response_headers[
                    "x-ms-request-duration-ms"
                ]
            ),
            response_length=int(
                self.container.client_connection.last_response_headers["Content-Length"]
            ),
        )

    def exception_fire(self, request_type, name, exception):
        events.request.fire(
            request_type=request_type,
            name=name,
            response_time=float(
                self.container.client_connection.last_response_headers[
                    "x-ms-request-duration-ms"
                ]
            ),
            response_length=int(
                self.container.client_connection.last_response_headers["Content-Length"]
            ),
            exception=exception,
        )

    @task(1)
    @tag("GET")
    def fetch_test(self):
        if len(self.chat_id) == 0:
            return
        chat_id = random.choice(self.chat_id)

        try:
            document = self.container.query_items(
                query="SELECT TOP @n * FROM c WHERE c.chat_id = @chat_id OREDER BY c.timestamp DESC",
                parameters=[
                    {"name": "@chat_id", "value": chat_id},
                    {"name": "@n", "value": 10},
                ],
                enable_cross_partition_query=False,
                populate_query_metrics=True,
                partition_key=self.user_id,
            )

            self.success_fire(
                "GET", "Retrieve all messages from a user's specific chat"
            )
        except Exception as e:
            self.exception_fire(
                "GET", "Retrieve all messages from a user's specific chat", e
            )

    @task(4)
    @tag("POST")
    def post_test(self):
        document = random.choice(self.documents)
        chat_id = random.randint(0, 100)
        document["user_id"] = self.user_id
        document["chat_id"] = chat_id
        document["message_id"] = random.randint(0, 999)

        try:
            res = self.container.create_item(
                body=document,
                enable_automatic_id_generation=True,
                populate_query_metrics=True,
            )
            print(
                f"Document created successfully: {res['id']} for partition key {self.user_id}"
            )
            if chat_id not in self.chat_id and res["id"] is not None:
                self.chat_id.append(chat_id)
            if res["id"] is not None:
                self.item_ids.append(res["id"])
            self.success_fire("POST", "Create a new message in a user's specific chat")
        except Exception as e:
            self.exception_fire(
                "POST", "Create a new message in a user's specific chat", e
            )

    @task(1)
    @tag("PUT")
    def put_test(self):
        if len(self.item_ids) == 0:
            return
        item_id = random.choice(self.item_ids)
        partitionKey = self.user_id

        try:
            # Retrieve the document
            document = self.container.patch_item(
                item_id,
                partition_key=partitionKey,
                patch_operations=[
                    {"op": "add", "path": "/feedback_rating", "value": 4},
                    {
                        "op": "add",
                        "path": "/feedback_content",
                        "value": "Updated feedback content",
                    },
                    {
                        "op": "add",
                        "path": "/feedback_timestamp",
                        "value": datetime.now(timezone.utc).isoformat(),
                    },
                ],
            )

            self.success_fire("PUT", "Update feedback-related fields")
        except Exception as e:
            self.exception_fire("PUT", "Update feedback-related fields", e)

    @task(2)
    @tag("SEARCH")
    def search_data_test(self):
        if len(self.item_ids) == 0:
            return
        documents = random.choice(self.documents)

        query_embed = documents["question_embeddings"]

        try:
            # Define query
            query = (
                "SELECT TOP 5 c.user_id, c.timestamp, c.chat_id, c.message_id, c.body"
                "FROM c"
                "WHERE VectorDistance(c.answer_embeddings, {0}) > 0.5"
                "ORDER BY VectorDistance(c.answer_embeddings, {0})"
            ).format(query_embed)

            # Query items
            query_iterable = self.container.query_items(
                query=query,
                enable_cross_partition_query=True,
                populate_query_metrics=True,
                partition_key=self.user_id,
            )

            all_fetched_res = []
            for page in query_iterable.by_page():
                fetched_res = list(page)
                all_fetched_res.extend(fetched_res)

            for res in all_fetched_res:
                print(res)

            self.success_fire("SEARCH", "Search for similar questions")
        except Exception as e:
            self.exception_fire("SEARCH", "Search for similar questions", e)

class Singleton(type):
    """
    This class is a metaclass that implements the singleton pattern.
    """

    instance = None

    def __call__(cls, *args, **kwargs):
        if cls.instance is None:
            cls.instance = super(Singleton, cls).__call__(*args, **kwargs)
        return cls.instance

class ConfigLoader(metaclass=Singleton):
    """
    This class is responsible for loading environment variables from Azure Key Vault.
    """

    def __init__(self):
        load_dotenv()
        self.key_vault_name = os.environ.get("KEY_VAULT_NAME", "")
        self.key_vault_prefixes = os.environ.get("KEY_VAULT_PREFIXES", "").split(",")

        self.client = SecretClient(
            vault_url=f"https://{self.key_vault_name}.vault.azure.net",
            credential=DefaultAzureCredential(),
        )

    def load(self, key: str, default_value: str = None):
        env_value = os.environ.get(key)
        for key_vault_prefix in self.key_vault_prefixes:
            if (
                env_value
                and key_vault_prefix
                and env_value.lower().startswith(key_vault_prefix.lower())
                and self.key_vault_name
            ):
                return self.client.get_secret(env_value).value
        if env_value is None and default_value:
            return default_value
        return env_value

    @classmethod
    def reset_instance(cls):
        cls.instance = None

class CredentialManager:
    def __init__(self, config_loader: ConfigLoader):
        self.config_loader = config_loader

    def get_credential(self) -> TokenCredential:
        if self.config_loader.load("ENVIRONMENT") == "local":
            tenant_id = self.config_loader.load("AZURE_TENANT_ID")
            client_id = self.config_loader.load("AZURE_CLIENT_ID")
            certificate_path = self.config_loader.load("CERT_FILE_PATH")
            return CertificateCredential(
                tenant_id=tenant_id,
                client_id=client_id,
                certificate_path=certificate_path,
                send_certificate_chain=True,
            )
        return DefaultAzureCredential()

    def get_bearer_token_provider(
        self, credential: TokenCredential, scope: str
    ) -> Callable[[], str]:
        ad_token_provider = get_bearer_token_provider(
            credential,
            scope,
        )
        return ad_token_provider

credential = CredentialManager(ConfigLoader()).get_credential()

ad_token_provider = CredentialManager(ConfigLoader()).get_bearer_token_provider(
    credential, "https://cognitiveservices.azure.com/.default"
)
AOAI_API_BASE_WUS = ConfigLoader().load("AOAI_API_BASE_WUS")
embedding_model = AzureTextEmbedding(
    deployment_name="text-embedding",
    ad_token_provider=ad_token_provider,
    endpoint=AOAI_API_BASE_WUS,
)

Finally, you will have the beautiful and insightful charts indicating the performance:

I didn’t find a formal guide on this, so this is starting from my need. Please feel free to share what you think about how database performance tests should be carried out, cheers!

• You want to deploy azure function app through azure pipelines with automatic CI/CD.
• You have extra component like a transformer model to download for your function app.
• You are using python for azure function app
• You meet the problem : Zip deployment successful, but no function is shown on the azure function app portal

The Azure Function app is a Platform-as-a-Service (PaaS) offering that provides serverless compute for running event-driven code without the need for explicit provisioning or infrastructure management. In my scenario, I aim to host a simple service that loads an open source embedding model and provides vectorizing functionality for a list of texts.

For this project, I’ve chosen to use Python 3.10 because it includes the sentence-transformer package necessary for downloading and saving the model. As for the virtual machine image, I've opted for ubuntu-latest. To achieve continuous delivery, I rely on Azure DevOps agents, utilizing Azure Pipelines for CI/CD.

One challenge I encountered is downloading the model, which can be over 10 gigabytes in size. To avoid storing such a massive file in Git, I’ve explored two options: First, having the Azure Pipeline download it during the initial stage, or second, implementing a deployment post-hook so that after successful deployment, the Azure Function app triggers a script to download the model.

Although I extensively researched online resources, I couldn’t find any solutions regarding the second approach. Therefore, I decided to proceed with the first option, downloading the model during the pipeline process and then packaging all files into a zip deployment for the Azure Function app.

During the development and deployment phases of the function app, I encountered considerable confusion due to scattered settings and documentation. It took me two days to finally get the function to work. The challenge with Microsoft technology is the scattered nature of the documentation and resources, which often require trial and error to navigate effectively.

In an effort to bring more organization to the process, I aim to provide clearer explanations of each configuration, rather than relying solely on memorizing configurations as if they were magic spells.

Downloading model

Leveraging sentence_transformers package, you can easily download the open-source model on hugging-face

import os

model_parent_dir_name = "models"
default_embedding_model = "Your Model"
model_env_var = "SENTENCE_TRANSFORMERS_TEXT_EMBEDDING_MODEL"


def main():
    model_parent_dir = os.path.join(os.getcwd(), model_parent_dir_name)
    # Check if the directory exists
    if not os.path.exists(model_parent_dir):
        # If it doesn't exist, create it
        os.makedirs(model_parent_dir)

    # Check if model is already downloaded
    model_name = os.getenv(model_env_var, default_embedding_model)
    model_dir = os.path.join(model_parent_dir, model_name)
    print(f"Model directory: {model_dir}")
    if os.path.exists(model_dir):
        print(f"Model {model_name} already downloaded")
        return

    # Initialize and download the model
    print(f"Downloading {model_name}...")
    from sentence_transformers import SentenceTransformer

    model = SentenceTransformer(model_name)
    model.save(model_dir)


if __name__ == "__main__":
    main()

sentence_transformer might not work for all models, check on the model card on the hugging face to see which package can support downloading it.

Azure function app code development

I chose vscode + azure function extension for my development. You can first create function and trigger template through extension, add your logic in your function_app.py

The azure function app is a sub project inside my project which leverages the function app service, so you can see I have a nested folder for organizing the structure. The key idea is to put your function_app.py, and host.json under your working space for your azure function app.

import json
import logging
import os

import azure.functions as func
from sentence_transformers import SentenceTransformer

app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)

@app.function_name(name="GetTextEmbedding")
@app.route(route="embed")
def GetTextEmbedding(req: func.HttpRequest) -> func.HttpResponse:
    logging.info("Python HTTP trigger function processed a request.")

    input_text = []
    try:
        req_body = req.get_json()
        if "text" in req_body:
            input_text.append(req_body["text"])
        elif "values" in req_body:
            for value in req_body["values"]:
                input_text.append(value["data"]["text"])
    except Exception as e:
        logging.exception(e, "Could not get text from request body")

    if len(input_text) == 0:
        return func.HttpResponse("No input text found", status_code=400)

    model = os.getenv(
        "SENTENCE_TRANSFORMERS_TEXT_EMBEDDING_MODEL",
        "mixedbread-ai/mxbai-embed-large-v1",
    )
    
    model_path = os.path.join(os.getcwd(), "models", model)
    #Load the model when the function app is initialized
    model = SentenceTransformer(model_path)

    embeddings = model.encode(input_text)
    response = {
        "values": [
            {"recordId": i, "data": {"vector": embedding.tolist()}}
            for i, embedding in enumerate(embeddings)
        ]
    }

    return func.HttpResponse(
        body=json.dumps(response), mimetype="application/json", status_code=200
    )

The code deployment is straightforward, in the generated template, you can add your logic in the route.

Deployment

Different service plans have different deployment limits — for example, consumption plan for linux only support run from package with external package URL

You can utilize the AzureFunctionApp task within your pipeline to update your function app, requiring a zip file as the input package. This necessitates zipping the working directory of your content in the previous step.

When it comes to deployment methods, two options typically meet the needs:

1. Zip Deployment + Remote Build
2. Zip Deployment + Run From Package

You can employ the AzureFunctionApp@2 task and specify your preferred deployment method in the configuration.

For “Remote Build” to work on Linux, you need to set the following application settings:

• ENABLE_ORYX_BUILD=true
• SCM_DO_BUILD_DURING_DEPLOYMENT=true

Remote Build

To enable remote build on Linux, you must set these application settings:

• ENABLE_ORYX_BUILD=true
• SCM_DO_BUILD_DURING_DEPLOYMENT=true

It’s important to note that remote build isn’t executed when an app is using “run-from-package.”

For my case, I don’t need to run things during the build phase, so I chose run_from_package with local reference to the deployed zip package.

Yml file for azure pipelines

Pay attention to the working directory and the file path of your script, the working directory on your local might differ from the one on the agent.

trigger:
  branches:
    # Cannot have variables in triggers
    include:
      - main
  paths:
    include:
      - src/azure_functions/*
variables:
  # Azure service connection established during pipeline creation
  azureSubscription: '----'
  appName: '---'
  # Agent VM image name
  vmImageName: 'ubuntu-latest'
  
jobs:
- job: BuildAndDeploy
  displayName: 'Build and Deploy Azure Function'
  pool:
    vmImage: $(vmImageName)

  steps:
  - task: UsePythonVersion@0
    displayName: "Set Python version to 3.10"
    inputs:
      versionSpec: '3.10'
      architecture: 'x64'

  - task: Bash@3
    displayName: 'Install pip requirements'
    inputs:
      targetType: 'inline'
      script: |
        python -m pip install --upgrade pip
        pip install --target="./.python_packages/lib/site-packages" -r ./requirements.txt
      workingDirectory: '$(Build.SourcesDirectory)/src/azure_functions/embedding_function'
      
  - task: ShellScript@2
    displayName: 'Download Model'
    inputs:
      scriptPath: './src/azure_functions/embedding_function/scripts/download_embedding_model.sh'
      disableAutoCwd: true
      cwd: '$(Build.SourcesDirectory)/src/azure_functions/embedding_function'

  # Zip function contents
  - task: ArchiveFiles@2
    displayName: Zip function contents
    inputs:
      rootFolderOrFile: '$(Build.SourcesDirectory)/src/azure_functions/embedding_function'
      includeRootFolder: false
      archiveType: 'zip'
      archiveFile: '$(Build.ArtifactStagingDirectory)/$(Build.BuildId).zip'
      replaceExistingArchive: true

  - task: AzureFunctionApp@2 # Add this at the end of your file
    inputs:
      azureSubscription: $(azureSubscription)
      appType: functionAppLinux # default is functionApp
      appName: $(appName)
      package: '$(Build.ArtifactStagingDirectory)/$(Build.BuildId).zip'
      deploymentMethod: runFromPackage

• Build.SourcesDirectory: Represents the local path on the agent where your source code files are downloaded.
• Build.ArtifactStagingDirectory: Denotes the local path on the agent where any artifacts are copied to before being pushed to their destination.

Ensure Python packages are installed in the target environment according to the guidelines provided in the official documentation. Exercise caution when considering the use of virtual environments, as they may not behave as expected.

#!/bin/bash
# Custom Python package directory
python_packages_dir="./.python_packages/lib/site-packages"

# Add the custom directory to the Python path
export PYTHONPATH="$PYTHONPATH:$python_packages_dir"
# Run the Python script
python_script="./scripts/download_embedding_models.py"  # Replace with the path to your Python script
if [ -f "$python_script" ]; then
    echo "Running Python script: $python_script"
    python "$python_script"
else
    echo "Error: Python script not found at $python_script"
    exit 1
fi

For the shell script, remember to add the python dependencies path to python sys path. So that your dependencies can be referenced.

Downloading model can cost a huge disk space, and that is limited on the agent, which is 10GB. So adjust your model size in case you met the space not enough error

import os

model_parent_dir_name = "models"
default_embedding_model = "mixedbread-ai/mxbai-embed-large-v1"
model_env_var = "SENTENCE_TRANSFORMERS_TEXT_EMBEDDING_MODEL"


def main():
    model_parent_dir = os.path.join(os.getcwd(), model_parent_dir_name)
    # Check if the directory exists
    if not os.path.exists(model_parent_dir):
        # If it doesn't exist, create it
        os.makedirs(model_parent_dir)

    # Check if model is already downloaded
    model_name = os.getenv(model_env_var, default_embedding_model)
    model_dir = os.path.join(model_parent_dir, model_name)
    print(f"Model directory: {model_dir}")
    if os.path.exists(model_dir):
        print(f"Model {model_name} already downloaded")
        return

    # Initialize and download the model
    print(f"Downloading {model_name}...")
    from sentence_transformers import SentenceTransformer

    model = SentenceTransformer(model_name)
    model.save(model_dir)


if __name__ == "__main__":
    main()

Your Python script should now be ready to utilize the dependencies. As a final step, use zip deployment to deploy your Azure Function. Remember to archive your agent’s working directory, storing it in the staging directory. Then, reference the zip file in the publish task.

Configure app settings on azure function app portal

Azure pipeline with AzureFunctionApp@2 task will overwrite the app settings depending on your deployment method, so you don’t need to configure it yourself on the portal. But it is good to check on the settings.

Debug on the portal and make fixes

Diagnosing errors can be challenging without access to error logs. I’ve found the “Function App Down” or “Reporting” section in the Diagnose section to be extremely helpful when deploying an app. This section provides a summarized view of errors or exceptions encountered during deployment.

Additionally, consider upgrading to a premium plan to enable Application Insights. This allows you to access detailed traces and logs, providing valuable insights into any issues that arise.

That’s all I got, feel free to post your thoughts and your way of doing it in the comments, I’d love to hear.

Some Traps:

1. Don’t name your function app too long, that will fail the deployment and gives you no warning!
2. Deploying through portal interface is error-prone, try deploying it using pipeline with Azure CLI where you explicitly specify your “deploymentMethod: runFromPackage” for python package, as it doesn’t need build. This is helpful because some practices are automatically enforced like the example below:

3. remember to add the target pip install — target=”./.python_packages/lib/site-packages” -r ./requirements.txt

4. Function app uses storage accounts for hosting files and runtime triggers. When switching to identity-based connections, remember that only dedicated plans and not consumption plan that is dynamically scaling can support the identity-based connections. Because share file will be enabled for dynamically scaling.

First of all, let’s see what happen if we throwing an error

this will cause the thread stop working, we can see the console never gets executed.

Now if we add a try catch, we can see the thread will keeping working and let you handle the error in the catch block.

And even if it is the nested function throwing the error, the catch can handle the error at the outer function level, the thread can still work.

Okay, now let’s see another style of error handling (The Golang way)

So here we handle error by returning it instead of throwing. You can see you have more granularities of error message here, and it is safer, it won’t accidentally stop the whole thread if you forget to try catch, and it is more readable. But we also have a drawback, if we want to add a new error in the very inner function call like inner10(), then we gonna return the error level by level, that’s error-prone and unpleasant to do so.

Better solutions? Yes, we could see that these two type of errors handling are not conflicting each other at all, we can choose to the second style if we want more granularity control of messages and we can choose to the first style if we just want to catch the error inside multiple depth of functions.

This example shows, I really don’t care the error details beyond inner2, if any layer throws, just catch it in the inner1, and in the main(), we can clearly test result.error to see if any error returned.

Overall, the two ways of error handling really provide flexibilities and conveniences on how you want to handling your errors. I am not saying the best way to do error handling, because there is no such case. Find out what works in your case and what kind of error you want to log or recover from.

Welcome any thoughts on this, always appreciate better ideas :)

After some googling, I really don’t see the very detailed tutorial on how to easily test, that’s the exact reason why I am writing this story.

How the sqs service actually works? It is fairly simple, producer sends the messages to the aws queue and consumer (either short polling or long polling) fetch it back to do the processing. Only after the message gets successfully processed, did the message get destroyed in the queue.

First, you need four actions in Postman

Let me explain why we need these four, for the first two, as the name suggests, it is the core actions to send and fetch messages in the queue. For the third one, you can get metadata on the queue for troubleshooting, and last one is somewhat important, because as I mentioned earlier, Only after the message gets successfully processed, did the message get destroyed in the queue. So this method is to actually purge all the messages in the queue, in case your message cannot get processed in the consumer, and your consumer repeatedly gets the same message.

Here I only take sendMessage as an example, and other actions just follow the pattern, and you could check the aws sqs docs to see how to play with those.

First, set up the aws authorization:

Second, filling up the parameters:

Things in the MessageBody are the test data you want to define, you could define it in the pre-request script. MessageAttribute is the metadata for your consumer to perform different tasks.

Examine carefully with the params’ keys, since HMAC hash examine the request data integrity. More info : https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-api-request-authentication.html

Last, it is time to start your consumer, and see the result from there, and you could debug it with the built in vscode debugger, remember to attach the process pid. Happy debugging!