Repository: redis/redis-py
Pull request #2258 (2022-06-28 10:33:05)

Fix warnings and resource usage problems in asyncio unittests
### Pull Request check-list

_Please make sure to review and check all of these items:_

- [x] Does `$ tox` pass with this change (including linting)?
- [x] Do the CI tests pass with this change (enable it first in your forked repo and wait for the github action build to finish)?
- [x] Is the new or changed code fully tested?
- [ ] Is a documentation update included (if this change modifies existing APIs, or introduces new ones)?
- [ ] Is there an example added to the examples folder (if applicable)?
- [ ] Was the change added to CHANGES file?

### Description of change

`pytest-asyncio` _auto_ mode is recommended unless there is a need to run asyncio tests using different async frameworks.
Using asyncio and removing the overly generous `pytestmark` reduces warning noise in the testsuite.  (Python 3.6 still needs it since the 'auto' mode isn't supported for it)

Perform various cleanup of asyncio connections within the fixture framework of pytest-asyncio, thus avoiding having
to run them via the general pytest non-async cleanup mechanism.

Continued work on making unittests more robust so that we don't get sporadic test failures for unrelateds PRs

codecov-commenter: # [Codecov](https://codecov.io/gh/redis/redis-py/pull/2258?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) Report
> Merging [#2258](https://codecov.io/gh/redis/redis-py/pull/2258?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) (24770e3) into [master](https://codecov.io/gh/redis/redis-py/commit/e6cd4fdf3b159d7f118f154e28e884069da89d7e?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) (e6cd4fd) will **increase** coverage by `0.02%`.
> The diff coverage is `98.70%`.

```diff
@@            Coverage Diff             @@
##           master    #2258      +/-   ##
==========================================
+ Coverage   91.85%   91.88%   +0.02%     
==========================================
  Files         108      108              
  Lines       27783    27786       +3     
==========================================
+ Hits        25521    25532      +11     
+ Misses       2262     2254       -8     
```


| [Impacted Files](https://codecov.io/gh/redis/redis-py/pull/2258?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) | Coverage Δ | |
|---|---|---|
| [tests/test\_asyncio/test\_json.py](https://codecov.io/gh/redis/redis-py/pull/2258/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfanNvbi5weQ==) | `100.00% <ø> (ø)` | |
| [tests/test\_asyncio/conftest.py](https://codecov.io/gh/redis/redis-py/pull/2258/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL2NvbmZ0ZXN0LnB5) | `94.06% <93.54%> (+0.67%)` | :arrow_up: |
| [tests/test\_asyncio/test\_bloom.py](https://codecov.io/gh/redis/redis-py/pull/2258/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfYmxvb20ucHk=) | `99.55% <100.00%> (+3.62%)` | :arrow_up: |
| [tests/test\_asyncio/test\_cluster.py](https://codecov.io/gh/redis/redis-py/pull/2258/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfY2x1c3Rlci5weQ==) | `97.44% <100.00%> (ø)` | |
| [tests/test\_asyncio/test\_commands.py](https://codecov.io/gh/redis/redis-py/pull/2258/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfY29tbWFuZHMucHk=) | `98.39% <100.00%> (+0.32%)` | :arrow_up: |
| [tests/test\_asyncio/test\_connection.py](https://codecov.io/gh/redis/redis-py/pull/2258/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfY29ubmVjdGlvbi5weQ==) | `97.50% <100.00%> (+0.13%)` | :arrow_up: |
| [tests/test\_asyncio/test\_connection\_pool.py](https://codecov.io/gh/redis/redis-py/pull/2258/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfY29ubmVjdGlvbl9wb29sLnB5) | `99.39% <100.00%> (+<0.01%)` | :arrow_up: |
| [tests/test\_asyncio/test\_encoding.py](https://codecov.io/gh/redis/redis-py/pull/2258/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfZW5jb2RpbmcucHk=) | `98.92% <100.00%> (ø)` | |
| [tests/test\_asyncio/test\_lock.py](https://codecov.io/gh/redis/redis-py/pull/2258/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfbG9jay5weQ==) | `100.00% <100.00%> (ø)` | |
| [tests/test\_asyncio/test\_monitor.py](https://codecov.io/gh/redis/redis-py/pull/2258/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfbW9uaXRvci5weQ==) | `91.22% <100.00%> (+0.31%)` | :arrow_up: |
| ... and [12 more](https://codecov.io/gh/redis/redis-py/pull/2258/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/redis/redis-py/pull/2258?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/redis/redis-py/pull/2258?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis). Last update [e6cd4fd...24770e3](https://codecov.io/gh/redis/redis-py/pull/2258?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).

kristjanvalur: Right.  Btw, seems like almost every PR I submit needs to have the unit-tests tweaked to avoid spurious failures :)


Patch:
diff --git a/docs/examples.rst b/docs/examples.rst
index 08526ff6..3fed8b41 100644
--- a/docs/examples.rst
+++ b/docs/examples.rst
@@ -12,4 +12,5 @@ Examples
    examples/set_and_get_examples
    examples/search_vector_similarity_examples
    examples/pipeline_examples
+   examples/timeseries_examples
    examples/redis-stream-example.ipynb
diff --git a/docs/examples/timeseries_examples.ipynb b/docs/examples/timeseries_examples.ipynb
new file mode 100644
index 00000000..fefc0c8f
--- /dev/null
+++ b/docs/examples/timeseries_examples.ipynb
@@ -0,0 +1,631 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Timeseries\n",
+    "\n",
+    "`redis-py` supports [RedisTimeSeries](https://github.com/RedisTimeSeries/RedisTimeSeries/) which is a time-series-database module for Redis.\n",
+    "\n",
+    "This example shows how to handle timeseries data with `redis-py`."
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Health check"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 1,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "True"
+      ]
+     },
+     "execution_count": 1,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "import redis \n",
+    "\n",
+    "r = redis.Redis(decode_responses=True)\n",
+    "ts = r.ts()\n",
+    "\n",
+    "r.ping()"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Simple example"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### Create a timeseries"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 2,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "True"
+      ]
+     },
+     "execution_count": 2,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "ts.create(\"ts_key\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### Add samples to the timeseries\n",
+    "\n",
+    "We can either set the timestamp with an UNIX timestamp in milliseconds or use * to set the timestamp based en server's clock."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 3,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "1657272304448"
+      ]
+     },
+     "execution_count": 3,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "ts.add(\"ts_key\", 1657265437756, 1)\n",
+    "ts.add(\"ts_key\", \"1657265437757\", 2)\n",
+    "ts.add(\"ts_key\", \"*\", 3)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### Get the last sample"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 4,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "(1657272304448, 3.0)"
+      ]
+     },
+     "execution_count": 4,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "ts.get(\"ts_key\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### Get samples between two timestamps\n",
+    "\n",
+    "The minimum and maximum possible timestamps can be expressed with respectfully - and +."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 5,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "[(1657265437756, 1.0), (1657265437757, 2.0), (1657272304448, 3.0)]"
+      ]
+     },
+     "execution_count": 5,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "ts.range(\"ts_key\", \"-\", \"+\")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 6,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "[(1657265437756, 1.0), (1657265437757, 2.0)]"
+      ]
+     },
+     "execution_count": 6,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "ts.range(\"ts_key\", 1657265437756, 1657265437757)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### Delete samples between two timestamps"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 7,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "Before deletion:  [(1657265437756, 1.0), (1657265437757, 2.0), (1657272304448, 3.0)]\n",
+      "After deletion:   [(1657272304448, 3.0)]\n"
+     ]
+    }
+   ],
+   "source": [
+    "print(\"Before deletion: \", ts.range(\"ts_key\", \"-\", \"+\"))\n",
+    "ts.delete(\"ts_key\", 1657265437756, 1657265437757)\n",
+    "print(\"After deletion:  \", ts.range(\"ts_key\", \"-\", \"+\"))"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Multiple timeseries with labels"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 8,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "True"
+      ]
+     },
+     "execution_count": 8,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "ts.create(\"ts_key1\")\n",
+    "ts.create(\"ts_key2\", labels={\"label1\": 1, \"label2\": 2})"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### Add samples to multiple timeseries"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 9,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "[1657272306147, 1657272306147]"
+      ]
+     },
+     "execution_count": 9,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "ts.madd([(\"ts_key1\", \"*\", 1), (\"ts_key2\", \"*\", 2)])"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### Add samples with labels"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 10,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "1657272306457"
+      ]
+     },
+     "execution_count": 10,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "ts.add(\"ts_key2\", \"*\", 2,  labels={\"label1\": 1, \"label2\": 2})\n",
+    "ts.add(\"ts_key2\", \"*\", 2,  labels={\"label1\": 3, \"label2\": 4})"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### Get the last sample matching specific label\n",
+    "\n",
+    "Get the last sample that matches \"label1=1\", see [Redis documentation](https://redis.io/commands/ts.mget/) to see the posible filter values."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 11,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "[{'ts_key2': [{}, 1657272306457, 2.0]}]"
+      ]
+     },
+     "execution_count": 11,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "ts.mget([\"label1=1\"])"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "Get also the label-value pairs of the sample:"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 12,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "[{'ts_key2': [{'label1': '1', 'label2': '2'}, 1657272306457, 2.0]}]"
+      ]
+     },
+     "execution_count": 12,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "ts.mget([\"label1=1\"], with_labels=True)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Retention period\n",
+    "\n",
+    "You can specify a retention period when creating timeseries objects or when adding a sample timeseries object. Once the retention period has elapsed, the sample is removed from the timeseries."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 13,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "True"
+      ]
+     },
+     "execution_count": 13,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "retention_time = 1000\n",
+    "ts.create(\"ts_key_ret\", retention_msecs=retention_time)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 14,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "Base timeseries:                      [(1657272307670, 1.0)]\n",
+      "Timeseries after 1000 milliseconds:   [(1657272307670, 1.0)]\n"
+     ]
+    }
+   ],
+   "source": [
+    "import time\n",
+    "# this will be deleted in 1000 milliseconds\n",
+    "ts.add(\"ts_key_ret\", \"*\", 1, retention_msecs=retention_time)\n",
+    "print(\"Base timeseries:                     \", ts.range(\"ts_key_ret\", \"-\", \"+\"))\n",
+    "# sleeping for 1000 milliseconds (1 second)\n",
+    "time.sleep(1)\n",
+    "print(\"Timeseries after 1000 milliseconds:  \", ts.range(\"ts_key_ret\", \"-\", \"+\"))"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "The two lists are the same, this is because the oldest values are deleted when a new sample is added."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 15,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "1657272308849"
+      ]
+     },
+     "execution_count": 15,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "ts.add(\"ts_key_ret\", \"*\", 10)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 16,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "[(1657272308849, 10.0)]"
+      ]
+     },
+     "execution_count": 16,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "ts.range(\"ts_key_ret\", \"-\", \"+\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "Here the first sample has been deleted."
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Specify duplicate policies\n",
+    "\n",
+    "By default, the policy for duplicates timestamp keys is set to \"BLOCK\", we cannot create two samples with the same timestamp:"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 17,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "TSDB: Error at upsert, update is not supported when DUPLICATE_POLICY is set to BLOCK mode\n"
+     ]
+    }
+   ],
+   "source": [
+    "ts.add(\"ts_key\", 123456789, 1)\n",
+    "try:\n",
+    "    ts.add(\"ts_key\", 123456789, 2)\n",
+    "except Exception as err:\n",
+    "    print(err)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "You can change this default behaviour using `duplicate_policy` parameter, for instance:"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 18,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "[(123456789, 2.0), (1657272304448, 3.0)]"
+      ]
+     },
+     "execution_count": 18,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "# using policy \"LAST\", we keep the last added sample\n",
+    "ts.add(\"ts_key\", 123456789, 2, duplicate_policy=\"LAST\")\n",
+    "ts.range(\"ts_key\", \"-\", \"+\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "For more informations about duplicate policies, see [Redis documentation](https://redis.io/commands/ts.add/)."
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Using Redis TSDB to keep track of a value"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 19,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "1657272310241"
+      ]
+     },
+     "execution_count": 19,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "ts.add(\"ts_key_incr\", \"*\", 0)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "Increment the value:"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 20,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "for _ in range(10):\n",
+    "    ts.incrby(\"ts_key_incr\", 1)\n",
+    "    # sleeping a bit so the timestamp are not duplicates\n",
+    "    time.sleep(0.01)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 21,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "[(1657272310241, 0.0),\n",
+       " (1657272310533, 1.0),\n",
+       " (1657272310545, 2.0),\n",
+       " (1657272310556, 3.0),\n",
+       " (1657272310567, 4.0),\n",
+       " (1657272310578, 5.0),\n",
+       " (1657272310589, 6.0),\n",
+       " (1657272310600, 7.0),\n",
+       " (1657272310611, 8.0),\n",
+       " (1657272310622, 9.0),\n",
+       " (1657272310632, 10.0)]"
+      ]
+     },
+     "execution_count": 21,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "ts.range(\"ts_key_incr\", \"-\", \"+\")"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3.9.2 64-bit",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.9.2"
+  },
+  "orig_nbformat": 4,
+  "vscode": {
+   "interpreter": {
+    "hash": "916dbcbb3f70747c44a77c7bcd40155683ae19c65e1c03b4aa3499c5328201f1"
+   }
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}
diff --git a/tox.ini b/tox.ini
index 0ceb008c..d1aeb02a 100644
--- a/tox.ini
+++ b/tox.ini
@@ -9,6 +9,7 @@ markers =
     asyncio: marker for async tests
     replica: replica tests
     experimental: run only experimental tests
+asyncio_mode = auto
 
 [tox]
 minversion = 3.2.0


Test patch:
diff --git a/tests/test_asyncio/conftest.py b/tests/test_asyncio/conftest.py
index 8166588d..04fbf62c 100644
--- a/tests/test_asyncio/conftest.py
+++ b/tests/test_asyncio/conftest.py
@@ -1,14 +1,9 @@
-import asyncio
+import functools
 import random
 import sys
 from typing import Union
 from urllib.parse import urlparse
 
-if sys.version_info[0:2] == (3, 6):
-    import pytest as pytest_asyncio
-else:
-    import pytest_asyncio
-
 import pytest
 from packaging.version import Version
 
@@ -26,6 +21,13 @@ from tests.conftest import REDIS_INFO
 
 from .compat import mock
 
+if sys.version_info[0:2] == (3, 6):
+    import pytest as pytest_asyncio
+
+    pytestmark = pytest.mark.asyncio
+else:
+    import pytest_asyncio
+
 
 async def _get_info(redis_url):
     client = redis.Redis.from_url(redis_url)
@@ -69,11 +71,13 @@ async def _get_info(redis_url):
         "pool-hiredis",
     ],
 )
-def create_redis(request, event_loop: asyncio.BaseEventLoop):
+async def create_redis(request):
     """Wrapper around redis.create_redis."""
     single_connection, parser_cls = request.param
 
-    async def f(
+    teardown_clients = []
+
+    async def client_factory(
         url: str = request.config.getoption("--redis-url"),
         cls=redis.Redis,
         flushdb=True,
@@ -95,56 +99,50 @@ def create_redis(request, event_loop: asyncio.BaseEventLoop):
             client = client.client()
             await client.initialize()
 
-        def teardown():
-            async def ateardown():
-                if not cluster_mode:
-                    if "username" in kwargs:
-                        return
-                    if flushdb:
-                        try:
-                            await client.flushdb()
-                        except redis.ConnectionError:
-                            # handle cases where a test disconnected a client
-                            # just manually retry the flushdb
-                            await client.flushdb()
-                    await client.close()
-                    await client.connection_pool.disconnect()
-                else:
-                    if flushdb:
-                        try:
-                            await client.flushdb(target_nodes="primaries")
-                        except redis.ConnectionError:
-                            # handle cases where a test disconnected a client
-                            # just manually retry the flushdb
-                            await client.flushdb(target_nodes="primaries")
-                    await client.close()
-
-            if event_loop.is_running():
-                event_loop.create_task(ateardown())
+        async def teardown():
+            if not cluster_mode:
+                if flushdb and "username" not in kwargs:
+                    try:
+                        await client.flushdb()
+                    except redis.ConnectionError:
+                        # handle cases where a test disconnected a client
+                        # just manually retry the flushdb
+                        await client.flushdb()
+                await client.close()
+                await client.connection_pool.disconnect()
             else:
-                event_loop.run_until_complete(ateardown())
-
-        request.addfinalizer(teardown)
-
+                if flushdb:
+                    try:
+                        await client.flushdb(target_nodes="primaries")
+                    except redis.ConnectionError:
+                        # handle cases where a test disconnected a client
+                        # just manually retry the flushdb
+                        await client.flushdb(target_nodes="primaries")
+                await client.close()
+
+        teardown_clients.append(teardown)
         return client
 
-    return f
+    yield client_factory
+
+    for teardown in teardown_clients:
+        await teardown()
 
 
 @pytest_asyncio.fixture()
-async def r(request, create_redis):
-    yield await create_redis()
+async def r(create_redis):
+    return await create_redis()
 
 
 @pytest_asyncio.fixture()
 async def r2(create_redis):
     """A second client for tests that need multiple"""
-    yield await create_redis()
+    return await create_redis()
 
 
 @pytest_asyncio.fixture()
 async def modclient(request, create_redis):
-    yield await create_redis(
+    return await create_redis(
         url=request.config.getoption("--redismod-url"), decode_responses=True
     )
 
@@ -222,7 +220,7 @@ async def mock_cluster_resp_slaves(create_redis, **kwargs):
 def master_host(request):
     url = request.config.getoption("--redis-url")
     parts = urlparse(url)
-    yield parts.hostname
+    return parts.hostname
 
 
 async def wait_for_command(
@@ -246,3 +244,41 @@ async def wait_for_command(
             return monitor_response
         if key in monitor_response["command"]:
             return None
+
+
+# python 3.6 doesn't have the asynccontextmanager decorator.  Provide it here.
+class AsyncContextManager:
+    def __init__(self, async_generator):
+        self.gen = async_generator
+
+    async def __aenter__(self):
+        try:
+            return await self.gen.__anext__()
+        except StopAsyncIteration as err:
+            raise RuntimeError("Pickles") from err
+
+    async def __aexit__(self, exc_type, exc_inst, tb):
+        if exc_type:
+            await self.gen.athrow(exc_type, exc_inst, tb)
+            return True
+        try:
+            await self.gen.__anext__()
+        except StopAsyncIteration:
+            return
+        raise RuntimeError("More pickles")
+
+
+if sys.version_info[0:2] == (3, 6):
+
+    def asynccontextmanager(func):
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            return AsyncContextManager(func(*args, **kwargs))
+
+        return wrapper
+
+else:
+    from contextlib import asynccontextmanager as _asynccontextmanager
+
+    def asynccontextmanager(func):
+        return _asynccontextmanager(func)
diff --git a/tests/test_asyncio/test_bloom.py b/tests/test_asyncio/test_bloom.py
index feb98cc4..2bf4e030 100644
--- a/tests/test_asyncio/test_bloom.py
+++ b/tests/test_asyncio/test_bloom.py
@@ -1,10 +1,13 @@
+import sys
+
 import pytest
 
 import redis.asyncio as redis
 from redis.exceptions import ModuleError, RedisError
 from redis.utils import HIREDIS_AVAILABLE
 
-pytestmark = pytest.mark.asyncio
+if sys.version_info[0:2] == (3, 6):
+    pytestmark = pytest.mark.asyncio
 
 
 def intlist(obj):
@@ -91,7 +94,7 @@ async def test_bf_scandump_and_loadchunk(modclient: redis.Redis):
             res += rv == x
         assert res < 5
 
-    do_verify()
+    await do_verify()
     cmds = []
     if HIREDIS_AVAILABLE:
         with pytest.raises(ModuleError):
@@ -120,7 +123,7 @@ async def test_bf_scandump_and_loadchunk(modclient: redis.Redis):
 
     cur_info = await modclient.bf().execute_command("bf.debug", "myBloom")
     assert prev_info == cur_info
-    do_verify()
+    await do_verify()
 
     await modclient.bf().client.delete("myBloom")
     await modclient.bf().create("myBloom", "0.0001", "10000000")
diff --git a/tests/test_asyncio/test_cluster.py b/tests/test_asyncio/test_cluster.py
index 0d0ea33d..8766cbf0 100644
--- a/tests/test_asyncio/test_cluster.py
+++ b/tests/test_asyncio/test_cluster.py
@@ -11,6 +11,8 @@ from .compat import mock
 
 if sys.version_info[0:2] == (3, 6):
     import pytest as pytest_asyncio
+
+    pytestmark = pytest.mark.asyncio
 else:
     import pytest_asyncio
 
@@ -39,8 +41,6 @@ from tests.conftest import (
     skip_unless_arch_bits,
 )
 
-pytestmark = pytest.mark.asyncio
-
 default_host = "127.0.0.1"
 default_port = 7000
 default_cluster_slots = [
diff --git a/tests/test_asyncio/test_commands.py b/tests/test_asyncio/test_commands.py
index e128ac40..913f05b3 100644
--- a/tests/test_asyncio/test_commands.py
+++ b/tests/test_asyncio/test_commands.py
@@ -12,6 +12,8 @@ import pytest
 
 if sys.version_info[0:2] == (3, 6):
     import pytest as pytest_asyncio
+
+    pytestmark = pytest.mark.asyncio
 else:
     import pytest_asyncio
 
@@ -27,11 +29,24 @@ from tests.conftest import (
 REDIS_6_VERSION = "5.9.0"
 
 
-pytestmark = pytest.mark.asyncio
+@pytest_asyncio.fixture()
+async def r_teardown(r: redis.Redis):
+    """
+    A special fixture which removes the provided names from the database after use
+    """
+    usernames = []
+
+    def factory(username):
+        usernames.append(username)
+        return r
+
+    yield factory
+    for username in usernames:
+        await r.acl_deluser(username)
 
 
 @pytest_asyncio.fixture()
-async def slowlog(r: redis.Redis, event_loop):
+async def slowlog(r: redis.Redis):
     current_config = await r.config_get()
     old_slower_than_value = current_config["slowlog-log-slower-than"]
     old_max_legnth_value = current_config["slowlog-max-len"]
@@ -94,17 +109,9 @@ class TestRedisCommands:
         assert "get" in commands
 
     @skip_if_server_version_lt(REDIS_6_VERSION)
-    async def test_acl_deluser(self, r: redis.Redis, request, event_loop):
+    async def test_acl_deluser(self, r_teardown):
         username = "redis-py-user"
-
-        def teardown():
-            coro = r.acl_deluser(username)
-            if event_loop.is_running():
-                event_loop.create_task(coro)
-            else:
-                event_loop.run_until_complete(coro)
-
-        request.addfinalizer(teardown)
+        r = r_teardown(username)
 
         assert await r.acl_deluser(username) == 0
         assert await r.acl_setuser(username, enabled=False, reset=True)
@@ -117,18 +124,9 @@ class TestRedisCommands:
 
     @skip_if_server_version_lt(REDIS_6_VERSION)
     @skip_if_server_version_gte("7.0.0")
-    async def test_acl_getuser_setuser(self, r: redis.Redis, request, event_loop):
+    async def test_acl_getuser_setuser(self, r_teardown):
         username = "redis-py-user"
-
-        def teardown():
-            coro = r.acl_deluser(username)
-            if event_loop.is_running():
-                event_loop.create_task(coro)
-            else:
-                event_loop.run_until_complete(coro)
-
-        request.addfinalizer(teardown)
-
+        r = r_teardown(username)
         # test enabled=False
         assert await r.acl_setuser(username, enabled=False, reset=True)
         assert await r.acl_getuser(username) == {
@@ -233,17 +231,9 @@ class TestRedisCommands:
 
     @skip_if_server_version_lt(REDIS_6_VERSION)
     @skip_if_server_version_gte("7.0.0")
-    async def test_acl_list(self, r: redis.Redis, request, event_loop):
+    async def test_acl_list(self, r_teardown):
         username = "redis-py-user"
-
-        def teardown():
-            coro = r.acl_deluser(username)
-            if event_loop.is_running():
-                event_loop.create_task(coro)
-            else:
-                event_loop.run_until_complete(coro)
-
-        request.addfinalizer(teardown)
+        r = r_teardown(username)
 
         assert await r.acl_setuser(username, enabled=False, reset=True)
         users = await r.acl_list()
@@ -251,17 +241,9 @@ class TestRedisCommands:
 
     @skip_if_server_version_lt(REDIS_6_VERSION)
     @pytest.mark.onlynoncluster
-    async def test_acl_log(self, r: redis.Redis, request, event_loop, create_redis):
+    async def test_acl_log(self, r_teardown, create_redis):
         username = "redis-py-user"
-
-        def teardown():
-            coro = r.acl_deluser(username)
-            if event_loop.is_running():
-                event_loop.create_task(coro)
-            else:
-                event_loop.run_until_complete(coro)
-
-        request.addfinalizer(teardown)
+        r = r_teardown(username)
         await r.acl_setuser(
             username,
             enabled=True,
@@ -294,55 +276,25 @@ class TestRedisCommands:
         assert await r.acl_log_reset()
 
     @skip_if_server_version_lt(REDIS_6_VERSION)
-    async def test_acl_setuser_categories_without_prefix_fails(
-        self, r: redis.Redis, request, event_loop
-    ):
+    async def test_acl_setuser_categories_without_prefix_fails(self, r_teardown):
         username = "redis-py-user"
-
-        def teardown():
-            coro = r.acl_deluser(username)
-            if event_loop.is_running():
-                event_loop.create_task(coro)
-            else:
-                event_loop.run_until_complete(coro)
-
-        request.addfinalizer(teardown)
+        r = r_teardown(username)
 
         with pytest.raises(exceptions.DataError):
             await r.acl_setuser(username, categories=["list"])
 
     @skip_if_server_version_lt(REDIS_6_VERSION)
-    async def test_acl_setuser_commands_without_prefix_fails(
-        self, r: redis.Redis, request, event_loop
-    ):
+    async def test_acl_setuser_commands_without_prefix_fails(self, r_teardown):
         username = "redis-py-user"
-
-        def teardown():
-            coro = r.acl_deluser(username)
-            if event_loop.is_running():
-                event_loop.create_task(coro)
-            else:
-                event_loop.run_until_complete(coro)
-
-        request.addfinalizer(teardown)
+        r = r_teardown(username)
 
         with pytest.raises(exceptions.DataError):
             await r.acl_setuser(username, commands=["get"])
 
     @skip_if_server_version_lt(REDIS_6_VERSION)
-    async def test_acl_setuser_add_passwords_and_nopass_fails(
-        self, r: redis.Redis, request, event_loop
-    ):
+    async def test_acl_setuser_add_passwords_and_nopass_fails(self, r_teardown):
         username = "redis-py-user"
-
-        def teardown():
-            coro = r.acl_deluser(username)
-            if event_loop.is_running():
-                event_loop.create_task(coro)
-            else:
-                event_loop.run_until_complete(coro)
-
-        request.addfinalizer(teardown)
+        r = r_teardown(username)
 
         with pytest.raises(exceptions.DataError):
             await r.acl_setuser(username, passwords="+mypass", nopass=True)
diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
index 78a3efd2..8030f7e6 100644
--- a/tests/test_asyncio/test_connection.py
+++ b/tests/test_asyncio/test_connection.py
@@ -1,5 +1,6 @@
 import asyncio
 import socket
+import sys
 import types
 from unittest.mock import patch
 
@@ -18,7 +19,8 @@ from tests.conftest import skip_if_server_version_lt
 
 from .compat import mock
 
-pytestmark = pytest.mark.asyncio
+if sys.version_info[0:2] == (3, 6):
+    pytestmark = pytest.mark.asyncio
 
 
 @pytest.mark.onlynoncluster
diff --git a/tests/test_asyncio/test_connection_pool.py b/tests/test_asyncio/test_connection_pool.py
index 6c56558d..c8eb918e 100644
--- a/tests/test_asyncio/test_connection_pool.py
+++ b/tests/test_asyncio/test_connection_pool.py
@@ -7,6 +7,8 @@ import pytest
 
 if sys.version_info[0:2] == (3, 6):
     import pytest as pytest_asyncio
+
+    pytestmark = pytest.mark.asyncio
 else:
     import pytest_asyncio
 
@@ -15,10 +17,9 @@ from redis.asyncio.connection import Connection, to_bool
 from tests.conftest import skip_if_redis_enterprise, skip_if_server_version_lt
 
 from .compat import mock
+from .conftest import asynccontextmanager
 from .test_pubsub import wait_for_message
 
-pytestmark = pytest.mark.asyncio
-
 
 @pytest.mark.onlynoncluster
 class TestRedisAutoReleaseConnectionPool:
@@ -114,7 +115,8 @@ class DummyConnection(Connection):
 
 
 class TestConnectionPool:
-    def get_pool(
+    @asynccontextmanager
+    async def get_pool(
         self,
         connection_kwargs=None,
         max_connections=None,
@@ -126,71 +128,77 @@ class TestConnectionPool:
             max_connections=max_connections,
             **connection_kwargs,
         )
-        return pool
+        try:
+            yield pool
+        finally:
+            await pool.disconnect(inuse_connections=True)
 
     async def test_connection_creation(self):
         connection_kwargs = {"foo": "bar", "biz": "baz"}
-        pool = self.get_pool(
+        async with self.get_pool(
             connection_kwargs=connection_kwargs, connection_class=DummyConnection
-        )
-        connection = await pool.get_connection("_")
-        assert isinstance(connection, DummyConnection)
-        assert connection.kwargs == connection_kwargs
+        ) as pool:
+            connection = await pool.get_connection("_")
+            assert isinstance(connection, DummyConnection)
+            assert connection.kwargs == connection_kwargs
 
     async def test_multiple_connections(self, master_host):
         connection_kwargs = {"host": master_host}
-        pool = self.get_pool(connection_kwargs=connection_kwargs)
-        c1 = await pool.get_connection("_")
-        c2 = await pool.get_connection("_")
-        assert c1 != c2
+        async with self.get_pool(connection_kwargs=connection_kwargs) as pool:
+            c1 = await pool.get_connection("_")
+            c2 = await pool.get_connection("_")
+            assert c1 != c2
 
     async def test_max_connections(self, master_host):
         connection_kwargs = {"host": master_host}
-        pool = self.get_pool(max_connections=2, connection_kwargs=connection_kwargs)
-        await pool.get_connection("_")
-        await pool.get_connection("_")
-        with pytest.raises(redis.ConnectionError):
+        async with self.get_pool(
+            max_connections=2, connection_kwargs=connection_kwargs
+        ) as pool:
+            await pool.get_connection("_")
             await pool.get_connection("_")
+            with pytest.raises(redis.ConnectionError):
+                await pool.get_connection("_")
 
     async def test_reuse_previously_released_connection(self, master_host):
         connection_kwargs = {"host": master_host}
-        pool = self.get_pool(connection_kwargs=connection_kwargs)
-        c1 = await pool.get_connection("_")
-        await pool.release(c1)
-        c2 = await pool.get_connection("_")
-        assert c1 == c2
+        async with self.get_pool(connection_kwargs=connection_kwargs) as pool:
+            c1 = await pool.get_connection("_")
+            await pool.release(c1)
+            c2 = await pool.get_connection("_")
+            assert c1 == c2
 
-    def test_repr_contains_db_info_tcp(self):
+    async def test_repr_contains_db_info_tcp(self):
         connection_kwargs = {
             "host": "localhost",
             "port": 6379,
             "db": 1,
             "client_name": "test-client",
         }
-        pool = self.get_pool(
+        async with self.get_pool(
             connection_kwargs=connection_kwargs, connection_class=redis.Connection
-        )
-        expected = (
-            "ConnectionPool<Connection<"
-            "host=localhost,port=6379,db=1,client_name=test-client>>"
-        )
-        assert repr(pool) == expected
+        ) as pool:
+            expected = (
+                "ConnectionPool<Connection<"
+                "host=localhost,port=6379,db=1,client_name=test-client>>"
+            )
+            assert repr(pool) == expected
 
-    def test_repr_contains_db_info_unix(self):
+    async def test_repr_contains_db_info_unix(self):
         connection_kwargs = {"path": "/abc", "db": 1, "client_name": "test-client"}
-        pool = self.get_pool(
+        async with self.get_pool(
             connection_kwargs=connection_kwargs,
             connection_class=redis.UnixDomainSocketConnection,
-        )
-        expected = (
-            "ConnectionPool<UnixDomainSocketConnection<"
-            "path=/abc,db=1,client_name=test-client>>"
-        )
-        assert repr(pool) == expected
+        ) as pool:
+            expected = (
+                "ConnectionPool<UnixDomainSocketConnection<"
+                "path=/abc,db=1,client_name=test-client>>"
+            )
+            assert repr(pool) == expected
 
 
 class TestBlockingConnectionPool:
-    def get_pool(self, connection_kwargs=None, max_connections=10, timeout=20):
+    @asynccontextmanager
+    async def get_pool(self, connection_kwargs=None, max_connections=10, timeout=20):
         connection_kwargs = connection_kwargs or {}
         pool = redis.BlockingConnectionPool(
             connection_class=DummyConnection,
@@ -198,7 +206,10 @@ class TestBlockingConnectionPool:
             timeout=timeout,
             **connection_kwargs,
         )
-        return pool
+        try:
+            yield pool
+        finally:
+            await pool.disconnect(inuse_connections=True)
 
     async def test_connection_creation(self, master_host):
         connection_kwargs = {
@@ -207,10 +218,10 @@ class TestBlockingConnectionPool:
             "host": master_host[0],
             "port": master_host[1],
         }
-        pool = self.get_pool(connection_kwargs=connection_kwargs)
-        connection = await pool.get_connection("_")
-        assert isinstance(connection, DummyConnection)
-        assert connection.kwargs == connection_kwargs
+        async with self.get_pool(connection_kwargs=connection_kwargs) as pool:
+            connection = await pool.get_connection("_")
+            assert isinstance(connection, DummyConnection)
+            assert connection.kwargs == connection_kwargs
 
     async def test_disconnect(self, master_host):
         """A regression test for #1047"""
@@ -220,30 +231,31 @@ class TestBlockingConnectionPool:
             "host": master_host[0],
             "port": master_host[1],
         }
-        pool = self.get_pool(connection_kwargs=connection_kwargs)
-        await pool.get_connection("_")
-        await pool.disconnect()
+        async with self.get_pool(connection_kwargs=connection_kwargs) as pool:
+            await pool.get_connection("_")
+            await pool.disconnect()
 
     async def test_multiple_connections(self, master_host):
         connection_kwargs = {"host": master_host[0], "port": master_host[1]}
-        pool = self.get_pool(connection_kwargs=connection_kwargs)
-        c1 = await pool.get_connection("_")
-        c2 = await pool.get_connection("_")
-        assert c1 != c2
+        async with self.get_pool(connection_kwargs=connection_kwargs) as pool:
+            c1 = await pool.get_connection("_")
+            c2 = await pool.get_connection("_")
+            assert c1 != c2
 
     async def test_connection_pool_blocks_until_timeout(self, master_host):
         """When out of connections, block for timeout seconds, then raise"""
         connection_kwargs = {"host": master_host}
-        pool = self.get_pool(
+        async with self.get_pool(
             max_connections=1, timeout=0.1, connection_kwargs=connection_kwargs
-        )
-        await pool.get_connection("_")
+        ) as pool:
+            c1 = await pool.get_connection("_")
 
-        start = asyncio.get_event_loop().time()
-        with pytest.raises(redis.ConnectionError):
-            await pool.get_connection("_")
-        # we should have waited at least 0.1 seconds
-        assert asyncio.get_event_loop().time() - start >= 0.1
+            start = asyncio.get_event_loop().time()
+            with pytest.raises(redis.ConnectionError):
+                await pool.get_connection("_")
+            # we should have waited at least 0.1 seconds
+            assert asyncio.get_event_loop().time() - start >= 0.1
+            await c1.disconnect()
 
     async def test_connection_pool_blocks_until_conn_available(self, master_host):
         """
@@ -251,26 +263,26 @@ class TestBlockingConnectionPool:
         to the pool
         """
         connection_kwargs = {"host": master_host[0], "port": master_host[1]}
-        pool = self.get_pool(
+        async with self.get_pool(
             max_connections=1, timeout=2, connection_kwargs=connection_kwargs
-        )
-        c1 = await pool.get_connection("_")
+        ) as pool:
+            c1 = await pool.get_connection("_")
 
-        async def target():
-            await asyncio.sleep(0.1)
-            await pool.release(c1)
+            async def target():
+                await asyncio.sleep(0.1)
+                await pool.release(c1)
 
-        start = asyncio.get_event_loop().time()
-        await asyncio.gather(target(), pool.get_connection("_"))
-        assert asyncio.get_event_loop().time() - start >= 0.1
+            start = asyncio.get_event_loop().time()
+            await asyncio.gather(target(), pool.get_connection("_"))
+            assert asyncio.get_event_loop().time() - start >= 0.1
 
     async def test_reuse_previously_released_connection(self, master_host):
         connection_kwargs = {"host": master_host}
-        pool = self.get_pool(connection_kwargs=connection_kwargs)
-        c1 = await pool.get_connection("_")
-        await pool.release(c1)
-        c2 = await pool.get_connection("_")
-        assert c1 == c2
+        async with self.get_pool(connection_kwargs=connection_kwargs) as pool:
+            c1 = await pool.get_connection("_")
+            await pool.release(c1)
+            c2 = await pool.get_connection("_")
+            assert c1 == c2
 
     def test_repr_contains_db_info_tcp(self):
         pool = redis.ConnectionPool(
@@ -689,6 +701,8 @@ class TestHealthCheck:
         if r.connection:
             await r.get("foo")
             next_health_check = r.connection.next_health_check
+            # ensure that the event loop's `time()` advances a bit
+            await asyncio.sleep(0.001)
             await r.get("foo")
             assert next_health_check < r.connection.next_health_check
 
diff --git a/tests/test_asyncio/test_encoding.py b/tests/test_asyncio/test_encoding.py
index 133ea378..5db7187c 100644
--- a/tests/test_asyncio/test_encoding.py
+++ b/tests/test_asyncio/test_encoding.py
@@ -4,14 +4,14 @@ import pytest
 
 if sys.version_info[0:2] == (3, 6):
     import pytest as pytest_asyncio
+
+    pytestmark = pytest.mark.asyncio
 else:
     import pytest_asyncio
 
 import redis.asyncio as redis
 from redis.exceptions import DataError
 
-pytestmark = pytest.mark.asyncio
-
 
 @pytest.mark.onlynoncluster
 class TestEncoding:
diff --git a/tests/test_asyncio/test_json.py b/tests/test_asyncio/test_json.py
index a045dd7c..416a9f4a 100644
--- a/tests/test_asyncio/test_json.py
+++ b/tests/test_asyncio/test_json.py
@@ -5,8 +5,6 @@ from redis import exceptions
 from redis.commands.json.path import Path
 from tests.conftest import skip_ifmodversion_lt
 
-pytestmark = pytest.mark.asyncio
-
 
 @pytest.mark.redismod
 async def test_json_setbinarykey(modclient: redis.Redis):
diff --git a/tests/test_asyncio/test_lock.py b/tests/test_asyncio/test_lock.py
index 8ceb3bc9..86a8d62f 100644
--- a/tests/test_asyncio/test_lock.py
+++ b/tests/test_asyncio/test_lock.py
@@ -5,14 +5,14 @@ import pytest
 
 if sys.version_info[0:2] == (3, 6):
     import pytest as pytest_asyncio
+
+    pytestmark = pytest.mark.asyncio
 else:
     import pytest_asyncio
 
 from redis.asyncio.lock import Lock
 from redis.exceptions import LockError, LockNotOwnedError
 
-pytestmark = pytest.mark.asyncio
-
 
 @pytest.mark.onlynoncluster
 class TestLock:
diff --git a/tests/test_asyncio/test_monitor.py b/tests/test_asyncio/test_monitor.py
index 783ba262..9185bcd2 100644
--- a/tests/test_asyncio/test_monitor.py
+++ b/tests/test_asyncio/test_monitor.py
@@ -1,10 +1,13 @@
+import sys
+
 import pytest
 
 from tests.conftest import skip_if_redis_enterprise, skip_ifnot_redis_enterprise
 
 from .conftest import wait_for_command
 
-pytestmark = pytest.mark.asyncio
+if sys.version_info[0:2] == (3, 6):
+    pytestmark = pytest.mark.asyncio
 
 
 @pytest.mark.onlynoncluster
diff --git a/tests/test_asyncio/test_pipeline.py b/tests/test_asyncio/test_pipeline.py
index dfeb6646..33391d01 100644
--- a/tests/test_asyncio/test_pipeline.py
+++ b/tests/test_asyncio/test_pipeline.py
@@ -1,3 +1,5 @@
+import sys
+
 import pytest
 
 import redis
@@ -5,7 +7,8 @@ from tests.conftest import skip_if_server_version_lt
 
 from .conftest import wait_for_command
 
-pytestmark = pytest.mark.asyncio
+if sys.version_info[0:2] == (3, 6):
+    pytestmark = pytest.mark.asyncio
 
 
 class TestPipeline:
diff --git a/tests/test_asyncio/test_pubsub.py b/tests/test_asyncio/test_pubsub.py
index 6c76bf33..d6a817a6 100644
--- a/tests/test_asyncio/test_pubsub.py
+++ b/tests/test_asyncio/test_pubsub.py
@@ -8,6 +8,8 @@ import pytest
 
 if sys.version_info[0:2] == (3, 6):
     import pytest as pytest_asyncio
+
+    pytestmark = pytest.mark.asyncio(forbid_global_loop=True)
 else:
     import pytest_asyncio
 
@@ -18,8 +20,6 @@ from tests.conftest import skip_if_server_version_lt
 
 from .compat import mock
 
-pytestmark = pytest.mark.asyncio(forbid_global_loop=True)
-
 
 def with_timeout(t):
     def wrapper(corofunc):
@@ -80,6 +80,13 @@ def make_subscribe_test_data(pubsub, type):
     assert False, f"invalid subscribe type: {type}"
 
 
+@pytest_asyncio.fixture()
+async def pubsub(r: redis.Redis):
+    p = r.pubsub()
+    yield p
+    await p.close()
+
+
 @pytest.mark.onlynoncluster
 class TestPubSubSubscribeUnsubscribe:
     async def _test_subscribe_unsubscribe(
@@ -101,12 +108,12 @@ class TestPubSubSubscribeUnsubscribe:
             i = len(keys) - 1 - i
             assert await wait_for_message(p) == make_message(unsub_type, key, i)
 
-    async def test_channel_subscribe_unsubscribe(self, r: redis.Redis):
-        kwargs = make_subscribe_test_data(r.pubsub(), "channel")
+    async def test_channel_subscribe_unsubscribe(self, pubsub):
+        kwargs = make_subscribe_test_data(pubsub, "channel")
         await self._test_subscribe_unsubscribe(**kwargs)
 
-    async def test_pattern_subscribe_unsubscribe(self, r: redis.Redis):
-        kwargs = make_subscribe_test_data(r.pubsub(), "pattern")
+    async def test_pattern_subscribe_unsubscribe(self, pubsub):
+        kwargs = make_subscribe_test_data(pubsub, "pattern")
         await self._test_subscribe_unsubscribe(**kwargs)
 
     @pytest.mark.onlynoncluster
@@ -144,12 +151,12 @@ class TestPubSubSubscribeUnsubscribe:
         for channel in unique_channels:
             assert channel in keys
 
-    async def test_resubscribe_to_channels_on_reconnection(self, r: redis.Redis):
-        kwargs = make_subscribe_test_data(r.pubsub(), "channel")
+    async def test_resubscribe_to_channels_on_reconnection(self, pubsub):
+        kwargs = make_subscribe_test_data(pubsub, "channel")
         await self._test_resubscribe_on_reconnection(**kwargs)
 
-    async def test_resubscribe_to_patterns_on_reconnection(self, r: redis.Redis):
-        kwargs = make_subscribe_test_data(r.pubsub(), "pattern")
+    async def test_resubscribe_to_patterns_on_reconnection(self, pubsub):
+        kwargs = make_subscribe_test_data(pubsub, "pattern")
         await self._test_resubscribe_on_reconnection(**kwargs)
 
     async def _test_subscribed_property(
@@ -199,13 +206,13 @@ class TestPubSubSubscribeUnsubscribe:
         # now we're finally unsubscribed
         assert p.subscribed is False
 
-    async def test_subscribe_property_with_channels(self, r: redis.Redis):
-        kwargs = make_subscribe_test_data(r.pubsub(), "channel")
+    async def test_subscribe_property_with_channels(self, pubsub):
+        kwargs = make_subscribe_test_data(pubsub, "channel")
         await self._test_subscribed_property(**kwargs)
 
     @pytest.mark.onlynoncluster
-    async def test_subscribe_property_with_patterns(self, r: redis.Redis):
-        kwargs = make_subscribe_test_data(r.pubsub(), "pattern")
+    async def test_subscribe_property_with_patterns(self, pubsub):
+        kwargs = make_subscribe_test_data(pubsub, "pattern")
         await self._test_subscribed_property(**kwargs)
 
     async def test_ignore_all_subscribe_messages(self, r: redis.Redis):
@@ -224,9 +231,10 @@ class TestPubSubSubscribeUnsubscribe:
             assert p.subscribed is True
             assert await wait_for_message(p) is None
         assert p.subscribed is False
+        await p.close()
 
-    async def test_ignore_individual_subscribe_messages(self, r: redis.Redis):
-        p = r.pubsub()
+    async def test_ignore_individual_subscribe_messages(self, pubsub):
+        p = pubsub
 
         checks = (
             (p.subscribe, "foo"),
@@ -243,13 +251,13 @@ class TestPubSubSubscribeUnsubscribe:
             assert message is None
         assert p.subscribed is False
 
-    async def test_sub_unsub_resub_channels(self, r: redis.Redis):
-        kwargs = make_subscribe_test_data(r.pubsub(), "channel")
+    async def test_sub_unsub_resub_channels(self, pubsub):
+        kwargs = make_subscribe_test_data(pubsub, "channel")
         await self._test_sub_unsub_resub(**kwargs)
 
     @pytest.mark.onlynoncluster
-    async def test_sub_unsub_resub_patterns(self, r: redis.Redis):
-        kwargs = make_subscribe_test_data(r.pubsub(), "pattern")
+    async def test_sub_unsub_resub_patterns(self, pubsub):
+        kwargs = make_subscribe_test_data(pubsub, "pattern")
         await self._test_sub_unsub_resub(**kwargs)
 
     async def _test_sub_unsub_resub(
@@ -266,12 +274,12 @@ class TestPubSubSubscribeUnsubscribe:
         assert await wait_for_message(p) == make_message(sub_type, key, 1)
         assert p.subscribed is True
 
-    async def test_sub_unsub_all_resub_channels(self, r: redis.Redis):
-        kwargs = make_subscribe_test_data(r.pubsub(), "channel")
+    async def test_sub_unsub_all_resub_channels(self, pubsub):
+        kwargs = make_subscribe_test_data(pubsub, "channel")
         await self._test_sub_unsub_all_resub(**kwargs)
 
-    async def test_sub_unsub_all_resub_patterns(self, r: redis.Redis):
-        kwargs = make_subscribe_test_data(r.pubsub(), "pattern")
+    async def test_sub_unsub_all_resub_patterns(self, pubsub):
+        kwargs = make_subscribe_test_data(pubsub, "pattern")
         await self._test_sub_unsub_all_resub(**kwargs)
 
     async def _test_sub_unsub_all_resub(
@@ -300,8 +308,8 @@ class TestPubSubMessages:
     async def async_message_handler(self, message):
         self.async_message = message
 
-    async def test_published_message_to_channel(self, r: redis.Redis):
-        p = r.pubsub()
+    async def test_published_message_to_channel(self, r: redis.Redis, pubsub):
+        p = pubsub
         await p.subscribe("foo")
         assert await wait_for_message(p) == make_message("subscribe", "foo", 1)
         assert await r.publish("foo", "test message") == 1
@@ -310,8 +318,8 @@ class TestPubSubMessages:
         assert isinstance(message, dict)
         assert message == make_message("message", "foo", "test message")
 
-    async def test_published_message_to_pattern(self, r: redis.Redis):
-        p = r.pubsub()
+    async def test_published_message_to_pattern(self, r: redis.Redis, pubsub):
+        p = pubsub
         await p.subscribe("foo")
         await p.psubscribe("f*")
         assert await wait_for_message(p) == make_message("subscribe", "foo", 1)
@@ -340,6 +348,7 @@ class TestPubSubMessages:
         assert await r.publish("foo", "test message") == 1
         assert await wait_for_message(p) is None
         assert self.message == make_message("message", "foo", "test message")
+        await p.close()
 
     async def test_channel_async_message_handler(self, r):
         p = r.pubsub(ignore_subscribe_messages=True)
@@ -348,6 +357,7 @@ class TestPubSubMessages:
         assert await r.publish("foo", "test message") == 1
         assert await wait_for_message(p) is None
         assert self.async_message == make_message("message", "foo", "test message")
+        await p.close()
 
     async def test_channel_sync_async_message_handler(self, r):
         p = r.pubsub(ignore_subscribe_messages=True)
@@ -359,6 +369,7 @@ class TestPubSubMessages:
         assert await wait_for_message(p) is None
         assert self.message == make_message("message", "foo", "test message")
         assert self.async_message == make_message("message", "bar", "test message 2")
+        await p.close()
 
     @pytest.mark.onlynoncluster
     async def test_pattern_message_handler(self, r: redis.Redis):
@@ -370,6 +381,7 @@ class TestPubSubMessages:
         assert self.message == make_message(
             "pmessage", "foo", "test message", pattern="f*"
         )
+        await p.close()
 
     async def test_unicode_channel_message_handler(self, r: redis.Redis):
         p = r.pubsub(ignore_subscribe_messages=True)
@@ -380,6 +392,7 @@ class TestPubSubMessages:
         assert await r.publish(channel, "test message") == 1
         assert await wait_for_message(p) is None
         assert self.message == make_message("message", channel, "test message")
+        await p.close()
 
     @pytest.mark.onlynoncluster
     # see: https://redis-py-cluster.readthedocs.io/en/stable/pubsub.html
@@ -395,9 +408,10 @@ class TestPubSubMessages:
         assert self.message == make_message(
             "pmessage", channel, "test message", pattern=pattern
         )
+        await p.close()
 
-    async def test_get_message_without_subscribe(self, r: redis.Redis):
-        p = r.pubsub()
+    async def test_get_message_without_subscribe(self, r: redis.Redis, pubsub):
+        p = pubsub
         with pytest.raises(RuntimeError) as info:
             await p.get_message()
         expect = (
@@ -427,8 +441,8 @@ class TestPubSubAutoDecoding:
     async def r(self, create_redis):
         return await create_redis(decode_responses=True)
 
-    async def test_channel_subscribe_unsubscribe(self, r: redis.Redis):
-        p = r.pubsub()
+    async def test_channel_subscribe_unsubscribe(self, pubsub):
+        p = pubsub
         await p.subscribe(self.channel)
         assert await wait_for_message(p) == self.make_message(
             "subscribe", self.channel, 1
@@ -439,8 +453,8 @@ class TestPubSubAutoDecoding:
             "unsubscribe", self.channel, 0
         )
 
-    async def test_pattern_subscribe_unsubscribe(self, r: redis.Redis):
-        p = r.pubsub()
+    async def test_pattern_subscribe_unsubscribe(self, pubsub):
+        p = pubsub
         await p.psubscribe(self.pattern)
         assert await wait_for_message(p) == self.make_message(
             "psubscribe", self.pattern, 1
@@ -451,8 +465,8 @@ class TestPubSubAutoDecoding:
             "punsubscribe", self.pattern, 0
         )
 
-    async def test_channel_publish(self, r: redis.Redis):
-        p = r.pubsub()
+    async def test_channel_publish(self, r: redis.Redis, pubsub):
+        p = pubsub
         await p.subscribe(self.channel)
         assert await wait_for_message(p) == self.make_message(
             "subscribe", self.channel, 1
@@ -463,8 +477,8 @@ class TestPubSubAutoDecoding:
         )
 
     @pytest.mark.onlynoncluster
-    async def test_pattern_publish(self, r: redis.Redis):
-        p = r.pubsub()
+    async def test_pattern_publish(self, r: redis.Redis, pubsub):
+        p = pubsub
         await p.psubscribe(self.pattern)
         assert await wait_for_message(p) == self.make_message(
             "psubscribe", self.pattern, 1
@@ -490,6 +504,7 @@ class TestPubSubAutoDecoding:
         await r.publish(self.channel, new_data)
         assert await wait_for_message(p) is None
         assert self.message == self.make_message("message", self.channel, new_data)
+        await p.close()
 
     async def test_pattern_message_handler(self, r: redis.Redis):
         p = r.pubsub(ignore_subscribe_messages=True)
@@ -511,6 +526,7 @@ class TestPubSubAutoDecoding:
         assert self.message == self.make_message(
             "pmessage", self.channel, new_data, pattern=self.pattern
         )
+        await p.close()
 
     async def test_context_manager(self, r: redis.Redis):
         async with r.pubsub() as pubsub:
@@ -520,6 +536,7 @@ class TestPubSubAutoDecoding:
         assert pubsub.connection is None
         assert pubsub.channels == {}
         assert pubsub.patterns == {}
+        await pubsub.close()
 
 
 @pytest.mark.onlynoncluster
@@ -535,8 +552,8 @@ class TestPubSubRedisDown:
 class TestPubSubSubcommands:
     @pytest.mark.onlynoncluster
     @skip_if_server_version_lt("2.8.0")
-    async def test_pubsub_channels(self, r: redis.Redis):
-        p = r.pubsub()
+    async def test_pubsub_channels(self, r: redis.Redis, pubsub):
+        p = pubsub
         await p.subscribe("foo", "bar", "baz", "quux")
         for i in range(4):
             assert (await wait_for_message(p))["type"] == "subscribe"
@@ -560,6 +577,9 @@ class TestPubSubSubcommands:
 
         channels = [(b"foo", 1), (b"bar", 2), (b"baz", 3)]
         assert await r.pubsub_numsub("foo", "bar", "baz") == channels
+        await p1.close()
+        await p2.close()
+        await p3.close()
 
     @skip_if_server_version_lt("2.8.0")
     async def test_pubsub_numpat(self, r: redis.Redis):
@@ -568,6 +588,7 @@ class TestPubSubSubcommands:
         for i in range(3):
             assert (await wait_for_message(p))["type"] == "psubscribe"
         assert await r.pubsub_numpat() == 3
+        await p.close()
 
 
 @pytest.mark.onlynoncluster
@@ -580,6 +601,7 @@ class TestPubSubPings:
         assert await wait_for_message(p) == make_message(
             type="pong", channel=None, data="", pattern=None
         )
+        await p.close()
 
     @skip_if_server_version_lt("3.0.0")
     async def test_send_pubsub_ping_message(self, r: redis.Redis):
@@ -589,13 +611,16 @@ class TestPubSubPings:
         assert await wait_for_message(p) == make_message(
             type="pong", channel=None, data="hello world", pattern=None
         )
+        await p.close()
 
 
 @pytest.mark.onlynoncluster
 class TestPubSubConnectionKilled:
     @skip_if_server_version_lt("3.0.0")
-    async def test_connection_error_raised_when_connection_dies(self, r: redis.Redis):
-        p = r.pubsub()
+    async def test_connection_error_raised_when_connection_dies(
+        self, r: redis.Redis, pubsub
+    ):
+        p = pubsub
         await p.subscribe("foo")
         assert await wait_for_message(p) == make_message("subscribe", "foo", 1)
         for client in await r.client_list():
@@ -607,8 +632,8 @@ class TestPubSubConnectionKilled:
 
 @pytest.mark.onlynoncluster
 class TestPubSubTimeouts:
-    async def test_get_message_with_timeout_returns_none(self, r: redis.Redis):
-        p = r.pubsub()
+    async def test_get_message_with_timeout_returns_none(self, pubsub):
+        p = pubsub
         await p.subscribe("foo")
         assert await wait_for_message(p) == make_message("subscribe", "foo", 1)
         assert await p.get_message(timeout=0.01) is None
@@ -616,15 +641,13 @@ class TestPubSubTimeouts:
 
 @pytest.mark.onlynoncluster
 class TestPubSubReconnect:
-    # @pytest.mark.xfail
     @with_timeout(2)
-    async def test_reconnect_listen(self, r: redis.Redis):
+    async def test_reconnect_listen(self, r: redis.Redis, pubsub):
         """
         Test that a loop processing PubSub messages can survive
         a disconnect, by issuing a connect() call.
         """
         messages = asyncio.Queue()
-        pubsub = r.pubsub()
         interrupt = False
 
         async def loop():
@@ -698,12 +721,12 @@ class TestPubSubRun:
             ):
                 return
 
-    async def test_callbacks(self, r: redis.Redis):
+    async def test_callbacks(self, r: redis.Redis, pubsub):
         def callback(message):
             messages.put_nowait(message)
 
         messages = asyncio.Queue()
-        p = r.pubsub()
+        p = pubsub
         await self._subscribe(p, foo=callback)
         task = asyncio.get_event_loop().create_task(p.run())
         await r.publish("foo", "bar")
@@ -720,13 +743,13 @@ class TestPubSubRun:
             "type": "message",
         }
 
-    async def test_exception_handler(self, r: redis.Redis):
+    async def test_exception_handler(self, r: redis.Redis, pubsub):
         def exception_handler_callback(e, pubsub) -> None:
             assert pubsub == p
             exceptions.put_nowait(e)
 
         exceptions = asyncio.Queue()
-        p = r.pubsub()
+        p = pubsub
         await self._subscribe(p, foo=lambda x: None)
         with mock.patch.object(p, "get_message", side_effect=Exception("error")):
             task = asyncio.get_event_loop().create_task(
@@ -740,26 +763,25 @@ class TestPubSubRun:
                 pass
         assert str(e) == "error"
 
-    async def test_late_subscribe(self, r: redis.Redis):
+    async def test_late_subscribe(self, r: redis.Redis, pubsub):
         def callback(message):
             messages.put_nowait(message)
 
         messages = asyncio.Queue()
-        p = r.pubsub()
+        p = pubsub
         task = asyncio.get_event_loop().create_task(p.run())
         # wait until loop gets settled.  Add a subscription
         await asyncio.sleep(0.1)
         await p.subscribe(foo=callback)
         # wait tof the subscribe to finish.  Cannot use _subscribe() because
         # p.run() is already accepting messages
-        await asyncio.sleep(0.1)
-        await r.publish("foo", "bar")
-        message = None
-        try:
-            async with async_timeout.timeout(0.1):
-                message = await messages.get()
-        except asyncio.TimeoutError:
-            pass
+        while True:
+            n = await r.publish("foo", "bar")
+            if n == 1:
+                break
+            await asyncio.sleep(0.1)
+        async with async_timeout.timeout(0.1):
+            message = await messages.get()
         task.cancel()
         # we expect a cancelled error, not the Runtime error
         # ("did you forget to call subscribe()"")
diff --git a/tests/test_asyncio/test_scripting.py b/tests/test_asyncio/test_scripting.py
index 764525fb..406ab208 100644
--- a/tests/test_asyncio/test_scripting.py
+++ b/tests/test_asyncio/test_scripting.py
@@ -4,6 +4,8 @@ import pytest
 
 if sys.version_info[0:2] == (3, 6):
     import pytest as pytest_asyncio
+
+    pytestmark = pytest.mark.asyncio
 else:
     import pytest_asyncio
 
diff --git a/tests/test_asyncio/test_search.py b/tests/test_asyncio/test_search.py
index 5aaa56f1..bc3a212a 100644
--- a/tests/test_asyncio/test_search.py
+++ b/tests/test_asyncio/test_search.py
@@ -1,6 +1,7 @@
 import bz2
 import csv
 import os
+import sys
 import time
 from io import TextIOWrapper
 
@@ -18,7 +19,8 @@ from redis.commands.search.result import Result
 from redis.commands.search.suggestion import Suggestion
 from tests.conftest import skip_ifmodversion_lt
 
-pytestmark = pytest.mark.asyncio
+if sys.version_info[0:2] == (3, 6):
+    pytestmark = pytest.mark.asyncio
 
 
 WILL_PLAY_TEXT = os.path.abspath(
diff --git a/tests/test_asyncio/test_sentinel.py b/tests/test_asyncio/test_sentinel.py
index 4130e674..e77e07f9 100644
--- a/tests/test_asyncio/test_sentinel.py
+++ b/tests/test_asyncio/test_sentinel.py
@@ -5,6 +5,8 @@ import pytest
 
 if sys.version_info[0:2] == (3, 6):
     import pytest as pytest_asyncio
+
+    pytestmark = pytest.mark.asyncio
 else:
     import pytest_asyncio
 
@@ -17,8 +19,6 @@ from redis.asyncio.sentinel import (
     SlaveNotFoundError,
 )
 
-pytestmark = pytest.mark.asyncio
-
 
 @pytest_asyncio.fixture(scope="module")
 def master_ip(master_host):
diff --git a/tests/test_asyncio/test_timeseries.py b/tests/test_asyncio/test_timeseries.py
index ac2807fe..0e57c4f0 100644
--- a/tests/test_asyncio/test_timeseries.py
+++ b/tests/test_asyncio/test_timeseries.py
@@ -1,3 +1,4 @@
+import sys
 import time
 from time import sleep
 
@@ -6,7 +7,8 @@ import pytest
 import redis.asyncio as redis
 from tests.conftest import skip_ifmodversion_lt
 
-pytestmark = pytest.mark.asyncio
+if sys.version_info[0:2] == (3, 6):
+    pytestmark = pytest.mark.asyncio
 
 
 @pytest.mark.redismod
diff --git a/tests/test_json.py b/tests/test_json.py
index 1cc448c5..0965a93d 100644
--- a/tests/test_json.py
+++ b/tests/test_json.py
@@ -1411,7 +1411,8 @@ def test_set_path(client):
 
     with open(jsonfile, "w+") as fp:
         fp.write(json.dumps({"hello": "world"}))
-    open(nojsonfile, "a+").write("hello")
+    with open(nojsonfile, "a+") as fp:
+        fp.write("hello")
 
     result = {jsonfile: True, nojsonfile: False}
     assert client.json().set_path(Path.root_path(), root) == result
diff --git a/tests/test_ssl.py b/tests/test_ssl.py
index d029b80d..ed38a316 100644
--- a/tests/test_ssl.py
+++ b/tests/test_ssl.py
@@ -68,8 +68,8 @@ class TestSSL:
         assert r.ping()
 
     def test_validating_self_signed_string_certificate(self, request):
-        f = open(self.SERVER_CERT)
-        cert_data = f.read()
+        with open(self.SERVER_CERT) as f:
+            cert_data = f.read()
         ssl_url = request.config.option.redis_ssl_url
         p = urlparse(ssl_url)[1].split(":")
         r = redis.Redis(