130 lines
5.6 KiB
Python
130 lines
5.6 KiB
Python
import json
|
|
import unittest
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
|
|
import httpx
|
|
|
|
from nocobase_crm_mcp.config import load_config
|
|
from nocobase_crm_mcp.nocobase_http import NocoBaseHttpClient, NocoBaseHttpError
|
|
|
|
|
|
class ProjectConfigTests(unittest.TestCase):
|
|
def test_load_config_parses_project_defaults(self) -> None:
|
|
with TemporaryDirectory() as directory:
|
|
path = Path(directory) / "config.json"
|
|
path.write_text(json.dumps({"projects": {"resource": "t_projects"}}), encoding="utf-8")
|
|
|
|
config = load_config(path)
|
|
|
|
self.assertEqual(config.project.resource, "t_projects")
|
|
self.assertEqual(config.project.name_field, "f_project_name")
|
|
self.assertEqual(config.project.customer_field, "client_ID")
|
|
self.assertEqual(config.project.inquiry_date_field, "xm_project_inquiry_date")
|
|
self.assertEqual(config.project.progress_field, "f_projects_progress")
|
|
self.assertEqual(config.project.applications_field, "f_projects_applications")
|
|
self.assertEqual(config.project.part_numbers_field, "f_projects_part_numbers")
|
|
self.assertEqual(config.project.period_usage_field, "f_projects_period_usage")
|
|
self.assertEqual(config.project.details_field, "f_projects_details")
|
|
self.assertEqual(config.project.attachment_field, "f_projects_attachments")
|
|
self.assertEqual(config.project.progress_values["SOP"], "a9t9awhtt3r")
|
|
|
|
|
|
class NocoBaseHttpClientTests(unittest.TestCase):
|
|
def test_upload_posts_file_with_attachment_field(self) -> None:
|
|
seen: dict[str, str] = {}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
seen["url"] = str(request.url)
|
|
seen["auth"] = request.headers["authorization"]
|
|
return httpx.Response(200, json={"data": {"id": 42}})
|
|
|
|
with TemporaryDirectory() as directory:
|
|
path = Path(directory) / "quote.pdf"
|
|
path.write_bytes(b"pdf")
|
|
client = NocoBaseHttpClient(
|
|
"https://nb.example/api/", "jwt", transport=httpx.MockTransport(handler)
|
|
)
|
|
|
|
self.assertEqual(client.upload_attachment(path, "quote.pdf", "application/pdf")["id"], 42)
|
|
|
|
self.assertTrue(seen["url"].endswith("/attachments:create?attachmentField=t_projects.f_projects_attachments"))
|
|
self.assertEqual(seen["auth"], "Bearer jwt")
|
|
|
|
def test_attachment_associations_post_ids_to_fixed_relation(self) -> None:
|
|
requests: list[httpx.Request] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
requests.append(request)
|
|
return httpx.Response(204)
|
|
|
|
client = NocoBaseHttpClient("https://nb.example/api", "jwt", transport=httpx.MockTransport(handler))
|
|
client.add_attachment_ids(7, [8, 9])
|
|
client.remove_attachment_ids(7, [10])
|
|
|
|
self.assertEqual(str(requests[0].url), "https://nb.example/api/t_projects/7/f_projects_attachments:add")
|
|
self.assertEqual(json.loads(requests[0].content), {"tk": [8, 9]})
|
|
self.assertEqual(str(requests[1].url), "https://nb.example/api/t_projects/7/f_projects_attachments:remove")
|
|
self.assertEqual(json.loads(requests[1].content), {"tk": [10]})
|
|
|
|
def test_attachment_association_encodes_project_id_as_one_path_segment(self) -> None:
|
|
requests: list[httpx.Request] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
requests.append(request)
|
|
return httpx.Response(204)
|
|
|
|
client = NocoBaseHttpClient("https://nb.example/api", "jwt", transport=httpx.MockTransport(handler))
|
|
client.add_attachment_ids("../project/7?#", [8])
|
|
|
|
self.assertEqual(
|
|
str(requests[0].url),
|
|
"https://nb.example/api/t_projects/..%2Fproject%2F7%3F%23/f_projects_attachments:add",
|
|
)
|
|
|
|
def test_attachment_association_rejects_non_scalar_project_id_before_request(self) -> None:
|
|
client = NocoBaseHttpClient(
|
|
"https://nb.example/api",
|
|
"jwt",
|
|
transport=httpx.MockTransport(lambda request: self.fail("request must not be sent")),
|
|
)
|
|
|
|
with self.assertRaisesRegex(ValueError, "project_id"):
|
|
client.add_attachment_ids({"id": 7}, [8])
|
|
|
|
def test_attachment_association_rejects_dot_path_project_ids_before_request(self) -> None:
|
|
client = NocoBaseHttpClient(
|
|
"https://nb.example/api",
|
|
"jwt",
|
|
transport=httpx.MockTransport(lambda request: self.fail("request must not be sent")),
|
|
)
|
|
|
|
for project_id in (".", ".."):
|
|
with self.subTest(project_id=project_id), self.assertRaisesRegex(ValueError, "project_id"):
|
|
client.add_attachment_ids(project_id, [8])
|
|
|
|
def test_rejects_blank_upload_credentials(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
NocoBaseHttpClient(" ", "jwt")
|
|
with self.assertRaises(ValueError):
|
|
NocoBaseHttpClient("https://nb.example/api", " ")
|
|
|
|
def test_http_error_redacts_jwt(self) -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(401, text="Bearer super-secret-token is invalid")
|
|
|
|
client = NocoBaseHttpClient(
|
|
"https://nb.example/api", "super-secret-token", transport=httpx.MockTransport(handler)
|
|
)
|
|
|
|
with self.assertRaises(NocoBaseHttpError) as caught:
|
|
client.add_attachment_ids(7, [8])
|
|
|
|
self.assertEqual(caught.exception.status_code, 401)
|
|
self.assertNotIn("super-secret-token", str(caught.exception))
|
|
self.assertIn("<redacted>", str(caught.exception))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|