"""Local stack integration checks; creates and retains clearly named test orders. Run: python3 -m tests.smoke_test. Standard library only. Honors .env/environment. """ from concurrent.futures import ThreadPoolExecutor import hashlib import http.cookiejar import json import os from pathlib import Path import time from urllib.error import HTTPError from urllib.request import build_opener, HTTPCookieProcessor, Request, urlopen from uuid import uuid4 if Path('.env').exists(): for line in Path('.env').read_text().splitlines(): if line.strip() and not line.startswith('#') and '=' in line: key,value=line.split('=',1) os.environ.setdefault(key,value) # CI runs these from a container on the stack's own network, because a runner # container cannot reach ports published on the host's loopback. SITE_BASE_URL # points at the gateway by service name; SITE_HOST_HEADER keeps the Host the # gateway and TrustedHostMiddleware expect, so the security configuration under # test stays identical to a developer's localhost run. BASE=os.environ.get('SITE_BASE_URL') or 'http://localhost:'+os.environ.get('SITE_PORT','8080') HOST_HEADER=os.environ.get('SITE_HOST_HEADER') def with_host(headers=None): headers=dict(headers or {}) if HOST_HEADER and not any(k.lower()=='host' for k in headers): headers['Host']=HOST_HEADER return headers class Client: def __init__(self): self.jar=http.cookiejar.CookieJar() self.opener=build_opener(HTTPCookieProcessor(self.jar)) self.operator_client=None def call(self,path,body=None,operator=False,expected=200,method=None): headers=with_host({'Content-Type':'application/json'}) if operator: if self.operator_client is None: self.operator_client=Client() self.operator_client.call('/operator/login',{'email':os.environ.get('OPERATOR_EMAIL','operator@example.test'),'password':os.environ.get('OPERATOR_PASSWORD','local-operator-only')}) return self.operator_client.call(path,body,expected=expected,method=method) request=Request(BASE+'/api'+path,data=None if body is None else json.dumps(body).encode(), headers=headers,method=method) try: with self.opener.open(request,timeout=30) as response: assert response.status==expected,(path,response.status,expected) return json.load(response) except HTTPError as exc: if exc.code!=expected:raise AssertionError((path,exc.code,exc.read().decode())) from exc return json.load(exc) def wait_scan(client, uid, operator=False, expected='clean'): prefix='/operator/uploads' if operator else '/uploads' deadline=time.monotonic()+150 while time.monotonic()