zwitschi
b1a6df9f90
- Introduced Pydantic schemas for profitability calculations in `schemas/calculations.py`. - Implemented service functions for profitability calculations in `services/calculations.py`. - Added new exception class `ProfitabilityValidationError` for handling validation errors. - Created repositories for managing project and scenario profitability snapshots. - Developed a utility script for verifying authenticated routes. - Added a new HTML template for the profitability calculator interface. - Implemented a script to fix user ID sequence in the database.
113 lines
3.4 KiB
Python
113 lines
3.4 KiB
Python
"""Utility script to verify key authenticated routes respond without errors."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.parse
|
|
from http.client import HTTPConnection
|
|
from http.cookies import SimpleCookie
|
|
from typing import Dict, List, Tuple
|
|
|
|
HOST = "127.0.0.1"
|
|
PORT = 8000
|
|
|
|
cookies: Dict[str, str] = {}
|
|
|
|
|
|
def _update_cookies(headers: List[Tuple[str, str]]) -> None:
|
|
for name, value in headers:
|
|
if name.lower() != "set-cookie":
|
|
continue
|
|
cookie = SimpleCookie()
|
|
cookie.load(value)
|
|
for key, morsel in cookie.items():
|
|
cookies[key] = morsel.value
|
|
|
|
|
|
def _cookie_header() -> str | None:
|
|
if not cookies:
|
|
return None
|
|
return "; ".join(f"{key}={value}" for key, value in cookies.items())
|
|
|
|
|
|
def request(method: str, path: str, *, body: bytes | None = None, headers: Dict[str, str] | None = None) -> Tuple[int, Dict[str, str], bytes]:
|
|
conn = HTTPConnection(HOST, PORT, timeout=10)
|
|
prepared_headers = {"User-Agent": "route-checker"}
|
|
if headers:
|
|
prepared_headers.update(headers)
|
|
cookie_header = _cookie_header()
|
|
if cookie_header:
|
|
prepared_headers["Cookie"] = cookie_header
|
|
|
|
conn.request(method, path, body=body, headers=prepared_headers)
|
|
resp = conn.getresponse()
|
|
payload = resp.read()
|
|
status = resp.status
|
|
reason = resp.reason
|
|
response_headers = {name: value for name, value in resp.getheaders()}
|
|
_update_cookies(list(resp.getheaders()))
|
|
conn.close()
|
|
print(f"{method} {path} -> {status} {reason}")
|
|
return status, response_headers, payload
|
|
|
|
|
|
def main() -> int:
|
|
status, _, _ = request("GET", "/login")
|
|
if status != 200:
|
|
print("Unexpected status for GET /login", file=sys.stderr)
|
|
return 1
|
|
|
|
admin_username = os.getenv("CALMINER_SEED_ADMIN_USERNAME", "admin")
|
|
admin_password = os.getenv("CALMINER_SEED_ADMIN_PASSWORD", "M11ffpgm.")
|
|
login_payload = urllib.parse.urlencode(
|
|
{"username": admin_username, "password": admin_password}
|
|
).encode()
|
|
status, headers, _ = request(
|
|
"POST",
|
|
"/login",
|
|
body=login_payload,
|
|
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
|
)
|
|
if status not in {200, 303}:
|
|
print("Login failed", file=sys.stderr)
|
|
return 1
|
|
|
|
location = headers.get("Location", "/")
|
|
redirect_path = urllib.parse.urlsplit(location).path or "/"
|
|
request("GET", redirect_path)
|
|
|
|
request("GET", "/")
|
|
request("GET", "/projects/ui")
|
|
|
|
status, headers, body = request(
|
|
"GET",
|
|
"/projects",
|
|
headers={"Accept": "application/json"},
|
|
)
|
|
projects: List[dict] = []
|
|
if headers.get("Content-Type", "").startswith("application/json"):
|
|
projects = json.loads(body.decode())
|
|
|
|
if projects:
|
|
project_id = projects[0]["id"]
|
|
request("GET", f"/projects/{project_id}/view")
|
|
status, headers, body = request(
|
|
"GET",
|
|
f"/projects/{project_id}/scenarios",
|
|
headers={"Accept": "application/json"},
|
|
)
|
|
scenarios: List[dict] = []
|
|
if headers.get("Content-Type", "").startswith("application/json"):
|
|
scenarios = json.loads(body.decode())
|
|
if scenarios:
|
|
scenario_id = scenarios[0]["id"]
|
|
request("GET", f"/scenarios/{scenario_id}/view")
|
|
|
|
print("Cookies:", cookies)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|