diff --git a/tests/test_live_app.py b/tests/test_live_app.py index 132de42..0730a76 100644 --- a/tests/test_live_app.py +++ b/tests/test_live_app.py @@ -1,14 +1,23 @@ """Optional conformance test against a running Manifest app, never a permissive stub. MNFST_TEST_APP_URL must identify a disposable app allowing customer signup. +The app serves no patch until an operator approves one, so the test signs in +as an operator too: MNFST_TEST_OPERATOR_EMAIL and MNFST_TEST_OPERATOR_PASSWORD, +defaulting to the dev seed. """ import asyncio import json import os +import random +import string import threading +import urllib.request import uuid +from contextlib import closing from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import parse_qs, urlsplit +import aiohttp import httpx import pytest import requests @@ -16,73 +25,119 @@ from mnfst.outbound import uninstall_outbound from tests.helpers import wait_for +ERROR = {'error': {'message': 'range of limit should be [1, 100]', 'param': 'limit', + 'code': 'invalid_value', 'type': 'validation_error'}} + + +def signed_in(base, path, payload): + client = httpx.Client(base_url=base, headers={'origin': base}) + client.post(path, json=payload).raise_for_status() + return client + + +def approve_clamp(base, key, url): + """Open the issue for `url` with one heal, then approve the clamp for it. + The issue is keyed by the endpoint path, not the query, so every call + below shares it.""" + opened = httpx.post(f'{base}/v1/heal', headers={'authorization': f'Bearer {key}'}, json={ + 'traceId': uuid.uuid4().hex, + 'request': {'method': 'POST', 'url': url, + 'headers': {'content-type': 'application/json'}, 'body': {'limit': 500}}, + 'response': {'statusCode': 400, 'body': ERROR, 'truncated': False}, + 'responseTimeMs': 1}) + opened.raise_for_status() + assert opened.json()['status'] == 'no_patch', opened.text + with closing(signed_in(base, '/api/auth/sign-in/email', { + 'email': os.environ.get('MNFST_TEST_OPERATOR_EMAIL', 'admin@colibri.local'), + 'password': os.environ.get('MNFST_TEST_OPERATOR_PASSWORD', 'colibridev')})) as operator: + proposed = operator.post(f"/api/issues/{opened.json()['issueId']}/candidates", json={ + 'operations': [{'type': 'clamp', 'args': {'path': '/limit', 'max': 100}}], + 'rationale': 'SDK conformance: the provider states the range'}) + proposed.raise_for_status() + operator.post(f"/api/candidates/{proposed.json()['candidateId']}/approve", + json={}).raise_for_status() + def test_real_app_outcomes(monkeypatch): base = os.environ.get('MNFST_TEST_APP_URL') if not base: pytest.skip('set MNFST_TEST_APP_URL to a disposable Manifest app') - with httpx.Client(base_url=base, headers={'origin': base}) as customer: - signup = customer.post('/api/auth/sign-up/email', json={ + with closing(signed_in(base, '/api/auth/sign-up/email', { 'name': 'SDK conformance', 'email': f'sdk-{uuid.uuid4().hex}@example.test', - 'password': 'LocalSdkConformance-2026!'}) - signup.raise_for_status() + 'password': 'LocalSdkConformance-2026!'})) as customer: project = customer.post('/api/customer/projects', json={'name': 'SDK conformance'}) project.raise_for_status() project_id = project.json()['id'] key = customer.post(f'/api/customer/projects/{project_id}/keys') key.raise_for_status() - reports = [] - original_send = httpx.Client.send + key = key.json()['key'] + reports = [] + original_send = httpx.Client.send + + def observe(self, request, **kwargs): + response = original_send(self, request, **kwargs) + if '/v1/heal-attempts/' in request.url.path: + reports.append((response.status_code, response.json())) + return response - def observe(self, request, **kwargs): - response = original_send(self, request, **kwargs) - if '/v1/heal-attempts/' in request.url.path: - reports.append((response.status_code, response.json())) - return response + monkeypatch.setattr(httpx.Client, 'send', observe) - monkeypatch.setattr(httpx.Client, 'send', observe) + class Provider(BaseHTTPRequestHandler): + def log_message(self, *args): + pass - class Provider(BaseHTTPRequestHandler): - def log_message(self, *args): - pass + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers['content-length']))) + mode = parse_qs(urlsplit(self.path).query).get('mode', [''])[0] + if body['limit'] <= 100 and mode == 'transport': + self.connection.shutdown(2) + self.connection.close() + return + failed = body['limit'] > 100 or mode == 'ineffective' + raw = json.dumps(ERROR if failed else body).encode() + self.send_response(400 if failed else 200) + self.send_header('Content-Type', 'application/json') + self.send_header('Content-Length', str(len(raw))) + self.end_headers() + self.wfile.write(raw) - def do_POST(self): - body = json.loads(self.rfile.read(int(self.headers['content-length']))) - if body['limit'] <= 100 and self.path.endswith('/transport'): - self.connection.shutdown(2) - self.connection.close() - return - failed = body['limit'] > 100 or self.path.endswith('/ineffective') - data = {'error': {'message': 'range of limit should be [1, 100]', 'param': 'limit', - 'code': 'invalid_value', 'type': 'validation_error'}} if failed else body - raw = json.dumps(data).encode() - self.send_response(400 if failed else 200) - self.send_header('Content-Type', 'application/json') - self.send_header('Content-Length', str(len(raw))) - self.end_headers() - self.wfile.write(raw) + server = ThreadingHTTPServer(('127.0.0.1', 0), Provider) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + # A fresh issue per run: an approved patch serves every project, and the + # app keys issues by path, templating id-like segments (hex, digits) and + # ignoring the port. A letters-only segment stays literal. + token = ''.join(random.choices(string.ascii_lowercase, k=16)) + upstream = f'http://127.0.0.1:{server.server_port}/sdk-{token}/items' + try: + approve_clamp(base, key, upstream) + manifest(key=key, url=base) + # A single-slot pool must release the original error before retrying. + with httpx.Client(limits=httpx.Limits(max_connections=1), timeout=5) as client: + assert client.post(upstream + '?client=sync', json={'limit': 500}).json() == {'limit': 100} - server = ThreadingHTTPServer(('127.0.0.1', 0), Provider) - thread = threading.Thread(target=server.serve_forever, daemon=True) - thread.start() - upstream = f'http://127.0.0.1:{server.server_port}/sdk{uuid.uuid4().hex}' - try: - manifest(key=key.json()['key'], url=base) - # A single-slot pool must release the original error before retrying. - with httpx.Client(limits=httpx.Limits(max_connections=1), timeout=5) as client: - assert client.post(upstream + '/sync', json={'limit': 500}).json() == {'limit': 100} - async def run_async(): - async with httpx.AsyncClient(timeout=5) as client: - assert (await client.post(upstream + '/async', json={'limit': 500})).status_code == 200 - asyncio.run(run_async()) - assert requests.post(upstream + '/requests', json={'limit': 500}, timeout=5).status_code == 200 - assert httpx.post(upstream + '/ineffective', json={'limit': 500}).status_code == 400 - assert httpx.post(upstream + '/transport', json={'limit': 500}).status_code == 400 - assert wait_for(lambda: len(reports) == 5, timeout=10), json.dumps(reports) - assert all(code == 200 for code, _ in reports) - assert sorted(body['status'] for _, body in reports) == ['failed', 'inconclusive', 'succeeded', 'succeeded', 'succeeded'] - finally: - uninstall_outbound() - server.shutdown() - server.server_close() - thread.join() + async def run_async(): + async with httpx.AsyncClient(timeout=5) as client: + response = await client.post(upstream + '?client=async', json={'limit': 500}) + assert response.status_code == 200 + async with aiohttp.ClientSession() as session: + async with session.post(upstream + '?client=aiohttp', json={'limit': 500}) as response: + assert await response.json() == {'limit': 100} + asyncio.run(run_async()) + assert requests.post(upstream + '?client=requests', json={'limit': 500}, + timeout=5).json() == {'limit': 100} + with urllib.request.urlopen(urllib.request.Request( + upstream + '?client=urllib', data=json.dumps({'limit': 500}).encode(), + headers={'Content-Type': 'application/json'}), timeout=5) as response: + assert json.loads(response.read()) == {'limit': 100} + assert httpx.post(upstream + '?mode=ineffective', json={'limit': 500}).status_code == 400 + assert httpx.post(upstream + '?mode=transport', json={'limit': 500}).status_code == 400 + assert wait_for(lambda: len(reports) == 7, timeout=10), json.dumps(reports) + assert all(code == 200 for code, _ in reports) + assert sorted(body['status'] for _, body in reports) == [ + 'failed', 'inconclusive'] + ['succeeded'] * 5 + finally: + uninstall_outbound() + server.shutdown() + server.server_close() + thread.join()