Hermes CI Fix
a13a080381
- 新增8个测试文件(bot_bridge/kpi_causality/cash/predict/reports/tax_compliance/expenses/probe_cost) - 增强 budget/auth/users + conftest账套模式适配 - 测试驱动修复: bot_bridge导入batch_id→source_batch; cash_forecast extra空dict - 全量: 451 passed, 1 xfailed; 报告 docs/cma-test-coverage-report.md
246 lines
11 KiB
Python
246 lines
11 KiB
Python
"""税务合规模块测试 — 税负监控 + 发票校验 + 社保比对
|
||
|
||
覆盖 tax_compliance.py 核心端点：
|
||
records CRUD / burden / check / invoices CRUD / invoices/check /
|
||
ss CRUD / ss/check / dashboard / demo-data
|
||
"""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
|
||
from tests.conftest import create_test_user, get_token_for_user, auth_header
|
||
|
||
BASE = "/api/cma/tax"
|
||
|
||
|
||
class TestTaxRecords:
|
||
def test_list_empty(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.get(f"{BASE}/records", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"] == []
|
||
|
||
def test_create(self, client: TestClient, db: Session):
|
||
"""创建税务记录（自动算税负率）"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(f"{BASE}/records", headers=auth_header(token), json={
|
||
"period": "2026-07", "tax_type": "vat",
|
||
"tax_payable": 13, "tax_paid": 13, "income": 100,
|
||
})
|
||
assert resp.status_code == 200
|
||
data = resp.json()["data"]
|
||
assert data["tax_burden_rate"] == 13.0 # 13/100
|
||
|
||
def test_create_missing(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(f"{BASE}/records", headers=auth_header(token), json={"period": "2026-07"})
|
||
assert resp.status_code == 400
|
||
|
||
def test_create_invalid_tax_type(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(f"{BASE}/records", headers=auth_header(token),
|
||
json={"period": "2026-07", "tax_type": "garbage"})
|
||
assert resp.status_code == 400
|
||
|
||
def test_update(self, client: TestClient, db: Session):
|
||
"""更新税务记录（自动重算税负率）"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
cid = client.post(f"{BASE}/records", headers=auth_header(token),
|
||
json={"period": "2026-07", "tax_type": "vat",
|
||
"tax_paid": 10, "income": 100}).json()["data"]["id"]
|
||
resp = client.put(f"{BASE}/records/{cid}", headers=auth_header(token),
|
||
json={"tax_paid": 15, "income": 100})
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"]["tax_burden_rate"] == 15.0
|
||
|
||
def test_update_not_found(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.put(f"{BASE}/records/99999", headers=auth_header(token), json={})
|
||
assert resp.status_code == 404
|
||
|
||
def test_delete(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
cid = client.post(f"{BASE}/records", headers=auth_header(token),
|
||
json={"period": "2026-07", "tax_type": "vat"}).json()["data"]["id"]
|
||
resp = client.delete(f"{BASE}/records/{cid}", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
assert resp.json()["message"] == "税务记录已删除"
|
||
|
||
def test_delete_not_found(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.delete(f"{BASE}/records/99999", headers=auth_header(token))
|
||
assert resp.status_code == 404
|
||
|
||
def test_list_filters(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
client.post(f"{BASE}/records", headers=auth_header(token),
|
||
json={"period": "2026-06", "tax_type": "vat"})
|
||
client.post(f"{BASE}/records", headers=auth_header(token),
|
||
json={"period": "2026-07", "tax_type": "income"})
|
||
resp = client.get(f"{BASE}/records?period=2026-07", headers=auth_header(token))
|
||
assert resp.json()["total"] == 1
|
||
resp2 = client.get(f"{BASE}/records?tax_type=vat", headers=auth_header(token))
|
||
assert resp2.json()["total"] == 1
|
||
|
||
def test_burden_analysis(self, client: TestClient, db: Session):
|
||
"""税负分析"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
client.post(f"{BASE}/records", headers=auth_header(token),
|
||
json={"period": "2026-07", "tax_type": "vat",
|
||
"tax_paid": 13, "income": 100})
|
||
resp = client.get(f"{BASE}/burden", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
|
||
def test_check_no_data(self, client: TestClient, db: Session):
|
||
"""税负检查无数据"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(f"{BASE}/check", headers=auth_header(token), json={})
|
||
assert resp.status_code == 200
|
||
|
||
|
||
class TestInvoices:
|
||
def test_create_and_list(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(f"{BASE}/invoices", headers=auth_header(token), json={
|
||
"invoice_no": "INV001", "amount": 1000, "supplier": "供应商A",
|
||
})
|
||
assert resp.status_code == 200
|
||
assert resp.json()["message"] == "发票已录入并校验"
|
||
|
||
list_resp = client.get(f"{BASE}/invoices", headers=auth_header(token))
|
||
assert list_resp.status_code == 200
|
||
assert list_resp.json()["total"] == 1
|
||
|
||
def test_create_missing_no(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(f"{BASE}/invoices", headers=auth_header(token), json={"amount": 100})
|
||
assert resp.status_code == 400
|
||
|
||
def test_update_invoice(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
iid = client.post(f"{BASE}/invoices", headers=auth_header(token),
|
||
json={"invoice_no": "INV002", "amount": 100}).json()["data"]["id"]
|
||
resp = client.put(f"{BASE}/invoices/{iid}", headers=auth_header(token),
|
||
json={"amount": 200, "supplier": "供应商B"})
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"]["amount"] == 200
|
||
|
||
def test_delete_invoice(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
iid = client.post(f"{BASE}/invoices", headers=auth_header(token),
|
||
json={"invoice_no": "INV003", "amount": 100}).json()["data"]["id"]
|
||
resp = client.delete(f"{BASE}/invoices/{iid}", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
|
||
def test_check_invoices(self, client: TestClient, db: Session):
|
||
"""发票校验"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
client.post(f"{BASE}/invoices", headers=auth_header(token),
|
||
json={"invoice_no": "INV004", "amount": 100, "supplier": ""})
|
||
resp = client.post(f"{BASE}/invoices/check", headers=auth_header(token), json={})
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["total"] == 1
|
||
assert data["abnormal_count"] >= 0
|
||
|
||
def test_abnormal_invoices(self, client: TestClient, db: Session):
|
||
"""异常发票列表"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
client.post(f"{BASE}/invoices", headers=auth_header(token),
|
||
json={"invoice_no": "INV005", "amount": 100, "supplier": ""})
|
||
resp = client.get(f"{BASE}/invoices/abnormal", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
assert "data" in resp.json()
|
||
|
||
|
||
class TestSocialSecurity:
|
||
def test_crud(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(f"{BASE}/ss", headers=auth_header(token), json={
|
||
"employee": "张三", "period": "2026-07", "base_amount": 5000,
|
||
"company_amount": 1225, "salary": 6000,
|
||
})
|
||
assert resp.status_code == 200
|
||
ss_id = resp.json()["data"]["id"]
|
||
|
||
list_resp = client.get(f"{BASE}/ss", headers=auth_header(token))
|
||
assert list_resp.status_code == 200
|
||
assert list_resp.json()["total"] == 1
|
||
|
||
upd = client.put(f"{BASE}/ss/{ss_id}", headers=auth_header(token),
|
||
json={"base_amount": 6000})
|
||
assert upd.status_code == 200
|
||
|
||
resp = client.delete(f"{BASE}/ss/{ss_id}", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
|
||
def test_ss_check(self, client: TestClient, db: Session):
|
||
"""社保比对"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
client.post(f"{BASE}/ss", headers=auth_header(token),
|
||
json={"employee": "李四", "period": "2026-07",
|
||
"base_amount": 5000, "company_amount": 1225})
|
||
resp = client.post(f"{BASE}/ss/check", headers=auth_header(token), json={})
|
||
assert resp.status_code == 200
|
||
|
||
def test_ss_abnormal(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.get(f"{BASE}/ss/abnormal", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
|
||
|
||
class TestDashboardDemo:
|
||
def test_dashboard(self, client: TestClient, db: Session):
|
||
"""税务驾驶舱"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
client.post(f"{BASE}/records", headers=auth_header(token),
|
||
json={"period": "2026-07", "tax_type": "vat",
|
||
"tax_paid": 13, "income": 100})
|
||
resp = client.get(f"{BASE}/dashboard", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "burden" in data and "invoice" in data and "ss" in data
|
||
|
||
@pytest.mark.xfail(reason="SQLite DateTime 不接受字符串日期（生产MySQL可隐式转换）；demo-data 为MySQL-only端点", strict=False)
|
||
def test_demo_data(self, client: TestClient, db: Session):
|
||
"""生成演示数据（MySQL-only：demo日期为字符串，SQLite DateTime严格模式不兼容）"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(f"{BASE}/demo-data", headers=auth_header(token), json={})
|
||
assert resp.status_code == 200
|
||
|
||
|
||
class TestTaxPermissions:
|
||
def test_business_write_denied(self, client: TestClient, db: Session):
|
||
"""business用户写税务 → 403"""
|
||
import hashlib
|
||
from app.models import User
|
||
db.add(User(username="tax_biz", password_hash=hashlib.sha256("p".encode()).hexdigest(),
|
||
name="业务", role="business"))
|
||
db.commit()
|
||
token = get_token_for_user(client, username="tax_biz", password="p")
|
||
resp = client.post(f"{BASE}/records", headers=auth_header(token),
|
||
json={"period": "2026-07", "tax_type": "vat"})
|
||
assert resp.status_code == 403
|