kangshifu-board/scripts/verify-product-overview-browser.py

"""Verify product insights, exact quote navigation and responsive layout on real data."""
import argparse
import base64
import json
from pathlib import Path
import subprocess
import tempfile
import time
from urllib.request import urlopen
import websocket

ROOT = Path(__file__).resolve().parents[1]
WORK = ROOT / 'work/library-preview'
WORK.mkdir(parents=True, exist_ok=True)
p = argparse.ArgumentParser(description=__doc__)
p.add_argument('--url', default='http://127.0.0.1:4181')
args = p.parse_args()
profile = Path(tempfile.mkdtemp(prefix='overview-browser-', dir=WORK))
process = subprocess.Popen(['C:/Program Files (x86)/Microsoft/Edge/Application/msedge.exe', '--headless=new',
    '--disable-gpu', '--remote-debugging-port=0', '--disable-background-networking', '--disable-extensions',
    '--no-first-run', '--user-data-dir=' + str(profile), 'about:blank'], stdout=subprocess.DEVNULL,
    stderr=subprocess.DEVNULL, creationflags=subprocess.CREATE_NO_WINDOW)
ws = None
serial = 0
events = []
try:
    end = time.monotonic() + 20
    while not (profile / 'DevToolsActivePort').exists():
        assert time.monotonic() < end and process.poll() is None
        time.sleep(.1)
    port = (profile / 'DevToolsActivePort').read_text().splitlines()[0]
    page = next(x for x in json.load(urlopen('http://127.0.0.1:' + port + '/json/list')) if x['type'] == 'page')
    ws = websocket.create_connection(page['webSocketDebuggerUrl'], timeout=35, suppress_origin=True, http_no_proxy=['127.0.0.1'])
    def call(method, params=None):
        global serial
        serial += 1
        ws.send(json.dumps(dict(id=serial, method=method, params=params or {})))
        while True:
            r = json.loads(ws.recv())
            if r.get('id') == serial:
                assert 'error' not in r, r
                return r.get('result', {})
            events.append(r)
    def ev(expression, promise=False):
        r = call('Runtime.evaluate', dict(expression=expression, returnByValue=True, awaitPromise=promise))
        assert 'exceptionDetails' not in r, r
        return r['result'].get('value')
    def wait(condition):
        ev('new Promise((resolve,reject)=>{const end=Date.now()+25000;const tick=()=>{try{if(' + condition + '){resolve(true);return}}catch{}if(Date.now()>end)reject(new Error("timeout"));else setTimeout(tick,60)};tick()})', True)
    V = 'document.querySelector("kf-app")?.shadowRoot?.querySelector("view-library")'
    P = V + '?.renderRoot.querySelector("library-product-detail")'
    def action(body): return ev('(()=>{const p=' + P + ';' + body + '})()')
    def ready(): wait(P + '?.product?.overview && !' + P + '._loading')
    def shot(name):
        data = call('Page.captureScreenshot', dict(format='png', captureBeyondViewport=False))
        (WORK / (name + '.png')).write_bytes(base64.b64decode(data['data']))
    call('Page.enable'); call('Runtime.enable')
    call('Emulation.setDeviceMetricsOverride', dict(width=1440, height=1100, deviceScaleFactor=1, mobile=False))
    url = args.url.rstrip('/') + '/library/products/jd%3A100005702440'
    call('Page.navigate', dict(url=url)); ready()
    stats = action('return p.product.overview.reviews;')
    assert (stats['collected'], stats['eligible'], stats['distinctTexts']) == (19, 18, 17), stats
    themes = action('return p.product.overview.themes;')
    assert [t['distinctTextCount'] for t in themes] == sorted([t['distinctTextCount'] for t in themes], reverse=True)
    assert action('return p.renderRoot.querySelector("details.theme-item").open;')
    assert action('return Array.from(p.renderRoot.querySelectorAll(".overview-metrics strong")).map(x=>+x.textContent);')[:3] == [19, 18, 17]
    shot('product-overview-desktop')
    selected = themes[0]['examples'][0]['reviewId']
    action('p.renderRoot.querySelector(".theme-quotes .text-button").click();')
    wait(P + '?.panel==="reviews" && !' + P + '._loading && ' + P + '._data')
    assert action('return p._data.totalMatched;') == 1
    assert action('return p._data.records[0].record_id;') == selected
    call('Page.reload'); ready()
    assert action('return p._data.totalMatched;') == 1
    action('p.renderRoot.querySelector(".review-controls .text-button").click();')
    wait('!' + P + '.query.relatedReviewId && !' + P + '._loading && ' + P + '._data?.totalMatched===18')
    action('p.renderRoot.getElementById("tab-overview").click();'); ready()
    layouts = []
    for width in (1440, 1024, 390, 320):
        call('Emulation.setDeviceMetricsOverride', dict(width=width, height=1100, deviceScaleFactor=1, mobile=width < 600))
        ev('new Promise(resolve=>requestAnimationFrame(()=>requestAnimationFrame(resolve)))', True)
        if width == 390: shot('product-overview-mobile')
        assert ev('document.documentElement.scrollWidth<=innerWidth'), width
        sizes = action('return Array.from(p.renderRoot.querySelectorAll(".insights-grid,.overview-metrics,.theme-item,.aggregate-panel")).map(x=>({class:x.className,scroll:x.scrollWidth,width:x.clientWidth}));')
        assert all(x['scroll'] <= x['width'] + 1 for x in sizes), (width, sizes)
        layouts.append(dict(width=width, metricColumns=action('return getComputedStyle(p.renderRoot.querySelector(".overview-metrics")).gridTemplateColumns.split(" ").length;')))
    call('Emulation.setDeviceMetricsOverride', dict(width=1440, height=1100, deviceScaleFactor=1, mobile=False))
    call('Page.navigate', dict(url=args.url.rstrip('/') + '/library?sort=reviews_asc'))
    wait(V + '?._data?.records?.length && !' + V + '._loading')
    empty = ev(V + '._data.records.find(r=>!r.relatedCounts.reviews).record_id')
    call('Page.navigate', dict(url=args.url.rstrip('/') + '/library/products/' + empty.replace(':', '%3A'))); ready()
    assert action('return p.product.overview.reviews.collected;') == 0
    assert action('return p.renderRoot.querySelector(".insight-empty").textContent.includes("暂未收录评价");')
    assert not [e for e in events if e.get('method') == 'Runtime.exceptionThrown']
    result = dict(ok=True, stats=stats, layouts=layouts, exactQuoteNavigation=True, refreshPreservesQuote=True,
                  emptyProductVerified=True, browserVerified=True, productionBrowserVerified=False, databaseWrites=0)
    (WORK / 'product-overview-validation.json').write_text(json.dumps(result, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    print(json.dumps(result, ensure_ascii=False, indent=2))
finally:
    if ws: ws.close()
    process.terminate()
    try: process.wait(timeout=8)
    except subprocess.TimeoutExpired: process.kill()