# Асинхронное тестирование { #async-tests }

Вы уже видели как тестировать **FastAPI** приложение, используя имеющийся класс `TestClient`. К этому моменту вы видели только как писать тесты в синхронном стиле без использования `async` функций.

您已经了解了如何使用 `TestClient` 测试 **FastAPI** 应用程序。但是到目前为止，您只了解了如何编写同步测试，而没有使用 `async` 异步函数。

在测试中能够使用异步函数可能会很有用，比如当您需要异步查询数据库的时候。想象一下，您想要测试向 FastAPI 应用程序发送请求，然后验证您的后端是否成功在数据库中写入了正确的数据，与此同时您使用了异步的数据库的库。

让我们看看如何才能实现这一点。

## pytest.mark.anyio

如果我们想在测试中调用异步函数，那么我们的测试函数必须是异步的。 AnyIO 为此提供了一个简洁的插件，它允许我们指定一些测试函数要异步调用。

## HTTPX

即使您的 **FastAPI** 应用程序使用普通的 `def` 函数而不是 `async def` ，它本质上仍是一个 `async` 异步应用程序。

// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <http://www.gnu.org/licenses/>.

package dsync

import (
	"bytes"
	"context"
	"errors"
	"net/http"
	"net/url"
	"time"

	xhttp "github.com/minio/minio/internal/http"
	"github.com/minio/minio/internal/rest"
)

}

func (g *Granted) isLocked() bool {
	return isLocked(g.lockUID)
}

func isLocked(uid string) bool {
	return len(uid) > 0
}

// NewDRWMutex - initializes a new dsync RW mutex.
func NewDRWMutex(clnt *Dsync, names ...string) *DRWMutex {
	restClnts, _ := clnt.GetLockers()
	sort.Strings(names)
	return &DRWMutex{
		writeLocks:           make([]string, len(restClnts)),

// volume, path and operation ID.
func (n *nsLockMap) NewNSLock(lockers func() ([]dsync.NetLocker, string), volume string, paths ...string) RWLocker {
	sort.Strings(paths)
	opsID := mustGetUUID()
	if n.isDistErasure {
		drwmutex := dsync.NewDRWMutex(&dsync.Dsync{
			GetLockers: lockers,
			Timeouts:   dsync.DefaultTimeouts,
		}, pathsJoinPrefix(volume, paths...)...)
		return &distLockInstance{drwmutex, opsID}

replicate objects one at a time so as not to impede server load. Ideally, resync should not be initiated for multiple buckets simultaneously - progress of the syncing can be monitored by looking at `mc replicate resync status alias/bucket --remote-bucket <arn>`. In the event that resync operation failed to replicate some versions, they would be picked up by the healing mechanism in-built as part of the scanner. If the resync operation reports a failed status or in the event of a cluster restart while...

@app.get("/async/dict-no-response-model")
async def async_dict_no_response_model():
    return {"name": "foo", "value": 123}


@app.get("/async/dict-with-response-model", response_model=ItemOut)
async def async_dict_with_response_model(
    dep: Annotated[int, Depends(dep_b)],
):
    return {"name": "foo", "value": 123, "dep": dep}


@app.get("/async/model-no-response-model")
async def async_model_no_response_model(

			// Reset the timer for next interval
			monitor.Reset(monitorInterval)
		}
	}
}

func (s *erasureSets) GetLockers(setIndex int) func() ([]dsync.NetLocker, string) {
	return func() ([]dsync.NetLocker, string) {
		lockers := make([]dsync.NetLocker, len(s.erasureLockers[setIndex]))
		copy(lockers, s.erasureLockers[setIndex])
		return lockers, s.erasureLockOwner
	}
}

replicated objects to the DR cluster need to be re-replicated, `mc replicate resync start alias/bucket --remote-bucket <arn>` can be used to initiate a reset. This would initiate a re-sync between the two clusters by walking the bucket namespace and replicating eligible objects that satisfy the existing objects replication rule specified in the replication config. The status of the resync operation can be viewed with `mc replicate resync status alias/bucket --remote-bucket <arn>`.

Note that Ex...

const (
	siteResyncPrefix = bucketMetaPrefix + "/site-replication/resync"
)

type resyncState struct {
	resyncID  string
	LastSaved time.Time
}

//msgp:ignore siteResyncMetrics
type siteResyncMetrics struct {
	sync.RWMutex
	// resyncStatus maps resync ID to resync status for peer
	resyncStatus map[string]SiteResyncStatus
	// map peer deployment ID to resync ID
	peerResyncMap map[string]resyncState
}