retoor/devplacepy
1
0
Fork 1
Code Issues 44 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

fix: correct "bugs" to "issues" in routing table and README references across multiple documentation files

Browse Source
This commit is contained in:
retoor
2026-06-14 07:48:10 +00:00
parent f79a427f32
commit f2b910ea75
162 changed files with 7011 additions and 1134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/api/admin/users/index.py
+1 -1
View File
@@ -21,7 +21,7 @@ def _session_content_negotiation(password="secret123"):
allow_redirects=True,
)
return s, name
PUBLIC_PAGES = ["/feed", "/projects", "/gists", "/news", "/leaderboard", "/bugs"]
PUBLIC_PAGES = ["/feed", "/projects", "/gists", "/news", "/leaderboard", "/issues"]
def test_admin_pages_negotiate(seeded_db):
tests/api/auth/matrix.py
+39 -27
View File
@@ -85,6 +85,15 @@ def _call(endpoint, headers):
elif data:
kwargs["data"] = data
return requests.request(endpoint["method"], f"{BASE_URL}{path}", **kwargs)
def _clear_probe_jobs():
from devplacepy.database import refresh_snapshot
refresh_snapshot()
jobs = get_table("jobs")
for row in list(jobs.find()):
jobs.delete(uid=row["uid"])
def _redirects_to(response, target):
return response.status_code in (
302,
@@ -110,38 +119,41 @@ def test_documented_minimal_role_matches_enforcement(seeded_db):
assert len(endpoints) >= 50
failures = []
for endpoint in endpoints:
auth = endpoint["auth"]
label = (
f"{endpoint['method']} {endpoint['path']} ({endpoint['id']}, doc={auth})"
)
anon = _call(endpoint, JSON_auth_matrix)
if auth == "public":
if not _is_allowed(anon):
failures.append(
f"{label}: public but anonymous was rejected ({anon.status_code})"
)
continue
if not _is_auth_rejected(anon):
failures.append(
f"{label}: requires {auth} but anonymous was NOT rejected ({anon.status_code})"
try:
for endpoint in endpoints:
auth = endpoint["auth"]
label = (
f"{endpoint['method']} {endpoint['path']} ({endpoint['id']}, doc={auth})"
)
anon = _call(endpoint, JSON_auth_matrix)
if auth == "user" and endpoint["method"] == "GET":
mem = _call(endpoint, member)
if not _is_allowed(mem):
if auth == "public":
if not _is_allowed(anon):
failures.append(
f"{label}: public but anonymous was rejected ({anon.status_code})"
)
continue
if not _is_auth_rejected(anon):
failures.append(
f"{label}: documented user but a member was rejected ({mem.status_code})"
f"{label}: requires {auth} but anonymous was NOT rejected ({anon.status_code})"
)
if auth == "admin":
mem = _call(endpoint, member)
if not _is_role_rejected(mem):
failures.append(
f"{label}: documented admin but a non-admin member was NOT rejected ({mem.status_code})"
)
if auth == "user" and endpoint["method"] == "GET":
mem = _call(endpoint, member)
if not _is_allowed(mem):
failures.append(
f"{label}: documented user but a member was rejected ({mem.status_code})"
)
if auth == "admin":
mem = _call(endpoint, member)
if not _is_role_rejected(mem):
failures.append(
f"{label}: documented admin but a non-admin member was NOT rejected ({mem.status_code})"
)
finally:
_clear_probe_jobs()
assert not failures, "Auth enforcement does not match documentation:\n" + "\n".join(
failures
tests/api/content/negotiation.py
+1 -1
View File
@@ -21,7 +21,7 @@ def _session_content_negotiation(password="secret123"):
allow_redirects=True,
)
return s, name
PUBLIC_PAGES = ["/feed", "/projects", "/gists", "/news", "/leaderboard", "/bugs"]
PUBLIC_PAGES = ["/feed", "/projects", "/gists", "/news", "/leaderboard", "/issues"]
def test_public_pages_serve_json_and_html(app_server):
tests/api/feed.py
+20
View File
@@ -340,3 +340,23 @@ def test_x_robots_tag_header(app_server):
def test_x_content_type_options_header(app_server):
r = requests.get(f"{BASE_URL}/feed", allow_redirects=True)
assert r.headers.get("X-Content-Type-Options") == "nosniff"
def test_strict_transport_security_header(app_server):
r = requests.get(f"{BASE_URL}/feed", allow_redirects=True)
assert "includeSubDomains" in r.headers.get("Strict-Transport-Security", "")
def test_referrer_policy_header(app_server):
r = requests.get(f"{BASE_URL}/feed", allow_redirects=True)
assert r.headers.get("Referrer-Policy") == "strict-origin-when-cross-origin"
def test_x_frame_options_header(app_server):
r = requests.get(f"{BASE_URL}/feed", allow_redirects=True)
assert r.headers.get("X-Frame-Options") == "DENY"
def test_x_frame_options_excluded_for_ingress_proxy(app_server):
r = requests.get(f"{BASE_URL}/p/nonexistent-instance", allow_redirects=False)
assert "X-Frame-Options" not in r.headers
tests/api/bugs/__init__.py → tests/api/issues/__init__.py
View File
tests/api/bugs/create.py → tests/api/issues/create.py
+29 -29
View File
@@ -124,13 +124,13 @@ from devplacepy.services.gitea import runtime, store
from devplacepy.services.gitea.config import gitea_config
from devplacepy.services.gitea.enhance import enhance_ticket
from devplacepy.services.gitea.fake import FakeGiteaClient
from devplacepy.services.gitea.service import BugTrackerService
from devplacepy.services.gitea.service import IssueTrackerService
from devplacepy.services.jobs import queue
from devplacepy.services.jobs.bug_create_service import BugCreateService
from devplacepy.services.jobs.issue_create_service import IssueCreateService
from tests.conftest import run_async
_counter_bugs_gitea = [0]
_counter_issues_gitea = [0]
@pytest.fixture(autouse=True)
def _init_db_bugs_gitea():
def _init_db_issues_gitea():
init_db()
yield
@pytest.fixture
@@ -141,20 +141,20 @@ def gitea_env():
set_setting("gitea_owner", "retoor")
set_setting("gitea_repo", "pydevplace")
set_setting("gitea_token", "test-token")
set_setting("bug_ai_enhance", "0")
set_setting("issue_ai_enhance", "0")
yield fake
runtime.set_client(None)
for key in ("gitea_base_url", "gitea_owner", "gitea_repo", "gitea_token"):
set_setting(key, "")
for table in ("bug_tickets", "bug_comment_authors"):
for table in ("issue_tickets", "issue_comment_authors"):
for row in list(get_table(table).find()):
get_table(table).delete(uid=row["uid"])
for row in list(get_table("jobs").find(kind="bug_create")):
for row in list(get_table("jobs").find(kind="issue_create")):
get_table("jobs").delete(uid=row["uid"])
def _make_user_bugs_gitea():
_counter_bugs_gitea[0] += 1
uid = f"bugtest-user-{_counter_bugs_gitea[0]}"
username = f"bugtester{_counter_bugs_gitea[0]}"
def _make_user_issues_gitea():
_counter_issues_gitea[0] += 1
uid = f"issuetest-user-{_counter_issues_gitea[0]}"
username = f"issuetester{_counter_issues_gitea[0]}"
get_table("users").insert(
{
"uid": uid,
@@ -170,17 +170,17 @@ def _unread(user_uid):
return [
n
for n in get_table("notifications").find(user_uid=user_uid)
if n.get("type") == "bug"
if n.get("type") == "issue"
]
def _drive_jobs():
async def run():
svc = BugCreateService()
svc = IssueCreateService()
for _ in range(200):
await svc.run_once()
refresh_snapshot()
pending = [
r
for r in get_table("jobs").find(kind="bug_create")
for r in get_table("jobs").find(kind="issue_create")
if r["status"] in ("pending", "running")
]
if not pending and not svc._inflight:
@@ -189,7 +189,7 @@ def _drive_jobs():
run_async(run())
def _role_user(role: str):
uid, username = _make_user_bugs_gitea()
uid, username = _make_user_issues_gitea()
get_table("users").update({"uid": uid, "role": role}, ["uid"])
return uid, username
def _open_ticket(fake, role: str):
@@ -214,14 +214,14 @@ def _open_ticket(fake, role: str):
return number, client
def test_bug_create_recorded(seeded_db):
# Filing is async + Gitea-gated: POST /bugs/create only enqueues a job (returns
# {uid, status_url}) and records the synchronous "bug.create.request" event; the
# final "bug.create" fires from BugCreateService, which the tests never run
def test_issue_create_recorded(seeded_db):
# Filing is async + Gitea-gated: POST /issues/create only enqueues a job (returns
# {uid, status_url}) and records the synchronous "issue.create.request" event; the
# final "issue.create" fires from IssueCreateService, which the tests never run
# (DEVPLACE_DISABLE_SERVICES=1). Configure Gitea so the enqueue path is taken
# (no network call happens here) and assert the request event. Restore after.
s, _ = _member()
title = _unique("aubug")
title = _unique("auissue")
for key, value in {
"gitea_base_url": "https://gitea.test",
"gitea_owner": "o",
@@ -231,9 +231,9 @@ def test_bug_create_recorded(seeded_db):
set_setting(key, value)
try:
resp = s.post(
f"{BASE_URL}/bugs/create",
f"{BASE_URL}/issues/create",
headers=JSON_audit_log,
data={"title": title, "description": "audited bug description text"},
data={"title": title, "description": "audited issue description text"},
)
assert resp.status_code == 200, resp.text[:300]
assert resp.json().get("uid")
@@ -241,7 +241,7 @@ def test_bug_create_recorded(seeded_db):
assert (
_find(
admin,
"bug.create.request",
"issue.create.request",
lambda e: title in (e.get("summary") or ""),
)
is not None
@@ -251,24 +251,24 @@ def test_bug_create_recorded(seeded_db):
set_setting(key, "")
def test_bug_detail_renders_for_member(gitea_env):
def test_issue_detail_renders_for_member(gitea_env):
number, client = _open_ticket(gitea_env, "Member")
html = client.get(f"/bugs/{number}")
html = client.get(f"/issues/{number}")
assert html.status_code == 200, html.text[:300]
assert "Submit does nothing" in html.text
assert "Close ticket" not in html.text
data = client.get(
f"/bugs/{number}", headers={"Accept": "application/json"}
f"/issues/{number}", headers={"Accept": "application/json"}
).json()
assert data["viewer_is_admin"] is False
def test_bug_detail_renders_for_admin(gitea_env):
def test_issue_detail_renders_for_admin(gitea_env):
number, client = _open_ticket(gitea_env, "Admin")
html = client.get(f"/bugs/{number}")
html = client.get(f"/issues/{number}")
assert html.status_code == 200, html.text[:300]
assert "Close ticket" in html.text
data = client.get(
f"/bugs/{number}", headers={"Accept": "application/json"}
f"/issues/{number}", headers={"Accept": "application/json"}
).json()
assert data["viewer_is_admin"] is True
tests/api/messages/index.py
+1 -1
View File
@@ -131,7 +131,7 @@ def _session_content_negotiation(password="secret123"):
allow_redirects=True,
)
return s, name
PUBLIC_PAGES = ["/feed", "/projects", "/gists", "/news", "/leaderboard", "/bugs"]
PUBLIC_PAGES = ["/feed", "/projects", "/gists", "/news", "/leaderboard", "/issues"]
def test_security_authz_denied_recorded(seeded_db):
tests/api/posts/create.py
+1 -1
View File
@@ -131,7 +131,7 @@ def _session_content_negotiation(password="secret123"):
allow_redirects=True,
)
return s, name
PUBLIC_PAGES = ["/feed", "/projects", "/gists", "/news", "/leaderboard", "/bugs"]
PUBLIC_PAGES = ["/feed", "/projects", "/gists", "/news", "/leaderboard", "/issues"]
_counter_polls = [0]
AJAX_polls = {"X-Requested-With": "fetch"}
def _session_polls():
tests/api/posts/index.py
+1 -1
View File
@@ -21,7 +21,7 @@ def _session_content_negotiation(password="secret123"):
allow_redirects=True,
)
return s, name
PUBLIC_PAGES = ["/feed", "/projects", "/gists", "/news", "/leaderboard", "/bugs"]
PUBLIC_PAGES = ["/feed", "/projects", "/gists", "/news", "/leaderboard", "/issues"]
def test_json_404_error_envelope(app_server):
tests/e2e/bugs/index.py
-66
View File
@@ -1,66 +0,0 @@
# retoor <retoor@molodetz.nl>
from tests.conftest import BASE_URL
def test_bugs_page_loads(alice):
page, _ = alice
page.goto(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
assert page.is_visible("h1:has-text('Bug Reports')")
def test_bugs_filters_visible(alice):
page, _ = alice
page.goto(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
assert page.locator(".bugs-filter:has-text('Open')").is_visible()
assert page.locator(".bugs-filter:has-text('Closed')").is_visible()
assert page.locator(".bugs-filter:has-text('All')").is_visible()
def test_bugs_not_configured_notice(alice):
page, _ = alice
page.goto(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
assert page.is_visible("text=The bug tracker is not configured yet.")
def test_bugs_create_modal(alice):
page, _ = alice
page.goto(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
page.click("button:has-text('Report Bug')")
assert page.is_visible("h3:has-text('Report a Bug')")
def test_bugs_create_cancel(alice):
page, _ = alice
page.goto(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
page.click("button:has-text('Report Bug')")
page.click("button:has-text('Cancel')")
modal = page.locator("#create-bug-modal")
assert not modal.is_visible()
def test_bugs_unauth(page):
page.goto(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
assert page.is_visible("h1:has-text('Bug Reports')")
assert page.locator("a.login-required:has-text('Report Bug')").is_visible()
assert page.locator("#create-bug-modal").count() == 0
def test_bug_button_icon_spacing(alice):
page, _ = alice
page.goto(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
html = page.locator("button:has-text('Report Bug')").inner_html()
assert "</span> Report Bug" in html
page.goto(f"{BASE_URL}/", wait_until="domcontentloaded")
landing = page.locator("a:has-text('Bug Report')").inner_html()
assert "</span> Bug Report" in landing
page.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded")
footer = page.locator("a:has-text('Bug Report')").inner_html()
assert "</span> Bug Report" in footer
def test_bug_detail_not_configured(alice):
page, _ = alice
page.goto(f"{BASE_URL}/bugs/999", wait_until="domcontentloaded")
assert page.is_visible("h1.error-code:has-text('404')")
assert page.is_visible("text=Page not found")
tests/e2e/docs.py
+1 -1
View File
@@ -121,7 +121,7 @@ def test_docs_panel_html_format_updates_snippet(page, app_server):
def test_docs_panel_live_send_returns_json(page, app_server):
page.goto(f"{BASE_URL}/docs/bugs.html", wait_until="domcontentloaded")
page.goto(f"{BASE_URL}/docs/issues.html", wait_until="domcontentloaded")
panel = page.locator(".api-tester").first
panel.wait_for(state="visible")
send = panel.locator(".try-send").first
tests/e2e/feed.py
+3 -3
View File
@@ -383,13 +383,13 @@ def _seed_feed_posts(count):
return topic
def test_bugs_footer_link(alice):
def test_issues_footer_link(alice):
page, _ = alice
page.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded")
link = page.locator("a:has-text('Bug Report')")
link = page.locator("a:has-text('Issue Report')")
assert link.is_visible()
link.click()
page.wait_for_url(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
page.wait_for_url(f"{BASE_URL}/issues", wait_until="domcontentloaded")
def test_global_toggle_suppresses_rendering_and_persists(bob):
tests/e2e/bugs/__init__.py → tests/e2e/issues/__init__.py
View File
tests/e2e/issues/create.py
+88
View File
@@ -0,0 +1,88 @@
# retoor <retoor@molodetz.nl>
from tests.conftest import BASE_URL
SUBMIT = "#create-issue-modal button[type='submit']"
SPINNER = "#create-issue-modal button[type='submit'] .btn-spinner"
def _open_form(page):
page.goto(f"{BASE_URL}/issues", wait_until="domcontentloaded")
page.click("button:has-text('Report Issue')")
page.fill("#issue-title", "Spinner regression title")
page.fill("#issue-description", "Steps to reproduce the spinner regression")
def test_submit_disables_button_and_shows_spinner(alice):
page, _ = alice
page.route(
"**/issues/create",
lambda route: route.fulfill(
status=200,
content_type="application/json",
body='{"uid":"spin-job","status_url":"/issues/jobs/spin-job"}',
),
)
page.route(
"**/issues/jobs/spin-job",
lambda route: route.fulfill(
status=200,
content_type="application/json",
body='{"uid":"spin-job","kind":"issue_create","status":"pending"}',
),
)
_open_form(page)
page.click(SUBMIT)
page.locator(f"{SUBMIT}.is-loading").wait_for(state="visible")
assert page.locator(SUBMIT).is_disabled()
assert page.locator(SPINNER).is_visible()
def test_submit_prevents_double_posting(alice):
page, _ = alice
create_calls = []
page.route(
"**/issues/create",
lambda route: (
create_calls.append(1),
route.fulfill(
status=200,
content_type="application/json",
body='{"uid":"once-job","status_url":"/issues/jobs/once-job"}',
),
)[-1],
)
page.route(
"**/issues/jobs/once-job",
lambda route: route.fulfill(
status=200,
content_type="application/json",
body='{"uid":"once-job","kind":"issue_create","status":"pending"}',
),
)
_open_form(page)
page.click(SUBMIT)
page.locator(f"{SUBMIT}.is-loading").wait_for(state="visible")
page.evaluate(
"document.querySelector('form[data-issue-create]')"
".dispatchEvent(new Event('submit', {cancelable: true, bubbles: true}))"
)
page.wait_for_timeout(300)
assert len(create_calls) == 1
def test_submit_reenables_button_on_error(alice):
page, _ = alice
page.route(
"**/issues/create",
lambda route: route.fulfill(
status=503,
content_type="application/json",
body='{"ok":false,"message":"The issue tracker is not configured"}',
),
)
_open_form(page)
page.click(SUBMIT)
page.locator(f"{SUBMIT}:not(.is-loading)").wait_for(state="visible")
assert not page.locator(SUBMIT).is_disabled()
assert not page.locator(SPINNER).is_visible()
tests/e2e/issues/index.py
+66
View File
@@ -0,0 +1,66 @@
# retoor <retoor@molodetz.nl>
from tests.conftest import BASE_URL
def test_issues_page_loads(alice):
page, _ = alice
page.goto(f"{BASE_URL}/issues", wait_until="domcontentloaded")
assert page.is_visible("h1:has-text('Issue Reports')")
def test_issues_filters_visible(alice):
page, _ = alice
page.goto(f"{BASE_URL}/issues", wait_until="domcontentloaded")
assert page.locator(".issues-filter:has-text('Open')").is_visible()
assert page.locator(".issues-filter:has-text('Closed')").is_visible()
assert page.locator(".issues-filter:has-text('All')").is_visible()
def test_issues_not_configured_notice(alice):
page, _ = alice
page.goto(f"{BASE_URL}/issues", wait_until="domcontentloaded")
assert page.is_visible("text=The issue tracker is not configured yet.")
def test_issues_create_modal(alice):
page, _ = alice
page.goto(f"{BASE_URL}/issues", wait_until="domcontentloaded")
page.click("button:has-text('Report Issue')")
assert page.is_visible("h3:has-text('Report an Issue')")
def test_issues_create_cancel(alice):
page, _ = alice
page.goto(f"{BASE_URL}/issues", wait_until="domcontentloaded")
page.click("button:has-text('Report Issue')")
page.click("button:has-text('Cancel')")
modal = page.locator("#create-issue-modal")
assert not modal.is_visible()
def test_issues_unauth(page):
page.goto(f"{BASE_URL}/issues", wait_until="domcontentloaded")
assert page.is_visible("h1:has-text('Issue Reports')")
assert page.locator("a.login-required:has-text('Report Issue')").is_visible()
assert page.locator("#create-issue-modal").count() == 0
def test_issue_button_icon_spacing(alice):
page, _ = alice
page.goto(f"{BASE_URL}/issues", wait_until="domcontentloaded")
html = page.locator("button:has-text('Report Issue')").inner_html()
assert "</span> Report Issue" in html
page.goto(f"{BASE_URL}/", wait_until="domcontentloaded")
landing = page.locator("a:has-text('Issue Report')").inner_html()
assert "</span> Issue Report" in landing
page.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded")
footer = page.locator("a:has-text('Issue Report')").inner_html()
assert "</span> Issue Report" in footer
def test_issue_detail_not_configured(alice):
page, _ = alice
page.goto(f"{BASE_URL}/issues/999", wait_until="domcontentloaded")
assert page.is_visible("h1.error-code:has-text('404')")
assert page.is_visible("text=Page not found")
tests/e2e/bugs/jobs.py → tests/e2e/issues/jobs.py
+2 -2
View File
@@ -3,8 +3,8 @@
from tests.conftest import BASE_URL
def test_bug_job_status_not_found(alice):
def test_issue_job_status_not_found(alice):
page, _ = alice
page.goto(f"{BASE_URL}/bugs/jobs/nonexistent-job", wait_until="domcontentloaded")
page.goto(f"{BASE_URL}/issues/jobs/nonexistent-job", wait_until="domcontentloaded")
assert page.is_visible("h1.error-code:has-text('404')")
assert page.is_visible("text=Page not found")
tests/e2e/root.py
+1 -1
View File
@@ -199,7 +199,7 @@ def test_landing_nav_links(page, app_server):
def test_landing_footer_links(page, app_server):
page.goto(f"{BASE_URL}/", wait_until="domcontentloaded")
assert page.is_visible("a:has-text('Bug Report')")
assert page.is_visible("a:has-text('Issue Report')")
def test_landing_hero_present(page, app_server):
tests/unit/curl_transport.py
+136
View File
@@ -0,0 +1,136 @@
# retoor <retoor@molodetz.nl>
import httpx
import pytest
from curl_cffi.requests.exceptions import RequestException, Timeout
import devplacepy.curl_transport as ct
from devplacepy.curl_transport import CurlTransport, resolve_timeout
from tests.conftest import run_async
class FakeCurlResponse:
def __init__(self, status=200, headers=None, chunks=None, http_version=3):
self.status_code = status
self.headers = headers or {}
self.http_version = http_version
self._chunks = chunks if chunks is not None else [b"hello"]
self.closed = False
async def aiter_content(self):
for chunk in self._chunks:
yield chunk
async def aclose(self):
self.closed = True
class FakeSession:
def __init__(self, response=None, raise_exc=None):
self.calls = []
self.closed = False
self.response = response if response is not None else FakeCurlResponse()
self.raise_exc = raise_exc
async def request(self, method, url, **kwargs):
self.calls.append({"method": method, "url": url, **kwargs})
if self.raise_exc is not None:
raise self.raise_exc
return self.response
async def close(self):
self.closed = True
def make_transport(monkeypatch, response=None, raise_exc=None):
fake = FakeSession(response=response, raise_exc=raise_exc)
monkeypatch.setattr(ct, "AsyncSession", lambda *a, **k: fake)
return CurlTransport(), fake
def test_request_headers_stripped_and_forwarded(monkeypatch):
transport, fake = make_transport(monkeypatch)
request = httpx.Request(
"POST",
"https://api.example.com/v1/x",
headers={
"user-agent": "leak/1.0",
"accept-encoding": "identity",
"authorization": "Bearer k",
"content-type": "application/json",
},
content=b'{"a":1}',
)
run_async(transport.handle_async_request(request))
sent = {key.lower(): value for key, value in fake.calls[0]["headers"].items()}
assert "user-agent" not in sent
assert "accept-encoding" not in sent
assert "host" not in sent
assert "content-length" not in sent
assert sent["authorization"] == "Bearer k"
assert sent["content-type"] == "application/json"
assert fake.calls[0]["data"] == b'{"a":1}'
assert fake.calls[0]["allow_redirects"] is False
assert fake.calls[0]["stream"] is True
assert fake.calls[0]["impersonate"] == ct.IMPERSONATE_TARGET
run_async(transport.aclose())
def test_response_headers_stripped_and_body_streamed(monkeypatch):
response = FakeCurlResponse(
headers={
"content-type": "text/html",
"content-encoding": "gzip",
"content-length": "5",
"x-foo": "bar",
},
chunks=[b"he", b"llo"],
)
transport, fake = make_transport(monkeypatch, response=response)
request = httpx.Request("GET", "https://example.com/")
out = run_async(transport.handle_async_request(request))
assert out.status_code == 200
assert "content-encoding" not in out.headers
assert "content-length" not in out.headers
assert out.headers["x-foo"] == "bar"
body = run_async(out.aread())
assert body == b"hello"
assert response.closed is True
@pytest.mark.parametrize(
"curl_value,label",
[(2, "HTTP/1.1"), (3, "HTTP/2"), (4, "HTTP/2"), (30, "HTTP/3"), (999, "HTTP/2")],
)
def test_http_version_mapping(monkeypatch, curl_value, label):
response = FakeCurlResponse(http_version=curl_value)
transport, _fake = make_transport(monkeypatch, response=response)
out = run_async(transport.handle_async_request(httpx.Request("GET", "https://example.com/")))
assert out.http_version == label
def test_resolve_timeout_prefers_read_then_connect_then_default():
read = httpx.Request("GET", "https://x/", extensions={"timeout": {"read": 12.0, "connect": 3.0}})
assert resolve_timeout(read) == 12.0
connect = httpx.Request("GET", "https://x/", extensions={"timeout": {"connect": 3.0}})
assert resolve_timeout(connect) == 3.0
none = httpx.Request("GET", "https://x/")
assert resolve_timeout(none) == ct.DEFAULT_TIMEOUT_SECONDS
def test_curl_timeout_maps_to_httpx_connecttimeout(monkeypatch):
transport, _fake = make_transport(monkeypatch, raise_exc=Timeout("slow"))
with pytest.raises(httpx.ConnectTimeout):
run_async(transport.handle_async_request(httpx.Request("GET", "https://example.com/")))
def test_curl_error_maps_to_httpx_connecterror(monkeypatch):
transport, _fake = make_transport(monkeypatch, raise_exc=RequestException("boom"))
with pytest.raises(httpx.ConnectError):
run_async(transport.handle_async_request(httpx.Request("GET", "https://example.com/")))
def test_aclose_closes_session(monkeypatch):
transport, fake = make_transport(monkeypatch)
run_async(transport.aclose())
assert fake.closed is True
tests/unit/database.py
+12
View File
@@ -40,6 +40,8 @@ def _post(owner, **extra):
"image": None,
"stars": 0,
"created_at": _now_db_helpers(),
"deleted_at": None,
"deleted_by": None,
**extra,
}
)
@@ -54,6 +56,8 @@ def _comment(owner, target_uid):
"target_type": "post",
"content": "db helper comment",
"created_at": _now_db_helpers(),
"deleted_at": None,
"deleted_by": None,
}
)
return uid
@@ -66,6 +70,8 @@ def _vote(target_type, target_uid, value):
"target_type": target_type,
"value": value,
"created_at": _now_db_helpers(),
"deleted_at": None,
"deleted_by": None,
}
)
from uuid import uuid4
@@ -94,6 +100,8 @@ def test_get_vote_counts_groups_up_and_down(local_db):
"target_type": "post",
"value": value,
"created_at": _now_db_helpers(),
"deleted_at": None,
"deleted_by": None,
}
)
ups, downs = get_vote_counts([target])
@@ -112,6 +120,8 @@ def test_get_user_votes_returns_user_value(local_db):
"target_type": "post",
"value": 1,
"created_at": _now_db_helpers(),
"deleted_at": None,
"deleted_by": None,
}
)
assert get_user_votes(user, [target]) == {target: 1}
@@ -129,6 +139,8 @@ def test_get_comment_counts_by_post_uids(local_db):
"target_uid": post,
"content": "hi",
"created_at": _now_db_helpers(),
"deleted_at": None,
"deleted_by": None,
}
)
assert get_comment_counts_by_post_uids([post]) == {post: 2}
tests/unit/net_guard.py
+53
View File
@@ -1,12 +1,27 @@
# retoor <retoor@molodetz.nl>
import ipaddress
import httpx
import pytest
from devplacepy import net_guard
from devplacepy.curl_transport import CurlTransport
from devplacepy.net_guard import BlockedAddressError
from tests.conftest import run_async
class _RecordingInner(httpx.AsyncBaseTransport):
def __init__(self):
self.called = False
self.closed = False
async def handle_async_request(self, request):
self.called = True
return httpx.Response(200, content=b"ok", request=request)
async def aclose(self):
self.closed = True
def test_loopback_is_blocked():
assert net_guard.is_blocked_address(ipaddress.ip_address("127.0.0.1"))
@@ -67,3 +82,41 @@ def test_guard_allow_private_skips_resolution():
net_guard.guard_public_url("http://127.0.0.1:9000/x", allow_private=True)
)
assert host == "127.0.0.1"
def test_guarded_client_wraps_stealth_transport():
client = net_guard.guarded_async_client(timeout=5.0)
transport = client._transport
assert isinstance(transport, net_guard._GuardedTransport)
assert isinstance(transport._inner, CurlTransport)
run_async(client.aclose())
def test_guarded_transport_blocks_before_delegating():
inner = _RecordingInner()
transport = net_guard._GuardedTransport(inner)
request = httpx.Request("GET", "http://127.0.0.1/admin")
with pytest.raises(BlockedAddressError):
run_async(transport.handle_async_request(request))
assert inner.called is False
def test_guarded_transport_delegates_when_allowed(monkeypatch):
async def _allow(url, **kwargs):
return "example.com"
monkeypatch.setattr(net_guard, "guard_public_url", _allow)
inner = _RecordingInner()
transport = net_guard._GuardedTransport(inner)
response = run_async(
transport.handle_async_request(httpx.Request("GET", "https://example.com/x"))
)
assert inner.called is True
assert response.status_code == 200
def test_guarded_transport_aclose_closes_inner():
inner = _RecordingInner()
transport = net_guard._GuardedTransport(inner)
run_async(transport.aclose())
assert inner.closed is True
tests/unit/services/audit.py
+4 -3
View File
@@ -157,9 +157,10 @@ def test_record_never_raises(monkeypatch):
raise RuntimeError("db exploded")
monkeypatch.setattr(store, "insert_event", boom)
# must swallow the error and return None rather than propagate
assert audit.record_system("test.unit.raise", actor_kind="system") is None
assert audit.record(None, "test.unit.raise2", user=None) is None
# persistence is deferred to the background task queue (inline under tests); a
# failing write is swallowed there and must never propagate into the caller
audit.record_system("test.unit.raise", actor_kind="system")
audit.record(None, "test.unit.raise2", user=None)
def test_record_with_none_request_uses_explicit_user():
tests/unit/services/bot/__init__.py
+1
View File
@@ -0,0 +1 @@
# retoor <retoor@molodetz.nl>
tests/unit/services/bot/handles.py
+96
View File
@@ -0,0 +1,96 @@
# retoor <retoor@molodetz.nl>
import random
from devplacepy.services.bot.config import SEARCH_TERMS
from devplacepy.services.bot.handles import (
MAX_HANDLE_LEN,
MIN_HANDLE_LEN,
apply_leet,
make_handle,
sanitize_handle,
)
SAMPLE_COUNT = 300
def _is_valid_username(handle: str) -> bool:
return (
handle.isascii()
and bool(handle)
and all(character.isalnum() or character in ("-", "_") for character in handle)
)
def test_make_handle_only_valid_username_characters():
generator = random.Random(1)
for _ in range(SAMPLE_COUNT):
handle = make_handle([], generator)
assert _is_valid_username(handle), handle
def test_make_handle_never_contains_dot():
generator = random.Random(2)
for _ in range(SAMPLE_COUNT):
handle = make_handle(["first app", "raspberry pi"], generator)
assert "." not in handle
def test_make_handle_respects_length_bounds():
generator = random.Random(3)
for _ in range(SAMPLE_COUNT):
handle = make_handle([], generator)
assert MIN_HANDLE_LEN <= len(handle) <= MAX_HANDLE_LEN
def test_make_handle_valid_for_every_persona():
generator = random.Random(4)
for persona, interests in SEARCH_TERMS.items():
for _ in range(SAMPLE_COUNT):
handle = make_handle(interests, generator)
assert _is_valid_username(handle), (persona, handle)
assert MIN_HANDLE_LEN <= len(handle) <= MAX_HANDLE_LEN
def test_make_handle_is_deterministic_for_seeded_rng():
first = [make_handle(["rust", "kubernetes"], random.Random(99)) for _ in range(20)]
second = [make_handle(["rust", "kubernetes"], random.Random(99)) for _ in range(20)]
assert first == second
def test_make_handle_produces_variety():
generator = random.Random(5)
handles = {make_handle([], generator) for _ in range(SAMPLE_COUNT)}
assert len(handles) > SAMPLE_COUNT // 2
def test_sanitize_handle_strips_dots_and_invalid_characters():
assert sanitize_handle("john.smith.dev!!") == "johnsmithdev"
assert sanitize_handle("hello world") == "helloworld"
assert sanitize_handle("a@b#c$d") == "abcd"
def test_sanitize_handle_collapses_and_trims_separators():
assert sanitize_handle("__null__byte__") == "null_byte"
assert sanitize_handle("--seg--fault--") == "seg-fault"
def test_sanitize_handle_caps_length():
assert len(sanitize_handle("a" * 50)) == MAX_HANDLE_LEN
def test_sanitize_handle_empty_input():
assert sanitize_handle("...") == ""
assert sanitize_handle("") == ""
def test_apply_leet_only_substitutes_mapped_characters():
generator = random.Random(7)
result = apply_leet("xyzq", generator)
assert result == "xyzq"
def test_apply_leet_can_substitute_known_characters():
always = random.Random(0)
seen = {apply_leet("aeiost", always) for _ in range(50)}
assert any(character.isdigit() for handle in seen for character in handle)
tests/unit/services/bot/llm.py
+50
View File
@@ -0,0 +1,50 @@
# retoor <retoor@molodetz.nl>
import json
from devplacepy.services.bot.handles import MAX_HANDLE_LEN
from devplacepy.services.bot.llm import LLMClient
def _client(reply: str) -> LLMClient:
client = LLMClient(
api_key="test",
api_url="http://example.invalid",
model="molodetz",
input_cost_per_1m=0.0,
output_cost_per_1m=0.0,
)
client._raw_call = lambda system, prompt, temperature=1.0: reply
return client
def test_generate_handle_candidates_parses_and_sanitizes():
reply = json.dumps({"handles": ["null_byte", "john.smith", "seg fault!!", "c0d3r"]})
handles = _client(reply).generate_handle_candidates("grumpy_senior")
assert handles == ["null_byte", "johnsmith", "segfault", "c0d3r"]
def test_generate_handle_candidates_dedupes_case_insensitively():
reply = json.dumps({"handles": ["NullByte", "nullbyte", "Daemon", "daemon"]})
handles = _client(reply).generate_handle_candidates("minimalist")
assert [handle.lower() for handle in handles] == ["nullbyte", "daemon"]
def test_generate_handle_candidates_drops_empty_results():
reply = json.dumps({"handles": ["", "...", "!!!", "kernelpanic"]})
handles = _client(reply).generate_handle_candidates("rebel")
assert handles == ["kernelpanic"]
def test_generate_handle_candidates_caps_length():
reply = json.dumps({"handles": ["x" * 40]})
handles = _client(reply).generate_handle_candidates("mentor")
assert len(handles[0]) == MAX_HANDLE_LEN
def test_generate_handle_candidates_handles_bad_json():
assert _client("not json at all").generate_handle_candidates("academic_type") == []
def test_generate_handle_candidates_handles_missing_key():
assert _client(json.dumps({"other": []})).generate_handle_candidates("mentor") == []
tests/unit/services/devii/store.py
+4 -4
View File
@@ -10,7 +10,7 @@ def _msgs(text):
def test_channels_isolate_one_owner(local_db):
store = ConversationStore()
owner = f"chan-{generate_uid()[:8]}"
owner = f"chan-{generate_uid()[-12:]}"
store.save("user", owner, _msgs("main thread"), "main")
store.save("user", owner, _msgs("docs thread"), "docs")
assert store.load("user", owner, "main") == _msgs("main thread")
@@ -19,7 +19,7 @@ def test_channels_isolate_one_owner(local_db):
def test_clear_one_channel_keeps_other(local_db):
store = ConversationStore()
owner = f"chan-{generate_uid()[:8]}"
owner = f"chan-{generate_uid()[-12:]}"
store.save("user", owner, _msgs("main"), "main")
store.save("user", owner, _msgs("docs"), "docs")
store.clear("user", owner, "docs")
@@ -29,7 +29,7 @@ def test_clear_one_channel_keeps_other(local_db):
def test_save_updates_in_place_per_channel(local_db):
store = ConversationStore()
owner = f"chan-{generate_uid()[:8]}"
owner = f"chan-{generate_uid()[-12:]}"
store.save("user", owner, _msgs("first"), "docs")
store.save("user", owner, _msgs("second"), "docs")
assert store.load("user", owner, "docs") == _msgs("second")
@@ -37,7 +37,7 @@ def test_save_updates_in_place_per_channel(local_db):
def test_default_channel_is_main(local_db):
store = ConversationStore()
owner = f"chan-{generate_uid()[:8]}"
owner = f"chan-{generate_uid()[-12:]}"
store.save("user", owner, _msgs("default"))
assert store.load("user", owner) == _msgs("default")
assert store.load("user", owner, "main") == _msgs("default")
tests/unit/services/gitea.py
+32 -32
View File
@@ -12,13 +12,13 @@ from devplacepy.services.gitea import runtime, store
from devplacepy.services.gitea.config import gitea_config
from devplacepy.services.gitea.enhance import enhance_ticket
from devplacepy.services.gitea.fake import FakeGiteaClient
from devplacepy.services.gitea.service import BugTrackerService
from devplacepy.services.gitea.service import IssueTrackerService
from devplacepy.services.jobs import queue
from devplacepy.services.jobs.bug_create_service import BugCreateService
from devplacepy.services.jobs.issue_create_service import IssueCreateService
from tests.conftest import run_async
_counter_bugs_gitea = [0]
_counter_issues_gitea = [0]
@pytest.fixture(autouse=True)
def _init_db_bugs_gitea():
def _init_db_issues_gitea():
init_db()
yield
@pytest.fixture
@@ -29,18 +29,18 @@ def gitea_env():
set_setting("gitea_owner", "retoor")
set_setting("gitea_repo", "pydevplace")
set_setting("gitea_token", "test-token")
set_setting("bug_ai_enhance", "0")
set_setting("issue_ai_enhance", "0")
yield fake
runtime.set_client(None)
for table in ("bug_tickets", "bug_comment_authors"):
for table in ("issue_tickets", "issue_comment_authors"):
for row in list(get_table(table).find()):
get_table(table).delete(uid=row["uid"])
for row in list(get_table("jobs").find(kind="bug_create")):
for row in list(get_table("jobs").find(kind="issue_create")):
get_table("jobs").delete(uid=row["uid"])
def _make_user_bugs_gitea():
_counter_bugs_gitea[0] += 1
uid = f"bugtest-user-{_counter_bugs_gitea[0]}"
username = f"bugtester{_counter_bugs_gitea[0]}"
def _make_user_issues_gitea():
_counter_issues_gitea[0] += 1
uid = f"issuetest-user-{_counter_issues_gitea[0]}"
username = f"issuetester{_counter_issues_gitea[0]}"
get_table("users").insert(
{
"uid": uid,
@@ -56,17 +56,17 @@ def _unread(user_uid):
return [
n
for n in get_table("notifications").find(user_uid=user_uid)
if n.get("type") == "bug"
if n.get("type") == "issue"
]
def _drive_jobs():
async def run():
svc = BugCreateService()
svc = IssueCreateService()
for _ in range(200):
await svc.run_once()
refresh_snapshot()
pending = [
r
for r in get_table("jobs").find(kind="bug_create")
for r in get_table("jobs").find(kind="issue_create")
if r["status"] in ("pending", "running")
]
if not pending and not svc._inflight:
@@ -75,7 +75,7 @@ def _drive_jobs():
run_async(run())
def _role_user(role: str):
uid, username = _make_user_bugs_gitea()
uid, username = _make_user_issues_gitea()
get_table("users").update({"uid": uid, "role": role}, ["uid"])
return uid, username
def _open_ticket(fake, role: str):
@@ -130,7 +130,7 @@ def test_fake_client_state_and_comments(gitea_env):
def test_store_records_and_maps(gitea_env):
uid, _ = _make_user_bugs_gitea()
uid, _ = _make_user_issues_gitea()
store.record_ticket(7, uid, "orig", "desc", "Better", "https://x/7", "open")
assert store.author_uid_for_issue(7) == uid
assert store.author_map([7, 8]) == {7: uid}
@@ -149,9 +149,9 @@ def test_enhance_fallback_when_disabled(gitea_env):
def test_create_job_files_issue(gitea_env):
uid, username = _make_user_bugs_gitea()
uid, username = _make_user_issues_gitea()
job_uid = queue.enqueue(
"bug_create",
"issue_create",
{"author_uid": uid, "title": "Login broken", "description": "cannot log in"},
"user",
uid,
@@ -171,9 +171,9 @@ def test_create_job_files_issue(gitea_env):
def test_create_job_fails_without_config(gitea_env):
set_setting("gitea_token", "")
uid, _ = _make_user_bugs_gitea()
uid, _ = _make_user_issues_gitea()
job_uid = queue.enqueue(
"bug_create",
"issue_create",
{"author_uid": uid, "title": "x", "description": "y"},
"user",
uid,
@@ -186,45 +186,45 @@ def test_create_job_fails_without_config(gitea_env):
def test_poller_notifies_on_developer_reply(gitea_env):
uid, _ = _make_user_bugs_gitea()
issue = run_async(gitea_env.create_issue("bug", "body"))
uid, _ = _make_user_issues_gitea()
issue = run_async(gitea_env.create_issue("issue", "body"))
number = issue["number"]
store.record_ticket(number, uid, "bug", "body", "bug", issue["html_url"], "open")
store.record_ticket(number, uid, "issue", "body", "issue", issue["html_url"], "open")
store.update_ticket_cache(number, "open", 0)
gitea_env.add_external_comment(number, "developer", "looking into it")
before = len(_unread(uid))
run_async(BugTrackerService().run_once())
run_async(IssueTrackerService().run_once())
refresh_snapshot()
assert len(_unread(uid)) == before + 1
assert int(store.get_ticket(number)["last_comment_count"]) == 1
def test_poller_ignores_local_comment(gitea_env):
uid, _ = _make_user_bugs_gitea()
issue = run_async(gitea_env.create_issue("bug", "body"))
uid, _ = _make_user_issues_gitea()
issue = run_async(gitea_env.create_issue("issue", "body"))
number = issue["number"]
store.record_ticket(number, uid, "bug", "body", "bug", issue["html_url"], "open")
store.record_ticket(number, uid, "issue", "body", "issue", issue["html_url"], "open")
store.update_ticket_cache(number, "open", 0)
comment = run_async(gitea_env.create_comment(number, "me again"))
store.record_comment_author(comment["id"], number, uid)
before = len(_unread(uid))
run_async(BugTrackerService().run_once())
run_async(IssueTrackerService().run_once())
refresh_snapshot()
assert len(_unread(uid)) == before
def test_poller_notifies_on_status_change(gitea_env):
uid, _ = _make_user_bugs_gitea()
issue = run_async(gitea_env.create_issue("bug", "body"))
uid, _ = _make_user_issues_gitea()
issue = run_async(gitea_env.create_issue("issue", "body"))
number = issue["number"]
store.record_ticket(number, uid, "bug", "body", "bug", issue["html_url"], "open")
store.record_ticket(number, uid, "issue", "body", "issue", issue["html_url"], "open")
store.update_ticket_cache(number, "open", 0)
run_async(gitea_env.set_state(number, "closed"))
before = len(_unread(uid))
run_async(BugTrackerService().run_once())
run_async(IssueTrackerService().run_once())
refresh_snapshot()
assert len(_unread(uid)) == before + 1
assert store.get_ticket(number)["last_status"] == "closed"
tests/unit/services/jobs/deepsearch/pdf.py
+62
View File
@@ -0,0 +1,62 @@
# retoor <retoor@molodetz.nl>
from devplacepy.services.jobs.deepsearch.pdf import extract_pdf_text, is_pdf
def _make_pdf(message: str) -> bytes:
content = b"BT /F1 24 Tf 72 720 Td (" + message.encode("latin-1") + b") Tj ET"
bodies = [
b"<< /Type /Catalog /Pages 2 0 R >>",
b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>",
b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] "
b"/Contents 4 0 R /Resources << /Font << /F1 5 0 R >> >> >>",
b"<< /Length " + str(len(content)).encode() + b" >>\nstream\n" + content + b"\nendstream",
b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>",
]
pdf = bytearray(b"%PDF-1.4\n")
offsets: list[int] = []
for index, body in enumerate(bodies, start=1):
offsets.append(len(pdf))
pdf += str(index).encode() + b" 0 obj\n" + body + b"\nendobj\n"
xref_pos = len(pdf)
pdf += b"xref\n0 " + str(len(bodies) + 1).encode() + b"\n"
pdf += b"0000000000 65535 f \n"
for offset in offsets:
pdf += ("%010d 00000 n \n" % offset).encode()
pdf += b"trailer\n<< /Size " + str(len(bodies) + 1).encode() + b" /Root 1 0 R >>\n"
pdf += b"startxref\n" + str(xref_pos).encode() + b"\n%%EOF"
return bytes(pdf)
def test_is_pdf_detects_content_type():
assert is_pdf("application/pdf", "https://example.com/file", b"") is True
assert is_pdf("application/x-pdf; charset=binary", "https://example.com/file", b"") is True
def test_is_pdf_detects_url_suffix():
assert is_pdf("text/html", "https://example.com/report.PDF", b"<html>") is True
assert is_pdf("", "https://example.com/docs/spec.pdf?v=2", b"") is True
def test_is_pdf_detects_magic_bytes():
assert is_pdf("application/octet-stream", "https://example.com/download", b"%PDF-1.7") is True
def test_is_pdf_rejects_html():
assert is_pdf("text/html", "https://example.com/page", b"<!DOCTYPE html>") is False
def test_extract_pdf_text_returns_content():
title, text = extract_pdf_text(_make_pdf("Hello DeepSearch PDF indexing"))
assert "Hello DeepSearch PDF indexing" in text
def test_extract_pdf_text_rejects_non_pdf():
assert extract_pdf_text(b"this is not a pdf at all") == ("", "")
assert extract_pdf_text(b"") == ("", "")
def test_extract_pdf_text_handles_corrupt_pdf():
title, text = extract_pdf_text(b"%PDF-1.4\ngarbage that is not valid pdf structure")
assert title == ""
assert text == ""
tests/unit/services/jobs/fork_service.py
+1 -1
View File
@@ -16,7 +16,7 @@ def _init_db_fork_jobs():
@pytest.fixture
def fork_env(tmp_path, monkeypatch):
monkeypatch.setattr(
"devplacepy.services.jobs.fork_service.STAGING_DIR", tmp_path / "staging"
"devplacepy.services.jobs.fork_service.FORK_STAGING_DIR", tmp_path / "staging"
)
monkeypatch.setattr("devplacepy.project_files.PROJECT_FILES_DIR", tmp_path / "pf")
yield tmp_path
tests/unit/services/jobs/zip_service.py
+1 -1
View File
@@ -22,7 +22,7 @@ def zip_env(tmp_path, monkeypatch):
"devplacepy.services.jobs.zip_service.ZIPS_DIR", tmp_path / "zips"
)
monkeypatch.setattr(
"devplacepy.services.jobs.zip_service.STAGING_DIR", tmp_path / "staging"
"devplacepy.services.jobs.zip_service.ZIP_STAGING_DIR", tmp_path / "staging"
)
monkeypatch.setattr("devplacepy.project_files.PROJECT_FILES_DIR", tmp_path / "pf")
yield tmp_path
tests/unit/stealth.py
+50
View File
@@ -0,0 +1,50 @@
# retoor <retoor@molodetz.nl>
import httpx
from devplacepy import stealth
from devplacepy.curl_transport import CurlTransport
from tests.conftest import run_async
def test_factory_returns_async_client():
client = stealth.stealth_async_client(timeout=5.0)
assert isinstance(client, httpx.AsyncClient)
run_async(client.aclose())
def test_curl_active_true():
assert stealth.curl_active() is True
def test_stealth_transport_is_curl_when_active():
transport = stealth.stealth_transport()
assert isinstance(transport, CurlTransport)
run_async(transport.aclose())
def test_curl_mode_forwards_headers_without_chrome_base():
client = stealth.stealth_async_client(headers={"Authorization": "Bearer k"})
assert client.headers.get("authorization") == "Bearer k"
assert client.headers.get("sec-ch-ua") is None
run_async(client.aclose())
def test_chrome_base_headers_match_chrome_149():
headers = stealth.chrome_base_headers()
assert "Chrome/149.0.0.0" in headers["user-agent"]
assert headers["sec-ch-ua"] == (
'"Google Chrome";v="149", "Chromium";v="149", "Not)A;Brand";v="24"'
)
assert headers["accept-encoding"] == "gzip, deflate, br, zstd"
def test_fallback_path_injects_chrome_headers(monkeypatch):
monkeypatch.setattr(stealth, "CURL_AVAILABLE", False)
transport = stealth.stealth_transport()
assert isinstance(transport, httpx.AsyncHTTPTransport)
client = stealth.stealth_async_client()
assert client.headers.get("sec-ch-ua") is not None
assert "Chrome/149.0.0.0" in client.headers.get("user-agent", "")
run_async(client.aclose())
run_async(transport.aclose())