flan
22e6499a8e
Move the authenticated drift check into the repository as tools/compat_auth.py, run daily by .github/workflows/compat-auth.yml on the maintainer's forge only. Credentials come from FH_EMAIL and FH_PASSWORD, the output is pass/fail labels only because the run log is public, and a failed run pushes the report to ntfy. Exit 5 means all hold, 10 drift, anything else that it could not run. Point compat.yml, tools/compat.py and the docs at it, add a README section, and cut 0.5.1.
217 lines
7.4 KiB
Python
217 lines
7.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Check what this integration assumes about freshharvest.com against the live site.
|
|
|
|
There is no API and no stability contract here — the integration reads HTML and
|
|
posts to form endpoints, and any redesign can silently change the meaning of a
|
|
value rather than breaking loudly. A sensor that quietly reports last week's
|
|
total is worse than one that goes unavailable, so this asserts the contract on a
|
|
schedule and fails when the site moves.
|
|
|
|
WHAT THIS CAN AND CANNOT SEE
|
|
----------------------------
|
|
Only the *unauthenticated* surface is checked here: the login handshake and the
|
|
Algolia catalogue, so it runs anywhere. The authenticated contract — dashboard
|
|
markup, cart add hashes, skip popups, subscribe forms — needs a real session.
|
|
tools/compat_auth.py checks that half, from a workflow that runs only on the
|
|
maintainer's forge, where the account credentials are; see README.
|
|
|
|
Exit code is the number of FAILED checks, so CI fails loudly on drift.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
BASE = "https://freshharvest.com"
|
|
UA = (
|
|
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) "
|
|
"Chrome/126.0 Safari/537.36"
|
|
)
|
|
|
|
OK, FAIL, WARN = "ok", "FAIL", "warn"
|
|
|
|
|
|
def fetch(url: str, data: bytes | None = None, headers: dict | None = None) -> str:
|
|
req = urllib.request.Request(
|
|
url, data=data, headers={"User-Agent": UA, **(headers or {})}
|
|
)
|
|
with urllib.request.urlopen(req, timeout=45) as resp:
|
|
return resp.read().decode("utf-8", "replace")
|
|
|
|
|
|
class Checks:
|
|
"""Each check returns (status, detail). Assumption text mirrors the code."""
|
|
|
|
def __init__(self) -> None:
|
|
self.results: list[tuple[str, str, str, str]] = []
|
|
|
|
def record(self, area, assumption, status, detail):
|
|
self.results.append((area, assumption, status, detail))
|
|
|
|
# -- login handshake ----------------------------------------------------
|
|
|
|
def login_form(self):
|
|
area = "Login"
|
|
try:
|
|
html = fetch(f"{BASE}/s/popup/login")
|
|
except urllib.error.URLError as err:
|
|
self.record(area, "login popup reachable", FAIL, str(err))
|
|
return
|
|
self.record(area, "`/s/popup/login` serves the form", OK, f"{len(html)} bytes")
|
|
|
|
for field in ("LoginSecurity", "SubmitToken"):
|
|
found = re.search(rf"name='{field}'[^>]*value='([^']+)'", html)
|
|
self.record(
|
|
area,
|
|
f"hidden `{field}` is minted",
|
|
OK if found else FAIL,
|
|
f"{len(found.group(1))} chars" if found else "absent — login will break",
|
|
)
|
|
act = re.search(r"action='([^']*submit/login)'", html)
|
|
self.record(
|
|
area,
|
|
"posts to `/s/submit/login`",
|
|
OK if act else FAIL,
|
|
act.group(1) if act else "form action changed",
|
|
)
|
|
for field in ("LoginEmail", "LoginPassword"):
|
|
self.record(
|
|
area,
|
|
f"field `{field}` present",
|
|
OK if f"name='{field}'" in html else FAIL,
|
|
"",
|
|
)
|
|
|
|
# -- catalogue ----------------------------------------------------------
|
|
|
|
def algolia(self):
|
|
area = "Catalogue"
|
|
try:
|
|
js = fetch(f"{BASE}/_home/JavaScript/search_algolia.js")
|
|
except urllib.error.URLError as err:
|
|
self.record(area, "search_algolia.js reachable", FAIL, str(err))
|
|
return
|
|
|
|
creds = re.search(
|
|
r'algoliasearch\(\s*"([^"]+)"\s*,\s*"([^"]+)"', js, re.S
|
|
)
|
|
index = re.search(r'indexName:\s*"([^"]+)"', js)
|
|
self.record(
|
|
area,
|
|
"Algolia credentials readable from site JS",
|
|
OK if creds else FAIL,
|
|
"app id + search key found" if creds else "pattern changed",
|
|
)
|
|
self.record(
|
|
area,
|
|
"index name readable",
|
|
OK if index else FAIL,
|
|
index.group(1) if index else "not found",
|
|
)
|
|
if not (creds and index):
|
|
return
|
|
|
|
app, key = creds.groups()
|
|
try:
|
|
body = fetch(
|
|
f"https://{app}-dsn.algolia.net/1/indexes/{index.group(1)}/query",
|
|
data=json.dumps({"params": "query=&hitsPerPage=1"}).encode(),
|
|
headers={
|
|
"X-Algolia-API-Key": key,
|
|
"X-Algolia-Application-Id": app,
|
|
"Content-Type": "application/json",
|
|
},
|
|
)
|
|
except urllib.error.URLError as err:
|
|
self.record(area, "index queryable", FAIL, str(err))
|
|
return
|
|
|
|
data = json.loads(body)
|
|
total = data.get("nbHits", 0)
|
|
# A collapse to near-zero means the index moved or emptied; the exact
|
|
# count drifts constantly as stock changes, so only the floor is checked.
|
|
self.record(
|
|
area,
|
|
"index returns a plausible catalogue",
|
|
OK if total > 100 else FAIL,
|
|
f"{total} records",
|
|
)
|
|
hit = (data.get("hits") or [{}])[0]
|
|
for fieldname in ("ID", "Name", "Price", "Measurement", "Categories"):
|
|
self.record(
|
|
area,
|
|
f"record field `{fieldname}`",
|
|
OK if fieldname in hit else FAIL,
|
|
"present" if fieldname in hit else "MISSING — parser reads this",
|
|
)
|
|
|
|
# -- action endpoints ---------------------------------------------------
|
|
|
|
def endpoints(self):
|
|
"""The write endpoints must still exist; they are never *called* here."""
|
|
area = "Endpoints"
|
|
try:
|
|
js = fetch(f"{BASE}/_home/Ajax/order-manage.js")
|
|
except urllib.error.URLError as err:
|
|
self.record(area, "order-manage.js reachable", FAIL, str(err))
|
|
return
|
|
pattern = "/p/Ajax/order-manage/"
|
|
self.record(
|
|
area,
|
|
"cart add/remove URL shape unchanged",
|
|
OK if pattern in js else FAIL,
|
|
pattern if pattern in js else "URL construction changed",
|
|
)
|
|
try:
|
|
popups = fetch(f"{BASE}/_home/JavaScript/popups.js")
|
|
except urllib.error.URLError as err:
|
|
self.record(area, "popups.js reachable", FAIL, str(err))
|
|
return
|
|
self.record(
|
|
area,
|
|
"popup route is `/x/popup/{type}/{token}`",
|
|
OK if "/x/popup/" in popups else FAIL,
|
|
"found" if "/x/popup/" in popups else "route changed",
|
|
)
|
|
|
|
def run(self):
|
|
self.login_form()
|
|
self.algolia()
|
|
self.endpoints()
|
|
return self.results
|
|
|
|
|
|
def render(results) -> str:
|
|
lines = [
|
|
"| Area | Assumption | Status | Detail |",
|
|
"| --- | --- | --- | --- |",
|
|
]
|
|
icon = {OK: "✅", FAIL: "❌", WARN: "⚠️"}
|
|
for area, assumption, status, detail in results:
|
|
lines.append(f"| {area} | {assumption} | {icon[status]} | {detail} |")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def main(argv):
|
|
results = Checks().run()
|
|
table = render(results)
|
|
failed = [r for r in results if r[2] == FAIL]
|
|
|
|
if "--markdown" in argv:
|
|
print(table)
|
|
else:
|
|
print(table)
|
|
print()
|
|
print(f"{len(results) - len(failed)}/{len(results)} checks passed")
|
|
for area, assumption, _, detail in failed:
|
|
print(f" FAIL {area}: {assumption} — {detail}")
|
|
return len(failed)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv[1:]))
|