200 lines
7.6 KiB
Python
200 lines
7.6 KiB
Python
import unittest
|
|
|
|
from nocobase_crm_mcp.config import ProjectConfig
|
|
from nocobase_crm_mcp.nocobase_http import NocoBaseHttpError
|
|
from nocobase_crm_mcp.projects import PROJECT_WHITELIST, ProjectService
|
|
|
|
|
|
class FakeNbClient:
|
|
def __init__(self) -> None:
|
|
self.create_calls: list[tuple[str, dict, tuple[str, ...]]] = []
|
|
self.update_calls: list[tuple[str, object, dict, tuple[str, ...]]] = []
|
|
|
|
def create_record(self, resource, values, whitelist):
|
|
self.create_calls.append((resource, dict(values), tuple(whitelist)))
|
|
return {"data": {"id": 41}}
|
|
|
|
def update_record(self, resource, project_id, values, whitelist):
|
|
self.update_calls.append((resource, project_id, dict(values), tuple(whitelist)))
|
|
return {"data": {"id": project_id}}
|
|
|
|
|
|
class FakeHttpClient:
|
|
def __init__(self) -> None:
|
|
self.add_calls: list[tuple[object, list[object]]] = []
|
|
self.remove_calls: list[tuple[object, list[object]]] = []
|
|
self.add_error: Exception | None = None
|
|
self.remove_error: Exception | None = None
|
|
|
|
def add_attachment_ids(self, project_id, attachment_ids) -> None:
|
|
self.add_calls.append((project_id, list(attachment_ids)))
|
|
if self.add_error:
|
|
raise self.add_error
|
|
|
|
def remove_attachment_ids(self, project_id, attachment_ids) -> None:
|
|
self.remove_calls.append((project_id, list(attachment_ids)))
|
|
if self.remove_error:
|
|
raise self.remove_error
|
|
|
|
|
|
class ProjectServiceTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.nb = FakeNbClient()
|
|
self.http = FakeHttpClient()
|
|
self.config = ProjectConfig()
|
|
self.service = ProjectService(self.config, self.nb, self.http)
|
|
|
|
def test_create_maps_fixed_fields_then_adds_attachment_ids(self) -> None:
|
|
result = self.service.create_project(
|
|
name=" EV inverter ", customer_id=3, progress="SOP", attachment_ids=[8, 9]
|
|
)
|
|
|
|
self.assertEqual(
|
|
self.nb.create_calls,
|
|
[
|
|
(
|
|
"t_projects",
|
|
{
|
|
"f_project_name": "EV inverter",
|
|
"client_ID": 3,
|
|
"f_projects_progress": "a9t9awhtt3r",
|
|
},
|
|
PROJECT_WHITELIST,
|
|
)
|
|
],
|
|
)
|
|
self.assertEqual(self.http.add_calls, [(41, [8, 9])])
|
|
self.assertEqual(result["project_id"], 41)
|
|
|
|
def test_whitelist_contains_only_the_configured_scalar_fields(self) -> None:
|
|
self.service.create_project(name="x")
|
|
|
|
self.assertEqual(
|
|
PROJECT_WHITELIST,
|
|
(
|
|
self.config.name_field,
|
|
self.config.customer_field,
|
|
self.config.inquiry_date_field,
|
|
self.config.progress_field,
|
|
self.config.applications_field,
|
|
self.config.part_numbers_field,
|
|
self.config.period_usage_field,
|
|
self.config.details_field,
|
|
),
|
|
)
|
|
self.assertNotIn(self.config.attachment_field, PROJECT_WHITELIST)
|
|
|
|
def test_update_rejects_overlapping_attachment_lists_before_writing(self) -> None:
|
|
result = self.service.update_project(41, add_attachment_ids=[8], remove_attachment_ids=[8])
|
|
|
|
self.assertEqual(result["error"]["code"], "invalid_project_input")
|
|
self.assertEqual(self.nb.update_calls, [])
|
|
self.assertEqual(self.http.add_calls, [])
|
|
self.assertEqual(self.http.remove_calls, [])
|
|
|
|
def test_invalid_date_progress_and_blank_name_have_stable_codes(self) -> None:
|
|
self.assertEqual(
|
|
self.service.create_project(name="x", inquiry_date="2026/07/12")["error"]["code"],
|
|
"invalid_project_input",
|
|
)
|
|
self.assertEqual(
|
|
self.service.create_project(name="x", inquiry_date="20260712")["error"]["code"],
|
|
"invalid_project_input",
|
|
)
|
|
self.assertEqual(
|
|
self.service.create_project(name="x", progress="草稿")["error"]["code"],
|
|
"invalid_progress",
|
|
)
|
|
self.assertEqual(self.service.create_project(name=" \t ")["error"]["code"], "invalid_project_input")
|
|
|
|
def test_update_rejects_an_empty_request_before_writing(self) -> None:
|
|
result = self.service.update_project(41)
|
|
|
|
self.assertEqual(result["error"]["code"], "invalid_project_input")
|
|
self.assertEqual(self.nb.update_calls, [])
|
|
|
|
def test_create_rejects_nested_scalar_inputs_before_writing(self) -> None:
|
|
cases = {
|
|
"name": {"name": "x"},
|
|
"customer_id": {"id": 3},
|
|
"inquiry_date": ["2026-07-12"],
|
|
"progress": ("SOP",),
|
|
"applications": ["EV"],
|
|
"part_numbers": ("123",),
|
|
"period_usage": {"annual"},
|
|
"details": {"note": "x"},
|
|
}
|
|
|
|
for field, value in cases.items():
|
|
with self.subTest(field=field):
|
|
self.setUp()
|
|
arguments = {"name": "x", **{field: value}}
|
|
result = self.service.create_project(**arguments)
|
|
|
|
self.assertEqual(result["error"]["code"], "invalid_project_input")
|
|
self.assertEqual(self.nb.create_calls, [])
|
|
self.assertEqual(self.http.add_calls, [])
|
|
|
|
def test_update_rejects_nested_project_id_before_any_client_call(self) -> None:
|
|
for project_id in ({"id": 41}, [41], {41}, (41,)):
|
|
with self.subTest(project_id=project_id):
|
|
self.setUp()
|
|
result = self.service.update_project(project_id, name="x", add_attachment_ids=[8])
|
|
|
|
self.assertEqual(result["error"]["code"], "invalid_project_input")
|
|
self.assertEqual(self.nb.update_calls, [])
|
|
self.assertEqual(self.http.add_calls, [])
|
|
|
|
def test_create_rejects_nested_attachment_id_before_any_client_call(self) -> None:
|
|
result = self.service.create_project(name="x", attachment_ids=[{"id": 8}])
|
|
|
|
self.assertEqual(result["error"]["code"], "invalid_project_input")
|
|
self.assertEqual(self.nb.create_calls, [])
|
|
self.assertEqual(self.http.add_calls, [])
|
|
|
|
def test_update_writes_scalars_then_deduplicated_attachment_changes_in_order(self) -> None:
|
|
result = self.service.update_project(
|
|
41,
|
|
inquiry_date="2026-07-12",
|
|
add_attachment_ids=[8, 8, 9],
|
|
remove_attachment_ids=[10, 10],
|
|
)
|
|
|
|
self.assertEqual(
|
|
self.nb.update_calls,
|
|
[
|
|
(
|
|
"t_projects",
|
|
41,
|
|
{"xm_project_inquiry_date": "2026-07-12"},
|
|
PROJECT_WHITELIST,
|
|
)
|
|
],
|
|
)
|
|
self.assertEqual(self.http.add_calls, [(41, [8, 9])])
|
|
self.assertEqual(self.http.remove_calls, [(41, [10])])
|
|
self.assertEqual(result, {"project_id": 41})
|
|
|
|
def test_association_errors_keep_project_id_and_map_to_stable_codes(self) -> None:
|
|
cases = [
|
|
(404, "attachment_not_found"),
|
|
(401, "nb_upload_unauthorized"),
|
|
(403, "nb_upload_unauthorized"),
|
|
(500, "nb_upload_failed"),
|
|
]
|
|
|
|
for status_code, error_code in cases:
|
|
with self.subTest(status_code=status_code):
|
|
self.setUp()
|
|
self.http.add_error = NocoBaseHttpError(status_code, "attachment request failed")
|
|
|
|
result = self.service.create_project(name="x", attachment_ids=[8])
|
|
|
|
self.assertEqual(result["project_id"], 41)
|
|
self.assertEqual(result["error"]["code"], error_code)
|
|
self.assertEqual(len(self.nb.create_calls), 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|