64 lines
3.3 KiB
Python
64 lines
3.3 KiB
Python
"""Run the fixed DOM observers in an isolated local browser, without any account or platform request."""
|
|
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from bs4 import BeautifulSoup
|
|
|
|
from backend.browser import tools
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def observed_discussion(tmp_path_factory):
|
|
mac_chrome = Path("/Applications/Google Chrome.app/Contents/MacOS/Google Chrome")
|
|
chrome = str(mac_chrome) if mac_chrome.is_file() else shutil.which("google-chrome") or shutil.which("chromium")
|
|
if not chrome:
|
|
pytest.skip("No local Chrome for the isolated DOM integration fixture")
|
|
directory = tmp_path_factory.mktemp("prospecting_dom")
|
|
fixture = directory / "discussion.html"
|
|
html = (Path(__file__).parent / "fixtures" / "prospecting-discussion.html").read_text(encoding="utf-8")
|
|
script = "\nconst observe=" + tools._OBSERVE_SOURCE + ";\nconst inspect=" + tools._INSPECT_SOURCE_CANDIDATES
|
|
script += ";\nconst profile=" + tools._INSPECT_PROFILE + ";\nconst control=" + tools._OBSERVE_REPLY_CONTROL + ";\n"
|
|
script += """
|
|
const all = inspect({max_comments: 5, saved_selectors: [], note_saved: false});
|
|
const onlyNew = inspect({max_comments: 5, saved_selectors: ['#root-content'], note_saved: true});
|
|
document.getElementById('result').textContent = JSON.stringify({all, onlyNew,
|
|
reply: observe('#reply-content-one'), root: observe('#root-content'),
|
|
profile: profile(), control: control({selector: '#expand'})});
|
|
"""
|
|
fixture.write_text(html + "<script>" + script + "</script>", encoding="utf-8")
|
|
process = subprocess.run([
|
|
chrome, "--headless=new", "--disable-gpu", "--disable-background-networking",
|
|
"--no-first-run", "--no-default-browser-check", "--user-data-dir=" + str(directory / "profile"),
|
|
"--dump-dom", fixture.as_uri(),
|
|
], capture_output=True, text=True, timeout=25)
|
|
result = BeautifulSoup(process.stdout, "html.parser").find(id="result")
|
|
assert process.returncode == 0 and result and result.text, "Local DOM fixture did not finish"
|
|
return json.loads(result.text)
|
|
|
|
|
|
def test_dom_discussion_keeps_replies_and_identical_text_from_separate_speakers(observed_discussion):
|
|
result = observed_discussion
|
|
assert [item["source_type"] for item in result["all"]["candidates"]] == ["note", "comment", "reply", "reply"]
|
|
assert [item["source_type"] for item in result["onlyNew"]["candidates"]] == ["reply", "reply"]
|
|
assert result["root"]["text"] == "我在杭州准备装修"
|
|
assert result["reply"]["text"] == "多少钱？"
|
|
|
|
|
|
def test_dom_author_region_and_thread_parent_are_observed_separately(observed_discussion):
|
|
result = observed_discussion
|
|
assert result["reply"]["author"] == "乙" and result["root"]["author"] == "甲"
|
|
assert result["reply"]["parent_source_selector"] == result["reply"]["root_source_selector"] == "#root-content"
|
|
assert result["reply"]["reply_to"] == "甲"
|
|
assert result["reply"]["ip_location"] == ""
|
|
|
|
|
|
def test_dom_profile_label_and_actual_expansion_entry_are_distinct_from_internal_ids(observed_discussion):
|
|
result = observed_discussion
|
|
assert result["profile"]["xiaohongshu_id"] == "visible_123"
|
|
assert result["all"]["reply_controls"] == [{"source_selector": "#expand", "text": "展开 2 条回复", "in_view": True}]
|
|
assert result["control"]["text"] == "展开 2 条回复"
|