84 lines
3.2 KiB
Python
84 lines
3.2 KiB
Python
"""Deciding whether a local copy is still the current one."""
|
|
|
|
import datetime
|
|
import os
|
|
|
|
import requests
|
|
|
|
from apworld_tester.core.model.instant import Instant
|
|
|
|
|
|
class LocalFile:
|
|
"""One file on disk, against what the remote currently offers.
|
|
|
|
"It exists" is not "it is current": an author can replace an
|
|
asset's content in place, same filename, same version. Two signals
|
|
catch that - the asset's own timestamp, and its size - and both are
|
|
needed. A replacement that happens to be the same length is
|
|
invisible to size alone, and a remote that reports no length at all
|
|
makes size no signal whatsoever.
|
|
"""
|
|
|
|
#: The statuses that mean the file has to be fetched. The other
|
|
#: two - "verified" and "assumed" - mean the copy on disk stands.
|
|
NEEDS_DOWNLOAD = ("missing", "stale")
|
|
|
|
def __init__(self, session, path):
|
|
self.session = session
|
|
self.path = path
|
|
|
|
def outdated(self, remote_updated_at):
|
|
"""Whether the remote was touched after this copy arrived.
|
|
|
|
mtime is when download() wrote this file, so a remote timestamp
|
|
later than it means the copy on disk predates what is being
|
|
offered now. Without this, a release that replaced its asset
|
|
with one of the same length was never fetched, and the recorded
|
|
test result went on describing content that no longer exists -
|
|
while the page showed the new release's date beside it.
|
|
"""
|
|
stamp = Instant.of(remote_updated_at)
|
|
if stamp is None:
|
|
return False
|
|
arrived = datetime.datetime.fromtimestamp(
|
|
os.path.getmtime(self.path), datetime.timezone.utc)
|
|
return arrived < stamp
|
|
|
|
def status(self, remote_size, remote_updated_at=None):
|
|
"""Whether this file is missing, stale, verified or assumed.
|
|
|
|
The timestamp is checked before the size, and before a missing
|
|
size falls through to "assumed": it is the stronger signal, and
|
|
the two cases it catches are exactly the ones size cannot.
|
|
"""
|
|
if not os.path.exists(self.path):
|
|
return "missing"
|
|
if self.outdated(remote_updated_at):
|
|
return "stale"
|
|
if remote_size is None:
|
|
return "assumed"
|
|
if os.path.getsize(self.path) == remote_size:
|
|
return "verified"
|
|
return "stale"
|
|
|
|
def remote_size(self, url, headers=None):
|
|
"""The size a HEAD request reports, or None."""
|
|
try:
|
|
response = self.session.head(url, headers=headers,
|
|
allow_redirects=True)
|
|
response.raise_for_status()
|
|
except requests.RequestException:
|
|
return None
|
|
length = response.headers.get("Content-Length")
|
|
return int(length) if length and length.isdigit() else None
|
|
|
|
def download(self, url, headers=None):
|
|
"""Stream a URL into this path, creating its directory."""
|
|
response = self.session.get(url, stream=True, headers=headers)
|
|
response.raise_for_status()
|
|
os.makedirs(os.path.dirname(self.path), exist_ok=True)
|
|
with open(self.path, "wb") as handle:
|
|
for chunk in response.iter_content(chunk_size=65536):
|
|
handle.write(chunk)
|
|
return self.path
|