prompt: 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()
  
  ```
  
golden_diff: 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:
  
verification_info: {"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"}]}