We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
`unittest` framework raises `ResourceWarning`
## Description
<!-- A clear description of the bug -->
I'm using [`unittest`](https://docs.python.org/3/library/unittest.html) framework for testing purposes and then a task is running inside a `unittest.TestCase`, a `ResourceWarning` is raised. 


## Expected Behavior
<!-- What did you expect to happen instead? -->
Run testcase without raising any warnings.




## Reproduction
<!-- A minimal example that exhibits the behavior. -->

`test.py` file:

```python
import typing
import unittest

from prefect.engine.task_runner import TaskRunner
from prefect.tasks.shell import ShellTask


class ShellTestCase(unittest.TestCase):
    def test_shell_command(self) -> typing.NoReturn:
        runner = TaskRunner(ShellTask('ls'))
        runner.run()
        self.assertTrue(True)
```

execution:

```bash
$ python -m unittest test.py 
[2020-07-20 16:13:20] INFO - prefect.TaskRunner | Task 'ShellTask': Starting task run...
/home/psimakis/.local/share/virtualenvs/data-workflows-GfPV92cZ/lib/python3.7/site-packages/prefect/utilities/tasks.py:444: ResourceWarning: unclosed file <_io.BufferedReader name=9>
  return run_method(self, *args, **kwargs)
ResourceWarning: Enable tracemalloc to get the object allocation traceback
[2020-07-20 16:13:21] INFO - prefect.TaskRunner | Task 'ShellTask': finished task run for task with final state: 'Success'
.
----------------------------------------------------------------------
Ran 1 test in 0.026s
```




## Environment
<!-- Any additional information about your environment

Optionally run `prefect diagnostics` from the command line and paste the information here. -->
```json
{
  "config_overrides": {},
  "env_vars": [
    "PREFECT__CONTEXT__SECRETS__....",
    "PREFECT__CONTEXT__SECRETS__...."
  ],
  "system_information": {
    "platform": "Linux-5.3.0-28-generic-x86_64-with-debian-buster-sid",
    "prefect_version": "0.12.3",
    "python_version": "3.7.3"
  }
}
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/prefect/tasks/shell.py`
Content:
```
1 import os
2 import tempfile
3 from subprocess import PIPE, STDOUT, Popen
4 from typing import Any
5 
6 import prefect
7 from prefect.utilities.tasks import defaults_from_attrs
8 
9 
10 class ShellTask(prefect.Task):
11     """
12     Task for running arbitrary shell commands.
13 
14     Args:
15         - command (string, optional): shell command to be executed; can also be
16             provided post-initialization by calling this task instance
17         - env (dict, optional): dictionary of environment variables to use for
18             the subprocess; can also be provided at runtime
19         - helper_script (str, optional): a string representing a shell script, which
20             will be executed prior to the `command` in the same process. Can be used to
21             change directories, define helper functions, etc. when re-using this Task
22             for different commands in a Flow
23         - shell (string, optional): shell to run the command with; defaults to "bash"
24         - return_all (bool, optional): boolean specifying whether this task
25             should return all lines of stdout as a list, or just the last line
26             as a string; defaults to `False`
27         - log_stderr (bool, optional): boolean specifying whether this task
28             should log the output from stderr in the case of a non-zero exit code;
29             defaults to `False`
30         - **kwargs: additional keyword arguments to pass to the Task constructor
31 
32     Example:
33         ```python
34         from prefect import Flow
35         from prefect.tasks.shell import ShellTask
36 
37         task = ShellTask(helper_script="cd ~")
38         with Flow("My Flow") as f:
39             # both tasks will be executed in home directory
40             contents = task(command='ls')
41             mv_file = task(command='mv .vimrc /.vimrc')
42 
43         out = f.run()
44         ```
45     """
46 
47     def __init__(
48         self,
49         command: str = None,
50         env: dict = None,
51         helper_script: str = None,
52         shell: str = "bash",
53         return_all: bool = False,
54         log_stderr: bool = False,
55         **kwargs: Any
56     ):
57         self.command = command
58         self.env = env
59         self.helper_script = helper_script
60         self.shell = shell
61         self.return_all = return_all
62         self.log_stderr = log_stderr
63         super().__init__(**kwargs)
64 
65     @defaults_from_attrs("command", "env")
66     def run(self, command: str = None, env: dict = None) -> str:
67         """
68         Run the shell command.
69 
70         Args:
71             - command (string): shell command to be executed; can also be
72                 provided at task initialization. Any variables / functions defined in
73                 `self.helper_script` will be available in the same process this command
74                 runs in
75             - env (dict, optional): dictionary of environment variables to use for
76                 the subprocess
77 
78         Returns:
79             - stdout (string): if `return_all` is `False` (the default), only
80                 the last line of stdout is returned, otherwise all lines are
81                 returned, which is useful for passing result of shell command
82                 to other downstream tasks. If there is no output, `None` is
83                 returned.
84 
85         Raises:
86             - prefect.engine.signals.FAIL: if command has an exit code other
87                 than 0
88         """
89         if command is None:
90             raise TypeError("run() missing required argument: 'command'")
91 
92         current_env = os.environ.copy()
93         current_env.update(env or {})
94         with tempfile.NamedTemporaryFile(prefix="prefect-") as tmp:
95             if self.helper_script:
96                 tmp.write(self.helper_script.encode())
97                 tmp.write("\n".encode())
98             tmp.write(command.encode())
99             tmp.flush()
100             sub_process = Popen(
101                 [self.shell, tmp.name], stdout=PIPE, stderr=STDOUT, env=current_env
102             )
103             lines = []
104             line = None
105             for raw_line in iter(sub_process.stdout.readline, b""):
106                 line = raw_line.decode("utf-8").rstrip()
107                 if self.return_all:
108                     lines.append(line)
109                 else:
110                     # if we're returning all, we don't log every line
111                     self.logger.debug(line)
112             sub_process.wait()
113             if sub_process.returncode:
114                 msg = "Command failed with exit code {}".format(sub_process.returncode,)
115                 self.logger.error(msg)
116 
117                 if self.log_stderr:
118                     self.logger.error("\n".join(lines))
119 
120                 raise prefect.engine.signals.FAIL(msg) from None  # type: ignore
121         if self.return_all:
122             return lines
123         else:
124             return line
125
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/prefect/tasks/shell.py b/src/prefect/tasks/shell.py
--- a/src/prefect/tasks/shell.py
+++ b/src/prefect/tasks/shell.py
@@ -97,27 +97,29 @@
                 tmp.write("\n".encode())
             tmp.write(command.encode())
             tmp.flush()
-            sub_process = Popen(
+            with Popen(
                 [self.shell, tmp.name], stdout=PIPE, stderr=STDOUT, env=current_env
-            )
-            lines = []
-            line = None
-            for raw_line in iter(sub_process.stdout.readline, b""):
-                line = raw_line.decode("utf-8").rstrip()
-                if self.return_all:
-                    lines.append(line)
-                else:
-                    # if we're returning all, we don't log every line
-                    self.logger.debug(line)
-            sub_process.wait()
-            if sub_process.returncode:
-                msg = "Command failed with exit code {}".format(sub_process.returncode,)
-                self.logger.error(msg)
+            ) as sub_process:
+                lines = []
+                line = None
+                for raw_line in iter(sub_process.stdout.readline, b""):
+                    line = raw_line.decode("utf-8").rstrip()
+                    if self.return_all:
+                        lines.append(line)
+                    else:
+                        # if we're returning all, we don't log every line
+                        self.logger.debug(line)
+                sub_process.wait()
+                if sub_process.returncode:
+                    msg = "Command failed with exit code {}".format(
+                        sub_process.returncode,
+                    )
+                    self.logger.error(msg)
 
-                if self.log_stderr:
-                    self.logger.error("\n".join(lines))
+                    if self.log_stderr:
+                        self.logger.error("\n".join(lines))
 
-                raise prefect.engine.signals.FAIL(msg) from None  # type: ignore
+                    raise prefect.engine.signals.FAIL(msg) from None  # type: ignore
         if self.return_all:
             return lines
         else:

{"golden_diff": "diff --git a/src/prefect/tasks/shell.py b/src/prefect/tasks/shell.py\n--- a/src/prefect/tasks/shell.py\n+++ b/src/prefect/tasks/shell.py\n@@ -97,27 +97,29 @@\n                 tmp.write(\"\\n\".encode())\n             tmp.write(command.encode())\n             tmp.flush()\n-            sub_process = Popen(\n+            with Popen(\n                 [self.shell, tmp.name], stdout=PIPE, stderr=STDOUT, env=current_env\n-            )\n-            lines = []\n-            line = None\n-            for raw_line in iter(sub_process.stdout.readline, b\"\"):\n-                line = raw_line.decode(\"utf-8\").rstrip()\n-                if self.return_all:\n-                    lines.append(line)\n-                else:\n-                    # if we're returning all, we don't log every line\n-                    self.logger.debug(line)\n-            sub_process.wait()\n-            if sub_process.returncode:\n-                msg = \"Command failed with exit code {}\".format(sub_process.returncode,)\n-                self.logger.error(msg)\n+            ) as sub_process:\n+                lines = []\n+                line = None\n+                for raw_line in iter(sub_process.stdout.readline, b\"\"):\n+                    line = raw_line.decode(\"utf-8\").rstrip()\n+                    if self.return_all:\n+                        lines.append(line)\n+                    else:\n+                        # if we're returning all, we don't log every line\n+                        self.logger.debug(line)\n+                sub_process.wait()\n+                if sub_process.returncode:\n+                    msg = \"Command failed with exit code {}\".format(\n+                        sub_process.returncode,\n+                    )\n+                    self.logger.error(msg)\n \n-                if self.log_stderr:\n-                    self.logger.error(\"\\n\".join(lines))\n+                    if self.log_stderr:\n+                        self.logger.error(\"\\n\".join(lines))\n \n-                raise prefect.engine.signals.FAIL(msg) from None  # type: ignore\n+                    raise prefect.engine.signals.FAIL(msg) from None  # type: ignore\n         if self.return_all:\n             return lines\n         else:\n", "issue": "`unittest` framework raises `ResourceWarning`\n## Description\r\n<!-- A clear description of the bug -->\r\nI'm using [`unittest`](https://docs.python.org/3/library/unittest.html) framework for testing purposes and then a task is running inside a `unittest.TestCase`, a `ResourceWarning` is raised. \r\n\r\n\r\n## Expected Behavior\r\n<!-- What did you expect to happen instead? -->\r\nRun testcase without raising any warnings.\r\n\r\n\r\n\r\n\r\n## Reproduction\r\n<!-- A minimal example that exhibits the behavior. -->\r\n\r\n`test.py` file:\r\n\r\n```python\r\nimport typing\r\nimport unittest\r\n\r\nfrom prefect.engine.task_runner import TaskRunner\r\nfrom prefect.tasks.shell import ShellTask\r\n\r\n\r\nclass ShellTestCase(unittest.TestCase):\r\n    def test_shell_command(self) -> typing.NoReturn:\r\n        runner = TaskRunner(ShellTask('ls'))\r\n        runner.run()\r\n        self.assertTrue(True)\r\n```\r\n\r\nexecution:\r\n\r\n```bash\r\n$ python -m unittest test.py \r\n[2020-07-20 16:13:20] INFO - prefect.TaskRunner | Task 'ShellTask': Starting task run...\r\n/home/psimakis/.local/share/virtualenvs/data-workflows-GfPV92cZ/lib/python3.7/site-packages/prefect/utilities/tasks.py:444: ResourceWarning: unclosed file <_io.BufferedReader name=9>\r\n  return run_method(self, *args, **kwargs)\r\nResourceWarning: Enable tracemalloc to get the object allocation traceback\r\n[2020-07-20 16:13:21] INFO - prefect.TaskRunner | Task 'ShellTask': finished task run for task with final state: 'Success'\r\n.\r\n----------------------------------------------------------------------\r\nRan 1 test in 0.026s\r\n```\r\n\r\n\r\n\r\n\r\n## Environment\r\n<!-- Any additional information about your environment\r\n\r\nOptionally run `prefect diagnostics` from the command line and paste the information here. -->\r\n```json\r\n{\r\n  \"config_overrides\": {},\r\n  \"env_vars\": [\r\n    \"PREFECT__CONTEXT__SECRETS__....\",\r\n    \"PREFECT__CONTEXT__SECRETS__....\"\r\n  ],\r\n  \"system_information\": {\r\n    \"platform\": \"Linux-5.3.0-28-generic-x86_64-with-debian-buster-sid\",\r\n    \"prefect_version\": \"0.12.3\",\r\n    \"python_version\": \"3.7.3\"\r\n  }\r\n}\r\n```\n", "before_files": [{"content": "import os\nimport tempfile\nfrom subprocess import PIPE, STDOUT, Popen\nfrom typing import Any\n\nimport prefect\nfrom prefect.utilities.tasks import defaults_from_attrs\n\n\nclass ShellTask(prefect.Task):\n    \"\"\"\n    Task for running arbitrary shell commands.\n\n    Args:\n        - command (string, optional): shell command to be executed; can also be\n            provided post-initialization by calling this task instance\n        - env (dict, optional): dictionary of environment variables to use for\n            the subprocess; can also be provided at runtime\n        - helper_script (str, optional): a string representing a shell script, which\n            will be executed prior to the `command` in the same process. Can be used to\n            change directories, define helper functions, etc. when re-using this Task\n            for different commands in a Flow\n        - shell (string, optional): shell to run the command with; defaults to \"bash\"\n        - return_all (bool, optional): boolean specifying whether this task\n            should return all lines of stdout as a list, or just the last line\n            as a string; defaults to `False`\n        - log_stderr (bool, optional): boolean specifying whether this task\n            should log the output from stderr in the case of a non-zero exit code;\n            defaults to `False`\n        - **kwargs: additional keyword arguments to pass to the Task constructor\n\n    Example:\n        ```python\n        from prefect import Flow\n        from prefect.tasks.shell import ShellTask\n\n        task = ShellTask(helper_script=\"cd ~\")\n        with Flow(\"My Flow\") as f:\n            # both tasks will be executed in home directory\n            contents = task(command='ls')\n            mv_file = task(command='mv .vimrc /.vimrc')\n\n        out = f.run()\n        ```\n    \"\"\"\n\n    def __init__(\n        self,\n        command: str = None,\n        env: dict = None,\n        helper_script: str = None,\n        shell: str = \"bash\",\n        return_all: bool = False,\n        log_stderr: bool = False,\n        **kwargs: Any\n    ):\n        self.command = command\n        self.env = env\n        self.helper_script = helper_script\n        self.shell = shell\n        self.return_all = return_all\n        self.log_stderr = log_stderr\n        super().__init__(**kwargs)\n\n    @defaults_from_attrs(\"command\", \"env\")\n    def run(self, command: str = None, env: dict = None) -> str:\n        \"\"\"\n        Run the shell command.\n\n        Args:\n            - command (string): shell command to be executed; can also be\n                provided at task initialization. Any variables / functions defined in\n                `self.helper_script` will be available in the same process this command\n                runs in\n            - env (dict, optional): dictionary of environment variables to use for\n                the subprocess\n\n        Returns:\n            - stdout (string): if `return_all` is `False` (the default), only\n                the last line of stdout is returned, otherwise all lines are\n                returned, which is useful for passing result of shell command\n                to other downstream tasks. If there is no output, `None` is\n                returned.\n\n        Raises:\n            - prefect.engine.signals.FAIL: if command has an exit code other\n                than 0\n        \"\"\"\n        if command is None:\n            raise TypeError(\"run() missing required argument: 'command'\")\n\n        current_env = os.environ.copy()\n        current_env.update(env or {})\n        with tempfile.NamedTemporaryFile(prefix=\"prefect-\") as tmp:\n            if self.helper_script:\n                tmp.write(self.helper_script.encode())\n                tmp.write(\"\\n\".encode())\n            tmp.write(command.encode())\n            tmp.flush()\n            sub_process = Popen(\n                [self.shell, tmp.name], stdout=PIPE, stderr=STDOUT, env=current_env\n            )\n            lines = []\n            line = None\n            for raw_line in iter(sub_process.stdout.readline, b\"\"):\n                line = raw_line.decode(\"utf-8\").rstrip()\n                if self.return_all:\n                    lines.append(line)\n                else:\n                    # if we're returning all, we don't log every line\n                    self.logger.debug(line)\n            sub_process.wait()\n            if sub_process.returncode:\n                msg = \"Command failed with exit code {}\".format(sub_process.returncode,)\n                self.logger.error(msg)\n\n                if self.log_stderr:\n                    self.logger.error(\"\\n\".join(lines))\n\n                raise prefect.engine.signals.FAIL(msg) from None  # type: ignore\n        if self.return_all:\n            return lines\n        else:\n            return line\n", "path": "src/prefect/tasks/shell.py"}], "after_files": [{"content": "import os\nimport tempfile\nfrom subprocess import PIPE, STDOUT, Popen\nfrom typing import Any\n\nimport prefect\nfrom prefect.utilities.tasks import defaults_from_attrs\n\n\nclass ShellTask(prefect.Task):\n    \"\"\"\n    Task for running arbitrary shell commands.\n\n    Args:\n        - command (string, optional): shell command to be executed; can also be\n            provided post-initialization by calling this task instance\n        - env (dict, optional): dictionary of environment variables to use for\n            the subprocess; can also be provided at runtime\n        - helper_script (str, optional): a string representing a shell script, which\n            will be executed prior to the `command` in the same process. Can be used to\n            change directories, define helper functions, etc. when re-using this Task\n            for different commands in a Flow\n        - shell (string, optional): shell to run the command with; defaults to \"bash\"\n        - return_all (bool, optional): boolean specifying whether this task\n            should return all lines of stdout as a list, or just the last line\n            as a string; defaults to `False`\n        - log_stderr (bool, optional): boolean specifying whether this task\n            should log the output from stderr in the case of a non-zero exit code;\n            defaults to `False`\n        - **kwargs: additional keyword arguments to pass to the Task constructor\n\n    Example:\n        ```python\n        from prefect import Flow\n        from prefect.tasks.shell import ShellTask\n\n        task = ShellTask(helper_script=\"cd ~\")\n        with Flow(\"My Flow\") as f:\n            # both tasks will be executed in home directory\n            contents = task(command='ls')\n            mv_file = task(command='mv .vimrc /.vimrc')\n\n        out = f.run()\n        ```\n    \"\"\"\n\n    def __init__(\n        self,\n        command: str = None,\n        env: dict = None,\n        helper_script: str = None,\n        shell: str = \"bash\",\n        return_all: bool = False,\n        log_stderr: bool = False,\n        **kwargs: Any\n    ):\n        self.command = command\n        self.env = env\n        self.helper_script = helper_script\n        self.shell = shell\n        self.return_all = return_all\n        self.log_stderr = log_stderr\n        super().__init__(**kwargs)\n\n    @defaults_from_attrs(\"command\", \"env\")\n    def run(self, command: str = None, env: dict = None) -> str:\n        \"\"\"\n        Run the shell command.\n\n        Args:\n            - command (string): shell command to be executed; can also be\n                provided at task initialization. Any variables / functions defined in\n                `self.helper_script` will be available in the same process this command\n                runs in\n            - env (dict, optional): dictionary of environment variables to use for\n                the subprocess\n\n        Returns:\n            - stdout (string): if `return_all` is `False` (the default), only\n                the last line of stdout is returned, otherwise all lines are\n                returned, which is useful for passing result of shell command\n                to other downstream tasks. If there is no output, `None` is\n                returned.\n\n        Raises:\n            - prefect.engine.signals.FAIL: if command has an exit code other\n                than 0\n        \"\"\"\n        if command is None:\n            raise TypeError(\"run() missing required argument: 'command'\")\n\n        current_env = os.environ.copy()\n        current_env.update(env or {})\n        with tempfile.NamedTemporaryFile(prefix=\"prefect-\") as tmp:\n            if self.helper_script:\n                tmp.write(self.helper_script.encode())\n                tmp.write(\"\\n\".encode())\n            tmp.write(command.encode())\n            tmp.flush()\n            with Popen(\n                [self.shell, tmp.name], stdout=PIPE, stderr=STDOUT, env=current_env\n            ) as sub_process:\n                lines = []\n                line = None\n                for raw_line in iter(sub_process.stdout.readline, b\"\"):\n                    line = raw_line.decode(\"utf-8\").rstrip()\n                    if self.return_all:\n                        lines.append(line)\n                    else:\n                        # if we're returning all, we don't log every line\n                        self.logger.debug(line)\n                sub_process.wait()\n                if sub_process.returncode:\n                    msg = \"Command failed with exit code {}\".format(\n                        sub_process.returncode,\n                    )\n                    self.logger.error(msg)\n\n                    if self.log_stderr:\n                        self.logger.error(\"\\n\".join(lines))\n\n                    raise prefect.engine.signals.FAIL(msg) from None  # type: ignore\n        if self.return_all:\n            return lines\n        else:\n            return line\n", "path": "src/prefect/tasks/shell.py"}]}