"""HTTP helpers for public black-box tests; Python 3.9+, standard library only.""" import base64 import datetime as dt import hashlib import json import os from pathlib import Path import re import ssl import time import urllib.error import urllib.request import uuid ROOT = Path(__file__).resolve().parents[1] LAB = json.loads((ROOT / 'lab.json').read_text())['number'] SPEC = json.loads((ROOT / 'contracts/openapi.json').read_text()) class Failure(AssertionError): pass def check(condition, message): if not condition: raise Failure(message) def timestamp(value): check(isinstance(value, str) and value.endswith('Z'), 'timestamp must be UTC ending in Z') try: return dt.datetime.fromisoformat(value[:-1] + '+00:00') except ValueError as exc: raise Failure('invalid RFC3339 timestamp') from exc def validate(value, schema, where='response'): """Validate the JSON Schema subset used by the supplied contract, not arbitrary OpenAPI.""" if '$ref' in schema: target = SPEC for part in schema['$ref'].split('/')[1:]: target = target[part] return validate(value, target, where) kinds = schema.get('type', []) if isinstance(kinds, str): kinds = [kinds] matches = {'null':value is None, 'boolean':isinstance(value,bool), 'integer':isinstance(value,int) and not isinstance(value,bool), 'number':isinstance(value,(int,float)) and not isinstance(value,bool), 'string':isinstance(value,str), 'array':isinstance(value,list), 'object':isinstance(value,dict)} if kinds: check(any(matches.get(k,False) for k in kinds), where + ': incorrect type') if 'enum' in schema: check(value in schema['enum'], where + ': invalid enum') if 'const' in schema: check(value == schema['const'], where + ': invalid const') if isinstance(value,dict): props = schema.get('properties',{}) check(all(k in value for k in schema.get('required',[])), where + ': missing required fields') if schema.get('additionalProperties') is False: check(not set(value).difference(props), where + ': unexpected fields') for key, sub in props.items(): if key in value: validate(value[key], sub, where + '.' + key) if isinstance(value,list): check(len(value) >= schema.get('minItems',0), where + ': too few items') check(len(value) <= schema.get('maxItems',float('inf')), where + ': too many items') if schema.get('uniqueItems'): encoded = [json.dumps(x,sort_keys=True) for x in value] check(len(encoded)==len(set(encoded)), where + ': duplicate items') if 'items' in schema: for item in value: validate(item, schema['items'], where + '[]') if isinstance(value,str): check(len(value) >= schema.get('minLength',0), where + ': too short') check(len(value) <= schema.get('maxLength',float('inf')), where + ': too long') if 'pattern' in schema: check(re.search(schema['pattern'], value) is not None, where + ': pattern mismatch') if schema.get('format') == 'uuid': try: uuid.UUID(value) except ValueError as exc: raise Failure(where + ': invalid UUID') from exc if schema.get('format') == 'date-time': timestamp(value) if isinstance(value,(int,float)) and not isinstance(value,bool): check(value >= schema.get('minimum',-float('inf')), where + ': below minimum') check(value <= schema.get('maximum',float('inf')), where + ': above maximum') if 'anyOf' in schema: for alternative in schema['anyOf']: try: validate(value, alternative, where) break except Failure: pass else: raise Failure(where + ': no anyOf branch matches') def shape(value, name): validate(value, SPEC['components']['schemas'][name]) return value class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None class Client: def __init__(self, base_url=None, ca_file=None, timeout=10): self.base = (base_url or os.environ.get('BASE_URL') or ('https://localhost:8443' if LAB==7 else 'http://localhost:8080')).rstrip('/') check(self.base.startswith(('http://','https://')), 'BASE_URL must be HTTP(S)') check(LAB != 7 or self.base.startswith('https://'), 'lab7 requires HTTPS') self.timeout = timeout context = ssl.create_default_context(cafile=ca_file or os.environ.get('CA_FILE') or None) self.opener = urllib.request.build_opener(NoRedirect(), urllib.request.HTTPSHandler(context=context)) def call(self, method, path, expected=200, body=None, actor=None, headers=None, raw=None, schema=None): h = {'Accept':'application/json'} if actor: h['X-User-Id' if LAB==1 else 'Authorization'] = actor['id'] if LAB==1 else 'Bearer ' + actor['token'] if body is not None: raw = json.dumps(body,ensure_ascii=False).encode('utf-8') h['Content-Type'] = 'application/json' h.update(headers or {}) req = urllib.request.Request(self.base + path, data=raw, headers=h, method=method) start = time.monotonic() try: response = self.opener.open(req, timeout=self.timeout) except urllib.error.HTTPError as error: response = error except (urllib.error.URLError,TimeoutError,OSError) as exc: # Do not print URL query, headers, request body, or credentials. raise Failure(method + ' ' + path.split('?')[0] + ': connection/TLS failure (' + type(exc).__name__ + ')') from None with response: status = response.code rh = response.headers data = response.read(16*1024*1024 + 1) check(len(data) <= 16*1024*1024, 'response exceeds test safety limit') allowed = [expected] if isinstance(expected,int) else expected check(status in allowed, method + ' ' + path.split('?')[0] + ': expected ' + str(allowed) + ', received ' + str(status)) check(bool(rh.get('X-Request-Id')), 'missing X-Request-Id') if LAB>=3: check(bool(rh.get('X-Instance-Id')), 'missing X-Instance-Id') payload = None if schema or status>=400: check(rh.get_content_type()=='application/json', 'expected application/json') try: payload = json.loads(data) except (ValueError,UnicodeError): raise Failure('invalid response JSON') from None shape(payload, 'Error' if status>=400 else schema) if status>=400: check(payload['error']['request_id']==rh.get('X-Request-Id'),'error request_id differs from header') if status==401 and LAB>=2: scheme = 'Basic' if path=='/api/v1/auth/sessions' else 'Bearer' check(rh.get('WWW-Authenticate','').lower().startswith(scheme.lower()), 'incorrect WWW-Authenticate challenge') if status==204: check(not data, '204 must not contain body') return {'status':status,'headers':rh,'json':payload,'bytes':data,'elapsed':time.monotonic()-start} def user(self, label): username = 'u_' + label + '_' + uuid.uuid4().hex[:12] password = 'T3st-' + uuid.uuid4().hex body = {'username':username,'display_name':'Студент ' + label} if LAB>=2: body['password'] = password user = self.call('POST','/api/v1/users',201,body=body,schema='User')['json'] check(user['username']==username,'username changed') actor = {'id':user['id'],'user':user,'registration':body} if LAB>=2: credential = base64.b64encode((username+':'+password).encode()).decode() session = self.call('POST','/api/v1/auth/sessions',201,headers={'Authorization':'Basic '+credential},schema='Session')['json'] check(session['user']==user,'session user mismatch') check(timestamp(session['expires_at']) > dt.datetime.now(dt.timezone.utc),'session already expired') actor['token'] = session['token'] return actor def bootstrap(self): self.call('GET','/health/live',schema='Health') self.call('GET','/health/ready',schema='Health') alice, bob, eve = (self.user(label) for label in ('alice','bob','eve')) chat = self.call('POST','/api/v1/chats',201,actor=alice,body={'title':'Контрактный тест','member_ids':[alice['id'],bob['id']]},schema='Chat')['json'] check(set(chat['member_ids'])=={alice['id'],bob['id']},'chat member mismatch') return alice,bob,eve,chat def message(self, actor, chat, text, key=None, attachments=None): body = {'text':text} if attachments is not None: body['attachment_ids'] = attachments headers = {'Idempotency-Key':key or uuid.uuid4().hex} if LAB>=3 else {} return self.call('POST','/api/v1/chats/'+chat['id']+'/messages',201,actor=actor,body=body,headers=headers,schema='Message') def attachment(self, actor, chat): content = (ROOT / 'tests/fixtures/sample.png').read_bytes() att = self.call('POST','/api/v1/chats/'+chat['id']+'/attachments?filename=sample.png',201 if LAB==4 else 202,actor=actor,raw=content,headers={'Content-Type':'image/png'},schema='Attachment')['json'] check(att['status']==('ready' if LAB==4 else 'queued'),'incorrect upload status') check(att['chat_id']==chat['id'] and att['owner_id']==actor['id'],'attachment ownership mismatch') check(att['sha256']==hashlib.sha256(content).hexdigest() and att['size_bytes']==len(content),'original checksum/size mismatch') return att,content def wait_ready(self, actor, att, wait_seconds=60): deadline = time.monotonic()+wait_seconds while time.monotonic()