#!/usr/bin/env python3 """Run API assertions against an existing ES 8 service; never start/reconfigure it.""" import argparse import copy import json from pathlib import Path import sys import urllib.error import urllib.parse import urllib.request import uuid def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--url', default='http://127.0.0.1:19200') parser.add_argument('--server-version', default='8.15.1', help='exact expected runtime version (8.x); source baseline remains 8.15.1/9.11.1') args = parser.parse_args() url = urllib.parse.urlsplit(args.url) if (url.scheme != 'http' or url.hostname not in ('127.0.0.1', 'localhost', '::1') or url.username or url.password or url.path not in ('', '/') or url.query or url.fragment): parser.error('--url must be a loopback HTTP service root without credentials/query') base = args.url.rstrip('/') report = {'kind': 'service-run', 'url': base, 'source_baseline': '8.15.1 / Lucene 9.11.1', 'expected_runtime': args.server_version, 'status': 'failed', 'checks': [], 'cleanup': {}} index = 'study-t18-es-' + uuid.uuid4().hex attempted, pit_id, failure = False, None, None def request(method, path, body=None, expected=(200,)): data = None if body is None else json.dumps(body).encode() req = urllib.request.Request(base + path, data=data, method=method, headers={'Content-Type': 'application/json'}) try: with urllib.request.urlopen(req, timeout=5) as response: code, raw = response.status, response.read() except urllib.error.HTTPError as exc: code, raw = exc.code, exc.read() result = json.loads(raw) if raw else {} if code not in expected: raise AssertionError(f'{method} {path}: HTTP {code}: {result}') return code, result def search(query): return request('POST', f'/{index}/_search', {'query': query, 'size': 10})[1] def total(query): return search(query)['hits']['total']['value'] try: info = request('GET', '/')[1] report['version'] = info['version'] assert args.server_version.startswith('8.'), 'this lab supports ES 8.x API assertions only' assert info['version']['number'] == args.server_version, 'read-only preflight: runtime version mismatch' if args.server_version == '8.15.1': assert info['version']['lucene_version'] == '9.11.1', 'expected bundled Lucene 9.11.1' fixture = json.loads(Path(__file__).with_name('fixture.json').read_text()) attr = {'properties': {'key': {'type': 'keyword'}, 'value': {'type': 'keyword'}}} mappings = {'_meta': {'study_t18_owner': index}, 'properties': { 'item_id': {'type': 'keyword'}, 'created_at': {'type': 'long'}, 'version': {'type': 'long'}, 'name': {'type': 'text', 'fields': {'raw': {'type': 'keyword'}}}, 'tenant': {'type': 'keyword'}, 'attrs': {'type': 'object', **attr}, 'nested_attrs': {'type': 'nested', **attr}, }} code, _ = request('GET', f'/{index}', expected=(200, 404)) assert code == 404, 'refuse a pre-existing namespace' attempted = True report['namespace'] = index code, created = request('PUT', f'/{index}', {'settings': {'number_of_shards': 1, 'number_of_replicas': 0, 'refresh_interval': '-1'}, 'mappings': mappings}, (200, 400)) if code == 400 and created.get('error', {}).get('type') == 'resource_already_exists_exception': attempted = False # a concurrent name conflict is never ours to delete assert code == 200 and created.get('acknowledged'), f'index creation failed: {created}' def put(doc, query=''): body = copy.deepcopy(doc) body['nested_attrs'] = body['attrs'] body.setdefault('version', 1) return request('PUT', f'/{index}/_doc/{doc["item_id"]}' + query, body, (200, 201))[1] for doc in fixture['documents']: put(doc) current = request('GET', f'/{index}/_doc/a')[1] assert current['found'] and total({'match_all': {}}) == 0 request('POST', f'/{index}/_refresh') assert total({'match_all': {}}) == 3 report['checks'].append('realtime GET=found; search before refresh=0; after=3') assert total({'match': {'name': 'apple'}}) == 2 assert total({'term': {'name': 'Red Apple'}}) == 0 assert total({'term': {'name.raw': 'Red Apple'}}) == 1 report['checks'].append('match apple=2; term text full string=0; keyword exact=1') loose = {'bool': {'filter': [{'term': {'attrs.key': 'color'}}, {'term': {'attrs.value': 'red'}}]}} nested = {'nested': {'path': 'nested_attrs', 'query': {'bool': {'filter': [ {'term': {'nested_attrs.key': 'color'}}, {'term': {'nested_attrs.value': 'red'}}]}}}} assert total(loose) == 2 and total(nested) == 1 report['checks'].append('object cross-element=2; nested same-element=1') condition = urllib.parse.urlencode({'if_seq_no': current['_seq_no'], 'if_primary_term': current['_primary_term']}) changed = {**fixture['documents'][0], 'version': 2} put(changed, '?' + condition) stale = {**fixture['documents'][0], 'version': 1, 'nested_attrs': fixture['documents'][0]['attrs']} _, conflict = request('PUT', f'/{index}/_doc/a?{condition}', stale, (409,)) assert conflict['error']['type'] == 'version_conflict_engine_exception' assert request('GET', f'/{index}/_doc/a')[1]['_source']['version'] == 2 request('POST', f'/{index}/_refresh') report['checks'].append('first OCC update succeeds; stale seqNo+term=409; version stays 2') pit_id = request('POST', f'/{index}/_pit?keep_alive=1m')[1]['id'] put(fixture['late_document']) request('POST', f'/{index}/_refresh') assert total({'match_all': {}}) == 4 seen, after, sorts = [], None, [] while True: body = {'size': 2, 'pit': {'id': pit_id, 'keep_alive': '1m'}, 'sort': [{'created_at': 'asc'}, {'item_id': 'asc'}]} if after is not None: body['search_after'] = after # keep the complete server sort, including _shard_doc page = request('POST', '/_search', body)[1] pit_id = page.get('pit_id', pit_id) hits = page['hits']['hits'] if not hits: break seen.extend(hit['_source']['item_id'] for hit in hits) after = hits[-1]['sort'] assert len(after) >= 3, 'PIT adds a tie-breaker; never trim the returned sort array' sorts.append(after) assert len(seen) <= 3, 'PIT should contain only the pre-open search view' assert seen == ['a', 'b', 'c'] report['checks'].append('PIT ids=a,b,c despite live d; full returned sort reused') report['pit_sort_arrays'] = sorts report['status'] = 'passed' except Exception as exc: failure = exc report['error'] = f'{type(exc).__name__}: {exc}' finally: if pit_id is not None: try: closed = request('DELETE', '/_pit', {'id': pit_id})[1] assert closed['succeeded'] report['cleanup']['pit_closed'] = True except Exception as exc: failure = failure or exc report['cleanup']['pit_error'] = str(exc) if attempted: try: code, state = request('GET', f'/{index}', expected=(200, 404)) if code == 200: assert state[index]['mappings'].get('_meta', {}).get('study_t18_owner') == index, \ 'ownership marker differs: refuse cleanup' request('DELETE', f'/{index}') request('GET', f'/{index}', expected=(404,)) report['cleanup']['owned_index_absent'] = True except Exception as exc: failure = failure or exc report['cleanup']['index_error'] = str(exc) report['status'] = 'failed' if failure else 'passed' print(json.dumps(report, ensure_ascii=False, indent=2)) return 1 if failure else 0 if __name__ == '__main__': sys.exit(main())