⚡️ Speed up method UsersGroupsDataSource.users_user_presence_clear_presence by 9%

[codeflash-ai]

📄 9% (0.09x) speedup for UsersGroupsDataSource.users_user_presence_clear_presence in backend/python/app/sources/external/microsoft/users_groups/users_groups.py

⏱️ Runtime : 1.11 milliseconds → 1.02 milliseconds (best of 125 runs)

📝 Explanation and details

The optimized code achieves an 8% runtime improvement by eliminating object allocation overhead in the users_user_presence_clear_presence method.

Key optimizations:

1. Reduced object instantiation: Instead of creating two separate RequestConfiguration() instances (one for query_params and one for config), the optimized version creates only one instance and reuses it via query_params = config.

2. Streamlined header handling: The original code unconditionally created a headers dictionary when search was present. The optimized version uses a more efficient pattern - it checks if headers already exist before creating a new dictionary, reducing unnecessary allocations.

3. Direct attribute assignment: Parameters are assigned directly to the single config object instead of being assigned to a separate query_params object and then copied over.

Why this matters for performance:

• Object creation in Python involves memory allocation and initialization overhead
• The line profiler shows the original RequestConfiguration() call took 561,013 nanoseconds vs 600,856 in the optimized version, but the overall method execution improved due to eliminating the second instantiation
• Memory pressure is reduced, leading to better cache locality and fewer garbage collection cycles

Test case performance:
The optimization is most effective for high-frequency scenarios like the throughput tests (test_users_user_presence_clear_presence_throughput_*) where the same method is called repeatedly. While throughput remained stable at 71,125 operations/second, the 8% runtime reduction means lower CPU utilization per operation, allowing for better resource efficiency under sustained load.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 647 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	96.4%

🌀 Generated Regression Tests and Runtime

import asyncio  # used to run async functions
# --- Function under test (EXACT COPY) ---
import logging
from typing import Any, Dict, List, Mapping, Optional

import pytest  # used for our unit tests
from app.sources.external.microsoft.users_groups.users_groups import \
    UsersGroupsDataSource

# --- Minimal stubs for dependencies and types ---

class UsersGroupsResponse:
    def __init__(self, success: bool, data: Any = None, error: Optional[str] = None):
        self.success = success
        self.data = data
        self.error = error

class DummyClearPresence:
    def __init__(self, response: Any = None, raise_exc: Exception = None):
        self._response = response
        self._raise_exc = raise_exc
        self.called_with = []

    async def post(self, body=None, request_configuration=None):
        self.called_with.append((body, request_configuration))
        if self._raise_exc:
            raise self._raise_exc
        return self._response

class DummyPresence:
    def __init__(self, clear_presence: DummyClearPresence):
        self.clear_presence = clear_presence

class DummyByUserId:
    def __init__(self, clear_presence: DummyClearPresence):
        self.presence = DummyPresence(clear_presence)

class DummyUsers:
    def __init__(self, clear_presence: DummyClearPresence):
        self._clear_presence = clear_presence

    def by_user_id(self, user_id):
        return DummyByUserId(self._clear_presence)

class DummyClient:
    def __init__(self, clear_presence: DummyClearPresence):
        self.users = DummyUsers(clear_presence)

class DummyMSGraphClient:
    def __init__(self, clear_presence: DummyClearPresence):
        self._client = DummyClient(clear_presence)

    def get_client(self):
        return self

    def get_ms_graph_service_client(self):
        return self._client

# --- TESTS BEGIN HERE ---

# 1. BASIC TEST CASES

@pytest.mark.asyncio
async def test_users_user_presence_clear_presence_basic_success():
    """Test basic successful response with minimal required arguments."""
    dummy_response = object()
    clear_presence = DummyClearPresence(response=dummy_response)
    datasource = UsersGroupsDataSource(DummyMSGraphClient(clear_presence))
    resp = await datasource.users_user_presence_clear_presence("user123")

@pytest.mark.asyncio
async def test_users_user_presence_clear_presence_basic_with_all_args():
    """Test function with all optional parameters set."""
    dummy_response = object()
    clear_presence = DummyClearPresence(response=dummy_response)
    datasource = UsersGroupsDataSource(DummyMSGraphClient(clear_presence))
    resp = await datasource.users_user_presence_clear_presence(
        "user123",
        select=["id", "displayName"],
        expand=["manager"],
        filter="startswith(displayName,'A')",
        orderby="displayName",
        search="Alice",
        top=10,
        skip=2,
        request_body={"foo": "bar"},
        headers={"Authorization": "Bearer token"}
    )

@pytest.mark.asyncio
async def test_users_user_presence_clear_presence_basic_async_behavior():
    """Test that the function is awaitable and returns a coroutine."""
    dummy_response = object()
    clear_presence = DummyClearPresence(response=dummy_response)
    datasource = UsersGroupsDataSource(DummyMSGraphClient(clear_presence))
    codeflash_output = datasource.users_user_presence_clear_presence("user123"); coro = codeflash_output
    resp = await coro

# 2. EDGE TEST CASES

@pytest.mark.asyncio
async def test_users_user_presence_clear_presence_handles_none_response():
    """Test handling when the underlying API returns None."""
    clear_presence = DummyClearPresence(response=None)
    datasource = UsersGroupsDataSource(DummyMSGraphClient(clear_presence))
    resp = await datasource.users_user_presence_clear_presence("user123")

@pytest.mark.asyncio
async def test_users_user_presence_clear_presence_handles_error_attr():
    """Test handling when API response has an 'error' attribute."""
    class ErrorResponse:
        error = "Some API error"
    clear_presence = DummyClearPresence(response=ErrorResponse())
    datasource = UsersGroupsDataSource(DummyMSGraphClient(clear_presence))
    resp = await datasource.users_user_presence_clear_presence("user123")

@pytest.mark.asyncio
async def test_users_user_presence_clear_presence_handles_error_dict():
    """Test handling when API response is a dict with 'error' key (dict value)."""
    error_dict = {"error": {"code": "BadRequest", "message": "Invalid user"}}
    clear_presence = DummyClearPresence(response=error_dict)
    datasource = UsersGroupsDataSource(DummyMSGraphClient(clear_presence))
    resp = await datasource.users_user_presence_clear_presence("user123")

@pytest.mark.asyncio
async def test_users_user_presence_clear_presence_handles_error_dict_str():
    """Test handling when API response is a dict with 'error' key (string value)."""
    error_dict = {"error": "An error occurred"}
    clear_presence = DummyClearPresence(response=error_dict)
    datasource = UsersGroupsDataSource(DummyMSGraphClient(clear_presence))
    resp = await datasource.users_user_presence_clear_presence("user123")

@pytest.mark.asyncio
async def test_users_user_presence_clear_presence_handles_code_message_attr():
    """Test handling when API response has 'code' and 'message' attributes."""
    class CodeMessageResponse:
        code = "401"
        message = "Unauthorized"
    clear_presence = DummyClearPresence(response=CodeMessageResponse())
    datasource = UsersGroupsDataSource(DummyMSGraphClient(clear_presence))
    resp = await datasource.users_user_presence_clear_presence("user123")

@pytest.mark.asyncio
async def test_users_user_presence_clear_presence_handles_exception():
    """Test handling when the underlying API raises an exception."""
    clear_presence = DummyClearPresence(raise_exc=ValueError("Simulated failure"))
    datasource = UsersGroupsDataSource(DummyMSGraphClient(clear_presence))
    resp = await datasource.users_user_presence_clear_presence("user123")

@pytest.mark.asyncio
async def test_users_user_presence_clear_presence_concurrent_calls():
    """Test concurrent execution of multiple calls with different users."""
    dummy_response1 = object()
    dummy_response2 = object()
    clear_presence1 = DummyClearPresence(response=dummy_response1)
    clear_presence2 = DummyClearPresence(response=dummy_response2)
    datasource1 = UsersGroupsDataSource(DummyMSGraphClient(clear_presence1))
    datasource2 = UsersGroupsDataSource(DummyMSGraphClient(clear_presence2))
    results = await asyncio.gather(
        datasource1.users_user_presence_clear_presence("userA"),
        datasource2.users_user_presence_clear_presence("userB")
    )

@pytest.mark.asyncio
async def test_users_user_presence_clear_presence_headers_and_search_consistency_level():
    """Test that ConsistencyLevel header is set when search is used."""
    dummy_response = object()
    clear_presence = DummyClearPresence(response=dummy_response)
    datasource = UsersGroupsDataSource(DummyMSGraphClient(clear_presence))
    await datasource.users_user_presence_clear_presence(
        "user123",
        search="test",
        headers={"Authorization": "Bearer token"}
    )
    # The DummyClearPresence object records calls for inspection
    called_args = clear_presence.called_with[-1]
    body, request_configuration = called_args

@pytest.mark.asyncio

async def test_users_user_presence_clear_presence_many_concurrent_calls():
    """Test function with a large number of concurrent calls (scalability)."""
    N = 50  # keep under 1000 as per guidelines
    responses = [object() for _ in range(N)]
    clear_presences = [DummyClearPresence(response=resp) for resp in responses]
    datasources = [UsersGroupsDataSource(DummyMSGraphClient(cp)) for cp in clear_presences]
    coros = [
        ds.users_user_presence_clear_presence(f"user{i}", top=i, skip=i)
        for i, ds in enumerate(datasources)
    ]
    results = await asyncio.gather(*coros)
    for i, result in enumerate(results):
        pass

# 4. THROUGHPUT TEST CASES

@pytest.mark.asyncio
async def test_users_user_presence_clear_presence_throughput_small_load():
    """Throughput test: small batch of requests."""
    N = 5
    clear_presence = DummyClearPresence(response={"ok": True})
    datasource = UsersGroupsDataSource(DummyMSGraphClient(clear_presence))
    coros = [
        datasource.users_user_presence_clear_presence(f"user{i}", top=1)
        for i in range(N)
    ]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_users_user_presence_clear_presence_throughput_medium_load():
    """Throughput test: medium batch of requests."""
    N = 25
    clear_presence = DummyClearPresence(response={"ok": True})
    datasource = UsersGroupsDataSource(DummyMSGraphClient(clear_presence))
    coros = [
        datasource.users_user_presence_clear_presence(f"user{i}", top=2)
        for i in range(N)
    ]
    results = await asyncio.gather(*coros)

@pytest.mark.asyncio
async def test_users_user_presence_clear_presence_throughput_high_volume():
    """Throughput test: higher volume of concurrent requests."""
    N = 100  # keep under 1000 as per guidelines
    clear_presence = DummyClearPresence(response={"ok": True})
    datasource = UsersGroupsDataSource(DummyMSGraphClient(clear_presence))
    coros = [
        datasource.users_user_presence_clear_presence(f"user{i}", top=3)
        for i in range(N)
    ]
    results = await asyncio.gather(*coros)
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
import asyncio  # used to run async functions
import logging
from typing import Any, Dict, List, Mapping, Optional

import pytest  # used for our unit tests
from app.sources.external.microsoft.users_groups.users_groups import \
    UsersGroupsDataSource

# --- Minimal stubs for dependencies ---

class UsersGroupsResponse:
    """Simple response wrapper for Users Groups API."""
    def __init__(self, success: bool, data: Any = None, error: Optional[str] = None):
        self.success = success
        self.data = data
        self.error = error

class DummyClearPresence:
    """Stub for clear_presence endpoint."""
    async def post(self, body=None, request_configuration=None):
        # Simulate a successful response with echo of body
        return {'status': 'cleared', 'body': body}

class DummyPresence:
    """Stub for presence property."""
    def __init__(self):
        self.clear_presence = DummyClearPresence()

class DummyUsersByUserId:
    """Stub for users.by_user_id."""
    def __init__(self, user_id):
        self.user_id = user_id
        self.presence = DummyPresence()

class DummyUsers:
    """Stub for users property."""
    def by_user_id(self, user_id):
        return DummyUsersByUserId(user_id)

class DummyMSGraphClient:
    """Stub for the MSGraphClient."""
    def __init__(self):
        self.users = DummyUsers()

class DummyClientWrapper:
    """Stub for .get_client().get_ms_graph_service_client() chain."""
    def get_client(self):
        return self
    def get_ms_graph_service_client(self):
        return DummyMSGraphClient()

# --- Unit tests ---

@pytest.fixture
def datasource():
    """Fixture to provide a UsersGroupsDataSource with dummy client."""
    return UsersGroupsDataSource(DummyClientWrapper())

# 1. Basic Test Cases

@pytest.mark.asyncio
async def test_clear_presence_basic_success(datasource):
    """Test basic async/await behavior and success response."""
    response = await datasource.users_user_presence_clear_presence("user123")

@pytest.mark.asyncio
async def test_clear_presence_with_request_body(datasource):
    """Test passing a request_body argument."""
    body = {"reason": "manual"}
    response = await datasource.users_user_presence_clear_presence("user456", request_body=body)

@pytest.mark.asyncio
async def test_clear_presence_with_select_and_expand(datasource):
    """Test select and expand arguments."""
    select = ["id", "displayName"]
    expand = ["manager"]
    response = await datasource.users_user_presence_clear_presence("user789", select=select, expand=expand)

@pytest.mark.asyncio
async def test_clear_presence_with_headers(datasource):
    """Test custom headers argument."""
    headers = {"Authorization": "Bearer token"}
    response = await datasource.users_user_presence_clear_presence("user999", headers=headers)

# 2. Edge Test Cases

@pytest.mark.asyncio
async def test_clear_presence_with_search_and_consistency(datasource):
    """Test that ConsistencyLevel header is set when search is used."""
    response = await datasource.users_user_presence_clear_presence("user321", search="John Doe")

@pytest.mark.asyncio
async def test_clear_presence_with_top_and_skip(datasource):
    """Test top and skip arguments."""
    response = await datasource.users_user_presence_clear_presence("user654", top=5, skip=2)

@pytest.mark.asyncio
async def test_clear_presence_with_orderby_and_filter(datasource):
    """Test orderby and filter arguments."""
    response = await datasource.users_user_presence_clear_presence("user777", orderby="displayName", filter="accountEnabled eq true")

@pytest.mark.asyncio
async def test_clear_presence_invalid_user_id(datasource):
    """Test with an invalid user_id (simulate error in stub)."""
    # Patch the stub to raise an exception for this user_id
    orig_by_user_id = datasource.client.users.by_user_id
    def error_by_user_id(user_id):
        if user_id == "invalid":
            raise Exception("User not found!")
        return orig_by_user_id(user_id)
    datasource.client.users.by_user_id = error_by_user_id
    response = await datasource.users_user_presence_clear_presence("invalid")
    # Restore original for other tests
    datasource.client.users.by_user_id = orig_by_user_id

@pytest.mark.asyncio
async def test_clear_presence_none_request_body(datasource):
    """Test with None as request_body."""
    response = await datasource.users_user_presence_clear_presence("user222", request_body=None)

@pytest.mark.asyncio
async def test_clear_presence_concurrent_calls(datasource):
    """Test concurrent execution of multiple async calls."""
    user_ids = [f"user{i}" for i in range(5)]
    coros = [datasource.users_user_presence_clear_presence(uid) for uid in user_ids]
    responses = await asyncio.gather(*coros)
    for resp in responses:
        pass

# 3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_clear_presence_many_concurrent_calls(datasource):
    """Test with many concurrent calls (scalability, up to 100)."""
    user_ids = [f"user{i}" for i in range(100)]
    coros = [datasource.users_user_presence_clear_presence(uid, request_body={"reason": "bulk"}) for uid in user_ids]
    responses = await asyncio.gather(*coros)
    for resp in responses:
        pass

@pytest.mark.asyncio
async def test_clear_presence_large_select_expand(datasource):
    """Test with large select/expand lists (up to 50 items)."""
    select = [f"prop{i}" for i in range(50)]
    expand = [f"rel{i}" for i in range(50)]
    response = await datasource.users_user_presence_clear_presence("userlarge", select=select, expand=expand)

# 4. Throughput Test Cases

@pytest.mark.asyncio
async def test_users_user_presence_clear_presence_throughput_small_load(datasource):
    """Throughput test: small load (10 requests)."""
    user_ids = [f"user{i}" for i in range(10)]
    coros = [datasource.users_user_presence_clear_presence(uid) for uid in user_ids]
    responses = await asyncio.gather(*coros)
    for resp in responses:
        pass

@pytest.mark.asyncio
async def test_users_user_presence_clear_presence_throughput_medium_load(datasource):
    """Throughput test: medium load (50 requests)."""
    user_ids = [f"user{i}" for i in range(50)]
    coros = [datasource.users_user_presence_clear_presence(uid, request_body={"source": "medium"}) for uid in user_ids]
    responses = await asyncio.gather(*coros)
    for resp in responses:
        pass

@pytest.mark.asyncio
async def test_users_user_presence_clear_presence_throughput_high_load(datasource):
    """Throughput test: high load (200 requests, bounded for test speed)."""
    user_ids = [f"user{i}" for i in range(200)]
    coros = [datasource.users_user_presence_clear_presence(uid, request_body={"source": "high"}) for uid in user_ids]
    responses = await asyncio.gather(*coros)
    for resp in responses:
        pass
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

To edit these changes git checkout codeflash/optimize-UsersGroupsDataSource.users_user_presence_clear_presence-mh5zd4oz and push.