99 lines
3.6 KiB
Python
99 lines
3.6 KiB
Python
|
import sys
|
||
|
|
import tempfile
|
||
|
|
import unittest
|
||
|
|
from pathlib import Path
|
||
|
|
from types import ModuleType, SimpleNamespace
|
||
|
|
from unittest.mock import patch
|
||
|
|
|
||
|
|
|
||
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "api"))
|
||
|
|
|
||
|
|
from insurance.admin import ppt_admin_service as service_module
|
||
|
|
from insurance.admin.ppt_admin_service import PptAdminService, _storage_key
|
||
|
|
from insurance.admin.ppt_admin_routes import _is_within
|
||
|
|
|
||
|
|
|
||
|
|
class _Query:
|
||
|
|
def __init__(self, product):
|
||
|
|
self.product = product
|
||
|
|
|
||
|
|
def get(self, _product_id):
|
||
|
|
return self.product
|
||
|
|
|
||
|
|
|
||
|
|
class _Product:
|
||
|
|
def __init__(self, filepath, status="pending"):
|
||
|
|
self.manual_file_url = filepath
|
||
|
|
self.manual_parse_status = status
|
||
|
|
self.manual_parse_message = ""
|
||
|
|
self.manual_parse_task_id = None
|
||
|
|
self.manual_parse_error = None
|
||
|
|
self.manual_parse_started_at = None
|
||
|
|
self.manual_parse_finished_at = None
|
||
|
|
self.manual_parsed_rules = None
|
||
|
|
self.manual_reviewed_by = None
|
||
|
|
self.manual_reviewed_at = None
|
||
|
|
|
||
|
|
def to_dict(self):
|
||
|
|
return {
|
||
|
|
"manualParseStatus": self.manual_parse_status,
|
||
|
|
"manualParseTaskId": self.manual_parse_task_id,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
class PptAdminManualTest(unittest.TestCase):
|
||
|
|
def test_storage_key_never_contains_path_segments(self):
|
||
|
|
key = _storage_key("../../another-company")
|
||
|
|
self.assertEqual(len(key), 32)
|
||
|
|
self.assertNotIn("/", key)
|
||
|
|
self.assertNotIn("..", key)
|
||
|
|
|
||
|
|
def test_file_guard_rejects_sibling_directory(self):
|
||
|
|
root = str(Path(tempfile.gettempdir()) / "company-logos")
|
||
|
|
sibling = str(Path(tempfile.gettempdir()) / "company-logos-backup" / "logo.png")
|
||
|
|
self.assertFalse(_is_within(root, sibling))
|
||
|
|
|
||
|
|
def test_parse_commits_queued_before_dispatch(self):
|
||
|
|
with tempfile.NamedTemporaryFile(suffix=".pdf") as pdf:
|
||
|
|
product = _Product(pdf.name)
|
||
|
|
fake_model = SimpleNamespace(query=_Query(product))
|
||
|
|
commits = []
|
||
|
|
|
||
|
|
def commit():
|
||
|
|
commits.append(product.manual_parse_status)
|
||
|
|
|
||
|
|
def dispatch(*, args, queue):
|
||
|
|
self.assertEqual(product.manual_parse_status, "queued")
|
||
|
|
self.assertEqual(args, ["product-1"])
|
||
|
|
self.assertEqual(queue, "insurance")
|
||
|
|
return SimpleNamespace(id="task-1")
|
||
|
|
|
||
|
|
fake_tasks = ModuleType("insurance.generation.celery_tasks")
|
||
|
|
fake_tasks.parse_product_manual_task = SimpleNamespace(apply_async=dispatch)
|
||
|
|
with patch.object(service_module, "PptProduct", fake_model), \
|
||
|
|
patch.object(service_module.db, "session", SimpleNamespace(commit=commit)), \
|
||
|
|
patch.dict(sys.modules, {"insurance.generation.celery_tasks": fake_tasks}):
|
||
|
|
result = PptAdminService().parse_manual("product-1")
|
||
|
|
|
||
|
|
self.assertEqual(result["code"], 0)
|
||
|
|
self.assertEqual(commits[0], "queued")
|
||
|
|
self.assertEqual(product.manual_parse_task_id, "task-1")
|
||
|
|
|
||
|
|
def test_save_rules_does_not_confirm_review(self):
|
||
|
|
product = _Product("manual.pdf", status="parsed")
|
||
|
|
fake_model = SimpleNamespace(query=_Query(product))
|
||
|
|
|
||
|
|
with patch.object(service_module, "PptProduct", fake_model), \
|
||
|
|
patch.object(service_module.db, "session", SimpleNamespace(commit=lambda: None)):
|
||
|
|
result = PptAdminService().review_manual(
|
||
|
|
"product-1", {"rules": {"features": []}, "confirm": False}
|
||
|
|
)
|
||
|
|
|
||
|
|
self.assertEqual(result["code"], 0)
|
||
|
|
self.assertEqual(product.manual_parse_status, "parsed")
|
||
|
|
self.assertIsNone(product.manual_reviewed_at)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|