112 lines
4.6 KiB
Python
112 lines
4.6 KiB
Python
"""Additional isolated regressions; never connects to a real 1C or database."""
|
||
import importlib.util
|
||
import json
|
||
import os
|
||
import sys
|
||
import threading
|
||
import types
|
||
import unittest
|
||
import urllib.error
|
||
import urllib.request
|
||
from http.server import HTTPServer
|
||
from unittest.mock import Mock
|
||
|
||
sys.modules["requests"] = types.ModuleType("requests")
|
||
sys.modules["psycopg2"] = types.ModuleType("psycopg2")
|
||
os.environ.update(ONEC_MCP_TOKEN="test-only", BRIDGE_API_KEY="test-only")
|
||
spec = importlib.util.spec_from_file_location("candidate_bridge", os.environ["BRIDGE_TEST_MODULE"])
|
||
bridge = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(bridge)
|
||
bridge.call_1c = Mock(return_value=({}, None))
|
||
server = HTTPServer(("127.0.0.1", 0), bridge.BridgeHandler)
|
||
threading.Thread(target=server.serve_forever, daemon=True).start()
|
||
|
||
|
||
class ExtraSecurityTests(unittest.TestCase):
|
||
def setUp(self):
|
||
bridge.call_1c.reset_mock()
|
||
|
||
def post(self, body, path="/appointments", headers=None, method=None):
|
||
request = urllib.request.Request(
|
||
f"http://127.0.0.1:{server.server_port}" + path,
|
||
data=json.dumps(body).encode(),
|
||
headers={"X-Api-Key": "test-only"} if headers is None else headers,
|
||
method=method,
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=5) as response:
|
||
return response.status
|
||
except urllib.error.HTTPError as error:
|
||
return error.code
|
||
|
||
def test_uuid_rejects_trailing_newline(self):
|
||
self.assertFalse(bridge.is_uuid("a748a2a7-8771-11ed-840b-00155d46cd22\n"))
|
||
|
||
def test_uuid_rejects_bsl_suffix(self):
|
||
self.assertFalse(bridge.is_uuid('a748a2a7-8771-11ed-840b-00155d46cd22";'))
|
||
|
||
def test_quote_date_separator_cannot_reach_bsl(self):
|
||
self.assertEqual(self.post({
|
||
"doctor_id": "a748a2a7-8771-11ed-840b-00155d46cd22",
|
||
"start": '2026-10-20"10:00:00+03:00',
|
||
}), 201)
|
||
code = bridge.call_1c.call_args.args[0]
|
||
self.assertIn('РазобратьISO8601("2026-10-20T10:00:00+03:00")', code)
|
||
self.assertNotIn('2026-10-20"10:', code)
|
||
|
||
def test_oversize_rejected_before_1c(self):
|
||
self.assertEqual(self.post({"comment": "x" * (64 * 1024)}), 413)
|
||
bridge.call_1c.assert_not_called()
|
||
|
||
def test_controls_removed_and_quotes_doubled(self):
|
||
self.assertEqual(bridge.bsl_text('a\r\n"b\x00'), 'a""b')
|
||
|
||
def test_nested_patient_text_escaped(self):
|
||
self.assertEqual(self.post({
|
||
"doctor_id": "a748a2a7-8771-11ed-840b-00155d46cd22",
|
||
"start": "2026-10-20T10:00:00+03:00",
|
||
"patient": {"name": 'Test"name', "phone": '+7"999', "comment": 'Test"comment'},
|
||
}), 201)
|
||
code = bridge.call_1c.call_args.args[0]
|
||
self.assertIn('Наименование = "Test""name";', code)
|
||
self.assertIn('Представление = "+7999";', code)
|
||
self.assertIn('"Онлайн-запись: Test""comment";', code)
|
||
|
||
def test_non_object_json_rejected(self):
|
||
self.assertEqual(self.post([]), 400)
|
||
bridge.call_1c.assert_not_called()
|
||
|
||
def test_non_ascii_bad_key_rejected(self):
|
||
self.assertEqual(self.post({}, headers={"X-Api-Key": "é"}), 401)
|
||
bridge.call_1c.assert_not_called()
|
||
|
||
def test_bitrix_link_preserved_and_escaped(self):
|
||
self.assertEqual(self.post({"appointment_id": 'id"value', "bitrix_lead_id": 'lead"value'}, "/appointments/link-bitrix"), 200)
|
||
code = bridge.call_1c.call_args.args[0]
|
||
self.assertIn('БитриксИД = "lead""value";', code)
|
||
self.assertIn('УникальныйИдентификатор("id""value")', code)
|
||
|
||
def test_crm_link_preserved_and_escaped(self):
|
||
self.assertEqual(self.post({"appointment_id": "00001234", "crm_deal_id": 'deal"value'}, "/appointments/link-crm"), 200)
|
||
code = bridge.call_1c.call_args.args[0]
|
||
self.assertIn('[CRM: deal""value]', code)
|
||
self.assertIn('НайтиПоНомеру("00001234",', code)
|
||
|
||
def test_delete_requires_auth_before_1c(self):
|
||
self.assertEqual(self.post({}, "/appointments/test", headers={}, method="DELETE"), 401)
|
||
bridge.call_1c.assert_not_called()
|
||
|
||
def test_delete_preserved_and_escaped(self):
|
||
self.assertEqual(self.post({}, "/appointments/test%22value", method="DELETE"), 200)
|
||
code = bridge.call_1c.call_args.args[0]
|
||
self.assertIn('УникальныйИдентификатор("test""value")', code)
|
||
self.assertIn('ПометкаУдаления = Истина;', code)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
unittest.main(verbosity=2)
|
||
finally:
|
||
server.shutdown()
|
||
server.server_close()
|