Spaces:
Sleeping
Sleeping
File size: 6,533 Bytes
ce8f04a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 | """Structural + unit: API handlers call shipped eligibility/strategy/tax functions."""
from __future__ import annotations
import inspect
import pytest
from fastapi import HTTPException
from endpoints import company as company_ep
from core.eligibility.spine import (
build_eligibility_report,
eligibility_override_is_complete,
)
from core.strategy.recommend import PATH_TAX, recommend_strategy_paths
from core.strategy.tax_paths import (
PARENT_STRATEGY_PATH_ID,
TAX_PATH_IP_BOX,
TAX_PATH_PSI,
TAX_PATH_ULGA_BR,
)
def test_company_router_has_eligibility_and_strategy_routes():
paths = {getattr(r, "path", "") for r in company_ep.router.routes}
# FastAPI may store path without prefix on route
joined = " ".join(sorted(paths))
assert "eligibility" in joined
assert "strategy" in joined or "recommend" in joined
assert "tax" in joined
def test_tax_router_has_paths_route():
"""Plan §8 GET /api/tax/paths + company /tax/paths both exposed."""
company_paths = " ".join(
sorted(getattr(r, "path", "") for r in company_ep.router.routes)
)
assert "tax" in company_paths and "paths" in company_paths
tax_paths = {getattr(r, "path", "") for r in company_ep.tax_router.routes}
joined = " ".join(sorted(tax_paths))
assert "paths" in joined
assert any(
getattr(r, "path", "").endswith("/paths") or getattr(r, "path", "") == "/paths"
for r in company_ep.tax_router.routes
)
def test_get_eligibility_handler_uses_build_report():
src = inspect.getsource(company_ep.get_company_eligibility)
assert "build_eligibility_report" in src
def test_strategy_handler_uses_recommend():
src = inspect.getsource(company_ep.post_strategy_recommend)
assert "recommend_strategy_paths" in src
# POST must load profile spine when eligibility omitted (parity with GET)
assert "company_data_from_profile" in src
assert "get_db" in src or "db" in src
# Incomplete body.eligibility must rebuild (not fail-open)
assert "eligibility_override_is_complete" in src
src2 = inspect.getsource(company_ep.get_strategy_recommend)
assert "recommend_strategy_paths" in src2
def test_handlers_are_callable_with_pure_backend():
# Direct pure path the handlers wrap
rep = build_eligibility_report(
nip="8721990161",
company_data={"msp_status": "mikro", "de_minimis_manual_eur": 100},
)
strat = recommend_strategy_paths(eligibility=rep, goal="B+R", company_data={})
assert rep["nip"]
assert len(strat["paths"]) >= 4
tax = next(p for p in strat["paths"] if p["id"] == PATH_TAX)
assert tax.get("tax_paths") and len(tax["tax_paths"]) == 3
assert tax.get("never_say_submit_to_parp") is True
def test_get_tax_paths_api_surface():
"""GET tax-paths returns PSI / ulga_br / ip_box with PSI PARP guardrail."""
out = company_ep.get_tax_paths(path_id=None, token_data={"sub": "user_tax_api"})
assert out["status"] == "ok"
assert out["parent_strategy_path_id"] == PARENT_STRATEGY_PATH_ID
assert out["checklist_ids"] == [TAX_PATH_PSI, TAX_PATH_ULGA_BR, TAX_PATH_IP_BOX]
assert len(out["paths"]) == 3
ids = [p["id"] for p in out["paths"]]
assert ids == [TAX_PATH_PSI, TAX_PATH_ULGA_BR, TAX_PATH_IP_BOX]
psi = next(p for p in out["paths"] if p["id"] == TAX_PATH_PSI)
assert psi["never_say_submit_to_parp"] is True
assert isinstance(psi["checklist"], list) and len(psi["checklist"]) >= 3
blob = " ".join(psi["checklist"] + psi.get("advisor_notes", [])).lower()
assert "parp" in blob
def test_get_tax_paths_single_id_and_404():
one = company_ep.get_tax_paths(path_id="psi", token_data={"sub": "user_tax_api"})
assert one["status"] == "ok"
assert one["path"]["id"] == TAX_PATH_PSI
assert one["checklist_ids"] == [TAX_PATH_PSI]
with pytest.raises(HTTPException) as ei:
company_ep.get_tax_paths(path_id="not_a_path", token_data={"sub": "user_tax_api"})
assert ei.value.status_code == 404
def test_get_tax_paths_plan_alias_delegates():
"""Plan-canonical /api/tax/paths handler shares response shape."""
a = company_ep.get_tax_paths(path_id=None, token_data={"sub": "u"})
b = company_ep.get_tax_paths_plan_alias(path_id=None, token_data={"sub": "u"})
assert a["checklist_ids"] == b["checklist_ids"]
assert len(a["paths"]) == len(b["paths"]) == 3
def test_get_tax_paths_compact_alias():
"""GET /api/company/tax-paths compact surface (enabled flag + catalog)."""
out = company_ep.get_tax_paths_compact(token_data={"sub": "user_tax_compact"})
assert out["status"] == "ok"
assert out["enabled"] is True
assert len(out["tax_paths"]) == 3
ids = [p["id"] for p in out["tax_paths"]]
assert ids == [TAX_PATH_PSI, TAX_PATH_ULGA_BR, TAX_PATH_IP_BOX]
psi = next(p for p in out["tax_paths"] if p["id"] == TAX_PATH_PSI)
assert psi["never_say_submit_to_parp"] is True
def test_post_strategy_incomplete_eligibility_rebuilds_and_includes_tax():
"""Thin body.eligibility rebuilds spine; strategy still embeds tax checklists."""
class _EmptyDB:
def query(self, *a, **k):
class _Q:
def filter(self, *a, **k):
return self
def first(self):
return None
return _Q()
body = company_ep.StrategyRecommendBody(
nip="8721990161",
goal="B+R",
eligibility={"msp": {"status": "mikro"}}, # incomplete — no de_minimis
company_data={"msp_status": "mikro", "de_minimis_manual_eur": 100},
)
assert eligibility_override_is_complete(body.eligibility) is False
out = company_ep.post_strategy_recommend(
body=body, token_data={"sub": "user_tax_strat"}, db=_EmptyDB()
)
elig = out["eligibility"]
assert elig.get("de_minimis", {}).get("used_eur") == 100.0
assert elig.get("msp", {}).get("status")
tax = next(p for p in out["strategy"]["paths"] if p["id"] == PATH_TAX)
assert tax.get("tax_paths") and len(tax["tax_paths"]) == 3
assert tax.get("never_say_submit_to_parp") is True
def test_post_strategy_incomplete_eligibility_rebuilds_in_handler_source():
"""Guard residual: thin eligibility must not skip spine rebuild."""
assert eligibility_override_is_complete({"msp": {"status": "mikro"}}) is False
src = inspect.getsource(company_ep.post_strategy_recommend)
assert "if not eligibility_override_is_complete" in src
assert "build_eligibility_report" in src
|