Function connect

tests/test_async_read_write_cancellation.py:478–508

(
        database: str,
        *,
        factory: type[sqlite3.Connection],
        timeout: float,
    )

    connection_count = 0

    def connect(
        database: str,
        *,
        factory: type[sqlite3.Connection],
        timeout: float,
    ) -> sqlite3.Connection:
        nonlocal connection_count
        real_connection = real_connect(
            database,
            check_same_thread=False,
            factory=factory,
            cached_statements=0,
            timeout=timeout,
        )
        connection_count += 1
        if connection_count != 2:
            return real_connection
        configuration: dict[str, Callable[[str], sqlite3.Cursor] | Callable[[], None]] = {}
        if executescript is not None:
            configuration["executescript.side_effect"] = functools.partial(executescript, real_connection)
        if rollback is not None:
            configuration["rollback.side_effect"] = functools.partial(rollback, real_connection)
        connection = mocker.MagicMock(spec_set=type(real_connection), wraps=real_connection, **configuration)
        mocker.patch.object(
            type(connection),
            "in_transaction",
            new_callable=mocker.PropertyMock,
            create=True,
            side_effect=lambda: real_connection.in_transaction,
        )
        return cast("sqlite3.Connection", connection)

    mocker.patch("filelock._read_write._connect", side_effect=connect)