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.
99 lines
2.7 KiB
Python
99 lines
2.7 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 production infrastructure (auth, db, observability, jobs)."""
|
|
|
|
import asyncio
|
|
|
|
from auth import AuthManager, base64_decode, base64_encode
|
|
from db import _has_sqlalchemy, get_db
|
|
from observability import track_llm_call, track_request, track_scrape
|
|
from tasks import JobQueue, JobStatus
|
|
|
|
|
|
def test_auth_hash_password() -> None:
|
|
a = AuthManager()
|
|
h, salt = a.hash_password("hunter2")
|
|
assert a.verify_password("hunter2", h, salt) is True
|
|
assert a.verify_password("wrong", h, salt) is False
|
|
|
|
|
|
def test_auth_api_key() -> None:
|
|
a = AuthManager()
|
|
user = a.create_user("test@example.com", "password")
|
|
key = a.create_api_key(user["id"])
|
|
assert key.startswith("pry_")
|
|
verified = a.verify_api_key(key)
|
|
assert verified is not None
|
|
assert verified["user_id"] == user["id"]
|
|
# Invalid key
|
|
assert a.verify_api_key("invalid") is None
|
|
|
|
|
|
def test_auth_rate_limit() -> None:
|
|
a = AuthManager()
|
|
a._rate_limits["test"] = {"window_start": 0, "count": 0}
|
|
allowed, remaining = a.check_rate_limit("test")
|
|
assert allowed is True
|
|
assert remaining >= 0
|
|
|
|
|
|
def test_auth_jwt_fallback() -> None:
|
|
a = AuthManager()
|
|
user = a.create_user("jwt@test.com", "pass")
|
|
token = a.create_jwt(user["id"])
|
|
payload = a.verify_jwt(token)
|
|
assert payload is not None
|
|
assert payload["sub"] == user["id"]
|
|
|
|
|
|
def test_base64_roundtrip() -> None:
|
|
assert base64_decode(base64_encode("hello world")) == "hello world"
|
|
|
|
|
|
def test_observability_track_request() -> None:
|
|
with track_request("/test"):
|
|
pass # Should not raise
|
|
|
|
|
|
def test_observability_track_scrape() -> None:
|
|
with track_scrape("direct"):
|
|
pass
|
|
|
|
|
|
def test_observability_track_llm() -> None:
|
|
track_llm_call("openai", "gpt-4o-mini", 0.001)
|
|
|
|
|
|
def test_db_is_available() -> None:
|
|
# Just check it doesn't crash; may be unavailable if no sqlalchemy
|
|
db = get_db()
|
|
if _has_sqlalchemy:
|
|
assert db is not None
|
|
|
|
|
|
def test_job_queue_submit() -> None:
|
|
async def _test():
|
|
q = JobQueue()
|
|
|
|
async def my_task(x):
|
|
return x * 2
|
|
|
|
job_id = await q.submit("double", my_task, args=(5,))
|
|
assert job_id in q.jobs
|
|
# Run inline
|
|
job = q.jobs[job_id]
|
|
await q._execute(job)
|
|
assert job.status == JobStatus.COMPLETED
|
|
assert job.result == 10
|
|
|
|
asyncio.run(_test())
|
|
|
|
|
|
def test_job_listing() -> None:
|
|
q = JobQueue()
|
|
q.jobs["test"] = type("J", (), {"to_dict": lambda self: {"id": "test", "name": "x"}})()
|
|
# Simple smoke test
|
|
assert hasattr(q, "list_jobs")
|