diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 53468ab..b46e53f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -6,7 +6,55 @@ on: pull_request: jobs: + # Quality gates run once on the current Python — they don't vary by runtime. + # Each is its own job so it shows as a separate check on the PR. + lint: + name: lint + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - run: pip install ruff + - run: ruff check src/bytecrawl tests + + format: + name: format + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - run: pip install ruff + - run: ruff format --check src/bytecrawl tests + + typecheck: + name: type check + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - name: Install (package + llm/mcp deps so mypy resolves imports) + run: pip install -e ".[llm,dev,mcp]" + - run: mypy src/bytecrawl + + security: + name: security + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - run: pip install bandit + - run: bandit -r src/bytecrawl -ll + test: + name: unit test ${{ matrix.python-version }} runs-on: ubuntu-latest strategy: matrix: @@ -23,7 +71,5 @@ jobs: - name: Install MCP extra (needs Python 3.10+) if: matrix.python-version != '3.9' run: pip install -e ".[mcp]" - - name: Lint - run: ruff check src/bytecrawl tests - name: Tests run: pytest --cov=bytecrawl --cov-report=term-missing diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 3bb1148..bee6e9d 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -14,6 +14,10 @@ jobs: runs-on: ubuntu-latest permissions: contents: write + outputs: + # true only when this push introduced a new version (no tag yet) — + # the publish job below keys off it so PyPI never gets a duplicate. + publish: ${{ steps.tag.outputs.exists == 'false' }} steps: - uses: actions/checkout@v4 with: @@ -66,3 +70,25 @@ jobs: gh release create "v${{ steps.v.outputs.version }}" \ --title "v${{ steps.v.outputs.version }}" \ --notes-file release-notes.md + + # Runs in the SAME workflow run right after `release`, only for a new version. + # (A separate workflow keyed on the tag/release would NOT fire: events made + # with GITHUB_TOKEN don't trigger further workflow runs.) + publish: + needs: release + if: needs.release.outputs.publish == 'true' + runs-on: ubuntu-latest + permissions: + # OIDC token for PyPI Trusted Publishing — no API token/secret needed. + id-token: write + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - name: Build sdist + wheel + run: | + pip install build + python -m build + - name: Publish to PyPI + uses: pypa/gh-action-pypi-publish@release/v1 diff --git a/CHANGELOG.md b/CHANGELOG.md index cede886..0be6d0c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,24 @@ # Changelog +## 1.3.0 — 2026-09-11 + +### Changed +- **CI is split into separate checks instead of one "test" job.** Lint + (`ruff check`), format (`ruff format --check`), type check (`mypy`) and + security (`bandit`) each run as their own top-level check alongside the + unit-test matrix (`unit test 3.9` / `unit test 3.12`). A red check now points + straight at the cause — a formatting slip no longer hides behind a failing + type check inside a single collapsed job. +- **Releases publish to PyPI automatically** through GitHub Actions Trusted + Publishing (OIDC) — no API tokens or stored secrets. A push to `main` with a + new `__version__` runs the tests, tags the release, and uploads to PyPI; + nothing publishes without a version bump. + +### Internal +- Version is single-sourced from `__version__`: a test now pins the top + CHANGELOG entry and both READMEs to it, so they can't drift. +- Type-checking fixes across the scraping and crawl modules and one audited + `# nosec` on the MCP HTTP bind. No behaviour change. + ## 1.2.2 — 2026-09-10 ### Fixed diff --git a/README.es.md b/README.es.md index c10d350..ec8097d 100644 --- a/README.es.md +++ b/README.es.md @@ -10,8 +10,8 @@ Shark-Search y OPIC en Python puro. - **Web**: https://bytecrawl.vercel.app/ - **Endpoint MCP hosteado**: https://bytecrawl.vercel.app/mcp - **Skill para agentes**: https://bytecrawl.vercel.app/agent-onboarding/SKILL.md -- **Última versión**: **1.2.2** — seis herramientas MCP, comparación de - estrategias, `Page.links()` absolutos ([changelog](https://github.com/abrahamperz/ByteCrawl/blob/main/CHANGELOG.md)) +- **Última versión**: **1.3.0** — publicación automática a PyPI y checks de CI + separados (lint, formato, tipos, seguridad) ([changelog](https://github.com/abrahamperz/ByteCrawl/blob/main/CHANGELOG.md)) - **PyPI**: https://pypi.org/project/bytecrawl/ - **GitHub**: https://github.com/abrahamperz/ByteCrawl diff --git a/README.md b/README.md index 1db6f2d..702fd92 100644 --- a/README.md +++ b/README.md @@ -16,8 +16,8 @@ and OPIC in pure Python. - **Webpage**: https://bytecrawl.vercel.app/ - **Hosted MCP endpoint**: https://bytecrawl.vercel.app/mcp - **Agent skill**: https://bytecrawl.vercel.app/agent-onboarding/SKILL.md -- **Latest release**: **1.2.2** — six MCP tools, strategy comparison, - absolute `Page.links()` ([changelog](https://github.com/abrahamperz/ByteCrawl/blob/main/CHANGELOG.md)) +- **Latest release**: **1.3.0** — automated PyPI publishing and separated CI + gates (lint, format, types, security) ([changelog](https://github.com/abrahamperz/ByteCrawl/blob/main/CHANGELOG.md)) ## Why focused crawling? diff --git a/pyproject.toml b/pyproject.toml index 397df76..79f2862 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,7 +29,7 @@ browser = ["playwright>=1.40"] llm = ["markdownify>=0.11", "trafilatura>=1.6"] mcp = ["mcp>=2; python_version >= '3.10'", "markdownify>=0.11", "trafilatura>=1.6"] all = ["playwright>=1.40", "markdownify>=0.11", "trafilatura>=1.6", "mcp>=2; python_version >= '3.10'"] -dev = ["pytest>=7", "pytest-cov", "ruff"] +dev = ["pytest>=7", "pytest-cov", "ruff", "mypy", "bandit"] [project.scripts] bytecrawl-mcp = "bytecrawl.mcp_server:main" @@ -60,5 +60,14 @@ target-version = "py39" [tool.ruff.lint] select = ["E", "F", "W", "I", "UP", "B"] +[tool.mypy] + +[[tool.mypy.overrides]] +# playwright is the optional [browser] extra, not installed for the type-check +# job — Scraper.browser() imports it lazily behind a try/except. Ignore its +# missing stubs instead of pulling the heavy browser stack into CI. +module = ["playwright.*"] +ignore_missing_imports = true + [tool.vercel] entrypoint = "landing.app:app" diff --git a/src/bytecrawl/__init__.py b/src/bytecrawl/__init__.py index 8a15b45..beca973 100644 --- a/src/bytecrawl/__init__.py +++ b/src/bytecrawl/__init__.py @@ -36,8 +36,17 @@ ) __all__ = [ - "Scraper", "Page", "Session", - "Crawler", "BFS", "SharkSearch", "OPIC", "CrawlResult", "STRATEGIES", - "pagerank", "cosine", "compare", + "Scraper", + "Page", + "Session", + "Crawler", + "BFS", + "SharkSearch", + "OPIC", + "CrawlResult", + "STRATEGIES", + "pagerank", + "cosine", + "compare", ] -__version__ = "1.2.2" +__version__ = "1.3.0" diff --git a/src/bytecrawl/crawl/__init__.py b/src/bytecrawl/crawl/__init__.py index 45bef4e..30bab17 100644 --- a/src/bytecrawl/crawl/__init__.py +++ b/src/bytecrawl/crawl/__init__.py @@ -27,6 +27,15 @@ from .text import cosine, relevance __all__ = [ - "Crawler", "BFS", "SharkSearch", "OPIC", "CrawlResult", "Frontier", - "STRATEGIES", "cosine", "relevance", "pagerank", "compare", + "Crawler", + "BFS", + "SharkSearch", + "OPIC", + "CrawlResult", + "Frontier", + "STRATEGIES", + "cosine", + "relevance", + "pagerank", + "compare", ] diff --git a/src/bytecrawl/crawl/base.py b/src/bytecrawl/crawl/base.py index 47f471b..090399d 100644 --- a/src/bytecrawl/crawl/base.py +++ b/src/bytecrawl/crawl/base.py @@ -31,15 +31,17 @@ class Crawler: name = "bfs" - def __init__(self, query: str = "", delay: float = 0.2, timeout: int = 10, - same_domain: bool = True): + def __init__( + self, query: str = "", delay: float = 0.2, timeout: int = 10, same_domain: bool = True + ): self.query = query self.same_domain = same_domain self.scraper = Scraper(delay=delay, timeout=timeout) # --- extension point ------------------------------------------------------- - def score_links(self, url: str, links: list[dict], page_relevance: float, - depth: int) -> list[tuple[str, float]]: + def score_links( + self, url: str, links: list[dict], page_relevance: float, depth: int + ) -> list[tuple[str, float]]: """BFS: the score only encodes depth (shallower = visited sooner). links: [{url, anchor}]. Returns [(url, score)] for the frontier. @@ -89,7 +91,7 @@ def crawl(self, start: str, max_pages: int = 50, max_depth: int = 10) -> CrawlRe links = [] for a in page.soup.select("a[href]"): - child = normalize(a["href"], url) + child = normalize(str(a["href"]), url) if not child or child == url: continue if self.same_domain and _root_domain(urlparse(child).netloc) != domain: @@ -97,11 +99,16 @@ def crawl(self, start: str, max_pages: int = 50, max_depth: int = 10) -> CrawlRe links.append({"url": child, "anchor": a.get_text(" ", strip=True)}) result.graph[url] = [link["url"] for link in links] - result.pages.append({ - "url": url, "title": title[:120], "score": round(score, 4), - "relevance": round(page_relevance, 4), "depth": depth, - "order": len(result.pages) + 1, - }) + result.pages.append( + { + "url": url, + "title": title[:120], + "score": round(score, 4), + "relevance": round(page_relevance, 4), + "depth": depth, + "order": len(result.pages) + 1, + } + ) self.on_visit(url, links) @@ -117,8 +124,8 @@ def crawl(self, start: str, max_pages: int = 50, max_depth: int = 10) -> CrawlRe "elapsed": round(time.perf_counter() - t0, 2), "frontier_left": len(frontier), "relevant_found": len(result.relevant()) if self.query else None, - "avg_relevance": round( - sum(p["relevance"] for p in result.pages) / len(result.pages), 4 - ) if result.pages else 0.0, + "avg_relevance": round(sum(p["relevance"] for p in result.pages) / len(result.pages), 4) + if result.pages + else 0.0, } return result diff --git a/src/bytecrawl/crawl/compare.py b/src/bytecrawl/crawl/compare.py index 432e650..7aca8d9 100644 --- a/src/bytecrawl/crawl/compare.py +++ b/src/bytecrawl/crawl/compare.py @@ -7,13 +7,15 @@ from .strategies import STRATEGIES -def _leg(name: str, start: str, query: str, max_pages: int, max_depth: int, - delay: float, timeout: int) -> dict: +def _leg( + name: str, start: str, query: str, max_pages: int, max_depth: int, delay: float, timeout: int +) -> dict: """One strategy's leg of a comparison. Never raises: a strategy that fails reports inside its own entry so the other two still come back.""" try: result = STRATEGIES[name](query=query, delay=delay, timeout=timeout).crawl( - start, max_pages=max_pages, max_depth=max_depth) + start, max_pages=max_pages, max_depth=max_depth + ) except Exception as e: return {"error": f"{type(e).__name__}: {e}"} return { @@ -23,8 +25,14 @@ def _leg(name: str, start: str, query: str, max_pages: int, max_depth: int, } -def compare(start: str, query: str, max_pages: int = 20, max_depth: int = 4, - delay: float = 0.2, timeout: int = 10) -> dict: +def compare( + start: str, + query: str, + max_pages: int = 20, + max_depth: int = 4, + delay: float = 0.2, + timeout: int = 10, +) -> dict: """Run every strategy over one site on the same budget, side by side. A single ordering looks like any other crawler's. The case for having @@ -49,13 +57,16 @@ def compare(start: str, query: str, max_pages: int = 20, max_depth: int = 4, if not query: raise ValueError( "compare needs a query: with nothing to be relevant to, the " - "strategies aren't comparable (use a single crawler instead)") + "strategies aren't comparable (use a single crawler instead)" + ) names = list(STRATEGIES) with ThreadPoolExecutor(max_workers=len(names)) as pool: - legs = list(pool.map( - lambda n: _leg(n, start, query, max_pages, max_depth, - delay * len(names), timeout), - names)) + legs = list( + pool.map( + lambda n: _leg(n, start, query, max_pages, max_depth, delay * len(names), timeout), + names, + ) + ) results = dict(zip(names, legs)) ok = {n: r for n, r in results.items() if "error" not in r} diff --git a/src/bytecrawl/crawl/graph.py b/src/bytecrawl/crawl/graph.py index 780068d..f10d2da 100644 --- a/src/bytecrawl/crawl/graph.py +++ b/src/bytecrawl/crawl/graph.py @@ -8,8 +8,9 @@ from __future__ import annotations -def pagerank(graph: dict[str, list[str]], damping: float = 0.85, - iterations: int = 30) -> dict[str, float]: +def pagerank( + graph: dict[str, list[str]], damping: float = 0.85, iterations: int = 30 +) -> dict[str, float]: """PageRank via power iteration over the crawled graph. Offline version of the same concept OPIC approximates online: comparing diff --git a/src/bytecrawl/crawl/result.py b/src/bytecrawl/crawl/result.py index 5fe6f75..9b400e8 100644 --- a/src/bytecrawl/crawl/result.py +++ b/src/bytecrawl/crawl/result.py @@ -10,7 +10,7 @@ class CrawlResult: """What a crawl returns: pages, graph and numbers for comparison.""" strategy: str - pages: list[dict] = field(default_factory=list) # url, title, score, relevance, depth, order + pages: list[dict] = field(default_factory=list) # url, title, score, relevance, depth, order graph: dict[str, list[str]] = field(default_factory=dict) stats: dict = field(default_factory=dict) diff --git a/src/bytecrawl/crawl/strategies.py b/src/bytecrawl/crawl/strategies.py index 02e1fdb..86a998b 100644 --- a/src/bytecrawl/crawl/strategies.py +++ b/src/bytecrawl/crawl/strategies.py @@ -51,7 +51,7 @@ def score_links(self, url, links, page_relevance, depth): scored = [] for link in links: url_words = " ".join(_tokens(urlparse(link["url"]).path)) - local = relevance(f'{link["anchor"]} {url_words}', self.query) + local = relevance(f"{link['anchor']} {url_words}", self.query) score = self.gamma * inherited + (1 - self.gamma) * local self._inherited[link["url"]] = inherited scored.append((link["url"], score)) @@ -104,5 +104,7 @@ def score_links(self, url, links, page_relevance, depth): # strategy, so a crawl requested over HTTP, over MCP or in Python means the # same thing everywhere. STRATEGIES: dict[str, type[Crawler]] = { - "bfs": BFS, "shark": SharkSearch, "opic": OPIC, + "bfs": BFS, + "shark": SharkSearch, + "opic": OPIC, } diff --git a/src/bytecrawl/crawl/text.py b/src/bytecrawl/crawl/text.py index 1783768..7a7473b 100644 --- a/src/bytecrawl/crawl/text.py +++ b/src/bytecrawl/crawl/text.py @@ -38,7 +38,7 @@ def cosine(text_a: str, text_b: str) -> float: @lru_cache(maxsize=4096) def _trigrams(word: str) -> frozenset: padded = f"~{word}~" - return frozenset(padded[i:i + 3] for i in range(len(padded) - 2)) + return frozenset(padded[i : i + 3] for i in range(len(padded) - 2)) def _dice(a: frozenset, b: frozenset) -> float: @@ -81,7 +81,7 @@ def _resolve_terms(query: str, vocabulary: set) -> str: score = _dice(grams, _trigrams(word)) if score > best_score: best, best_score = word, score - resolved.append(best if best_score >= _FUZZY_MIN else term) + resolved.append(best if best is not None and best_score >= _FUZZY_MIN else term) return " ".join(resolved) diff --git a/src/bytecrawl/mcp_http.py b/src/bytecrawl/mcp_http.py index 110c8f6..b51bcdb 100644 --- a/src/bytecrawl/mcp_http.py +++ b/src/bytecrawl/mcp_http.py @@ -38,13 +38,13 @@ from . import mcp_server as local from .security import assert_public_url -MAX_PAGES = 10 # hard cap per focused_crawl call (local allows 50) +MAX_PAGES = 10 # hard cap per focused_crawl call (local allows 50) # compare_strategies is three crawls that the rate limiter charges as one # request, so it gets a smaller budget: 3 x 6 = 18 fetches, about two # focused_crawls' worth. Same bounded-cost rule, applied to a heavier call. MAX_COMPARE_PAGES = 6 -RATE_LIMIT = 20 # requests per window per client IP -RATE_WINDOW = 60.0 # seconds +RATE_LIMIT = 20 # requests per window per client IP +RATE_WINDOW = 60.0 # seconds def _guarded(url: str) -> None: @@ -93,16 +93,19 @@ def _client_ip(scope) -> str: async def __call__(self, scope, receive, send): if scope["type"] == "http" and not self.limiter.allow(self._client_ip(scope)): - await send({ - "type": "http.response.start", - "status": 429, - "headers": [(b"content-type", b"application/json"), - (b"retry-after", b"60")], - }) - await send({ - "type": "http.response.body", - "body": b'{"error": "rate limit exceeded, try again in a minute"}', - }) + await send( + { + "type": "http.response.start", + "status": 429, + "headers": [(b"content-type", b"application/json"), (b"retry-after", b"60")], + } + ) + await send( + { + "type": "http.response.body", + "body": b'{"error": "rate limit exceeded, try again in a minute"}', + } + ) return await self.app(scope, receive, send) @@ -119,64 +122,75 @@ async def __call__(self, scope, receive, send): ) -@server.tool(description=( - "Fetch a public web page and return it as clean Markdown for LLM " - "consumption, with tokens_estimate against tokens_html so you can " - "state the saving. Static HTML only on the hosted server." -)) +@server.tool( + description=( + "Fetch a public web page and return it as clean Markdown for LLM " + "consumption, with tokens_estimate against tokens_html so you can " + "state the saving. Static HTML only on the hosted server." + ) +) def fetch_markdown(url: str) -> dict: _guarded(url) return local.fetch_markdown(url) -@server.tool(description=( - "Extract data from a public page with CSS selectors. Call it with ONLY the " - "url to discover what is extractable — it returns the page's repeated " - "blocks with ready-to-use item + fields and a sample record each, which is " - "what you need when you have not seen the markup. Then 'item'+'fields' for " - "structured records (::text / ::attr(name); a field name ending in [] " - "collects a list), or 'select' on its own for a flat list of one " - "selector's values." -)) -def extract(url: str, item: str = "", fields: dict[str, str] | None = None, - select: str = "") -> dict: +@server.tool( + description=( + "Extract data from a public page with CSS selectors. Call it with ONLY the " + "url to discover what is extractable — it returns the page's repeated " + "blocks with ready-to-use item + fields and a sample record each, which is " + "what you need when you have not seen the markup. Then 'item'+'fields' for " + "structured records (::text / ::attr(name); a field name ending in [] " + "collects a list), or 'select' on its own for a flat list of one " + "selector's values." + ) +) +def extract( + url: str, item: str = "", fields: dict[str, str] | None = None, select: str = "" +) -> dict: _guarded(url) return local.extract(url, item=item, fields=fields, select=select) -@server.tool(description=( - "Every outbound link on a public page as an absolute, deduplicated URL " - "(#fragments, mailto: and asset files dropped). raw=true returns the " - "untouched href values." -)) +@server.tool( + description=( + "Every outbound link on a public page as an absolute, deduplicated URL " + "(#fragments, mailto: and asset files dropped). raw=true returns the " + "untouched href values." + ) +) def list_links(url: str, raw: bool = False) -> dict: _guarded(url) return local.list_links(url, raw=raw) -@server.tool(description=( - "Focused crawl of a public site: visit the pages most relevant to 'query' " - "first. strategy: shark (default) | opic | bfs. Capped at " - f"{MAX_PAGES} pages per call on the hosted server." -)) -def focused_crawl(url: str, query: str = "", strategy: str = "shark", - max_pages: int = MAX_PAGES) -> dict: +@server.tool( + description=( + "Focused crawl of a public site: visit the pages most relevant to 'query' " + "first. strategy: shark (default) | opic | bfs. Capped at " + f"{MAX_PAGES} pages per call on the hosted server." + ) +) +def focused_crawl( + url: str, query: str = "", strategy: str = "shark", max_pages: int = MAX_PAGES +) -> dict: _guarded(url) - return local.focused_crawl(url, query=query, strategy=strategy, - max_pages=min(max_pages, MAX_PAGES)) - - -@server.tool(description=( - "Run Shark-Search, OPIC and BFS over the same public site on the same " - "budget and return them side by side with a winner. Three crawls in one " - f"call, so it is capped at {MAX_COMPARE_PAGES} pages per strategy on the " - "hosted server." -)) -def compare_strategies(url: str, query: str, - max_pages: int = MAX_COMPARE_PAGES) -> dict: + return local.focused_crawl( + url, query=query, strategy=strategy, max_pages=min(max_pages, MAX_PAGES) + ) + + +@server.tool( + description=( + "Run Shark-Search, OPIC and BFS over the same public site on the same " + "budget and return them side by side with a winner. Three crawls in one " + f"call, so it is capped at {MAX_COMPARE_PAGES} pages per strategy on the " + "hosted server." + ) +) +def compare_strategies(url: str, query: str, max_pages: int = MAX_COMPARE_PAGES) -> dict: _guarded(url) - return local.compare_strategies(url, query=query, - max_pages=min(max_pages, MAX_COMPARE_PAGES)) + return local.compare_strategies(url, query=query, max_pages=min(max_pages, MAX_COMPARE_PAGES)) @server.tool(description="Fetch a public JSON API endpoint.") @@ -204,12 +218,14 @@ def _transport_security() -> TransportSecuritySettings: and BYTECRAWL_ALLOWED_ORIGINS alongside it) to turn it back on — worth doing if you self-host on a domain that also serves authenticated apps. """ - hosts = [h.strip() for h in - os.environ.get("BYTECRAWL_ALLOWED_HOSTS", "").split(",") if h.strip()] + hosts = [ + h.strip() for h in os.environ.get("BYTECRAWL_ALLOWED_HOSTS", "").split(",") if h.strip() + ] if not hosts: return TransportSecuritySettings(enable_dns_rebinding_protection=False) - origins = [o.strip() for o in - os.environ.get("BYTECRAWL_ALLOWED_ORIGINS", "").split(",") if o.strip()] + origins = [ + o.strip() for o in os.environ.get("BYTECRAWL_ALLOWED_ORIGINS", "").split(",") if o.strip() + ] return TransportSecuritySettings( enable_dns_rebinding_protection=True, allowed_hosts=hosts, @@ -221,7 +237,9 @@ def create_app(limiter: RateLimiter | None = None): """ASGI app: streamable-HTTP MCP at /mcp, rate-limited. Stateless for serverless (json_response avoids long-lived SSE streams on Vercel).""" app = server.streamable_http_app( - streamable_http_path="/mcp", stateless_http=True, json_response=True, + streamable_http_path="/mcp", + stateless_http=True, + json_response=True, transport_security=_transport_security(), ) return RateLimitMiddleware(app, limiter) @@ -231,7 +249,9 @@ def main() -> None: """Entry point for bytecrawl-mcp-http (self-hosted).""" import uvicorn - uvicorn.run(create_app(), host="0.0.0.0", port=8000) + # nosec B104 — an MCP HTTP server must bind all interfaces to be reachable + # inside a container or serverless host. + uvicorn.run(create_app(), host="0.0.0.0", port=8000) # nosec B104 if __name__ == "__main__": diff --git a/src/bytecrawl/mcp_server.py b/src/bytecrawl/mcp_server.py index 57381b9..12931a2 100644 --- a/src/bytecrawl/mcp_server.py +++ b/src/bytecrawl/mcp_server.py @@ -16,9 +16,7 @@ try: from mcp.server.mcpserver import MCPServer except ImportError as e: # pragma: no cover - exercised only without the extra - raise ImportError( - "The MCP server needs the 'mcp' extra: pip install bytecrawl[mcp]" - ) from e + raise ImportError("The MCP server needs the 'mcp' extra: pip install bytecrawl[mcp]") from e from .core import Scraper from .crawler import STRATEGIES as _STRATEGIES @@ -79,8 +77,9 @@ def fetch_markdown(url: str) -> dict: "selector's values across the page, e.g. 'h3 a::attr(title)'." ) ) -def extract(url: str, item: str = "", fields: dict[str, str] | None = None, - select: str = "") -> dict: +def extract( + url: str, item: str = "", fields: dict[str, str] | None = None, select: str = "" +) -> dict: # Three shapes, one tool. Agents reach for a single selector far more # often than for a record schema, and an agent that has never seen the # page can produce neither: fetch_markdown strips exactly the classes a @@ -91,26 +90,28 @@ def extract(url: str, item: str = "", fields: dict[str, str] | None = None, if select and (item or fields): raise ValueError("pass either 'select' or 'item'+'fields', not both") if bool(item) != bool(fields): - raise ValueError("'item' and 'fields' go together — or pass neither " - "to see what this page offers") + raise ValueError( + "'item' and 'fields' go together — or pass neither to see what this page offers" + ) # Checked before the fetch: a malformed call shouldn't cost the site a # request. page = _scraper.fetch(url) if select: values = page.css_all(select) - return {"url": url, "select": select, "count": len(values), - "values": values} + return {"url": url, "select": select, "count": len(values), "values": values} if item: - records = page.extract(item, fields) + records = page.extract(item, fields or {}) return {"url": url, "count": len(records), "records": records} candidates = page.selectors() return { "url": url, "candidates": candidates, - "hint": ("call extract again with the item and fields of whichever " - "candidate holds what you want" if candidates else - "no repeated blocks here — this page is probably not a " - "listing; pass 'select' with a specific selector instead"), + "hint": ( + "call extract again with the item and fields of whichever candidate holds what you want" + if candidates + else "no repeated blocks here — this page is probably not a " + "listing; pass 'select' with a specific selector instead" + ), } @@ -136,8 +137,7 @@ def list_links(url: str, raw: bool = False) -> dict: "(level by level). Returns pages ranked by relevance plus crawl stats." ) ) -def focused_crawl(url: str, query: str = "", strategy: str = "shark", - max_pages: int = 20) -> dict: +def focused_crawl(url: str, query: str = "", strategy: str = "shark", max_pages: int = 20) -> dict: if strategy not in _STRATEGIES: raise ValueError(f"strategy must be one of {sorted(_STRATEGIES)}") max_pages = min(max_pages, 50) # keep agent calls bounded and polite @@ -148,9 +148,7 @@ def focused_crawl(url: str, query: str = "", strategy: str = "shark", "strategy": strategy, "stats": result.stats, "pages": result.top(max_pages) if query else result.pages, - "pagerank_top": [ - {"url": u, "rank": round(r, 4)} for u, r in list(ranks.items())[:10] - ], + "pagerank_top": [{"url": u, "rank": round(r, 4)} for u, r in list(ranks.items())[:10]], } @@ -170,9 +168,7 @@ def compare_strategies(url: str, query: str, max_pages: int = 20) -> dict: return {"url": url, "query": query, "max_pages": max_pages, **out} -@server.tool( - description="Fetch a JSON API endpoint (the 'hidden API' scraping technique)." -) +@server.tool(description="Fetch a JSON API endpoint (the 'hidden API' scraping technique).") def fetch_json_api(url: str, params: dict | None = None) -> dict: page = _scraper.api(url, params=params) return {"url": url, "status": page.status, "data": page.json()} diff --git a/src/bytecrawl/scraping/page.py b/src/bytecrawl/scraping/page.py index 92a4aa9..aef61bc 100644 --- a/src/bytecrawl/scraping/page.py +++ b/src/bytecrawl/scraping/page.py @@ -63,7 +63,7 @@ def extract(self, item: str, fields: dict[str, str]) -> list[dict]: """ records = [] for node in self.soup.select(item): - row = {} + row: dict[str, Any] = {} for name, spec in fields.items(): sel, op, arg = _split_selector(spec) if name.endswith("[]"): @@ -138,8 +138,11 @@ def sig(node) -> str | None: items = {s for s, _, _, _ in found} out = [] for s, count, fields, node in found: - if any(count == c2 and set(fields) <= set(f2) and node in n2.parents - for s2, c2, f2, n2 in found if s2 != s): + if any( + count == c2 and set(fields) <= set(f2) and node in n2.parents + for s2, c2, f2, n2 in found + if s2 != s + ): continue # The other tell of a wrapper: one of its columns is an entire # other record. `li.col-xs-6` offers `article.product_pod::text` @@ -147,8 +150,9 @@ def sig(node) -> str | None: if any(spec.split("::")[0] in items for spec in fields.values()): continue sample = self.extract(s, fields) - out.append({"item": s, "count": count, "fields": fields, - "sample": sample[0] if sample else {}}) + out.append( + {"item": s, "count": count, "fields": fields, "sample": sample[0] if sample else {}} + ) if len(out) == limit: break return out @@ -180,8 +184,10 @@ def _fields_of(nodes: list, coverage: float = 0.6) -> dict[str, str]: need = max(1, int(len(nodes) * coverage)) # Named classes before bare tags, then shortest: `p.price_color::text` # is a column, `p::text` is whichever paragraph happened to be first. - keep = sorted((s for s, n in hits.items() if n >= need), - key=lambda s: ("." not in s.split("::")[0], len(s))) + keep = sorted( + (s for s, n in hits.items() if n >= need), + key=lambda s: ("." not in s.split("::")[0], len(s)), + ) fields, used = {}, set() for spec in keep: @@ -195,8 +201,9 @@ def _fields_of(nodes: list, coverage: float = 0.6) -> dict[str, str]: # A selector that resolves everywhere but is blank everywhere is # not a column. Check against the instances, not the markup. op, arg = _split_selector(spec)[1:] - if not any(_value_from(n.select_one(sel), op, arg) - for n in nodes[:5] if n.select_one(sel)): + if not any( + _value_from(n.select_one(sel), op, arg) for n in nodes[:5] if n.select_one(sel) + ): continue used.add(sel) fields[name] = spec @@ -208,13 +215,19 @@ def _fields_of(nodes: list, coverage: float = 0.6) -> dict[str, str]: # "£51.77In stockAdd to basket" — three fields glued into one, and # never what the caller wanted from any of them. probe = nodes[0] - inner = {sel for sel in used - if (n := probe.select_one(sel)) is not None - and any(n is not m and m in n.descendants - for other in used - if (m := probe.select_one(other)) is not None)} - return {k: v for k, v in fields.items() - if _split_selector(v)[0] not in inner or "::attr" in v} + inner = { + sel + for sel in used + if (n := probe.select_one(sel)) is not None + and any( + n is not m and m in n.descendants + for other in used + if (m := probe.select_one(other)) is not None + ) + } + return { + k: v for k, v in fields.items() if _split_selector(v)[0] not in inner or "::attr" in v + } def links(self, raw: bool = False) -> list[str]: """Every link on the page as an absolute URL, deduplicated in order. @@ -225,7 +238,7 @@ def links(self, raw: bool = False) -> list[str]: use, so what you get back is the set of pages you could visit next. Pass raw=True for the untouched attribute values. """ - hrefs = [a["href"] for a in self.soup.select("a[href]")] + hrefs = [str(a["href"]) for a in self.soup.select("a[href]")] if raw: return hrefs out: list[str] = [] @@ -261,8 +274,11 @@ def _extraction_gutted_it(self, md: str) -> bool: visible = len(self.soup.get_text(" ", strip=True)) if visible and len(md) / visible < self._MAIN_CONTENT_FLOOR: return True - headings = [h.get_text(strip=True) - for h in self.soup.select("h1,h2,h3,h4") if h.get_text(strip=True)] + headings = [ + h.get_text(strip=True) + for h in self.soup.select("h1,h2,h3,h4") + if h.get_text(strip=True) + ] if len(headings) < self._MANY_HEADINGS: return False kept = sum(1 for h in headings if h[:30] in md) diff --git a/src/bytecrawl/scraping/scraper.py b/src/bytecrawl/scraping/scraper.py index 2631707..fd7129b 100644 --- a/src/bytecrawl/scraping/scraper.py +++ b/src/bytecrawl/scraping/scraper.py @@ -34,8 +34,13 @@ def static(self, url: str) -> Page: r = self._session.get(url, timeout=self.timeout) r.raise_for_status() self._wait() - return Page(url=url, html=_decoded_html(r), method="static", - elapsed=round(time.perf_counter() - t0, 3), status=r.status_code) + return Page( + url=url, + html=_decoded_html(r), + method="static", + elapsed=round(time.perf_counter() - t0, 3), + status=r.status_code, + ) def api(self, url: str, params: dict | None = None) -> Page: """Technique 3: requests an API and stores the JSON.""" @@ -48,8 +53,13 @@ def api(self, url: str, params: dict | None = None) -> Page: except ValueError as e: ctype = r.headers.get("content-type", "unknown") raise ValueError(f"{url} did not return JSON (content-type: {ctype})") from e - return Page(url=url, data=data, method="api", - elapsed=round(time.perf_counter() - t0, 3), status=r.status_code) + return Page( + url=url, + data=data, + method="api", + elapsed=round(time.perf_counter() - t0, 3), + status=r.status_code, + ) def browser(self, url: str, wait: str | None = None, scroll: bool = False) -> Page: """Technique 2: real browser (Playwright) for JS-rendered sites.""" @@ -75,8 +85,13 @@ def browser(self, url: str, wait: str | None = None, scroll: bool = False) -> Pa status = resp.status if resp else 0 nav.close() self._wait() - return Page(url=url, html=html, method="browser", - elapsed=round(time.perf_counter() - t0, 3), status=status) + return Page( + url=url, + html=html, + method="browser", + elapsed=round(time.perf_counter() - t0, 3), + status=status, + ) def fetch(self, url: str, strategy: str = "auto") -> Page: """Fetches the page. strategy: auto | static | browser. @@ -116,7 +131,7 @@ def crawl( """ from urllib.parse import urljoin - url = start + url: str | None = start results: list[dict] = [] n = 0 while url: diff --git a/src/bytecrawl/scraping/session.py b/src/bytecrawl/scraping/session.py index 61753ef..981a15f 100644 --- a/src/bytecrawl/scraping/session.py +++ b/src/bytecrawl/scraping/session.py @@ -49,5 +49,10 @@ def fetch(self, url: str) -> Page: t0 = time.perf_counter() r = self._s.get(url, timeout=self._scraper.timeout) r.raise_for_status() - return Page(url=url, html=_decoded_html(r), method="static", - elapsed=round(time.perf_counter() - t0, 3), status=r.status_code) + return Page( + url=url, + html=_decoded_html(r), + method="static", + elapsed=round(time.perf_counter() - t0, 3), + status=r.status_code, + ) diff --git a/src/bytecrawl/security.py b/src/bytecrawl/security.py index d0af9a2..437bd3d 100644 --- a/src/bytecrawl/security.py +++ b/src/bytecrawl/security.py @@ -32,6 +32,4 @@ def assert_public_url(url: str) -> None: for info in infos: ip = ipaddress.ip_address(info[4][0]) if not ip.is_global or ip.is_multicast: - raise ValueError( - f"URL resolves to a non-public address ({ip}) — refusing to fetch it" - ) + raise ValueError(f"URL resolves to a non-public address ({ip}) — refusing to fetch it") diff --git a/tests/conftest.py b/tests/conftest.py index 7620afe..d6ca5d4 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -16,8 +16,15 @@ class FakeResponse: """Minimal stand-in for requests.Response.""" - def __init__(self, text="", status=200, json_data=None, headers=None, - encoding="utf-8", apparent_encoding="utf-8"): + def __init__( + self, + text="", + status=200, + json_data=None, + headers=None, + encoding="utf-8", + apparent_encoding="utf-8", + ): self.status_code = status self.content = text.encode(apparent_encoding) self.encoding = encoding @@ -70,16 +77,22 @@ def _post(self, url, **kw): # /cats/* is not. One external link tests the same_domain filter. SITE = { "https://s.test/": ( - "Home", "welcome to the machine learning intro site", - [("/ml", "machine learning"), ("/cats", "cat pictures"), - ("https://other.test/x", "external")], + "Home", + "welcome to the machine learning intro site", + [ + ("/ml", "machine learning"), + ("/cats", "cat pictures"), + ("https://other.test/x", "external"), + ], ), "https://s.test/ml": ( - "ML", "machine learning and deep learning models machine learning", + "ML", + "machine learning and deep learning models machine learning", [("/ml/deep", "deep learning"), ("/", "home")], ), "https://s.test/cats": ( - "Cats", "cats cats cats pictures of cats", + "Cats", + "cats cats cats pictures of cats", [("/cats/more", "more cats")], ), "https://s.test/ml/deep": ("Deep", "deep neural networks training", []), @@ -90,8 +103,7 @@ def _post(self, url, **kw): def render(url: str) -> str: title, text, links = SITE[url] anchors = "".join(f'{a}' for h, a in links) - return (f"{title}" - f"

{text}

{anchors}") + return f"{title}

{text}

{anchors}" @pytest.fixture diff --git a/tests/test_browser_live.py b/tests/test_browser_live.py index f6e8992..5962aab 100644 --- a/tests/test_browser_live.py +++ b/tests/test_browser_live.py @@ -21,6 +21,7 @@ @pytest.fixture(scope="module") def chromium_available(): from playwright.sync_api import sync_playwright + try: with sync_playwright() as p: p.chromium.launch(headless=True).close() diff --git a/tests/test_core.py b/tests/test_core.py index 4ba5878..eaba67c 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -36,8 +36,7 @@ class TestEncoding: def test_iso_8859_1_fallback_uses_apparent_encoding(self): # requests defaults to ISO-8859-1 when the header has no charset # (RFC 2616); for a UTF-8 body that turns £ into £. - r = FakeResponse(text="price £10", encoding="iso-8859-1", - apparent_encoding="utf-8") + r = FakeResponse(text="price £10", encoding="iso-8859-1", apparent_encoding="utf-8") assert _decoded_html(r) == "price £10" def test_declared_encoding_is_respected(self): @@ -48,8 +47,8 @@ def test_declared_encoding_is_respected(self): class TestApi: def test_json_page(self, http): http.routes["https://api.test/items"] = FakeResponse( - json_data={"items": [1, 2, 3]}, - headers={"content-type": "application/json"}) + json_data={"items": [1, 2, 3]}, headers={"content-type": "application/json"} + ) page = Scraper().api("https://api.test/items") assert page.method == "api" assert page.json() == {"items": [1, 2, 3]} @@ -64,23 +63,24 @@ class TestFetchAuto: def test_content_rich_page_stays_static(self, http, monkeypatch): http.routes["https://x.test/"] = FakeResponse(text=LONG_HTML) called = [] - monkeypatch.setattr(Scraper, "browser", - lambda self, url, **kw: called.append(url)) + monkeypatch.setattr(Scraper, "browser", lambda self, url, **kw: called.append(url)) page = Scraper().fetch("https://x.test/") assert page.method == "static" assert called == [] def test_sparse_page_falls_back_to_browser(self, http, monkeypatch): http.routes["https://spa.test/"] = FakeResponse( - text="
") + text="
" + ) sentinel = object() monkeypatch.setattr(Scraper, "browser", lambda self, url, **kw: sentinel) assert Scraper().fetch("https://spa.test/") is sentinel def test_explicit_static_never_falls_back(self, http, monkeypatch): http.routes["https://spa.test/"] = FakeResponse(text="") - monkeypatch.setattr(Scraper, "browser", - lambda self, url, **kw: pytest.fail("browser called")) + monkeypatch.setattr( + Scraper, "browser", lambda self, url, **kw: pytest.fail("browser called") + ) page = Scraper().fetch("https://spa.test/", strategy="static") assert page.method == "static" @@ -101,21 +101,27 @@ def test_follows_next_page_link(self, http): http.routes["https://x.test/page1"] = FakeResponse(text=ITEMS_P1) http.routes["https://x.test/page2"] = FakeResponse(text=ITEMS_P2) rows = Scraper().crawl( - "https://x.test/page1", item="article.item", + "https://x.test/page1", + item="article.item", fields={"name": "h3::text"}, - next_page="li.next a::attr(href)") + next_page="li.next a::attr(href)", + ) assert [r["name"] for r in rows] == ["A", "B", "C"] def test_pages_limit_stops_early(self, http): http.routes["https://x.test/page1"] = FakeResponse(text=ITEMS_P1) rows = Scraper().crawl( - "https://x.test/page1", item="article.item", + "https://x.test/page1", + item="article.item", fields={"name": "h3::text"}, - next_page="li.next a::attr(href)", pages=1) + next_page="li.next a::attr(href)", + pages=1, + ) assert [r["name"] for r in rows] == ["A", "B"] def test_no_next_page_selector_single_page(self, http): http.routes["https://x.test/page1"] = FakeResponse(text=ITEMS_P1) - rows = Scraper().crawl("https://x.test/page1", item="article.item", - fields={"name": "h3::text"}) + rows = Scraper().crawl( + "https://x.test/page1", item="article.item", fields={"name": "h3::text"} + ) assert len(rows) == 2 diff --git a/tests/test_crawler_strategies.py b/tests/test_crawler_strategies.py index f432135..05a42b8 100644 --- a/tests/test_crawler_strategies.py +++ b/tests/test_crawler_strategies.py @@ -22,8 +22,7 @@ def test_visits_level_by_level(self, fake_site): # Seed first, then both depth-1 pages before any depth-2 page. assert fake_site[0] == SEED assert set(fake_site[1:3]) == {"https://s.test/ml", "https://s.test/cats"} - assert set(fake_site[3:5]) == {"https://s.test/ml/deep", - "https://s.test/cats/more"} + assert set(fake_site[3:5]) == {"https://s.test/ml/deep", "https://s.test/cats/more"} def test_max_depth_excludes_deeper_pages(self, fake_site): result = BFS(**crawler_kwargs()).crawl(SEED, max_pages=10, max_depth=1) @@ -53,29 +52,23 @@ def test_invalid_url_raises(self): class TestSharkSearch: def test_relevant_branch_visited_before_irrelevant(self, fake_site): - SharkSearch(query="machine learning", **crawler_kwargs()).crawl( - SEED, max_pages=4) - assert fake_site.index("https://s.test/ml") < \ - fake_site.index("https://s.test/cats") + SharkSearch(query="machine learning", **crawler_kwargs()).crawl(SEED, max_pages=4) + assert fake_site.index("https://s.test/ml") < fake_site.index("https://s.test/cats") def test_irrelevant_branch_decays_geometrically(self, fake_site): - shark = SharkSearch(query="machine learning", delta=0.5, - **crawler_kwargs()) + shark = SharkSearch(query="machine learning", delta=0.5, **crawler_kwargs()) shark.crawl(SEED, max_pages=5) # /cats has no query overlap, so /cats/more inherits delta * inherited(/cats). cats = shark._inherited["https://s.test/cats"] - assert shark._inherited["https://s.test/cats/more"] == \ - pytest.approx(shark.delta * cats) + assert shark._inherited["https://s.test/cats/more"] == pytest.approx(shark.delta * cats) def test_relevance_scores_favor_ml_pages(self, fake_site): - result = SharkSearch(query="machine learning", - **crawler_kwargs()).crawl(SEED, max_pages=5) + result = SharkSearch(query="machine learning", **crawler_kwargs()).crawl(SEED, max_pages=5) by_url = {p["url"]: p["relevance"] for p in result.pages} assert by_url["https://s.test/ml"] > by_url["https://s.test/cats"] def test_top_orders_by_relevance(self, fake_site): - result = SharkSearch(query="machine learning", - **crawler_kwargs()).crawl(SEED, max_pages=5) + result = SharkSearch(query="machine learning", **crawler_kwargs()).crawl(SEED, max_pages=5) top = result.top(3) assert top[0]["relevance"] >= top[1]["relevance"] >= top[2]["relevance"] @@ -101,13 +94,13 @@ def test_history_accumulates_importance(self, fake_site): def test_visits_all_reachable_pages(self, fake_site): result = OPIC(**crawler_kwargs()).crawl(SEED, max_pages=10) - assert {p["url"] for p in result.pages} == set( - u for u in fake_site) + assert {p["url"] for p in result.pages} == set(u for u in fake_site) class TestErrorHandling: def test_fetch_errors_counted_not_fatal(self, fake_site, monkeypatch): from bytecrawl.crawler import Crawler + original = Crawler.score_links def with_broken_link(self, url, links, relevance, depth): diff --git a/tests/test_graph_utils.py b/tests/test_graph_utils.py index 84a06f2..ea28998 100644 --- a/tests/test_graph_utils.py +++ b/tests/test_graph_utils.py @@ -1,6 +1,5 @@ """Pure graph/text utilities: cosine, normalize, _root_domain, pagerank.""" - import pytest from bytecrawl.crawler import ( @@ -43,7 +42,8 @@ def test_one_letter_typos_still_match(self, typo): # A slipped keystroke used to score 0.0, which reads as "broken" # rather than "no match". assert relevance(self.TEXT, typo, self.URL) == pytest.approx( - relevance(self.TEXT, "mystery", self.URL)) + relevance(self.TEXT, "mystery", self.URL) + ) @pytest.mark.parametrize("miss", ["travel", "poetry", "zzzzzz"]) def test_absent_topics_stay_zero(self, miss): @@ -67,8 +67,7 @@ def test_resolves_relative(self): assert normalize("/docs", "https://x.test/a/b") == "https://x.test/docs" def test_strips_fragment(self): - assert normalize("https://x.test/p#section", "https://x.test") == \ - "https://x.test/p" + assert normalize("https://x.test/p#section", "https://x.test") == "https://x.test/p" def test_bare_host_and_root_are_the_same_page(self): # Otherwise the crawler burns two budget slots on one page. @@ -124,11 +123,14 @@ def test_symmetric_graph_equal_ranks(self): class TestCrawlResult: def _result(self): - return CrawlResult(strategy="test", pages=[ - {"url": "a", "relevance": 0.9}, - {"url": "b", "relevance": 0.05}, - {"url": "c", "relevance": 0.5}, - ]) + return CrawlResult( + strategy="test", + pages=[ + {"url": "a", "relevance": 0.9}, + {"url": "b", "relevance": 0.05}, + {"url": "c", "relevance": 0.5}, + ], + ) def test_relevant_filters_by_threshold(self): assert [p["url"] for p in self._result().relevant(0.1)] == ["a", "c"] diff --git a/tests/test_markdown_clean.py b/tests/test_markdown_clean.py index 6269122..a6a4c0b 100644 --- a/tests/test_markdown_clean.py +++ b/tests/test_markdown_clean.py @@ -9,8 +9,9 @@ class TestCarouselDedup: def test_repeated_long_block_kept_once(self): - md = "\n\n".join([LONG_BLOCK, "Something else entirely different here", - LONG_BLOCK, LONG_BLOCK]) + md = "\n\n".join( + [LONG_BLOCK, "Something else entirely different here", LONG_BLOCK, LONG_BLOCK] + ) out = _clean_markdown(md) assert out.count(LONG_BLOCK) == 1 @@ -55,12 +56,19 @@ class TestMainContentFallback: that as a 144x saving. """ - LISTING = ("" + "".join( - f"

Book {i}

" - f"

£{i}.00

" for i in range(20)) + "") + LISTING = ( + "" + + "".join( + f"

Book {i}

" + f"

£{i}.00

" + for i in range(20) + ) + + "" + ) def test_falls_back_when_extraction_guts_the_page(self, monkeypatch): from bytecrawl.core import Page + page = Page(url="https://x.test/", html=self.LISTING) trafilatura = pytest.importorskip("trafilatura") # stand in for what it really does here: return a sliver of the page @@ -70,6 +78,7 @@ def test_falls_back_when_extraction_guts_the_page(self, monkeypatch): def test_keeps_extraction_when_it_looks_sane(self, monkeypatch): from bytecrawl.core import Page + page = Page(url="https://x.test/", html=self.LISTING) trafilatura = pytest.importorskip("trafilatura") full = page.soup.get_text(" ", strip=True) @@ -82,4 +91,5 @@ def test_the_floor_sits_well_below_real_pages(self): would fire on an article buried in navigation and hand back the navigation that was correctly removed.""" from bytecrawl.core import Page + assert Page._MAIN_CONTENT_FLOOR < 0.5 diff --git a/tests/test_mcp_http.py b/tests/test_mcp_http.py index ff064e1..087019a 100644 --- a/tests/test_mcp_http.py +++ b/tests/test_mcp_http.py @@ -14,22 +14,26 @@ class TestSSRFGuard: """IP literals only — no DNS needed except localhost (hosts file).""" - @pytest.mark.parametrize("url", [ - "http://127.0.0.1/admin", # loopback - "http://localhost:8080/", # loopback via hosts file - "http://10.0.0.5/internal", # RFC1918 - "http://172.16.3.1/", # RFC1918 - "http://192.168.1.1/router", # RFC1918 - "http://169.254.169.254/latest/meta-data/", # cloud metadata - "http://[::1]/", # IPv6 loopback - "http://0.0.0.0/", # unspecified - ]) + @pytest.mark.parametrize( + "url", + [ + "http://127.0.0.1/admin", # loopback + "http://localhost:8080/", # loopback via hosts file + "http://10.0.0.5/internal", # RFC1918 + "http://172.16.3.1/", # RFC1918 + "http://192.168.1.1/router", # RFC1918 + "http://169.254.169.254/latest/meta-data/", # cloud metadata + "http://[::1]/", # IPv6 loopback + "http://0.0.0.0/", # unspecified + ], + ) def test_rejects_non_public_addresses(self, url): with pytest.raises(ValueError, match="non-public|resolve"): assert_public_url(url) - @pytest.mark.parametrize("url", ["ftp://example.com/x", "file:///etc/passwd", - "javascript:alert(1)"]) + @pytest.mark.parametrize( + "url", ["ftp://example.com/x", "file:///etc/passwd", "javascript:alert(1)"] + ) def test_rejects_non_http_schemes(self, url): with pytest.raises(ValueError, match="http"): assert_public_url(url) @@ -43,28 +47,31 @@ def test_allows_global_address(self): def test_tools_guard_before_fetching(self, monkeypatch): from mcp.server.mcpserver.exceptions import ToolError + called = [] - monkeypatch.setattr(mcp_http.local, "fetch_markdown", - lambda url: called.append(url)) + monkeypatch.setattr(mcp_http.local, "fetch_markdown", lambda url: called.append(url)) # tools re-raise as ToolError so the client sees the reason with pytest.raises(ToolError, match="non-public"): mcp_http.fetch_markdown("http://169.254.169.254/") assert called == [] # the underlying fetch never ran - @pytest.mark.parametrize("tool,kwargs", [ - ("list_links", {}), - ("compare_strategies", {"query": "anything"}), - ("extract", {"select": "h3"}), - ("fetch_json_api", {}), - ("focused_crawl", {}), - ]) + @pytest.mark.parametrize( + "tool,kwargs", + [ + ("list_links", {}), + ("compare_strategies", {"query": "anything"}), + ("extract", {"select": "h3"}), + ("fetch_json_api", {}), + ("focused_crawl", {}), + ], + ) def test_every_tool_guards_before_fetching(self, monkeypatch, tool, kwargs): """A new tool that forgets _guarded() would be an open proxy into the host's network, so every one of them is checked, not just the first.""" from mcp.server.mcpserver.exceptions import ToolError + called = [] - monkeypatch.setattr(mcp_http.local, tool, - lambda *a, **kw: called.append(a)) + monkeypatch.setattr(mcp_http.local, tool, lambda *a, **kw: called.append(a)) with pytest.raises(ToolError, match="non-public"): getattr(mcp_http, tool)("http://10.0.0.5/internal", **kwargs) assert called == [] @@ -99,8 +106,11 @@ async def inner(scope, receive, send): await send({"type": "http.response.body", "body": b"ok"}) mw = RateLimitMiddleware(inner, RateLimiter(limit=1, window=60)) - scope = {"type": "http", "headers": [(b"x-forwarded-for", b"1.2.3.4")], - "client": ("9.9.9.9", 1234)} + scope = { + "type": "http", + "headers": [(b"x-forwarded-for", b"1.2.3.4")], + "client": ("9.9.9.9", 1234), + } async def receive(): # pragma: no cover - never called return {"type": "http.request"} @@ -110,6 +120,7 @@ async def run_once(): async def send(ev): events.append(ev) + await mw(scope, receive, send) return events[0]["status"] @@ -124,6 +135,7 @@ def test_focused_crawl_clamped(self, monkeypatch): def spy(url, query="", strategy="shark", max_pages=0): seen["max_pages"] = max_pages return {} + monkeypatch.setattr(mcp_http.local, "focused_crawl", spy) mcp_http.focused_crawl("http://8.8.8.8/", max_pages=9999) assert seen["max_pages"] == mcp_http.MAX_PAGES @@ -135,6 +147,7 @@ def test_compare_strategies_clamped_lower(self, monkeypatch): def spy(url, query="", max_pages=0): seen["max_pages"] = max_pages return {} + monkeypatch.setattr(mcp_http.local, "compare_strategies", spy) mcp_http.compare_strategies("http://8.8.8.8/", query="x", max_pages=9999) assert seen["max_pages"] == mcp_http.MAX_COMPARE_PAGES @@ -146,9 +159,16 @@ async def test_hosted_server_registers_same_tools(): """Same six as the local server: an agent that learned the hosted surface can point at a local install and every call still resolves.""" from bytecrawl import mcp_server as local_server + hosted = {t.name for t in await mcp_http.server.list_tools()} - assert hosted == {"fetch_markdown", "extract", "list_links", - "focused_crawl", "compare_strategies", "fetch_json_api"} + assert hosted == { + "fetch_markdown", + "extract", + "list_links", + "focused_crawl", + "compare_strategies", + "fetch_json_api", + } assert hosted == {t.name for t in await local_server.server.list_tools()} @@ -184,6 +204,7 @@ def test_app_passes_the_settings_through(self, monkeypatch): def spy(**kw): seen.update(kw) return orig(**kw) + monkeypatch.setattr(mcp_http.server, "streamable_http_app", spy) mcp_http.create_app() assert "transport_security" in seen, "app built without Host settings" diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 669fff9..5e64374 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -24,16 +24,26 @@ async def test_all_tools_registered(): names = {t.name for t in tools} # One per playground mode, with graph crawl split into the cheap # single-strategy call and the three-way comparison. - assert names == {"fetch_markdown", "extract", "list_links", "focused_crawl", - "compare_strategies", "fetch_json_api"} + assert names == { + "fetch_markdown", + "extract", + "list_links", + "focused_crawl", + "compare_strategies", + "fetch_json_api", + } def test_extract_tool(http): http.routes["https://x.test/"] = FakeResponse( - text="

" + "enough visible text to stay static " * 10 + "

" - + '

Widget

9
' * 20) - out = mcp_server.extract("https://x.test/", item="article.p", - fields={"name": "h3::text", "cost": "span.c::text"}) + text="

" + + "enough visible text to stay static " * 10 + + "

" + + '

Widget

9
' * 20 + ) + out = mcp_server.extract( + "https://x.test/", item="article.p", fields={"name": "h3::text", "cost": "span.c::text"} + ) assert out["count"] == 20 assert out["records"][0] == {"name": "Widget", "cost": "9"} @@ -41,8 +51,11 @@ def test_extract_tool(http): def test_extract_by_single_selector(http): """The playground's Extract chip: one selector, a flat list of values.""" http.routes["https://x.test/"] = FakeResponse( - text="

" + "enough visible text to stay static " * 10 + "

" - + "".join(f'

W{i}

' for i in range(5))) + text="

" + + "enough visible text to stay static " * 10 + + "

" + + "".join(f'

W{i}

' for i in range(5)) + ) out = mcp_server.extract("https://x.test/", select="article.p h3::text") assert out["values"] == ["W0", "W1", "W2", "W3", "W4"] assert out["count"] == 5 @@ -51,15 +64,15 @@ def test_extract_by_single_selector(http): def test_extract_rejects_both_shapes(http): with pytest.raises(ValueError, match="not both"): - mcp_server.extract("https://x.test/", item="a", fields={"x": "y"}, - select="h3") + mcp_server.extract("https://x.test/", item="a", fields={"x": "y"}, select="h3") def test_extract_with_nothing_is_discovery_not_an_error(http): """This used to raise. Refusing a bare call was refusing the only question an agent can ask about a page it has not seen.""" http.routes["https://x.test/"] = FakeResponse( - text="

" + "enough visible text to stay static " * 10 + "

") + text="

" + "enough visible text to stay static " * 10 + "

" + ) out = mcp_server.extract("https://x.test/") assert "candidates" in out @@ -68,7 +81,8 @@ def test_list_links_tool(http): http.routes["https://x.test/"] = FakeResponse( text="

" + "enough visible text to stay static " * 10 + "

" 'adupefrag' - 'mailasset') + 'mailasset' + ) out = mcp_server.list_links("https://x.test/") # Absolute and deduplicated; mailto: and assets gone. A bare "#cite" # collapses onto the page itself rather than becoming its own entry — @@ -80,19 +94,19 @@ def test_list_links_tool(http): def test_list_links_raw_passes_hrefs_through(http): http.routes["https://x.test/"] = FakeResponse( text="

" + "enough visible text to stay static " * 10 + "

" - 'afrag') + 'afrag' + ) out = mcp_server.list_links("https://x.test/", raw=True) assert out["links"] == ["/a", "#cite"] and out["raw"] is True def test_compare_strategies_tool(fake_site): - out = mcp_server.compare_strategies("https://s.test/", - query="machine learning", max_pages=3) + out = mcp_server.compare_strategies("https://s.test/", query="machine learning", max_pages=3) assert set(out["strategies"]) == {"bfs", "shark", "opic"} assert out["query"] == "machine learning" for leg in out["strategies"].values(): - assert leg["relevant"] == 2 # "/" and "/ml" on this site - assert leg["pages"][0]["url"] == "https://s.test/ml" # ranked, not visit order + assert leg["relevant"] == 2 # "/" and "/ml" on this site + assert leg["pages"][0]["url"] == "https://s.test/ml" # ranked, not visit order # Five pages is too small a site to separate them: all three find the same # two relevant pages, so this is a real tie and the tool has to say so # rather than crown whichever came first. @@ -103,10 +117,8 @@ def test_compare_strategies_tool(fake_site): def test_compare_strategies_spends_its_budget_differently(fake_site): """The tie is in the score, not the behaviour: with 3 pages Shark-Search stays in the relevant branch (/ml/deep) where BFS takes the level (/cats).""" - out = mcp_server.compare_strategies("https://s.test/", - query="machine learning", max_pages=3) - visited = {n: {p["url"] for p in leg["pages"]} - for n, leg in out["strategies"].items()} + out = mcp_server.compare_strategies("https://s.test/", query="machine learning", max_pages=3) + visited = {n: {p["url"] for p in leg["pages"]} for n, leg in out["strategies"].items()} assert "https://s.test/ml/deep" in visited["shark"] assert "https://s.test/cats" in visited["bfs"] @@ -125,35 +137,36 @@ def boom(*a, **kw): raise RuntimeError("frontier exploded") monkeypatch.setitem(crawler.STRATEGIES, "opic", boom) - out = mcp_server.compare_strategies("https://s.test/", - query="machine learning", max_pages=3) + out = mcp_server.compare_strategies("https://s.test/", query="machine learning", max_pages=3) assert "frontier exploded" in out["strategies"]["opic"]["error"] assert "stats" in out["strategies"]["shark"] - assert out["winner"] in ("bfs", "shark") # the failed leg can't win + assert out["winner"] in ("bfs", "shark") # the failed leg can't win def test_focused_crawl_tool(fake_site): - out = mcp_server.focused_crawl("https://s.test/", query="machine learning", - strategy="shark", max_pages=5) + out = mcp_server.focused_crawl( + "https://s.test/", query="machine learning", strategy="shark", max_pages=5 + ) assert out["strategy"] == "shark" assert out["stats"]["requests"] == len(SITE) assert out["pages"][0]["relevance"] >= out["pages"][-1]["relevance"] assert out["pagerank_top"] # pagerank computed over the crawled graph + def test_focused_crawl_rejects_unknown_strategy(): with pytest.raises(ValueError, match="strategy"): mcp_server.focused_crawl("https://s.test/", strategy="dfs") def test_focused_crawl_caps_max_pages(fake_site): - out = mcp_server.focused_crawl("https://s.test/", strategy="bfs", - max_pages=9999) + out = mcp_server.focused_crawl("https://s.test/", strategy="bfs", max_pages=9999) assert out["stats"]["requests"] <= 50 def test_fetch_json_api_tool(http): http.routes["https://api.test/x"] = FakeResponse( - json_data={"ok": True}, headers={"content-type": "application/json"}) + json_data={"ok": True}, headers={"content-type": "application/json"} + ) out = mcp_server.fetch_json_api("https://api.test/x") assert out["data"] == {"ok": True} @@ -163,9 +176,14 @@ def test_extract_with_only_a_url_discovers(http): an agent that has not seen the markup cannot invent a selector, and no tool here returns HTML for it to read.""" http.routes["https://x.test/"] = FakeResponse( - text="

" + "enough visible text to stay static " * 10 + "

" - + "".join(f"

W{i}

" - f"

£{i}

" for i in range(5))) + text="

" + + "enough visible text to stay static " * 10 + + "

" + + "".join( + f"

W{i}

£{i}

" + for i in range(5) + ) + ) out = mcp_server.extract("https://x.test/") assert "candidates" in out and out["candidates"] top = out["candidates"][0] @@ -181,10 +199,13 @@ def test_extract_rejects_item_without_fields(http): def test_fetch_markdown_tool(http): """The one tool with no offline test at all. The smoke suite only ever called it to check the SSRF guard refused it, never that it worked.""" - body = "".join(f"

Book {i}

" - f"

£{i}.00

" for i in range(20)) + body = "".join( + f"

Book {i}

£{i}.00

" + for i in range(20) + ) http.routes["https://x.test/"] = FakeResponse( - text=f"{body}") + text=f"{body}" + ) out = mcp_server.fetch_markdown("https://x.test/") assert out["method"] == "static" and out["status"] == 200 # a listing is all repetition, which is what an article extractor discards; diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 676ae14..70c16d7 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -1,14 +1,38 @@ """Package surface: version single-sourcing and top-level exports.""" import importlib.metadata +import re +from pathlib import Path import bytecrawl +_ROOT = Path(__file__).resolve().parent.parent + def test_version_matches_installed_metadata(): assert bytecrawl.__version__ == importlib.metadata.version("bytecrawl") +def test_changelog_top_entry_matches_version(): + """The newest CHANGELOG heading is the version being shipped — + __init__.py is the single source, everything else tracks it.""" + text = (_ROOT / "CHANGELOG.md").read_text(encoding="utf-8") + top = re.search(r"^## (\d+\.\d+\.\d+)", text, re.M) + assert top, "no `## X.Y.Z` heading found in CHANGELOG.md" + assert top.group(1) == bytecrawl.__version__, ( + f"CHANGELOG top is {top.group(1)} but __version__ is {bytecrawl.__version__}" + ) + + +def test_readmes_advertise_current_version(): + """Both READMEs' 'latest release' line must name the shipped version.""" + for name in ("README.md", "README.es.md"): + text = (_ROOT / name).read_text(encoding="utf-8") + assert bytecrawl.__version__ in text, ( + f"{name} does not mention the current version {bytecrawl.__version__}" + ) + + def test_top_level_exports(): from bytecrawl import ( # noqa: F401 BFS, @@ -41,11 +65,24 @@ def test_serverless_entrypoints_do_not_shadow_dependencies(): from pathlib import Path root = Path(__file__).resolve().parent.parent - third_party = {"mcp", "flask", "requests", "bs4", "lxml", "uvicorn", - "starlette", "anyio", "pydantic", "posthog", "dotenv", - "markdownify", "trafilatura", "playwright", "bytecrawl"} - clashes = [p.name for p in (root / "web/mcp").glob("*.py") - if p.stem in third_party] + third_party = { + "mcp", + "flask", + "requests", + "bs4", + "lxml", + "uvicorn", + "starlette", + "anyio", + "pydantic", + "posthog", + "dotenv", + "markdownify", + "trafilatura", + "playwright", + "bytecrawl", + } + clashes = [p.name for p in (root / "web/mcp").glob("*.py") if p.stem in third_party] assert not clashes, f"web/mcp/ entrypoints shadow installed packages: {clashes}" # and every path vercel.json points at has to exist @@ -71,14 +108,18 @@ def test_serverless_requirements_stay_in_sync(): root = Path(__file__).resolve().parent.parent def pkgs(path): - return [ln.strip() for ln in (root / path).read_text().splitlines() - if ln.strip() and not ln.startswith("#")] + return [ + ln.strip() + for ln in (root / path).read_text().splitlines() + if ln.strip() and not ln.startswith("#") + ] demo, api = pkgs("web/landing/requirements.txt"), pkgs("web/mcp/requirements.txt") assert demo == api, ( "web/landing/requirements.txt and web/mcp/requirements.txt must match — they " f"install into the same venv.\n demo only: {set(demo) - set(api)}\n" - f" api only: {set(api) - set(demo)}") + f" api only: {set(api) - set(demo)}" + ) # the hosted MCP server cannot start without these assert any(p.startswith("mcp") for p in api) assert "trafilatura" in api and "flask" in api @@ -96,8 +137,9 @@ def test_agent_skill_points_at_the_real_repo(): from pathlib import Path root = Path(__file__).resolve().parent.parent - canonical = re.search(r'Homepage = "(https://github\.com/[^"]+)"', - (root / "pyproject.toml").read_text()).group(1) + canonical = re.search( + r'Homepage = "(https://github\.com/[^"]+)"', (root / "pyproject.toml").read_text() + ).group(1) skill = (root / "web/landing/static/SKILL.md").read_text() found = set(re.findall(r"https://github\.com/[A-Za-z0-9._/-]+", skill)) wrong = {u for u in found if not u.startswith(canonical)} @@ -125,12 +167,13 @@ def test_the_skill_does_not_order_the_agent_around(): # nothing about writing to the machine before the reader knows what this is for probe in ("~/.claude/skills", "mcpServers", "mcp add"): - assert probe not in head + opening, \ + assert probe not in head + opening, ( f"{probe!r} appears before the skill has said what it does" + ) # the option still has to exist, and be framed as the person's choice assert "~/.claude/skills/bytecrawl" in skill, "no way to install it at all" - flat = " ".join(skill.split()) # the prose is hard-wrapped + flat = " ".join(skill.split()) # the prose is hard-wrapped assert "do not change their configuration on your own" in flat assert "ask first" in flat @@ -143,12 +186,14 @@ def test_the_skill_does_not_order_the_agent_around(): landing = (root / "web/landing/templates/landing.html").read_text() setup = re.search(r"const SETUP_CMD = (.+?);\n", landing, re.S).group(1) assert "agent-onboarding/SKILL.md" in setup, "the button lost the URL" - assert "MCP server" in setup and "skill" in setup, \ + assert "MCP server" in setup and "skill" in setup, ( "the button must ask for both — the server alone leaves /bytecrawl missing" + ) def _repo_version(): import bytecrawl + return bytecrawl.__version__ @@ -172,8 +217,10 @@ def test_every_version_in_the_repo_agrees(): wrong["CHANGELOG.md"] = top.group(1) # both READMEs advertise it to anyone who never opens the changelog - for name, label in (("README.md", r"\*\*Latest release\*\*: \*\*([\d.]+)\*\*"), - ("README.es.md", r"\*\*Última versión\*\*: \*\*([\d.]+)\*\*")): + for name, label in ( + ("README.md", r"\*\*Latest release\*\*: \*\*([\d.]+)\*\*"), + ("README.es.md", r"\*\*Última versión\*\*: \*\*([\d.]+)\*\*"), + ): m = re.search(label, (root / name).read_text()) assert m, f"{name} no longer states a version" if m.group(1) != version: @@ -208,9 +255,10 @@ def test_the_skill_can_tell_it_is_stale(): flat = " ".join(skill.split()) assert "skill.json" in skill, "the skill never mentions how to check" assert "shasum -a 256" in skill - assert "do not fetch it yourself" in flat, \ - "an agent overwriting a file in someone's home because a hash differed " \ + assert "do not fetch it yourself" in flat, ( + "an agent overwriting a file in someone's home because a hash differed " "is what the rest of this document tells it to refuse" + ) # and it has to look on its own, not wait for a symptom: the failure that # prompted this was a link that 404s, which errors nothing assert "Check once, the first time you use this in a session" in flat @@ -219,12 +267,14 @@ def test_the_skill_can_tell_it_is_stale(): # route rather than running it: importing the landing app drags in Flask, # which the package does not depend on and CI does not install — the landing # site is a separate application that happens to live in this repository. - route = app_py[app_py.index('"/agent-onboarding/skill.json"'):] - route = route[:route.index("@pages_bp.route", 1)] - assert 'read_bytes()' in route and "hashlib.sha256" in route, \ + route = app_py[app_py.index('"/agent-onboarding/skill.json"') :] + route = route[: route.index("@pages_bp.route", 1)] + assert "read_bytes()" in route and "hashlib.sha256" in route, ( "the endpoint must hash the file it serves, not a stored value" + ) assert "SKILL.md" in route # a stored hash would be the drift this is meant to prevent - assert hashlib.sha256(body).hexdigest()[:8] not in app_py, \ + assert hashlib.sha256(body).hexdigest()[:8] not in app_py, ( "the current hash is hardcoded somewhere — it has to be computed" + ) diff --git a/tests/test_pep8_vercel.py b/tests/test_pep8_vercel.py index 4f30737..d15ae1b 100644 --- a/tests/test_pep8_vercel.py +++ b/tests/test_pep8_vercel.py @@ -11,6 +11,7 @@ length (pyproject.toml, 100). Reading the file list from vercel.json means a new entrypoint is covered the moment it's deployed, with no edit here. """ + from __future__ import annotations import json @@ -34,7 +35,8 @@ def test_vercel_entrypoints_follow_pep8(): assert sources, "vercel.json declares no Python builds — nothing to check" proc = subprocess.run( - ["ruff", "check", "--select", "E,W", *sources], - cwd=ROOT, capture_output=True, text=True) + ["ruff", "check", "--select", "E,W", *sources], cwd=ROOT, capture_output=True, text=True + ) assert proc.returncode == 0, ( - "Vercel-deployed code violates PEP 8:\n" + proc.stdout + proc.stderr) + "Vercel-deployed code violates PEP 8:\n" + proc.stdout + proc.stderr + ) diff --git a/tests/test_selectors.py b/tests/test_selectors.py index c2ed2a0..13f806c 100644 --- a/tests/test_selectors.py +++ b/tests/test_selectors.py @@ -54,16 +54,17 @@ def test_css_all_no_match_is_empty(self): assert page().css_all("h1.missing") == [] def test_extract_records_with_list_field(self): - rows = page().extract("div.quote", { - "quote": "span.text::text", - "author": "small.author::text", - "tags[]": "a.tag::text", - }) + rows = page().extract( + "div.quote", + { + "quote": "span.text::text", + "author": "small.author::text", + "tags[]": "a.tag::text", + }, + ) assert rows == [ - {"quote": "To be or not to be", "author": "Shakespeare", - "tags": ["life", "doubt"]}, - {"quote": "I think therefore I am", "author": "Descartes", - "tags": ["mind"]}, + {"quote": "To be or not to be", "author": "Shakespeare", "tags": ["life", "doubt"]}, + {"quote": "I think therefore I am", "author": "Descartes", "tags": ["mind"]}, ] def test_extract_missing_field_is_none(self): @@ -81,12 +82,15 @@ def test_links_raw_keeps_the_attribute_values(self): assert page().links(raw=True) == ["/tag/life", "/tag/doubt", "/tag/mind"] def test_links_drops_fragments_mailto_and_duplicates(self): - p = Page(url="https://x.test/a/", html=""" + p = Page( + url="https://x.test/a/", + html=""" 1top maillogo bb again c - """) + """, + ) assert p.links() == ["https://x.test/a/", "https://x.test/b", "https://x.test/c"] def test_tokens_estimate(self): @@ -97,14 +101,21 @@ def test_tokens_estimate(self): class TestSelectorDiscovery: """page.selectors(): the answer to "I have never seen this markup".""" - LISTING = ("
" + "".join( - f"
  • " - f"

    Book {i}

    " - f"

    £{i}.00

    In stock

    " - f"
    • " for i in range(6)) + "
    ") + LISTING = ( + "
    " + + "".join( + f"
  • " + f"

    Book {i}

    " + f"

    £{i}.00

    In stock

    " + f"
    • " + for i in range(6) + ) + + "
    " + ) def _page(self): from bytecrawl.core import Page + return Page(url="https://x.test/", html=self.LISTING) def test_finds_the_repeating_record(self): @@ -136,5 +147,6 @@ def test_no_field_is_always_empty(self): def test_a_page_with_no_listing_returns_nothing(self): from bytecrawl.core import Page + page = Page(url="https://x.test/", html="

    Just prose.

    ") assert page.selectors() == [] diff --git a/tests/test_session.py b/tests/test_session.py index 8117a7a..8d44070 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -17,23 +17,20 @@ def test_csrf_token_read_and_posted(self, http): http.routes["https://x.test/login"] = FakeResponse(text=LOGIN_FORM) http.routes["POST https://x.test/login"] = FakeResponse(status=200) - Scraper().session().login("https://x.test/login", - {"user": "ana", "pass": "s3cret"}, - csrf_field="csrf_token") + Scraper().session().login( + "https://x.test/login", {"user": "ana", "pass": "s3cret"}, csrf_field="csrf_token" + ) method, url, kw = http.calls[-1] assert (method, url) == ("POST", "https://x.test/login") - assert kw["data"] == {"user": "ana", "pass": "s3cret", - "csrf_token": "tok123"} + assert kw["data"] == {"user": "ana", "pass": "s3cret", "csrf_token": "tok123"} def test_csrf_input_without_value_posts_empty(self, http): # Regression: token["value"] used to KeyError on with no value. - http.routes["https://x.test/login"] = FakeResponse( - text='') + http.routes["https://x.test/login"] = FakeResponse(text='') http.routes["POST https://x.test/login"] = FakeResponse(status=200) - Scraper().session().login("https://x.test/login", {"user": "ana"}, - csrf_field="csrf_token") + Scraper().session().login("https://x.test/login", {"user": "ana"}, csrf_field="csrf_token") assert http.calls[-1][2]["data"]["csrf_token"] == "" def test_failed_post_raises(self, http): @@ -42,8 +39,9 @@ def test_failed_post_raises(self, http): http.routes["POST https://x.test/login"] = FakeResponse(status=403) with pytest.raises(requests.HTTPError): - Scraper().session().login("https://x.test/login", {"user": "ana"}, - csrf_field="csrf_token") + Scraper().session().login( + "https://x.test/login", {"user": "ana"}, csrf_field="csrf_token" + ) def test_login_without_csrf_skips_get(self, http): http.routes["POST https://x.test/login"] = FakeResponse(status=200) @@ -58,7 +56,8 @@ def test_sets_authorization_header(self): def test_fetch_returns_page(self, http): http.routes["https://x.test/private"] = FakeResponse( - text="secret") + text="secret" + ) page = Scraper().session().fetch("https://x.test/private") assert "secret" in page.html assert page.status == 200 diff --git a/tests/test_smoke_production.py b/tests/test_smoke_production.py index e7f308c..f5af23f 100644 --- a/tests/test_smoke_production.py +++ b/tests/test_smoke_production.py @@ -39,12 +39,17 @@ def _get(path: str, **params) -> tuple[dict, str]: def _rpc(method: str, params: dict | None = None) -> dict: - body = json.dumps({"jsonrpc": "2.0", "id": 1, "method": method, - "params": params or {}}).encode() + body = json.dumps( + {"jsonrpc": "2.0", "id": 1, "method": method, "params": params or {}} + ).encode() req = urllib.request.Request( - f"{BASE}/mcp", data=body, - headers={"Content-Type": "application/json", - "Accept": "application/json, text/event-stream"}) + f"{BASE}/mcp", + data=body, + headers={ + "Content-Type": "application/json", + "Accept": "application/json, text/event-stream", + }, + ) with urllib.request.urlopen(req, timeout=TIMEOUT) as r: # noqa: S310 return json.loads(r.read().decode()) @@ -62,13 +67,15 @@ def test_served_as_markdown(self): def test_every_link_it_gives_out_resolves(self): """It shipped pointing at a repo that 404s, and nothing caught it.""" import re + req = urllib.request.Request(f"{BASE}/agent-onboarding/SKILL.md") with urllib.request.urlopen(req, timeout=TIMEOUT) as r: # noqa: S310 body = r.read().decode() for url in sorted(set(re.findall(r"https://github\.com/[A-Za-z0-9._/-]+", body))): try: urllib.request.urlopen( # noqa: S310 - urllib.request.Request(url, method="HEAD"), timeout=TIMEOUT) + urllib.request.Request(url, method="HEAD"), timeout=TIMEOUT + ) except urllib.error.HTTPError as e: pytest.fail(f"SKILL.md points at {url} — HTTP {e.code}") @@ -80,12 +87,15 @@ class TestHttpApi: `elapsed`, which no method has ever returned. """ - @pytest.mark.parametrize("method,extra,expected", [ - ("markdown", {}, {"markdown", "tokens"}), - ("text", {}, {"text"}), - ("links", {}, {"links"}), - ("extract", {"select": "h3 a::attr(title)"}, {"values", "select"}), - ]) + @pytest.mark.parametrize( + "method,extra,expected", + [ + ("markdown", {}, {"markdown", "tokens"}), + ("text", {}, {"text"}), + ("links", {}, {"links"}), + ("extract", {"select": "h3 a::attr(title)"}, {"values", "select"}), + ], + ) def test_documented_fields(self, method, extra, expected): out, _ = _get("/api", url="books.toscrape.com", method=method, **extra) assert {"url", "method"} <= set(out), "url and method are on every response" @@ -93,8 +103,7 @@ def test_documented_fields(self, method, extra, expected): assert "status" not in out and "elapsed" not in out def test_compare_returns_a_verdict(self): - out, _ = _get("/api", url="books.toscrape.com", method="compare", - query="fiction", pages=3) + out, _ = _get("/api", url="books.toscrape.com", method="compare", query="fiction", pages=3) assert set(out["strategies"]) == {"bfs", "shark", "opic"} assert out["winner"] in out["tied"] @@ -108,20 +117,35 @@ class TestHostedMcp: def test_six_tools(self): """Deployed once answering 421 to every request, and once not at all.""" names = {t["name"] for t in _rpc("tools/list")["result"]["tools"]} - assert names == {"fetch_markdown", "extract", "list_links", - "focused_crawl", "compare_strategies", "fetch_json_api"} - - @pytest.mark.parametrize("tool,args,expect", [ - ("fetch_markdown", {"url": BOOKS}, {"markdown", "tokens_html"}), - ("extract", {"url": BOOKS}, {"candidates"}), - ("extract", {"url": BOOKS, "select": "h3 a::attr(title)"}, {"values"}), - ("list_links", {"url": BOOKS}, {"links"}), - ("fetch_json_api", {"url": QUOTES_API}, {"data"}), - ("focused_crawl", {"url": BOOKS, "query": "fiction", "max_pages": 3}, - {"pages", "stats"}), - ("compare_strategies", {"url": BOOKS, "query": "fiction", "max_pages": 3}, - {"strategies", "winner"}), - ]) + assert names == { + "fetch_markdown", + "extract", + "list_links", + "focused_crawl", + "compare_strategies", + "fetch_json_api", + } + + @pytest.mark.parametrize( + "tool,args,expect", + [ + ("fetch_markdown", {"url": BOOKS}, {"markdown", "tokens_html"}), + ("extract", {"url": BOOKS}, {"candidates"}), + ("extract", {"url": BOOKS, "select": "h3 a::attr(title)"}, {"values"}), + ("list_links", {"url": BOOKS}, {"links"}), + ("fetch_json_api", {"url": QUOTES_API}, {"data"}), + ( + "focused_crawl", + {"url": BOOKS, "query": "fiction", "max_pages": 3}, + {"pages", "stats"}, + ), + ( + "compare_strategies", + {"url": BOOKS, "query": "fiction", "max_pages": 3}, + {"strategies", "winner"}, + ), + ], + ) def test_every_tool_actually_runs(self, tool, args, expect): """Listing six tools is not the same as six tools working. @@ -139,8 +163,10 @@ def test_every_tool_actually_runs(self, tool, args, expect): def test_ssrf_guard_is_live(self): """The one property that must never regress: a hosted scraper that forwards to link-local addresses is an open door to cloud metadata.""" - out = _rpc("tools/call", {"name": "fetch_markdown", - "arguments": {"url": "http://169.254.169.254/"}}) + out = _rpc( + "tools/call", + {"name": "fetch_markdown", "arguments": {"url": "http://169.254.169.254/"}}, + ) text = json.dumps(out) assert "non-public" in text, text[:200] @@ -167,7 +193,8 @@ def _mm(v: str) -> tuple[int, int]: def _pypi(self) -> str: with urllib.request.urlopen( # noqa: S310 - "https://pypi.org/pypi/bytecrawl/json", timeout=TIMEOUT) as r: + "https://pypi.org/pypi/bytecrawl/json", timeout=TIMEOUT + ) as r: return json.loads(r.read())["info"]["version"] def test_the_readme_version_is_installable(self): @@ -180,23 +207,27 @@ def test_the_readme_version_is_installable(self): import bytecrawl root = Path(__file__).resolve().parent.parent - advertised = re.search(r"\*\*Latest release\*\*: \*\*([\d.]+)\*\*", - (root / "README.md").read_text()).group(1) - assert advertised == bytecrawl.__version__ # offline suite covers this + advertised = re.search( + r"\*\*Latest release\*\*: \*\*([\d.]+)\*\*", (root / "README.md").read_text() + ).group(1) + assert advertised == bytecrawl.__version__ # offline suite covers this published = self._pypi() assert self._mm(advertised) == self._mm(published), ( f"README and the package say {advertised}, PyPI serves {published} — " - "publish it, or the docs describe a release nobody can install") + "publish it, or the docs describe a release nobody can install" + ) assert self._mm(published) <= self._mm(advertised), ( - f"PyPI is ahead of the repo: {published} vs {advertised}") + f"PyPI is ahead of the repo: {published} vs {advertised}" + ) def test_the_site_shows_the_published_version(self): """/docs renders it from the package, so a deploy behind PyPI shows a version people cannot get.""" import re + req = urllib.request.Request(f"{BASE}/docs") with urllib.request.urlopen(req, timeout=TIMEOUT) as r: # noqa: S310 - shown = re.search(r'v([\d.]+)', - r.read().decode()).group(1) - assert self._mm(shown) == self._mm(self._pypi()), \ + shown = re.search(r'v([\d.]+)', r.read().decode()).group(1) + assert self._mm(shown) == self._mm(self._pypi()), ( f"/docs shows v{shown}, PyPI serves {self._pypi()}" + )