We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Don't use file locking to protect selfcheck state file
**What's the problem this feature will solve?**

There are several issues around file locking that have been filed over the years, specifically related to:

1. Underlying OS/filesystem does not support hardlinks as used by the file lock (#2993, #5322, #6761)
2. Lingering lock files and/or lock files in an inconsistent state can cause pip to hang when attempting to acquire the lock (some of #3532, #5034)
3. lockfile uses hostname when creating its unique name, which can result in invalid paths when hostname includes a `/` (#6938)

**Describe the solution you'd like**

1. Write a selfcheck state file per-prefix, to remove the need to read and then write the file within a lock
2. Write the file atomically (write to a separate tmp file and then move into place) to avoid partial writes if the process is killed

This will satisfy the linked issues and help us progress on #4766 to remove lockfile entirely.

**Alternative Solutions**

1. Switch to `MkdirLockFile` as currently used in the HTTP cache - the downside of this approach is that it is not backwards-compatible, so we would need to use a separate file to track the information for modern pip versions. If we would need to use a separate file anyway, we might as well go one step further to progress #4766.

**Additional context**

* PR #6855 - writes per-prefix selfcheck state files
* PR #6879 - removes file locking

--- 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/pip/_internal/utils/outdated.py`
Content:
```
1 from __future__ import absolute_import
2 
3 import datetime
4 import hashlib
5 import json
6 import logging
7 import os.path
8 import sys
9 
10 from pip._vendor import lockfile, pkg_resources
11 from pip._vendor.packaging import version as packaging_version
12 from pip._vendor.six import ensure_binary
13 
14 from pip._internal.cli.cmdoptions import make_search_scope
15 from pip._internal.index import PackageFinder
16 from pip._internal.models.selection_prefs import SelectionPreferences
17 from pip._internal.utils.compat import WINDOWS
18 from pip._internal.utils.filesystem import check_path_owner
19 from pip._internal.utils.misc import ensure_dir, get_installed_version
20 from pip._internal.utils.packaging import get_installer
21 from pip._internal.utils.typing import MYPY_CHECK_RUNNING
22 
23 if MYPY_CHECK_RUNNING:
24     import optparse
25     from typing import Any, Dict, Text, Union
26     from pip._internal.download import PipSession
27 
28 
29 SELFCHECK_DATE_FMT = "%Y-%m-%dT%H:%M:%SZ"
30 
31 
32 logger = logging.getLogger(__name__)
33 
34 
35 def _get_statefile_name(key):
36     # type: (Union[str, Text]) -> str
37     key_bytes = ensure_binary(key)
38     name = hashlib.sha224(key_bytes).hexdigest()
39     return name
40 
41 
42 class SelfCheckState(object):
43     def __init__(self, cache_dir):
44         # type: (str) -> None
45         self.state = {}  # type: Dict[str, Any]
46         self.statefile_path = None
47 
48         # Try to load the existing state
49         if cache_dir:
50             self.statefile_path = os.path.join(
51                 cache_dir, "selfcheck", _get_statefile_name(self.key)
52             )
53             try:
54                 with open(self.statefile_path) as statefile:
55                     self.state = json.load(statefile)
56             except (IOError, ValueError, KeyError):
57                 # Explicitly suppressing exceptions, since we don't want to
58                 # error out if the cache file is invalid.
59                 pass
60 
61     @property
62     def key(self):
63         return sys.prefix
64 
65     def save(self, pypi_version, current_time):
66         # type: (str, datetime.datetime) -> None
67         # If we do not have a path to cache in, don't bother saving.
68         if not self.statefile_path:
69             return
70 
71         # Check to make sure that we own the directory
72         if not check_path_owner(os.path.dirname(self.statefile_path)):
73             return
74 
75         # Now that we've ensured the directory is owned by this user, we'll go
76         # ahead and make sure that all our directories are created.
77         ensure_dir(os.path.dirname(self.statefile_path))
78 
79         state = {
80             # Include the key so it's easy to tell which pip wrote the
81             # file.
82             "key": self.key,
83             "last_check": current_time.strftime(SELFCHECK_DATE_FMT),
84             "pypi_version": pypi_version,
85         }
86 
87         text = json.dumps(state, sort_keys=True, separators=(",", ":"))
88 
89         # Attempt to write out our version check file
90         with lockfile.LockFile(self.statefile_path):
91             # Since we have a prefix-specific state file, we can just
92             # overwrite whatever is there, no need to check.
93             with open(self.statefile_path, "w") as statefile:
94                 statefile.write(text)
95 
96 
97 def was_installed_by_pip(pkg):
98     # type: (str) -> bool
99     """Checks whether pkg was installed by pip
100 
101     This is used not to display the upgrade message when pip is in fact
102     installed by system package manager, such as dnf on Fedora.
103     """
104     try:
105         dist = pkg_resources.get_distribution(pkg)
106         return "pip" == get_installer(dist)
107     except pkg_resources.DistributionNotFound:
108         return False
109 
110 
111 def pip_version_check(session, options):
112     # type: (PipSession, optparse.Values) -> None
113     """Check for an update for pip.
114 
115     Limit the frequency of checks to once per week. State is stored either in
116     the active virtualenv or in the user's USER_CACHE_DIR keyed off the prefix
117     of the pip script path.
118     """
119     installed_version = get_installed_version("pip")
120     if not installed_version:
121         return
122 
123     pip_version = packaging_version.parse(installed_version)
124     pypi_version = None
125 
126     try:
127         state = SelfCheckState(cache_dir=options.cache_dir)
128 
129         current_time = datetime.datetime.utcnow()
130         # Determine if we need to refresh the state
131         if "last_check" in state.state and "pypi_version" in state.state:
132             last_check = datetime.datetime.strptime(
133                 state.state["last_check"],
134                 SELFCHECK_DATE_FMT
135             )
136             if (current_time - last_check).total_seconds() < 7 * 24 * 60 * 60:
137                 pypi_version = state.state["pypi_version"]
138 
139         # Refresh the version if we need to or just see if we need to warn
140         if pypi_version is None:
141             # Lets use PackageFinder to see what the latest pip version is
142             search_scope = make_search_scope(options, suppress_no_index=True)
143 
144             # Pass allow_yanked=False so we don't suggest upgrading to a
145             # yanked version.
146             selection_prefs = SelectionPreferences(
147                 allow_yanked=False,
148                 allow_all_prereleases=False,  # Explicitly set to False
149             )
150 
151             finder = PackageFinder.create(
152                 search_scope=search_scope,
153                 selection_prefs=selection_prefs,
154                 session=session,
155             )
156             best_candidate = finder.find_best_candidate("pip").best_candidate
157             if best_candidate is None:
158                 return
159             pypi_version = str(best_candidate.version)
160 
161             # save that we've performed a check
162             state.save(pypi_version, current_time)
163 
164         remote_version = packaging_version.parse(pypi_version)
165 
166         local_version_is_older = (
167             pip_version < remote_version and
168             pip_version.base_version != remote_version.base_version and
169             was_installed_by_pip('pip')
170         )
171 
172         # Determine if our pypi_version is older
173         if not local_version_is_older:
174             return
175 
176         # Advise "python -m pip" on Windows to avoid issues
177         # with overwriting pip.exe.
178         if WINDOWS:
179             pip_cmd = "python -m pip"
180         else:
181             pip_cmd = "pip"
182         logger.warning(
183             "You are using pip version %s, however version %s is "
184             "available.\nYou should consider upgrading via the "
185             "'%s install --upgrade pip' command.",
186             pip_version, pypi_version, pip_cmd
187         )
188     except Exception:
189         logger.debug(
190             "There was an error checking the latest version of pip",
191             exc_info=True,
192         )
193
```

Path: `src/pip/_internal/utils/filesystem.py`
Content:
```
1 import os
2 import os.path
3 import shutil
4 import stat
5 
6 from pip._internal.utils.compat import get_path_uid
7 
8 
9 def check_path_owner(path):
10     # type: (str) -> bool
11     # If we don't have a way to check the effective uid of this process, then
12     # we'll just assume that we own the directory.
13     if not hasattr(os, "geteuid"):
14         return True
15 
16     previous = None
17     while path != previous:
18         if os.path.lexists(path):
19             # Check if path is writable by current user.
20             if os.geteuid() == 0:
21                 # Special handling for root user in order to handle properly
22                 # cases where users use sudo without -H flag.
23                 try:
24                     path_uid = get_path_uid(path)
25                 except OSError:
26                     return False
27                 return path_uid == 0
28             else:
29                 return os.access(path, os.W_OK)
30         else:
31             previous, path = path, os.path.dirname(path)
32     return False  # assume we don't own the path
33 
34 
35 def copy2_fixed(src, dest):
36     # type: (str, str) -> None
37     """Wrap shutil.copy2() but map errors copying socket files to
38     SpecialFileError as expected.
39 
40     See also https://bugs.python.org/issue37700.
41     """
42     try:
43         shutil.copy2(src, dest)
44     except (OSError, IOError):
45         for f in [src, dest]:
46             try:
47                 is_socket_file = is_socket(f)
48             except OSError:
49                 # An error has already occurred. Another error here is not
50                 # a problem and we can ignore it.
51                 pass
52             else:
53                 if is_socket_file:
54                     raise shutil.SpecialFileError("`%s` is a socket" % f)
55 
56         raise
57 
58 
59 def is_socket(path):
60     # type: (str) -> bool
61     return stat.S_ISSOCK(os.lstat(path).st_mode)
62
```
--- 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/pip/_internal/utils/filesystem.py b/src/pip/_internal/utils/filesystem.py
--- a/src/pip/_internal/utils/filesystem.py
+++ b/src/pip/_internal/utils/filesystem.py
@@ -2,8 +2,26 @@
 import os.path
 import shutil
 import stat
+from contextlib import contextmanager
+from tempfile import NamedTemporaryFile
+
+# NOTE: retrying is not annotated in typeshed as on 2017-07-17, which is
+#       why we ignore the type on this import.
+from pip._vendor.retrying import retry  # type: ignore
+from pip._vendor.six import PY2
 
 from pip._internal.utils.compat import get_path_uid
+from pip._internal.utils.misc import cast
+from pip._internal.utils.typing import MYPY_CHECK_RUNNING
+
+if MYPY_CHECK_RUNNING:
+    from typing import BinaryIO, Iterator
+
+    class NamedTemporaryFileResult(BinaryIO):
+        @property
+        def file(self):
+            # type: () -> BinaryIO
+            pass
 
 
 def check_path_owner(path):
@@ -59,3 +77,39 @@
 def is_socket(path):
     # type: (str) -> bool
     return stat.S_ISSOCK(os.lstat(path).st_mode)
+
+
+@contextmanager
+def adjacent_tmp_file(path):
+    # type: (str) -> Iterator[NamedTemporaryFileResult]
+    """Given a path to a file, open a temp file next to it securely and ensure
+    it is written to disk after the context reaches its end.
+    """
+    with NamedTemporaryFile(
+        delete=False,
+        dir=os.path.dirname(path),
+        prefix=os.path.basename(path),
+        suffix='.tmp',
+    ) as f:
+        result = cast('NamedTemporaryFileResult', f)
+        try:
+            yield result
+        finally:
+            result.file.flush()
+            os.fsync(result.file.fileno())
+
+
+_replace_retry = retry(stop_max_delay=1000, wait_fixed=250)
+
+if PY2:
+    @_replace_retry
+    def replace(src, dest):
+        # type: (str, str) -> None
+        try:
+            os.rename(src, dest)
+        except OSError:
+            os.remove(dest)
+            os.rename(src, dest)
+
+else:
+    replace = _replace_retry(os.replace)
diff --git a/src/pip/_internal/utils/outdated.py b/src/pip/_internal/utils/outdated.py
--- a/src/pip/_internal/utils/outdated.py
+++ b/src/pip/_internal/utils/outdated.py
@@ -7,7 +7,7 @@
 import os.path
 import sys
 
-from pip._vendor import lockfile, pkg_resources
+from pip._vendor import pkg_resources
 from pip._vendor.packaging import version as packaging_version
 from pip._vendor.six import ensure_binary
 
@@ -15,7 +15,11 @@
 from pip._internal.index import PackageFinder
 from pip._internal.models.selection_prefs import SelectionPreferences
 from pip._internal.utils.compat import WINDOWS
-from pip._internal.utils.filesystem import check_path_owner
+from pip._internal.utils.filesystem import (
+    adjacent_tmp_file,
+    check_path_owner,
+    replace,
+)
 from pip._internal.utils.misc import ensure_dir, get_installed_version
 from pip._internal.utils.packaging import get_installer
 from pip._internal.utils.typing import MYPY_CHECK_RUNNING
@@ -86,12 +90,16 @@
 
         text = json.dumps(state, sort_keys=True, separators=(",", ":"))
 
-        # Attempt to write out our version check file
-        with lockfile.LockFile(self.statefile_path):
+        with adjacent_tmp_file(self.statefile_path) as f:
+            f.write(ensure_binary(text))
+
+        try:
             # Since we have a prefix-specific state file, we can just
             # overwrite whatever is there, no need to check.
-            with open(self.statefile_path, "w") as statefile:
-                statefile.write(text)
+            replace(f.name, self.statefile_path)
+        except OSError:
+            # Best effort.
+            pass
 
 
 def was_installed_by_pip(pkg):

{"golden_diff": "diff --git a/src/pip/_internal/utils/filesystem.py b/src/pip/_internal/utils/filesystem.py\n--- a/src/pip/_internal/utils/filesystem.py\n+++ b/src/pip/_internal/utils/filesystem.py\n@@ -2,8 +2,26 @@\n import os.path\n import shutil\n import stat\n+from contextlib import contextmanager\n+from tempfile import NamedTemporaryFile\n+\n+# NOTE: retrying is not annotated in typeshed as on 2017-07-17, which is\n+#       why we ignore the type on this import.\n+from pip._vendor.retrying import retry  # type: ignore\n+from pip._vendor.six import PY2\n \n from pip._internal.utils.compat import get_path_uid\n+from pip._internal.utils.misc import cast\n+from pip._internal.utils.typing import MYPY_CHECK_RUNNING\n+\n+if MYPY_CHECK_RUNNING:\n+    from typing import BinaryIO, Iterator\n+\n+    class NamedTemporaryFileResult(BinaryIO):\n+        @property\n+        def file(self):\n+            # type: () -> BinaryIO\n+            pass\n \n \n def check_path_owner(path):\n@@ -59,3 +77,39 @@\n def is_socket(path):\n     # type: (str) -> bool\n     return stat.S_ISSOCK(os.lstat(path).st_mode)\n+\n+\n+@contextmanager\n+def adjacent_tmp_file(path):\n+    # type: (str) -> Iterator[NamedTemporaryFileResult]\n+    \"\"\"Given a path to a file, open a temp file next to it securely and ensure\n+    it is written to disk after the context reaches its end.\n+    \"\"\"\n+    with NamedTemporaryFile(\n+        delete=False,\n+        dir=os.path.dirname(path),\n+        prefix=os.path.basename(path),\n+        suffix='.tmp',\n+    ) as f:\n+        result = cast('NamedTemporaryFileResult', f)\n+        try:\n+            yield result\n+        finally:\n+            result.file.flush()\n+            os.fsync(result.file.fileno())\n+\n+\n+_replace_retry = retry(stop_max_delay=1000, wait_fixed=250)\n+\n+if PY2:\n+    @_replace_retry\n+    def replace(src, dest):\n+        # type: (str, str) -> None\n+        try:\n+            os.rename(src, dest)\n+        except OSError:\n+            os.remove(dest)\n+            os.rename(src, dest)\n+\n+else:\n+    replace = _replace_retry(os.replace)\ndiff --git a/src/pip/_internal/utils/outdated.py b/src/pip/_internal/utils/outdated.py\n--- a/src/pip/_internal/utils/outdated.py\n+++ b/src/pip/_internal/utils/outdated.py\n@@ -7,7 +7,7 @@\n import os.path\n import sys\n \n-from pip._vendor import lockfile, pkg_resources\n+from pip._vendor import pkg_resources\n from pip._vendor.packaging import version as packaging_version\n from pip._vendor.six import ensure_binary\n \n@@ -15,7 +15,11 @@\n from pip._internal.index import PackageFinder\n from pip._internal.models.selection_prefs import SelectionPreferences\n from pip._internal.utils.compat import WINDOWS\n-from pip._internal.utils.filesystem import check_path_owner\n+from pip._internal.utils.filesystem import (\n+    adjacent_tmp_file,\n+    check_path_owner,\n+    replace,\n+)\n from pip._internal.utils.misc import ensure_dir, get_installed_version\n from pip._internal.utils.packaging import get_installer\n from pip._internal.utils.typing import MYPY_CHECK_RUNNING\n@@ -86,12 +90,16 @@\n \n         text = json.dumps(state, sort_keys=True, separators=(\",\", \":\"))\n \n-        # Attempt to write out our version check file\n-        with lockfile.LockFile(self.statefile_path):\n+        with adjacent_tmp_file(self.statefile_path) as f:\n+            f.write(ensure_binary(text))\n+\n+        try:\n             # Since we have a prefix-specific state file, we can just\n             # overwrite whatever is there, no need to check.\n-            with open(self.statefile_path, \"w\") as statefile:\n-                statefile.write(text)\n+            replace(f.name, self.statefile_path)\n+        except OSError:\n+            # Best effort.\n+            pass\n \n \n def was_installed_by_pip(pkg):\n", "issue": "Don't use file locking to protect selfcheck state file\n**What's the problem this feature will solve?**\r\n\r\nThere are several issues around file locking that have been filed over the years, specifically related to:\r\n\r\n1. Underlying OS/filesystem does not support hardlinks as used by the file lock (#2993, #5322, #6761)\r\n2. Lingering lock files and/or lock files in an inconsistent state can cause pip to hang when attempting to acquire the lock (some of #3532, #5034)\r\n3. lockfile uses hostname when creating its unique name, which can result in invalid paths when hostname includes a `/` (#6938)\r\n\r\n**Describe the solution you'd like**\r\n\r\n1. Write a selfcheck state file per-prefix, to remove the need to read and then write the file within a lock\r\n2. Write the file atomically (write to a separate tmp file and then move into place) to avoid partial writes if the process is killed\r\n\r\nThis will satisfy the linked issues and help us progress on #4766 to remove lockfile entirely.\r\n\r\n**Alternative Solutions**\r\n\r\n1. Switch to `MkdirLockFile` as currently used in the HTTP cache - the downside of this approach is that it is not backwards-compatible, so we would need to use a separate file to track the information for modern pip versions. If we would need to use a separate file anyway, we might as well go one step further to progress #4766.\r\n\r\n**Additional context**\r\n\r\n* PR #6855 - writes per-prefix selfcheck state files\r\n* PR #6879 - removes file locking\n", "before_files": [{"content": "from __future__ import absolute_import\n\nimport datetime\nimport hashlib\nimport json\nimport logging\nimport os.path\nimport sys\n\nfrom pip._vendor import lockfile, pkg_resources\nfrom pip._vendor.packaging import version as packaging_version\nfrom pip._vendor.six import ensure_binary\n\nfrom pip._internal.cli.cmdoptions import make_search_scope\nfrom pip._internal.index import PackageFinder\nfrom pip._internal.models.selection_prefs import SelectionPreferences\nfrom pip._internal.utils.compat import WINDOWS\nfrom pip._internal.utils.filesystem import check_path_owner\nfrom pip._internal.utils.misc import ensure_dir, get_installed_version\nfrom pip._internal.utils.packaging import get_installer\nfrom pip._internal.utils.typing import MYPY_CHECK_RUNNING\n\nif MYPY_CHECK_RUNNING:\n    import optparse\n    from typing import Any, Dict, Text, Union\n    from pip._internal.download import PipSession\n\n\nSELFCHECK_DATE_FMT = \"%Y-%m-%dT%H:%M:%SZ\"\n\n\nlogger = logging.getLogger(__name__)\n\n\ndef _get_statefile_name(key):\n    # type: (Union[str, Text]) -> str\n    key_bytes = ensure_binary(key)\n    name = hashlib.sha224(key_bytes).hexdigest()\n    return name\n\n\nclass SelfCheckState(object):\n    def __init__(self, cache_dir):\n        # type: (str) -> None\n        self.state = {}  # type: Dict[str, Any]\n        self.statefile_path = None\n\n        # Try to load the existing state\n        if cache_dir:\n            self.statefile_path = os.path.join(\n                cache_dir, \"selfcheck\", _get_statefile_name(self.key)\n            )\n            try:\n                with open(self.statefile_path) as statefile:\n                    self.state = json.load(statefile)\n            except (IOError, ValueError, KeyError):\n                # Explicitly suppressing exceptions, since we don't want to\n                # error out if the cache file is invalid.\n                pass\n\n    @property\n    def key(self):\n        return sys.prefix\n\n    def save(self, pypi_version, current_time):\n        # type: (str, datetime.datetime) -> None\n        # If we do not have a path to cache in, don't bother saving.\n        if not self.statefile_path:\n            return\n\n        # Check to make sure that we own the directory\n        if not check_path_owner(os.path.dirname(self.statefile_path)):\n            return\n\n        # Now that we've ensured the directory is owned by this user, we'll go\n        # ahead and make sure that all our directories are created.\n        ensure_dir(os.path.dirname(self.statefile_path))\n\n        state = {\n            # Include the key so it's easy to tell which pip wrote the\n            # file.\n            \"key\": self.key,\n            \"last_check\": current_time.strftime(SELFCHECK_DATE_FMT),\n            \"pypi_version\": pypi_version,\n        }\n\n        text = json.dumps(state, sort_keys=True, separators=(\",\", \":\"))\n\n        # Attempt to write out our version check file\n        with lockfile.LockFile(self.statefile_path):\n            # Since we have a prefix-specific state file, we can just\n            # overwrite whatever is there, no need to check.\n            with open(self.statefile_path, \"w\") as statefile:\n                statefile.write(text)\n\n\ndef was_installed_by_pip(pkg):\n    # type: (str) -> bool\n    \"\"\"Checks whether pkg was installed by pip\n\n    This is used not to display the upgrade message when pip is in fact\n    installed by system package manager, such as dnf on Fedora.\n    \"\"\"\n    try:\n        dist = pkg_resources.get_distribution(pkg)\n        return \"pip\" == get_installer(dist)\n    except pkg_resources.DistributionNotFound:\n        return False\n\n\ndef pip_version_check(session, options):\n    # type: (PipSession, optparse.Values) -> None\n    \"\"\"Check for an update for pip.\n\n    Limit the frequency of checks to once per week. State is stored either in\n    the active virtualenv or in the user's USER_CACHE_DIR keyed off the prefix\n    of the pip script path.\n    \"\"\"\n    installed_version = get_installed_version(\"pip\")\n    if not installed_version:\n        return\n\n    pip_version = packaging_version.parse(installed_version)\n    pypi_version = None\n\n    try:\n        state = SelfCheckState(cache_dir=options.cache_dir)\n\n        current_time = datetime.datetime.utcnow()\n        # Determine if we need to refresh the state\n        if \"last_check\" in state.state and \"pypi_version\" in state.state:\n            last_check = datetime.datetime.strptime(\n                state.state[\"last_check\"],\n                SELFCHECK_DATE_FMT\n            )\n            if (current_time - last_check).total_seconds() < 7 * 24 * 60 * 60:\n                pypi_version = state.state[\"pypi_version\"]\n\n        # Refresh the version if we need to or just see if we need to warn\n        if pypi_version is None:\n            # Lets use PackageFinder to see what the latest pip version is\n            search_scope = make_search_scope(options, suppress_no_index=True)\n\n            # Pass allow_yanked=False so we don't suggest upgrading to a\n            # yanked version.\n            selection_prefs = SelectionPreferences(\n                allow_yanked=False,\n                allow_all_prereleases=False,  # Explicitly set to False\n            )\n\n            finder = PackageFinder.create(\n                search_scope=search_scope,\n                selection_prefs=selection_prefs,\n                session=session,\n            )\n            best_candidate = finder.find_best_candidate(\"pip\").best_candidate\n            if best_candidate is None:\n                return\n            pypi_version = str(best_candidate.version)\n\n            # save that we've performed a check\n            state.save(pypi_version, current_time)\n\n        remote_version = packaging_version.parse(pypi_version)\n\n        local_version_is_older = (\n            pip_version < remote_version and\n            pip_version.base_version != remote_version.base_version and\n            was_installed_by_pip('pip')\n        )\n\n        # Determine if our pypi_version is older\n        if not local_version_is_older:\n            return\n\n        # Advise \"python -m pip\" on Windows to avoid issues\n        # with overwriting pip.exe.\n        if WINDOWS:\n            pip_cmd = \"python -m pip\"\n        else:\n            pip_cmd = \"pip\"\n        logger.warning(\n            \"You are using pip version %s, however version %s is \"\n            \"available.\\nYou should consider upgrading via the \"\n            \"'%s install --upgrade pip' command.\",\n            pip_version, pypi_version, pip_cmd\n        )\n    except Exception:\n        logger.debug(\n            \"There was an error checking the latest version of pip\",\n            exc_info=True,\n        )\n", "path": "src/pip/_internal/utils/outdated.py"}, {"content": "import os\nimport os.path\nimport shutil\nimport stat\n\nfrom pip._internal.utils.compat import get_path_uid\n\n\ndef check_path_owner(path):\n    # type: (str) -> bool\n    # If we don't have a way to check the effective uid of this process, then\n    # we'll just assume that we own the directory.\n    if not hasattr(os, \"geteuid\"):\n        return True\n\n    previous = None\n    while path != previous:\n        if os.path.lexists(path):\n            # Check if path is writable by current user.\n            if os.geteuid() == 0:\n                # Special handling for root user in order to handle properly\n                # cases where users use sudo without -H flag.\n                try:\n                    path_uid = get_path_uid(path)\n                except OSError:\n                    return False\n                return path_uid == 0\n            else:\n                return os.access(path, os.W_OK)\n        else:\n            previous, path = path, os.path.dirname(path)\n    return False  # assume we don't own the path\n\n\ndef copy2_fixed(src, dest):\n    # type: (str, str) -> None\n    \"\"\"Wrap shutil.copy2() but map errors copying socket files to\n    SpecialFileError as expected.\n\n    See also https://bugs.python.org/issue37700.\n    \"\"\"\n    try:\n        shutil.copy2(src, dest)\n    except (OSError, IOError):\n        for f in [src, dest]:\n            try:\n                is_socket_file = is_socket(f)\n            except OSError:\n                # An error has already occurred. Another error here is not\n                # a problem and we can ignore it.\n                pass\n            else:\n                if is_socket_file:\n                    raise shutil.SpecialFileError(\"`%s` is a socket\" % f)\n\n        raise\n\n\ndef is_socket(path):\n    # type: (str) -> bool\n    return stat.S_ISSOCK(os.lstat(path).st_mode)\n", "path": "src/pip/_internal/utils/filesystem.py"}], "after_files": [{"content": "from __future__ import absolute_import\n\nimport datetime\nimport hashlib\nimport json\nimport logging\nimport os.path\nimport sys\n\nfrom pip._vendor import pkg_resources\nfrom pip._vendor.packaging import version as packaging_version\nfrom pip._vendor.six import ensure_binary\n\nfrom pip._internal.cli.cmdoptions import make_search_scope\nfrom pip._internal.index import PackageFinder\nfrom pip._internal.models.selection_prefs import SelectionPreferences\nfrom pip._internal.utils.compat import WINDOWS\nfrom pip._internal.utils.filesystem import (\n    adjacent_tmp_file,\n    check_path_owner,\n    replace,\n)\nfrom pip._internal.utils.misc import ensure_dir, get_installed_version\nfrom pip._internal.utils.packaging import get_installer\nfrom pip._internal.utils.typing import MYPY_CHECK_RUNNING\n\nif MYPY_CHECK_RUNNING:\n    import optparse\n    from typing import Any, Dict, Text, Union\n    from pip._internal.download import PipSession\n\n\nSELFCHECK_DATE_FMT = \"%Y-%m-%dT%H:%M:%SZ\"\n\n\nlogger = logging.getLogger(__name__)\n\n\ndef _get_statefile_name(key):\n    # type: (Union[str, Text]) -> str\n    key_bytes = ensure_binary(key)\n    name = hashlib.sha224(key_bytes).hexdigest()\n    return name\n\n\nclass SelfCheckState(object):\n    def __init__(self, cache_dir):\n        # type: (str) -> None\n        self.state = {}  # type: Dict[str, Any]\n        self.statefile_path = None\n\n        # Try to load the existing state\n        if cache_dir:\n            self.statefile_path = os.path.join(\n                cache_dir, \"selfcheck\", _get_statefile_name(self.key)\n            )\n            try:\n                with open(self.statefile_path) as statefile:\n                    self.state = json.load(statefile)\n            except (IOError, ValueError, KeyError):\n                # Explicitly suppressing exceptions, since we don't want to\n                # error out if the cache file is invalid.\n                pass\n\n    @property\n    def key(self):\n        return sys.prefix\n\n    def save(self, pypi_version, current_time):\n        # type: (str, datetime.datetime) -> None\n        # If we do not have a path to cache in, don't bother saving.\n        if not self.statefile_path:\n            return\n\n        # Check to make sure that we own the directory\n        if not check_path_owner(os.path.dirname(self.statefile_path)):\n            return\n\n        # Now that we've ensured the directory is owned by this user, we'll go\n        # ahead and make sure that all our directories are created.\n        ensure_dir(os.path.dirname(self.statefile_path))\n\n        state = {\n            # Include the key so it's easy to tell which pip wrote the\n            # file.\n            \"key\": self.key,\n            \"last_check\": current_time.strftime(SELFCHECK_DATE_FMT),\n            \"pypi_version\": pypi_version,\n        }\n\n        text = json.dumps(state, sort_keys=True, separators=(\",\", \":\"))\n\n        with adjacent_tmp_file(self.statefile_path) as f:\n            f.write(ensure_binary(text))\n\n        try:\n            # Since we have a prefix-specific state file, we can just\n            # overwrite whatever is there, no need to check.\n            replace(f.name, self.statefile_path)\n        except OSError:\n            # Best effort.\n            pass\n\n\ndef was_installed_by_pip(pkg):\n    # type: (str) -> bool\n    \"\"\"Checks whether pkg was installed by pip\n\n    This is used not to display the upgrade message when pip is in fact\n    installed by system package manager, such as dnf on Fedora.\n    \"\"\"\n    try:\n        dist = pkg_resources.get_distribution(pkg)\n        return \"pip\" == get_installer(dist)\n    except pkg_resources.DistributionNotFound:\n        return False\n\n\ndef pip_version_check(session, options):\n    # type: (PipSession, optparse.Values) -> None\n    \"\"\"Check for an update for pip.\n\n    Limit the frequency of checks to once per week. State is stored either in\n    the active virtualenv or in the user's USER_CACHE_DIR keyed off the prefix\n    of the pip script path.\n    \"\"\"\n    installed_version = get_installed_version(\"pip\")\n    if not installed_version:\n        return\n\n    pip_version = packaging_version.parse(installed_version)\n    pypi_version = None\n\n    try:\n        state = SelfCheckState(cache_dir=options.cache_dir)\n\n        current_time = datetime.datetime.utcnow()\n        # Determine if we need to refresh the state\n        if \"last_check\" in state.state and \"pypi_version\" in state.state:\n            last_check = datetime.datetime.strptime(\n                state.state[\"last_check\"],\n                SELFCHECK_DATE_FMT\n            )\n            if (current_time - last_check).total_seconds() < 7 * 24 * 60 * 60:\n                pypi_version = state.state[\"pypi_version\"]\n\n        # Refresh the version if we need to or just see if we need to warn\n        if pypi_version is None:\n            # Lets use PackageFinder to see what the latest pip version is\n            search_scope = make_search_scope(options, suppress_no_index=True)\n\n            # Pass allow_yanked=False so we don't suggest upgrading to a\n            # yanked version.\n            selection_prefs = SelectionPreferences(\n                allow_yanked=False,\n                allow_all_prereleases=False,  # Explicitly set to False\n            )\n\n            finder = PackageFinder.create(\n                search_scope=search_scope,\n                selection_prefs=selection_prefs,\n                session=session,\n            )\n            best_candidate = finder.find_best_candidate(\"pip\").best_candidate\n            if best_candidate is None:\n                return\n            pypi_version = str(best_candidate.version)\n\n            # save that we've performed a check\n            state.save(pypi_version, current_time)\n\n        remote_version = packaging_version.parse(pypi_version)\n\n        local_version_is_older = (\n            pip_version < remote_version and\n            pip_version.base_version != remote_version.base_version and\n            was_installed_by_pip('pip')\n        )\n\n        # Determine if our pypi_version is older\n        if not local_version_is_older:\n            return\n\n        # Advise \"python -m pip\" on Windows to avoid issues\n        # with overwriting pip.exe.\n        if WINDOWS:\n            pip_cmd = \"python -m pip\"\n        else:\n            pip_cmd = \"pip\"\n        logger.warning(\n            \"You are using pip version %s, however version %s is \"\n            \"available.\\nYou should consider upgrading via the \"\n            \"'%s install --upgrade pip' command.\",\n            pip_version, pypi_version, pip_cmd\n        )\n    except Exception:\n        logger.debug(\n            \"There was an error checking the latest version of pip\",\n            exc_info=True,\n        )\n", "path": "src/pip/_internal/utils/outdated.py"}, {"content": "import os\nimport os.path\nimport shutil\nimport stat\nfrom contextlib import contextmanager\nfrom tempfile import NamedTemporaryFile\n\n# NOTE: retrying is not annotated in typeshed as on 2017-07-17, which is\n#       why we ignore the type on this import.\nfrom pip._vendor.retrying import retry  # type: ignore\nfrom pip._vendor.six import PY2\n\nfrom pip._internal.utils.compat import get_path_uid\nfrom pip._internal.utils.misc import cast\nfrom pip._internal.utils.typing import MYPY_CHECK_RUNNING\n\nif MYPY_CHECK_RUNNING:\n    from typing import BinaryIO, Iterator\n\n    class NamedTemporaryFileResult(BinaryIO):\n        @property\n        def file(self):\n            # type: () -> BinaryIO\n            pass\n\n\ndef check_path_owner(path):\n    # type: (str) -> bool\n    # If we don't have a way to check the effective uid of this process, then\n    # we'll just assume that we own the directory.\n    if not hasattr(os, \"geteuid\"):\n        return True\n\n    previous = None\n    while path != previous:\n        if os.path.lexists(path):\n            # Check if path is writable by current user.\n            if os.geteuid() == 0:\n                # Special handling for root user in order to handle properly\n                # cases where users use sudo without -H flag.\n                try:\n                    path_uid = get_path_uid(path)\n                except OSError:\n                    return False\n                return path_uid == 0\n            else:\n                return os.access(path, os.W_OK)\n        else:\n            previous, path = path, os.path.dirname(path)\n    return False  # assume we don't own the path\n\n\ndef copy2_fixed(src, dest):\n    # type: (str, str) -> None\n    \"\"\"Wrap shutil.copy2() but map errors copying socket files to\n    SpecialFileError as expected.\n\n    See also https://bugs.python.org/issue37700.\n    \"\"\"\n    try:\n        shutil.copy2(src, dest)\n    except (OSError, IOError):\n        for f in [src, dest]:\n            try:\n                is_socket_file = is_socket(f)\n            except OSError:\n                # An error has already occurred. Another error here is not\n                # a problem and we can ignore it.\n                pass\n            else:\n                if is_socket_file:\n                    raise shutil.SpecialFileError(\"`%s` is a socket\" % f)\n\n        raise\n\n\ndef is_socket(path):\n    # type: (str) -> bool\n    return stat.S_ISSOCK(os.lstat(path).st_mode)\n\n\n@contextmanager\ndef adjacent_tmp_file(path):\n    # type: (str) -> Iterator[NamedTemporaryFileResult]\n    \"\"\"Given a path to a file, open a temp file next to it securely and ensure\n    it is written to disk after the context reaches its end.\n    \"\"\"\n    with NamedTemporaryFile(\n        delete=False,\n        dir=os.path.dirname(path),\n        prefix=os.path.basename(path),\n        suffix='.tmp',\n    ) as f:\n        result = cast('NamedTemporaryFileResult', f)\n        try:\n            yield result\n        finally:\n            result.file.flush()\n            os.fsync(result.file.fileno())\n\n\n_replace_retry = retry(stop_max_delay=1000, wait_fixed=250)\n\nif PY2:\n    @_replace_retry\n    def replace(src, dest):\n        # type: (str, str) -> None\n        try:\n            os.rename(src, dest)\n        except OSError:\n            os.remove(dest)\n            os.rename(src, dest)\n\nelse:\n    replace = _replace_retry(os.replace)\n", "path": "src/pip/_internal/utils/filesystem.py"}]}