mirror of
https://github.com/Tria-plc/edr-platform.git
synced 2026-08-26 18:42:49 +00:00
505 lines
18 KiB
Python
505 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Seed freight contracts across every flow variant, driven end-to-end over HTTP.
|
|
|
|
For each flow the script logs in, creates a DRAFT contract, and pushes it through
|
|
the lifecycle up to and INCLUDING both signatures (customer sign + staff
|
|
counter-sign). It STOPS after the staff counter-sign — no clearance, no booking.
|
|
|
|
Flow dimensions (3 x 2 x 2 x 2 = 24 combos, but only the 20 real ones are built):
|
|
movement : intercity(DOMESTIC) | import(IMPORT) | export(EXPORT)
|
|
kind : one-time(ONE_TIME) | general(GENERAL)
|
|
customs : without | with (customsClearingEnabled)
|
|
freight : bulk(BULK) | container(CONTAINER)
|
|
|
|
intercity + customs is dropped (DOMESTIC ignores customs → no real combo), which
|
|
removes 4 dead combos and leaves 20 flows (16 working + 4 customs+bulk whose
|
|
break is downstream in clearance). Each is created twice → 40 contracts on `all`.
|
|
|
|
CLI (filter by movement, pass one or many):
|
|
python create_contracts.py # all 20 flows
|
|
python create_contracts.py all # all 20 flows
|
|
python create_contracts.py intercity # only DOMESTIC flows
|
|
python create_contracts.py import # only IMPORT flows
|
|
python create_contracts.py import export # IMPORT + EXPORT flows
|
|
|
|
Config comes from .env (see .env.example). Requires: requests, psycopg,
|
|
python-dotenv (see requirements.txt).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import os
|
|
import sys
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import psycopg
|
|
import requests
|
|
from dotenv import load_dotenv
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
load_dotenv(HERE / ".env")
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Config
|
|
# --------------------------------------------------------------------------- #
|
|
API_URL = os.getenv("FREIGHT_API_URL", "http://localhost:3001/api").rstrip("/")
|
|
|
|
CUSTOMER_EMAIL = os.getenv("CUSTOMER_EMAIL", "")
|
|
CUSTOMER_PASSWORD = os.getenv("CUSTOMER_PASSWORD", "")
|
|
ADMIN_EMAIL = os.getenv("ADMIN_EMAIL", "")
|
|
ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "")
|
|
|
|
# Phone the sign-OTP is sent to and read back from Postgres. Resolved at runtime
|
|
# from the customer's own IAM profile (GET /api/me → phoneNumber). OTP_PHONE is an
|
|
# optional override / fallback used only when the customer has no phone on file.
|
|
# The sign endpoint keys the OTP purely on this number, so it just has to be the
|
|
# same value for "send" and "read".
|
|
OTP_PHONE = os.getenv("OTP_PHONE", "")
|
|
OTP_PHONE_FALLBACK = os.getenv("OTP_PHONE_FALLBACK", "251900000000")
|
|
|
|
# DB connection used ONLY to read the plaintext sign-OTP from freight.otp_verifications.
|
|
DB_HOST = os.getenv("DB_HOST", "localhost")
|
|
DB_PORT = os.getenv("DB_PORT", "5432")
|
|
DB_NAME = os.getenv("DB_NAME", "edr_dev")
|
|
DB_USER = os.getenv("DB_USER", "postgres")
|
|
DB_PASSWORD = os.getenv("DB_PASSWORD", "")
|
|
DB_SCHEMA = os.getenv("DB_SCHEMA", "freight")
|
|
|
|
WAAFI = HERE / "waafi.jpeg"
|
|
|
|
VALIDITY_DAYS = int(os.getenv("VALIDITY_DAYS", "365"))
|
|
CONTRACTS_PER_FLOW = int(os.getenv("CONTRACTS_PER_FLOW", "2"))
|
|
REQUEST_TIMEOUT = int(os.getenv("REQUEST_TIMEOUT", "60"))
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Flow matrix — the 20 real flows
|
|
# --------------------------------------------------------------------------- #
|
|
@dataclass(frozen=True)
|
|
class Flow:
|
|
movement: str # intercity | import | export
|
|
trade_direction: str # DOMESTIC | IMPORT | EXPORT
|
|
kind: str # ONE_TIME | GENERAL
|
|
customs: bool # customsClearingEnabled
|
|
freight: str # BULK | CONTAINER
|
|
|
|
@property
|
|
def label(self) -> str:
|
|
return (
|
|
f"{self.movement}+{'general' if self.kind == 'GENERAL' else 'one-time'}"
|
|
f"+{'with' if self.customs else 'no'}-customs"
|
|
f"+{self.freight.lower()}"
|
|
)
|
|
|
|
|
|
def build_flow_matrix() -> list[Flow]:
|
|
movements = [
|
|
("intercity", "DOMESTIC"),
|
|
("import", "IMPORT"),
|
|
("export", "EXPORT"),
|
|
]
|
|
kinds = ["ONE_TIME", "GENERAL"]
|
|
freights = ["BULK", "CONTAINER"]
|
|
|
|
flows: list[Flow] = []
|
|
for movement, direction in movements:
|
|
# DOMESTIC ignores customs (no clearance gate) → customs=True is not a
|
|
# real combo. Only build without-customs for intercity.
|
|
customs_options = [False] if direction == "DOMESTIC" else [False, True]
|
|
for kind in kinds:
|
|
for customs in customs_options:
|
|
for freight in freights:
|
|
flows.append(Flow(movement, direction, kind, customs, freight))
|
|
return flows
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# HTTP client
|
|
# --------------------------------------------------------------------------- #
|
|
class ApiError(RuntimeError):
|
|
def __init__(self, method: str, path: str, resp: requests.Response):
|
|
body = resp.text
|
|
try:
|
|
body = resp.json()
|
|
except Exception:
|
|
pass
|
|
super().__init__(f"{method} {path} -> {resp.status_code}: {body}")
|
|
self.status_code = resp.status_code
|
|
|
|
|
|
class Client:
|
|
"""Thin wrapper that carries a bearer token."""
|
|
|
|
def __init__(self, name: str, token: str | None = None):
|
|
self.name = name
|
|
self.token = token
|
|
|
|
def _headers(self, extra: dict[str, str] | None = None) -> dict[str, str]:
|
|
h: dict[str, str] = {}
|
|
if self.token:
|
|
h["Authorization"] = f"Bearer {self.token}"
|
|
if extra:
|
|
h.update(extra)
|
|
return h
|
|
|
|
def get(self, path: str, params: dict | None = None) -> Any:
|
|
r = requests.get(
|
|
f"{API_URL}{path}",
|
|
headers=self._headers(),
|
|
params=params,
|
|
timeout=REQUEST_TIMEOUT,
|
|
)
|
|
if not r.ok:
|
|
raise ApiError("GET", path, r)
|
|
return r.json() if r.content else None
|
|
|
|
def post_json(self, path: str, body: dict | None = None) -> Any:
|
|
r = requests.post(
|
|
f"{API_URL}{path}",
|
|
headers=self._headers({"Content-Type": "application/json"}),
|
|
json=body or {},
|
|
timeout=REQUEST_TIMEOUT,
|
|
)
|
|
if not r.ok:
|
|
raise ApiError("POST", path, r)
|
|
return r.json() if r.content else None
|
|
|
|
def post_multipart(
|
|
self, path: str, data: dict[str, str], files: list[tuple] | None = None
|
|
) -> Any:
|
|
r = requests.post(
|
|
f"{API_URL}{path}",
|
|
headers=self._headers(), # requests sets multipart Content-Type
|
|
data=data,
|
|
files=files or [],
|
|
timeout=REQUEST_TIMEOUT,
|
|
)
|
|
if not r.ok:
|
|
raise ApiError("POST", path, r)
|
|
return r.json() if r.content else None
|
|
|
|
|
|
def login(email: str, password: str, who: str) -> Client:
|
|
r = requests.post(
|
|
f"{API_URL}/auth/login",
|
|
json={"email": email, "password": password},
|
|
timeout=REQUEST_TIMEOUT,
|
|
)
|
|
if not r.ok:
|
|
raise ApiError("POST", "/auth/login", r)
|
|
payload = r.json()
|
|
if payload.get("mfaRequired"):
|
|
raise RuntimeError(
|
|
f"{who} login requires MFA — this script cannot complete an MFA login. "
|
|
"Disable MFA for the seed account or supply a non-MFA account."
|
|
)
|
|
token = payload.get("token")
|
|
if not token:
|
|
raise RuntimeError(f"{who} login returned no token: {payload}")
|
|
return Client(who, token)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# OTP — send + read from Postgres
|
|
# --------------------------------------------------------------------------- #
|
|
def resolve_otp_phone(customer: Client) -> str:
|
|
"""Phone the sign-OTP is sent to. Prefer the customer's own IAM profile phone
|
|
(GET /api/me → phoneNumber); fall back to OTP_PHONE, then OTP_PHONE_FALLBACK.
|
|
The value only has to be consistent between send + DB read."""
|
|
phone = ""
|
|
try:
|
|
me = customer.get("/me") or {}
|
|
phone = (me.get("phoneNumber") or "").strip()
|
|
except Exception:
|
|
pass
|
|
phone = phone or OTP_PHONE or OTP_PHONE_FALLBACK
|
|
if not phone:
|
|
raise RuntimeError(
|
|
"Could not resolve an OTP phone (customer has none, and neither "
|
|
"OTP_PHONE nor OTP_PHONE_FALLBACK is set)."
|
|
)
|
|
return phone
|
|
|
|
|
|
def send_otp(customer: Client, phone: str) -> None:
|
|
# POST /api/otp/send is @Public — no token needed, but sending one is harmless.
|
|
customer.post_json("/otp/send", {"phone": phone})
|
|
|
|
|
|
def read_otp_from_db(phone: str) -> str:
|
|
"""Read the freshest plaintext OTP for `phone` from freight.otp_verifications."""
|
|
dsn = (
|
|
f"host={DB_HOST} port={DB_PORT} dbname={DB_NAME} "
|
|
f"user={DB_USER} password={DB_PASSWORD}"
|
|
)
|
|
with psycopg.connect(dsn) as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
f'SELECT otp FROM "{DB_SCHEMA}".otp_verifications '
|
|
"WHERE phone = %s ORDER BY updated_at DESC LIMIT 1",
|
|
(phone,),
|
|
)
|
|
row = cur.fetchone()
|
|
if not row:
|
|
raise RuntimeError(f"No OTP row found for phone {phone} in {DB_SCHEMA}.otp_verifications")
|
|
return str(row[0])
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Reference-data lookups (yards / service types / cargo types)
|
|
# --------------------------------------------------------------------------- #
|
|
@dataclass
|
|
class RefData:
|
|
yards: list[dict] = field(default_factory=list)
|
|
service_types: list[dict] = field(default_factory=list)
|
|
cargo_types: list[dict] = field(default_factory=list)
|
|
|
|
|
|
def _as_items(resp: Any) -> list[dict]:
|
|
if isinstance(resp, list):
|
|
return resp
|
|
if isinstance(resp, dict):
|
|
return resp.get("items") or resp.get("data") or []
|
|
return []
|
|
|
|
|
|
def load_ref_data(client: Client) -> RefData:
|
|
ref = RefData(
|
|
yards=_as_items(client.get("/yards")),
|
|
service_types=_as_items(client.get("/service-types")),
|
|
cargo_types=_as_items(client.get("/cargo-types")),
|
|
)
|
|
if len(ref.yards) < 2:
|
|
raise RuntimeError(f"Need >=2 yards, got {len(ref.yards)}. Seed yards first.")
|
|
if not ref.service_types:
|
|
raise RuntimeError("No service types found. Seed service types first.")
|
|
if not ref.cargo_types:
|
|
raise RuntimeError("No cargo types found. Seed cargo types first.")
|
|
return ref
|
|
|
|
|
|
def pick_service_type(ref: RefData, wants_customs: bool) -> str:
|
|
"""Prefer a service type whose includesCustoms matches the flow's customs need."""
|
|
for st in ref.service_types:
|
|
if bool(st.get("includesCustoms")) == wants_customs:
|
|
return st["id"]
|
|
# Fall back to any — customsClearingEnabled on the contract still drives the flow.
|
|
return ref.service_types[0]["id"]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Contract payload builder
|
|
# --------------------------------------------------------------------------- #
|
|
def build_create_payload(flow: Flow, ref: RefData, idx: int) -> dict[str, str]:
|
|
"""Return multipart form fields. Booleans as 'true'/'false' strings; nested
|
|
arrays as JSON strings (implicit conversion is off in the API)."""
|
|
import json
|
|
|
|
origin = ref.yards[0]["id"]
|
|
destination = ref.yards[1]["id"]
|
|
service_type_id = pick_service_type(ref, flow.customs)
|
|
|
|
# Cargo scope: CONTAINER -> >=1 size row; BULK -> exactly one cargo-type row.
|
|
if flow.freight == "CONTAINER":
|
|
cargo_scope = [{"containerSize": "20ft"}]
|
|
if flow.kind == "GENERAL":
|
|
cargo_scope[0]["quantityCap"] = 10
|
|
else: # BULK
|
|
cargo_scope = [{"cargoTypeId": ref.cargo_types[0]["id"]}]
|
|
if flow.kind == "GENERAL":
|
|
cargo_scope[0]["quantityCap"] = 1000
|
|
|
|
# Routes: ONE_TIME -> exactly 1; GENERAL -> 1..N (one is fine).
|
|
routes = [{"originYardId": origin, "destinationYardId": destination, "sortOrder": 0}]
|
|
|
|
fields: dict[str, str] = {
|
|
"contractKind": flow.kind,
|
|
"tradeDirection": flow.trade_direction,
|
|
"freightType": flow.freight,
|
|
"serviceTypeId": service_type_id,
|
|
"paymentCurrency": "ETB",
|
|
"customsClearingEnabled": "true" if flow.customs else "false",
|
|
"contractType": "SPOT",
|
|
"cargoScope": json.dumps(cargo_scope),
|
|
"routes": json.dumps(routes),
|
|
}
|
|
if flow.customs:
|
|
fields["customsClearingAgent"] = "Seed Agent"
|
|
return fields
|
|
|
|
|
|
def signature_b64() -> str:
|
|
return base64.b64encode(WAAFI.read_bytes()).decode()
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Lifecycle driver — create → submit → accept → approve → generate → sign x2
|
|
# --------------------------------------------------------------------------- #
|
|
def waafi_file_tuple(field_name: str) -> tuple:
|
|
return (field_name, (WAAFI.name, WAAFI.read_bytes(), "image/jpeg"))
|
|
|
|
|
|
def drive_flow(
|
|
flow: Flow, idx: int, customer: Client, admin: Client, ref: RefData, otp_phone: str
|
|
) -> dict[str, Any]:
|
|
result: dict[str, Any] = {"flow": flow.label, "n": idx, "status": None}
|
|
|
|
# S1 — create (customer, multipart, waafi attached as intake doc)
|
|
fields = build_create_payload(flow, ref, idx)
|
|
contract = customer.post_multipart(
|
|
"/contracts", data=fields, files=[waafi_file_tuple("intake_document")]
|
|
)
|
|
cid = contract["id"]
|
|
result["contractId"] = cid
|
|
result["reference"] = contract.get("reference")
|
|
|
|
# S2 — submit (customer). May go to PRICE_CHANGED_PENDING_CONFIRM → confirm.
|
|
contract = customer.post_json(f"/contracts/{cid}/submit")
|
|
if (contract or {}).get("status") == "PRICE_CHANGED_PENDING_CONFIRM":
|
|
contract = customer.post_json(f"/contracts/{cid}/confirm-submit")
|
|
|
|
# S3 — staff accept (admin) → PENDING_APPROVAL + approval chain
|
|
admin.post_json(f"/contracts/{cid}/staff/accept", {"validityDays": VALIDITY_DAYS})
|
|
|
|
# S4 — approve every pending step IN ORDER with its exact requiredRole (admin)
|
|
approve_all_steps(admin, cid)
|
|
|
|
# S5 — generate contract document (admin) → CONTRACT_READY
|
|
admin.post_json(f"/contracts/{cid}/contract/generate")
|
|
|
|
# S6 — customer sign (needs OTP) → SIGNED_CUSTOMER
|
|
send_otp(customer, otp_phone)
|
|
time.sleep(1.0) # let the OTP row land
|
|
otp = read_otp_from_db(otp_phone)
|
|
customer.post_json(
|
|
f"/contracts/{cid}/contract/sign",
|
|
{
|
|
"role": "CUSTOMER",
|
|
"signatureImageBase64": signature_b64(),
|
|
"signerDisplayName": "Seed Customer",
|
|
"consentText": "I agree.",
|
|
"otp": otp,
|
|
"otpPhone": otp_phone,
|
|
},
|
|
)
|
|
|
|
# S7 — staff counter-sign (admin) → FULLY_EXECUTED / CONTRACT_ACTIVE /
|
|
# AWAITING_CLEARANCE_DOCUMENTS depending on dimension. STOP HERE.
|
|
signed = admin.post_json(
|
|
f"/contracts/{cid}/contract/sign",
|
|
{
|
|
"role": "STAFF",
|
|
"signatureImageBase64": signature_b64(),
|
|
"signerDisplayName": "Seed Staff",
|
|
"consentText": "Countersigned.",
|
|
},
|
|
)
|
|
result["status"] = (signed or {}).get("status")
|
|
return result
|
|
|
|
|
|
def approve_all_steps(admin: Client, cid: str) -> None:
|
|
"""Read the contract, approve each PENDING approval step in order. Superadmin
|
|
can approve any role, but the endpoint still checks step.requiredRole == body,
|
|
so we echo the step's own requiredRole back."""
|
|
guard = 0
|
|
while True:
|
|
guard += 1
|
|
if guard > 12:
|
|
raise RuntimeError(f"Approval loop exceeded 12 iterations for {cid}")
|
|
contract = admin.get(f"/contracts/{cid}")
|
|
steps = contract.get("approvalSteps") or []
|
|
pending = [s for s in steps if s.get("status") == "PENDING"]
|
|
if not pending:
|
|
return
|
|
# findNextPendingApprovalStep orders by sequence; sort the same way.
|
|
pending.sort(key=lambda s: s.get("sequence", s.get("sortOrder", 0)))
|
|
step = pending[0]
|
|
admin.post_json(
|
|
f"/contracts/{cid}/approval-steps/{step['id']}/approve",
|
|
{"requiredRole": step["requiredRole"]},
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Main
|
|
# --------------------------------------------------------------------------- #
|
|
VALID_FILTERS = {"all", "intercity", "import", "export"}
|
|
|
|
|
|
def parse_filters(argv: list[str]) -> set[str]:
|
|
args = [a.lower() for a in argv[1:]]
|
|
if not args or "all" in args:
|
|
return {"intercity", "import", "export"}
|
|
unknown = set(args) - VALID_FILTERS
|
|
if unknown:
|
|
raise SystemExit(
|
|
f"Unknown filter(s): {', '.join(sorted(unknown))}. "
|
|
f"Valid: {', '.join(sorted(VALID_FILTERS))}"
|
|
)
|
|
return set(args)
|
|
|
|
|
|
def require_config() -> None:
|
|
missing = [
|
|
name
|
|
for name, val in [
|
|
("CUSTOMER_EMAIL", CUSTOMER_EMAIL),
|
|
("CUSTOMER_PASSWORD", CUSTOMER_PASSWORD),
|
|
("ADMIN_EMAIL", ADMIN_EMAIL),
|
|
("ADMIN_PASSWORD", ADMIN_PASSWORD),
|
|
]
|
|
if not val
|
|
]
|
|
if missing:
|
|
raise SystemExit(f"Missing required .env keys: {', '.join(missing)}")
|
|
if not WAAFI.exists():
|
|
raise SystemExit(f"Missing signature/upload image: {WAAFI}")
|
|
|
|
|
|
def main() -> None:
|
|
require_config()
|
|
wanted = parse_filters(sys.argv)
|
|
|
|
flows = [f for f in build_flow_matrix() if f.movement in wanted]
|
|
total = len(flows) * CONTRACTS_PER_FLOW
|
|
print(f"API : {API_URL}")
|
|
print(f"Filters : {', '.join(sorted(wanted))}")
|
|
print(f"Flows : {len(flows)} x {CONTRACTS_PER_FLOW} = {total} contracts\n")
|
|
|
|
print("Logging in...")
|
|
customer = login(CUSTOMER_EMAIL, CUSTOMER_PASSWORD, "customer")
|
|
admin = login(ADMIN_EMAIL, ADMIN_PASSWORD, "admin")
|
|
|
|
otp_phone = resolve_otp_phone(customer)
|
|
print(f"OTP phone: {otp_phone}")
|
|
|
|
print("Loading reference data...")
|
|
ref = load_ref_data(admin)
|
|
|
|
results: list[dict] = []
|
|
for flow in flows:
|
|
for n in range(1, CONTRACTS_PER_FLOW + 1):
|
|
tag = f"[{flow.label} #{n}]"
|
|
try:
|
|
res = drive_flow(flow, n, customer, admin, ref, otp_phone)
|
|
print(f" OK {tag} {res['reference']} -> {res['status']}")
|
|
results.append(res)
|
|
except Exception as exc: # noqa: BLE001 — report and continue
|
|
print(f" FAIL {tag} {exc}")
|
|
results.append({"flow": flow.label, "n": n, "error": str(exc)})
|
|
|
|
ok = [r for r in results if not r.get("error")]
|
|
bad = [r for r in results if r.get("error")]
|
|
print(f"\nDone. {len(ok)} created, {len(bad)} failed, {total} attempted.")
|
|
if bad:
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|