cryptorugmunch
a7c30b12cd
Mass ruff auto-fix:
- ruff check --fix: 109 issues fixed (F401 unused imports,
I001 unsorted imports, UP037 quoted annotations, SIM105
suppressible exception, RUF100 unused-noqa)
- ruff check --fix --unsafe-fixes: 22 additional issues
- ruff format: 70 files reformatted
- Manual pass: fix 16 misplaced import httpx lines
- Manual pass: fix remaining E402 (import-after-docstring)
Result: 283 errors -> 30 errors.
The remaining 30 are real issues that need manual review:
5 F401 unused-import (likely auto-generated stubs)
5 F821 undefined-name (real bugs in code that references
redis/pydantic/LLMRegistry without imports)
3 BLE001 (the compliance LLM fallback is intentional; the
other two are real)
3 RUF012 mutable-class-default
3 SIM105, 3 SIM117, 2 E722, 2 E741
1 B007, 1 B025, 1 E402, 1 RUF200 (pyproject.toml issue)
Tests: 436/437 pass (1 pre-existing SSE sandbox failure).
format check + import sort: now clean.
make ci: still gated on the 30 remaining real issues.
Follow-up: triage the 30 issues file-by-file.
100 lines
3.1 KiB
Python
100 lines
3.1 KiB
Python
# SPDX-License-Identifier: MIT
|
|
# Copyright (c) 2026 Rug Munch Media LLC
|
|
#
|
|
# Part of Pry — https://git.rugmunch.io/RugMunchMedia/pryscraper
|
|
# Licensed under MIT. See LICENSE.
|
|
"""Tests for marketplace, x402 middleware, and webhooks."""
|
|
|
|
import asyncio
|
|
|
|
from actor_marketplace import Actor, ActorMarketplace, ActorVisibility
|
|
from webhook_delivery import WebhookDelivery
|
|
from x402_middleware import PAID_ENDPOINTS, X402_PRICING
|
|
|
|
|
|
def test_x402_pricing() -> None:
|
|
assert "scrape" in X402_PRICING
|
|
assert X402_PRICING["scrape"]["price_usd"] > 0
|
|
for info in X402_PRICING.values():
|
|
assert "price_usd" in info
|
|
assert "description" in info
|
|
|
|
|
|
def test_x402_paid_endpoints() -> None:
|
|
assert "/v1/scrape" in PAID_ENDPOINTS
|
|
assert "/v1/crawl" in PAID_ENDPOINTS
|
|
assert len(PAID_ENDPOINTS) >= 5
|
|
|
|
|
|
def test_x402_middleware_signing() -> None:
|
|
w = WebhookDelivery(signing_secret="test-secret")
|
|
sig = w.sign_payload({"event": "test", "data": {"x": 1}})
|
|
assert isinstance(sig, str)
|
|
assert len(sig) == 64 # SHA-256 hex
|
|
|
|
|
|
def test_actor_marketplace_create() -> None:
|
|
m = ActorMarketplace()
|
|
actor = m.create(
|
|
"Test Actor",
|
|
"Test description",
|
|
template_id="amazon-product",
|
|
price_per_run=0.01,
|
|
visibility=ActorVisibility.PRIVATE,
|
|
tags=["e-commerce"],
|
|
)
|
|
assert actor.name == "Test Actor"
|
|
assert actor.template_id == "amazon-product"
|
|
assert actor.price_per_run == 0.01
|
|
assert "e-commerce" in actor.tags
|
|
|
|
|
|
def test_actor_marketplace_list() -> None:
|
|
m = ActorMarketplace()
|
|
m.create("Test 1", "Desc", tags=["test"])
|
|
actors = m.list()
|
|
assert len(actors) >= 1
|
|
|
|
|
|
def test_actor_marketplace_filter() -> None:
|
|
m = ActorMarketplace()
|
|
m.create("Public", "D", visibility=ActorVisibility.PUBLIC, tags=["free"])
|
|
m.create("Private", "D", visibility=ActorVisibility.PRIVATE, tags=["private"])
|
|
public = m.list(visibility="public")
|
|
private = m.list(visibility="private")
|
|
assert all(a["visibility"] == "public" for a in public)
|
|
assert all(a["visibility"] == "private" for a in private)
|
|
|
|
|
|
def test_actor_marketplace_run_missing() -> None:
|
|
m = ActorMarketplace()
|
|
result = asyncio.run(m.run("nonexistent"))
|
|
assert result["success"] is False
|
|
|
|
|
|
def test_actor_to_dict() -> None:
|
|
a = Actor("id1", "Name", "Desc")
|
|
d = a.to_dict()
|
|
assert d["name"] == "Name"
|
|
assert d["actor_id"] == "id1"
|
|
assert "created_at" in d
|
|
|
|
|
|
def test_webhook_signature_consistency() -> None:
|
|
w1 = WebhookDelivery(signing_secret="same-secret")
|
|
w2 = WebhookDelivery(signing_secret="same-secret")
|
|
payload = {"event": "test", "x": 1}
|
|
assert w1.sign_payload(payload) == w2.sign_payload(payload)
|
|
|
|
|
|
def test_webhook_different_secrets() -> None:
|
|
w1 = WebhookDelivery(signing_secret="secret1")
|
|
w2 = WebhookDelivery(signing_secret="secret2")
|
|
payload = {"event": "test"}
|
|
assert w1.sign_payload(payload) != w2.sign_payload(payload)
|
|
|
|
|
|
def test_actor_visibility_enum() -> None:
|
|
assert ActorVisibility.PRIVATE.value == "private"
|
|
assert ActorVisibility.PUBLIC.value == "public"
|
|
assert ActorVisibility.UNLISTED.value == "unlist"
|