99 lines
4.1 KiB
Python
99 lines
4.1 KiB
Python
"""Check the public signer and migration without connecting to a configured database."""
|
|
|
|
import importlib.util
|
|
from io import StringIO
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
from alembic.migration import MigrationContext
|
|
from alembic.operations import Operations
|
|
from app.services.integration_auth import signature_message
|
|
from fastapi import Request
|
|
from scripts.call_external_follow_up import sign_follow_up_request
|
|
from sqlalchemy import create_engine, inspect
|
|
|
|
|
|
def test_client_signature_matches_server_for_encoded_url_and_non_ascii_body():
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
|
|
url = httpx.URL("http://test/api/v1/integrations/risk-assessments/follow-up")
|
|
body = '{"content":"总结原报告"}'.encode()
|
|
timestamp, nonce, secret = "1793145600", "nonce-with-32-valid-characters-123", "test-secret-" * 4
|
|
request = Request(
|
|
{
|
|
"type": "http",
|
|
"method": "POST",
|
|
"scheme": "http",
|
|
"server": ("test", 80),
|
|
"path": url.path,
|
|
"raw_path": url.raw_path.partition(b"?")[0],
|
|
"query_string": url.query,
|
|
"headers": [],
|
|
}
|
|
)
|
|
signed = sign_follow_up_request("POST", url, "dieya-plm", secret, body, timestamp=timestamp, nonce=nonce)
|
|
canonical = signature_message(request, timestamp, nonce, body)
|
|
assert signed["X-Signature-Version"] == "2"
|
|
assert (
|
|
signed["X-Signature"]
|
|
== base64.b64encode(hmac.new(secret.encode(), canonical, hashlib.sha256).digest()).decode()
|
|
)
|
|
|
|
|
|
def test_documented_signature_example_is_executable_and_matches_server():
|
|
from unittest.mock import patch
|
|
|
|
root = Path(__file__).resolve().parents[1]
|
|
doc = (root / "docs" / "外部系统风险评估报告接口.md").read_text(encoding="utf-8")
|
|
examples = [part.split("```", 1)[0] for part in doc.split("```python\n")[1:]]
|
|
example = next(part for part in examples if "def sign_follow_up_request" in part)
|
|
scope = {}
|
|
exec(compile(example, "documented HMAC example", "exec"), scope)
|
|
body = '{"content":"请解释"}'.encode()
|
|
target = "/api/v1/integrations/risk-assessments/follow-up"
|
|
with patch("time.time", return_value=1793145600), patch("secrets.token_urlsafe", return_value="fixed-nonce-123"):
|
|
documented = scope["sign_follow_up_request"]("POST", target, body, "dieya-plm", "secret" * 8)
|
|
expected = sign_follow_up_request(
|
|
"POST",
|
|
httpx.URL("http://test" + target),
|
|
"dieya-plm",
|
|
"secret" * 8,
|
|
body,
|
|
timestamp="1793145600",
|
|
nonce="fixed-nonce-123",
|
|
)
|
|
assert documented == expected
|
|
|
|
|
|
def test_migration_postgresql_ddl_and_upgrade_downgrade_preserve_existing_tables():
|
|
root = Path(__file__).resolve().parents[1]
|
|
filename = root / "migrations" / "versions" / "20261028_0023_integration_conversations.py"
|
|
spec = importlib.util.spec_from_file_location("integration_chat_migration", filename)
|
|
migration = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(migration)
|
|
output = StringIO()
|
|
context = MigrationContext.configure(dialect_name="postgresql", opts={"as_sql": True, "output_buffer": output})
|
|
with Operations.context(context):
|
|
migration.upgrade()
|
|
sql = output.getvalue()
|
|
assert "CREATE TABLE integration_conversations" in sql
|
|
assert "REFERENCES integration_assessments (client_id, request_id)" in sql
|
|
assert "CREATE TABLE integration_conversation_requests" in sql
|
|
|
|
engine = create_engine("sqlite://")
|
|
try:
|
|
with engine.begin() as connection:
|
|
connection.exec_driver_sql("CREATE TABLE conversations (id BIGINT PRIMARY KEY)")
|
|
connection.exec_driver_sql(
|
|
"CREATE TABLE integration_assessments (client_id TEXT, request_id TEXT, UNIQUE(client_id, request_id))"
|
|
)
|
|
with Operations.context(MigrationContext.configure(connection)):
|
|
migration.upgrade()
|
|
assert len(inspect(connection).get_columns("integration_conversation_requests")) == 14
|
|
migration.downgrade()
|
|
assert set(inspect(connection).get_table_names()) == {"conversations", "integration_assessments"}
|
|
finally:
|
|
engine.dispose()
|