⚡️ Speed up method BitwardenService.unlock by 7%

[codeflash-ai]

📄 7% (0.07x) speedup for BitwardenService.unlock in skyvern/forge/sdk/services/bitwarden.py

⏱️ Runtime : 1.73 milliseconds → 1.61 milliseconds (best of 134 runs)

📝 Explanation and details

The optimization replaces the line-by-line parsing approach in _extract_session_key with direct string searching, achieving a 7% runtime improvement and 0.8% throughput increase.

Key optimization: Instead of splitting the output into lines and iterating through them, the optimized version uses two direct find() operations to locate the BW_SESSION=" marker and the closing quote. This eliminates the overhead of:

• Creating a list of strings via split("\n") (16.8% of original function time)
• Iterating through lines (19.5% of original function time)
• Multiple string operations per line

Performance impact: The line profiler shows the optimized version reduces total function time from 1.14ms to 1.08ms. The _extract_session_key function is called within the critical unlock method, which consumes 21% of the total unlock time. Since unlock is called in hot paths for Bitwarden authentication (as shown in the function references where it's used for retrieving secrets, identity information, and credit card data), this optimization compounds across multiple authentication flows.

Test case suitability: The optimization performs consistently well across all test scenarios - basic success cases, edge cases with varied output formats, and high-throughput concurrent scenarios (up to 200 concurrent operations). The direct string searching approach scales better with larger outputs since it avoids creating intermediate collections, making it particularly beneficial for scenarios with verbose Bitwarden CLI output.

The optimization maintains identical functionality while reducing memory allocations and CPU cycles, making authentication operations more efficient across all Bitwarden service workflows.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import asyncio  # used to run async functions
import os
# Mocks and helpers
from types import SimpleNamespace

import pytest  # used for our unit tests
from skyvern.forge.sdk.services.bitwarden import BitwardenService

# Exception class as per function
class BitwardenUnlockError(Exception):
    pass

# Dummy RunCommandResult to match expected output
class RunCommandResult:
    def __init__(self, stdout, stderr, returncode):
        self.stdout = stdout
        self.stderr = stderr
        self.returncode = returncode

# Helper to patch run_command with desired behavior
def set_run_command(monkeypatch, result: RunCommandResult, delay: float = 0):
    async def fake_run_command(command, additional_env=None, timeout=60):
        if delay > 0:
            await asyncio.sleep(delay)
        return result
    monkeypatch.setattr(BitwardenService, "run_command", fake_run_command)

# Helper to patch run_command to raise exception
def set_run_command_exception(monkeypatch, exc):
    async def fake_run_command(command, additional_env=None, timeout=60):
        raise exc
    monkeypatch.setattr(BitwardenService, "run_command", fake_run_command)

# 1. Basic Test Cases

@pytest.mark.asyncio
async def test_unlock_success(monkeypatch):
    """
    Basic: Should unlock and return session key when stdout contains success message and BW_SESSION.
    """
    session = "abc123"
    stdout = f'BW_SESSION="{session}"\nYour vault is now unlocked!'
    stderr = ""
    rc = 0
    set_run_command(monkeypatch, RunCommandResult(stdout, stderr, rc))
    result = await BitwardenService.unlock("correct_password")

@pytest.mark.asyncio
async def test_unlock_success_with_extra_lines(monkeypatch):
    """
    Basic: Should ignore extra lines and extract correct session key.
    """
    session = "xyz987"
    stdout = f"Some info\nBW_SESSION=\"{session}\"\nYour vault is now unlocked!\nOther stuff"
    stderr = ""
    rc = 0
    set_run_command(monkeypatch, RunCommandResult(stdout, stderr, rc))
    result = await BitwardenService.unlock("another_password")

@pytest.mark.asyncio
async def test_unlock_success_with_spaces(monkeypatch):
    """
    Basic: Should extract session key even if line has spaces.
    """
    session = "keywithspaces"
    stdout = f'  BW_SESSION="{session}"  \nYour vault is now unlocked!'
    stderr = ""
    rc = 0
    set_run_command(monkeypatch, RunCommandResult(stdout, stderr, rc))
    result = await BitwardenService.unlock("pw")

# 2. Edge Test Cases

@pytest.mark.asyncio

async def test_unlock_run_command_raises(monkeypatch):
    """
    Edge: Should propagate exception from run_command.
    """
    exc = RuntimeError("run_command failed")
    set_run_command_exception(monkeypatch, exc)
    with pytest.raises(RuntimeError) as e:
        await BitwardenService.unlock("pw")

@pytest.mark.asyncio

async def test_unlock_concurrent(monkeypatch):
    """
    Edge: Test concurrent unlocks with different passwords.
    """
    # Each unlock returns a different session key
    async def fake_run_command(command, additional_env=None, timeout=60):
        pw = additional_env["BW_PASSWORD"]
        return RunCommandResult(
            f'BW_SESSION="{pw}_session"\nYour vault is now unlocked!',
            "",
            0
        )
    monkeypatch.setattr(BitwardenService, "run_command", fake_run_command)
    passwords = ["pw1", "pw2", "pw3", "pw4"]
    results = await asyncio.gather(*(BitwardenService.unlock(pw) for pw in passwords))
    for i, session in enumerate(results):
        pass

# 3. Large Scale Test Cases

@pytest.mark.asyncio
async def test_unlock_many_concurrent(monkeypatch):
    """
    Large Scale: Test 100 concurrent unlocks.
    """
    async def fake_run_command(command, additional_env=None, timeout=60):
        pw = additional_env["BW_PASSWORD"]
        return RunCommandResult(
            f'BW_SESSION="{pw}_session"\nYour vault is now unlocked!',
            "",
            0
        )
    monkeypatch.setattr(BitwardenService, "run_command", fake_run_command)
    passwords = [f"pw{i}" for i in range(100)]
    results = await asyncio.gather(*(BitwardenService.unlock(pw) for pw in passwords))
    for i, session in enumerate(results):
        pass

@pytest.mark.asyncio
async def test_unlock_varied_outputs(monkeypatch):
    """
    Large Scale: Test unlock with varied stdout formats.
    """
    outputs = [
        'BW_SESSION="sessionA"\nYour vault is now unlocked!',
        'Some info\nBW_SESSION="sessionB"\nYour vault is now unlocked!',
        'BW_SESSION="sessionC"\nYour vault is now unlocked!\nExtra',
        'BW_SESSION="sessionD"\nYour vault is now unlocked!',
    ]
    async def fake_run_command(command, additional_env=None, timeout=60):
        idx = int(additional_env["BW_PASSWORD"].replace("pw", ""))
        return RunCommandResult(outputs[idx], "", 0)
    monkeypatch.setattr(BitwardenService, "run_command", fake_run_command)
    passwords = [f"pw{i}" for i in range(4)]
    expected = ["sessionA", "sessionB", "sessionC", "sessionD"]
    results = await asyncio.gather(*(BitwardenService.unlock(pw) for pw in passwords))

# 4. Throughput Test Cases

@pytest.mark.asyncio
async def test_unlock_throughput_small_load(monkeypatch):
    """
    Throughput: Test unlock under small load (10 concurrent).
    """
    async def fake_run_command(command, additional_env=None, timeout=60):
        pw = additional_env["BW_PASSWORD"]
        return RunCommandResult(
            f'BW_SESSION="{pw}_session"\nYour vault is now unlocked!',
            "",
            0
        )
    monkeypatch.setattr(BitwardenService, "run_command", fake_run_command)
    passwords = [f"pw{i}" for i in range(10)]
    results = await asyncio.gather(*(BitwardenService.unlock(pw) for pw in passwords))

@pytest.mark.asyncio
async def test_unlock_throughput_medium_load(monkeypatch):
    """
    Throughput: Test unlock under medium load (50 concurrent).
    """
    async def fake_run_command(command, additional_env=None, timeout=60):
        pw = additional_env["BW_PASSWORD"]
        return RunCommandResult(
            f'BW_SESSION="{pw}_session"\nYour vault is now unlocked!',
            "",
            0
        )
    monkeypatch.setattr(BitwardenService, "run_command", fake_run_command)
    passwords = [f"pw{i}" for i in range(50)]
    results = await asyncio.gather(*(BitwardenService.unlock(pw) for pw in passwords))

@pytest.mark.asyncio
async def test_unlock_throughput_high_volume(monkeypatch):
    """
    Throughput: Test unlock under high volume (200 concurrent).
    """
    async def fake_run_command(command, additional_env=None, timeout=60):
        pw = additional_env["BW_PASSWORD"]
        return RunCommandResult(
            f'BW_SESSION="{pw}_session"\nYour vault is now unlocked!',
            "",
            0
        )
    monkeypatch.setattr(BitwardenService, "run_command", fake_run_command)
    passwords = [f"pw{i}" for i in range(200)]
    results = await asyncio.gather(*(BitwardenService.unlock(pw) for pw in passwords))
# 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 os
from unittest.mock import AsyncMock, patch

import pytest  # used for our unit tests
from skyvern.exceptions import BitwardenUnlockError
from skyvern.forge.sdk.services.bitwarden import BitwardenService

class RunCommandResult:
    """Minimal stub for testing purposes."""
    def __init__(self, stdout, stderr, returncode):
        self.stdout = stdout
        self.stderr = stderr
        self.returncode = returncode

# --- TESTS BEGIN HERE ---

# Helper: mock run_command to return a successful unlock result
def mock_run_command_success(*args, **kwargs):
    stdout = (
        "Your vault is now unlocked!\n"
        'export BW_SESSION="test-session-key"\n'
    )
    stderr = ""
    return RunCommandResult(stdout=stdout, stderr=stderr, returncode=0)

# Helper: mock run_command to return a failed unlock result (wrong password)
def mock_run_command_fail(*args, **kwargs):
    stdout = "Invalid master password.\n"
    stderr = ""
    return RunCommandResult(stdout=stdout, stderr=stderr, returncode=1)

# Helper: mock run_command to return unlock result with missing session key
def mock_run_command_no_session(*args, **kwargs):
    stdout = "Your vault is now unlocked!\n"
    stderr = ""
    return RunCommandResult(stdout=stdout, stderr=stderr, returncode=0)

# Helper: mock run_command to return unlock result with malformed session key
def mock_run_command_malformed_session(*args, **kwargs):
    stdout = "Your vault is now unlocked!\nexport BW_SESSION=\n"
    stderr = ""
    return RunCommandResult(stdout=stdout, stderr=stderr, returncode=0)

# Helper: mock run_command to raise TimeoutError
async def mock_run_command_timeout(*args, **kwargs):
    raise asyncio.TimeoutError("Timed out")

# Helper: mock run_command to simulate large output
def mock_run_command_large_output(*args, **kwargs):
    session_key = "A" * 512  # Large session key
    stdout = (
        "Your vault is now unlocked!\n"
        f'export BW_SESSION="{session_key}"\n'
    )
    stderr = ""
    return RunCommandResult(stdout=stdout, stderr=stderr, returncode=0)

# 1. BASIC TEST CASES

@pytest.mark.asyncio
async def test_unlock_successful():
    """Test that unlock returns the correct session key when successful."""
    with patch.object(BitwardenService, "run_command", new=AsyncMock(side_effect=mock_run_command_success)):
        session_key = await BitwardenService.unlock("correct-password")

@pytest.mark.asyncio
async def test_unlock_basic_async_behavior():
    """Test that unlock is awaitable and returns a string session key."""
    with patch.object(BitwardenService, "run_command", new=AsyncMock(side_effect=mock_run_command_success)):
        result = await BitwardenService.unlock("any-password")

# 2. EDGE TEST CASES

@pytest.mark.asyncio
async def test_unlock_wrong_password_raises():
    """Test that unlock raises BitwardenUnlockError on wrong password."""
    with patch.object(BitwardenService, "run_command", new=AsyncMock(side_effect=mock_run_command_fail)):
        with pytest.raises(BitwardenUnlockError) as excinfo:
            await BitwardenService.unlock("wrong-password")

@pytest.mark.asyncio
async def test_unlock_missing_session_key_raises():
    """Test that unlock raises BitwardenUnlockError when session key is missing."""
    with patch.object(BitwardenService, "run_command", new=AsyncMock(side_effect=mock_run_command_no_session)):
        with pytest.raises(BitwardenUnlockError) as excinfo:
            await BitwardenService.unlock("password")

@pytest.mark.asyncio
async def test_unlock_malformed_session_key_raises():
    """Test that unlock raises BitwardenUnlockError when session key is malformed."""
    with patch.object(BitwardenService, "run_command", new=AsyncMock(side_effect=mock_run_command_malformed_session)):
        with pytest.raises(BitwardenUnlockError) as excinfo:
            await BitwardenService.unlock("password")

@pytest.mark.asyncio
async def test_unlock_run_command_timeout_raises():
    """Test that unlock propagates TimeoutError from run_command."""
    with patch.object(BitwardenService, "run_command", new=AsyncMock(side_effect=mock_run_command_timeout)):
        with pytest.raises(asyncio.TimeoutError):
            await BitwardenService.unlock("password")

@pytest.mark.asyncio
async def test_unlock_concurrent_success():
    """Test concurrent unlock calls all succeed and return correct session keys."""
    with patch.object(BitwardenService, "run_command", new=AsyncMock(side_effect=mock_run_command_success)):
        # Run 5 unlocks concurrently
        tasks = [BitwardenService.unlock(f"password-{i}") for i in range(5)]
        results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_unlock_concurrent_mixed_results():
    """Test concurrent unlock calls with mixed success and failure."""
    # First two succeed, next three fail
    side_effects = [
        mock_run_command_success, mock_run_command_success,
        mock_run_command_fail, mock_run_command_no_session, mock_run_command_malformed_session
    ]
    async def side_effect(*args, **kwargs):
        fn = side_effects.pop(0)
        if asyncio.iscoroutinefunction(fn):
            return await fn(*args, **kwargs)
        else:
            return fn(*args, **kwargs)
    with patch.object(BitwardenService, "run_command", new=AsyncMock(side_effect=side_effect)):
        tasks = [BitwardenService.unlock(f"pw-{i}") for i in range(5)]
        # Use asyncio.gather with return_exceptions=True to collect all results/exceptions
        results = await asyncio.gather(*tasks, return_exceptions=True)

# 3. LARGE SCALE TEST CASES

@pytest.mark.asyncio
async def test_unlock_large_session_key():
    """Test unlock with a very large session key."""
    with patch.object(BitwardenService, "run_command", new=AsyncMock(side_effect=mock_run_command_large_output)):
        session_key = await BitwardenService.unlock("large-key-password")

@pytest.mark.asyncio
async def test_unlock_many_concurrent():
    """Test unlock with many concurrent calls (50)."""
    with patch.object(BitwardenService, "run_command", new=AsyncMock(side_effect=mock_run_command_success)):
        tasks = [BitwardenService.unlock(f"pw-{i}") for i in range(50)]
        results = await asyncio.gather(*tasks)

# 4. THROUGHPUT TEST CASES

@pytest.mark.asyncio
async def test_unlock_throughput_small_load():
    """Throughput test: 5 concurrent unlocks, all succeed."""
    with patch.object(BitwardenService, "run_command", new=AsyncMock(side_effect=mock_run_command_success)):
        tasks = [BitwardenService.unlock(f"small-{i}") for i in range(5)]
        results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_unlock_throughput_medium_load():
    """Throughput test: 20 concurrent unlocks, all succeed."""
    with patch.object(BitwardenService, "run_command", new=AsyncMock(side_effect=mock_run_command_success)):
        tasks = [BitwardenService.unlock(f"medium-{i}") for i in range(20)]
        results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_unlock_throughput_high_volume():
    """Throughput test: 100 concurrent unlocks, all succeed."""
    with patch.object(BitwardenService, "run_command", new=AsyncMock(side_effect=mock_run_command_success)):
        tasks = [BitwardenService.unlock(f"high-{i}") for i in range(100)]
        results = await asyncio.gather(*tasks)

@pytest.mark.asyncio
async def test_unlock_throughput_mixed_load():
    """Throughput test: 10 unlocks, half succeed, half fail."""
    # First 5 succeed, next 5 fail
    side_effects = [mock_run_command_success] * 5 + [mock_run_command_fail] * 5
    async def side_effect(*args, **kwargs):
        fn = side_effects.pop(0)
        if asyncio.iscoroutinefunction(fn):
            return await fn(*args, **kwargs)
        else:
            return fn(*args, **kwargs)
    with patch.object(BitwardenService, "run_command", new=AsyncMock(side_effect=side_effect)):
        tasks = [BitwardenService.unlock(f"mixed-{i}") for i in range(10)]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        for i in range(5):
            pass
        for i in range(5, 10):
            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-BitwardenService.unlock-mirhlpj6 and push.