zhanghongyu_0626
f856ab4aa6
- Updated test files to import `AgentSessionLocal` from `advisor_db` instead of directly, preventing session binding to the real database during tests. - Fixed 6 test cases to use the new login token utility, ensuring consistency across authentication methods. - Adjusted customer risk codes in `test_convert_confirm.py` to reflect changes in customer classification (C3 to C4). - Verified that changes resulted in zero database pollution during test runs, maintaining integrity of the testing environment. - Documented findings and updates in the relevant test logs and memory files, ensuring clarity on the current state of tests and defects.
209 lines
6.9 KiB
Python
209 lines
6.9 KiB
Python
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from app import advisor_db
|
|
from app.advisor_exceptions import OwnershipDeniedError
|
|
from app.api.advisor_auth_adapter import AdvisorAuthContext
|
|
from app.model.entities import AuditLog
|
|
from app.service.audit_service import AuditService, InMemoryAuditRepository
|
|
from app.service.ownership_service import OwnershipService
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
def test_requirements_include_sprint0_approved_dependencies():
|
|
requirements = (ROOT / "requirements.txt").read_text(encoding="utf-8")
|
|
|
|
for package_name in [
|
|
"pytest",
|
|
"httpx",
|
|
]:
|
|
assert package_name in requirements
|
|
|
|
def test_advisor_agent_migration_sql_exists():
|
|
path = ROOT / "scripts" / "agent" / "migrate-advisor-agent-sprint1-3.sql"
|
|
assert path.exists()
|
|
text = path.read_text(encoding="utf-8")
|
|
assert "compliance_rule" in text
|
|
assert "kyc_session" in text
|
|
|
|
def test_core_reset_script_runs_core_seed_and_optional_neo4j_sync():
|
|
reset_script = (ROOT / "scripts" / "core" / "reset.ps1").read_text(encoding="utf-8")
|
|
|
|
assert "scripts/core/01-ddl.sql" in reset_script
|
|
assert "python scripts/sync/sync_advisor_rel.py" in reset_script
|
|
assert "[switch]$SkipNeo4j" in reset_script
|
|
assert "if (-not $SkipNeo4j)" in reset_script
|
|
assert "python scripts/sync/sync_neo4j.py" in reset_script
|
|
|
|
|
|
def test_core_reset_script_uses_run_sql_file_helper():
|
|
reset_script = (ROOT / "scripts" / "core" / "reset.ps1").read_text(encoding="utf-8")
|
|
|
|
assert "scripts/dev/run_sql_file.py" in reset_script
|
|
assert "fix_utf8_seed.py" in reset_script
|
|
|
|
|
|
def test_sprint0_test_suite_covers_required_quality_gates():
|
|
tests_text = "\n".join(
|
|
path.read_text(encoding="utf-8")
|
|
for path in sorted((ROOT / "tests").glob("test_sprint0_*.py"))
|
|
)
|
|
|
|
required_checks = [
|
|
"test_requirements_include_sprint0_approved_dependencies",
|
|
"test_advisor_agent_migration_sql_exists",
|
|
"test_core_reset_script_runs_core_seed_and_optional_neo4j_sync",
|
|
"test_sqlalchemy_audit_repository_persists_append_only_event",
|
|
"test_trace_id_header_is_reused_in_response_and_rbac_audit_event",
|
|
"test_missing_trace_id_generates_response_header_on_unauthenticated_request",
|
|
"test_platform_compliance_suitability_unchanged",
|
|
]
|
|
|
|
for check_name in required_checks:
|
|
assert check_name in tests_text
|
|
|
|
class FakeCoreRepo:
|
|
def __init__(self, assigned: bool) -> None:
|
|
self.assigned = assigned
|
|
|
|
def is_advisor_assigned(self, advisor_id: str, customer_id: str) -> bool:
|
|
return self.assigned and advisor_id == "ADV-TEST-001" and customer_id == "CUST-001"
|
|
|
|
def test_ownership_guard_allows_assigned_advisor_customer():
|
|
service = OwnershipService(FakeCoreRepo(assigned=True))
|
|
auth = AdvisorAuthContext(
|
|
user_id="STAFF-10086",
|
|
token_type="staff",
|
|
roles=["advisor"],
|
|
permissions=["advisor:workspace"],
|
|
advisor_id="ADV-TEST-001",
|
|
trace_id="trace-owner-ok",
|
|
)
|
|
|
|
service.assert_customer_access(auth, "CUST-001")
|
|
|
|
def test_ownership_guard_denies_unassigned_advisor_customer():
|
|
service = OwnershipService(FakeCoreRepo(assigned=False))
|
|
auth = AdvisorAuthContext(
|
|
user_id="STAFF-10086",
|
|
token_type="staff",
|
|
roles=["advisor"],
|
|
permissions=["advisor:workspace"],
|
|
advisor_id="ADV-TEST-001",
|
|
trace_id="trace-owner-deny",
|
|
)
|
|
|
|
with pytest.raises(OwnershipDeniedError):
|
|
service.assert_customer_access(auth, "CUST-999")
|
|
|
|
def test_audit_service_appends_events_without_mutating_prior_records():
|
|
repository = InMemoryAuditRepository()
|
|
service = AuditService(repository)
|
|
|
|
service.record(
|
|
trace_id="trace-audit-001",
|
|
event_type="auth_login",
|
|
actor_id="advisor_test",
|
|
decision="success",
|
|
)
|
|
service.record(
|
|
trace_id="trace-audit-002",
|
|
event_type="rbac_denied",
|
|
actor_id="advisor_test",
|
|
decision="audit:read",
|
|
)
|
|
|
|
events = service.list_events()
|
|
assert len(events) == 2
|
|
assert events[0].trace_id == "trace-audit-001"
|
|
assert events[1].trace_id == "trace-audit-002"
|
|
|
|
def test_sqlalchemy_audit_repository_persists_append_only_event():
|
|
from app.service import audit_service as audit_module
|
|
|
|
trace_id = f"trace-db-{uuid4().hex}"
|
|
repository_class = getattr(audit_module, "SqlAlchemyAuditRepository", None)
|
|
assert repository_class is not None
|
|
repository = repository_class(advisor_db.AgentSessionLocal)
|
|
service = AuditService(repository)
|
|
|
|
service.record(
|
|
trace_id=trace_id,
|
|
event_type="audit_repository_test",
|
|
actor_id="test_runner",
|
|
decision="persisted",
|
|
input_summary={"source": "pytest"},
|
|
)
|
|
|
|
with advisor_db.AgentSessionLocal() as session:
|
|
event = session.query(AuditLog).filter(AuditLog.trace_id == trace_id).one()
|
|
|
|
assert event.event_type == "audit_repository_test"
|
|
assert event.actor_id == "test_runner"
|
|
assert event.decision == "persisted"
|
|
assert event.input_summary == {"source": "pytest"}
|
|
|
|
def test_unknown_staff_login_is_rejected():
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
|
|
response = TestClient(app).post(
|
|
"/api/auth/login",
|
|
json={"actor_id": "STAFF-UNKNOWN-ACTOR", "token_type": "staff"},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
def test_trace_id_header_is_reused_in_response_and_rbac_audit_event():
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
|
|
trace_id = f"trace-rbac-db-{uuid4().hex}"
|
|
login = TestClient(app).post(
|
|
"/api/auth/login",
|
|
json={"actor_id": "STAFF-10086", "token_type": "staff"},
|
|
)
|
|
token = login.json()["data"]["access_token"]
|
|
|
|
response = TestClient(app).post(
|
|
"/api/advisor-agent/compliance/rules",
|
|
json={
|
|
"rule_type": "keyword",
|
|
"pattern": "trace-id-rbac-test",
|
|
"severity": "block",
|
|
"category": "return_promise",
|
|
"suggestion": "no promise",
|
|
},
|
|
headers={"Authorization": f"Bearer {token}", "X-Trace-Id": trace_id},
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
assert response.headers["X-Trace-Id"] == trace_id
|
|
assert response.json()["trace_id"] == trace_id
|
|
|
|
with advisor_db.AgentSessionLocal() as session:
|
|
event = (
|
|
session.query(AuditLog)
|
|
.filter(AuditLog.trace_id == trace_id, AuditLog.event_type == "rbac_denied")
|
|
.one()
|
|
)
|
|
|
|
assert event.actor_id == "STAFF-10086"
|
|
assert event.decision == "compliance:rule:write"
|
|
|
|
def test_missing_trace_id_generates_response_header_on_unauthenticated_request():
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
|
|
response = TestClient(app).get("/api/advisor-agent/compliance/ping")
|
|
|
|
assert response.status_code == 401
|
|
generated_trace_id = response.headers["X-Trace-Id"]
|
|
assert generated_trace_id
|
|
assert response.json()["trace_id"] == generated_trace_id
|