docs_to_postman.py

#!/usr/bin/env python
# python docs_to_postman.py | newman run /dev/stdin
from __future__ import print_function

import re
import os
import sys
import json
import copy
import email.message
import urllib.parse as urlparse


PARSER_RE = re.compile(r'\n```\s*([a-z]*)(.*?)\n```|\n(#+)\s*([^\n]+)|\n([a-z]+)[ \t]*:[ \t]*([^\n]+)', re.DOTALL)

BASE_DIR = os.path.dirname(os.path.abspath(__file__))

# The migrated docs site: the Astro content is the source of truth. Its pages carry
# the E2E requests (```rest / ```http / ```json) and assertions (in <!-- e2e:begin -->
# comments), and its sidebar encodes the reading order.
CONTENT_DIR = os.path.join(BASE_DIR, 'docs', 'src', 'content', 'docs')
SIDEBAR_PATH = os.path.join(BASE_DIR, 'docs', 'src', 'lib', 'sidebar.mjs')
ASSETS_DIR = os.path.join(BASE_DIR, 'docs', 'public', 'assets')


def content_order():
    """Absolute md paths in sidebar reading order (depth-first).

    The doc-driven E2E examples build on each other across pages (index a document,
    then get it, update it, delete it; the /bank/ queries assume a dataset state), so
    they must run in the order the docs are meant to be read. That order is the
    committed Starlight sidebar (src/lib/sidebar.mjs), which replaced the old Jekyll
    docs.yaml when the site migrated to Astro. Paths are lower-cased for matching so
    the case-insensitively-slugged test pages (tests/dataTypes/...) line up on
    case-sensitive filesystems too. Returns [] if the sidebar is absent (caller then
    falls back to a stable alphabetical order)."""
    try:
        txt = open(SIDEBAR_PATH).read()
    except OSError:
        return []
    m = re.search(r'export\s+const\s+sidebar\s*=\s*(\[[\s\S]*\])\s*;', txt)
    if not m:
        return []
    try:
        data = json.loads(m.group(1))
    except ValueError:
        return []
    order = []

    def walk(items):
        for it in items:
            slug = it.get('slug')
            link = it.get('link')
            if slug is not None:
                order.append(slug)
            elif link:
                order.append(link.lstrip('/'))
            if it.get('items'):
                walk(it['items'])

    walk(data)
    return [os.path.abspath(os.path.join(CONTENT_DIR, s + '.md')).lower() for s in order]


def parse_filename(filename, index, all_tests):
    filename_path, _ = os.path.splitext(filename)
    data = open(filename).read()
    fnp = filename_path
    file_context = {
        'filename': filename,
        'titles': [],
    }
    context = {}
    context.update(file_context)
    while fnp and fnp != BASE_DIR and fnp != '/':
        if fnp in index:
            file_context['titles'].insert(0, (0, index.get(fnp)))
        fnp = os.path.dirname(fnp)
    # print(filename_path, base_titles)

    def process(m):
        groups = m.groups()
        # print(groups)
        if groups[0] in ('json', 'rest', 'http'):
            # Flush:
            if context and 'request' in context:
                all_tests.append(copy.deepcopy(context))
            context.clear()
            context.update(file_context)
            context['request'] = groups[1].strip()
        elif groups[0] == 'js':
            context.setdefault('tests', []).append(groups[1].strip())
        elif groups[2]:
            # Flush:
            if context and 'request' in context:
                all_tests.append(copy.deepcopy(context))
            context.clear()
            context.update(file_context)
            # Add title:
            level = len(groups[2])
            file_context['titles'] = [title for title in file_context.get('titles', []) if title[0] < level]
            file_context['titles'].append((level, groups[3]))
            # Clear description:
            file_context.pop('description', None)
            context.pop('description', None)
        elif groups[5]:
            name = groups[4]
            if name == 'description':
                # Persist description:
                file_context[name] = groups[5]
                context[name] = groups[5]
            elif name == 'title':
                # Flush:
                if context and 'request' in context:
                    all_tests.append(copy.deepcopy(context))
                context.clear()
                context.update(file_context)
                # Add title:
                index[filename_path] = groups[5]
                file_context['titles'].append((1, groups[5]))
            else:
                context[name] = groups[5]
    PARSER_RE.sub(process, data)
    # Flush:
    if context and 'request' in context:
        all_tests.append(copy.deepcopy(context))


def parse_directory(directory, index, all_tests):
    # Collect every .md under the tree, then process in a DETERMINISTIC order: the
    # sidebar reading order first (so dependent examples run in the intended
    # sequence), then anything not in the sidebar, alphabetically. This replaces the
    # platform-dependent os.walk order (see content_order()).
    found = []
    for path, dirs, files in os.walk(directory):
        for f in files:
            if f.endswith('.md'):
                found.append(os.path.abspath(os.path.join(path, f)))
    rank = {p: i for i, p in enumerate(content_order())}
    big = len(rank)
    # Match case-insensitively: the sidebar slugs are lower-cased, the test pages sit
    # at camelCase paths on disk (tests/dataTypes/...).
    found.sort(key=lambda p: (rank.get(p.lower(), big), p))
    for filename in found:
        parse_filename(filename, index, all_tests)


def main():
    index = {}
    all_tests = []

    if len(sys.argv) > 1:
        for arg in sys.argv:
            if os.path.abspath(arg) != os.path.abspath(__file__):
                if os.path.isdir(arg):
                    parse_directory(arg, index, all_tests)
                else:
                    parse_filename(arg, index, all_tests)
    else:
        parse_directory(CONTENT_DIR, index, all_tests)

    # print(json.dumps(all_tests, indent=4))

    collection = {
        "info": {
            "name": "Xapiand",
            "description": "Xapiand is A Modern Highly Available Distributed RESTful Search and Storage Engine built for the Cloud and with Data Locality in mind.",
            "schema": "https://schema.getpostman.com/json/collection/v2.1.0/collection.json"
        },
        "variable": [
            {
                "key": "domain",
                "value": "localhost:8880",
                "type": "string"
            }
        ],
        "item": [],
    }

    for test in all_tests:
        items = collection["item"]
        title = []
        description = test.get('description')
        titles = test['titles']
        if not description:
            if titles:
                description = titles[-1][1]
                titles = titles[:-1]
        for _, name in titles:
            for item in items:
                if item.get("name") == name and 'item' in item:
                    break
            else:
                item = {
                    "name": name,
                    "item": []
                }
                items.append(item)
            title.append(item["name"])
            items = item["item"]
        url = ""
        method = None
        body = []
        headers = email.message.Message()
        for i, line in enumerate(test['request'].split('\n')):
            line = line.strip()
            if i == 0:
                method, _, url = line.partition(' ')
                if not re.match(r'[A-Z]+', method):
                    method = None
                    break
            elif not body:
                if not line:
                    body.append(line)
                else:
                    k, _, v = line.partition(':')
                    if not _:
                        raise ValueError("Malformed header in {} ({}): {}".format(test['filename'], ' / '.join(title), repr(line)))
                    headers.add_header(k.strip(), v.strip())
            else:
                body.append(line)
        if not method:
            continue
        for k, v in {
            "Content-Type": "application/json"
        }.items():
            if k not in headers:
                headers[k] = v
        header = []
        for k, v in headers.items():
            header.append({
                "type": "text",
                "key": k,
                "value": v,
            })
        parsed = urlparse.urlparse(url)
        path = parsed.path[1:] if parsed.path.startswith('/') else parsed.path
        path = path.split('/')
        qs = urlparse.parse_qs(parsed.query + '&' + test.get('params', ''))
        for k, v in {
            "commit": None,
            "volatile": None,
            "echo": None,
        }.items():
            if k not in qs:
                qs[k] = v
        query = []
        for k, v in qs.items():
            if v is None:
                # A valueless flag (commit / volatile / echo): emit as a bare param.
                query.append({
                    "key": k,
                    "value": None,
                })
            else:
                # urlparse.parse_qs returns each value as a list; Postman expects a
                # string per query entry, so emit one entry per value. (Previously the
                # raw list was passed through, which newman serialised as a bare
                # valueless param -- e.g. `sort=foo` became `?sort`, so every
                # sort/params-bearing request was sent malformed and 400'd.)
                for item in (v if isinstance(v, list) else [v]):
                    query.append({
                        "key": k,
                        "value": item,
                    })
        url = {
            "host": [
                "{{domain}}"
            ],
            "path": path,
        }
        if query:
            url["query"] = query
        body = '\n'.join(body).strip()
        request = {
            "method": method,
            "url": url,
        }
        if header:
            request["header"] = header
        if body:
            if body[0] == '@':
                request["body"] = {
                    "mode": "file",
                    "file": {
                        "src": os.path.join(ASSETS_DIR, body[1:])
                    }
                }
            else:
                request["body"] = {
                    "mode": "raw",
                    "raw": body,
                }
        item = {
            "request": request,
        }
        if description:
            item["name"] = description
        items.append(item)
        tests = test.get('tests')
        if not tests:
            # No explicit assertions in this doc block: synthesize a default so
            # every request is verified at least at the status level (this is what
            # makes the assertion-only e2e gate a complete regression net without a
            # response-body baseline). A `status: NNN` marker -- harness metadata
            # inside a {% comment %} block, invisible in the rendered docs -- pins a
            # specific expected status for the deliberate error/edge demos
            # (unimplemented CLOSE/OPEN -> 501, placeholder paths -> 404,
            # invalid-input examples -> 400); everything else asserts 2xx success.
            status = test.get('status')
            if status:
                tests = ['pm.test("Status is %d", function () {\n  pm.response.to.have.status(%d);\n});' % (int(status), int(status))]
            else:
                tests = ['pm.test("Response is success", function () {\n  pm.response.to.be.success;\n});']
        if tests:
            scripts = []
            item["event"] = [{
                "listen": "test",
                "script": {
                    "type": "text/javascript",
                    "exec": scripts,
                }
            }]
            for script in tests:
                scripts.append(script)
        # print(test)

    json.dump(collection, sys.stdout, indent=2)
    return 0

if __name__ == '__main__':
    sys.exit(main())