diff --git a/backend/application/settings.py b/backend/application/settings.py index 454c759..e9e961a 100644 --- a/backend/application/settings.py +++ b/backend/application/settings.py @@ -64,6 +64,7 @@ INSTALLED_APPS = [ "apps.pisadmin.basicinfo", "apps.pisadmin.dashboard", "apps.pissupplier", + "sync", ] MIDDLEWARE = [ @@ -428,7 +429,7 @@ SHARED_APPS = [] # from dvadmin_upgrade_center.settings import * # 升级中心 from dvadmin3_celery.settings import * # celery 异步任务 # from dvadmin_third.settings import * # 第三方用户管理 -# from dvadmin_ak_sk.settings import * # 秘钥管理管理 +from dvadmin_ak_sk.settings import * # 秘钥管理管理 # from dvadmin_tenants.settings import * # 租户管理 #from dvadmin_social_auth.settings import * #from dvadmin_uniapp.settings import * diff --git a/backend/application/urls.py b/backend/application/urls.py index 869f837..f8696dc 100644 --- a/backend/application/urls.py +++ b/backend/application/urls.py @@ -102,6 +102,7 @@ urlpatterns = ( schema_view.with_ui("redoc", cache_timeout=0), name="schema-redoc", ), + path("api/pisadmin/sync/", include("sync.urls")), path("api/pisadmin/miscprocurement/", include("apps.pisadmin.miscprocurement.urls")), path("api/pisadmin/basicinfo/", include("apps.pisadmin.basicinfo.urls")), path("api/pisadmin/dashboard/", include("apps.pisadmin.dashboard.urls")), diff --git a/backend/conf/env.py b/backend/conf/env.py index f22d39d..31a9f55 100644 --- a/backend/conf/env.py +++ b/backend/conf/env.py @@ -78,3 +78,14 @@ RUSTFS_BUCKET = "pis-media" RUSTFS_SECURE = False # True=HTTPS, False=HTTP RUSTFS_PATH_PREFIX = "" # 存储路径前缀，可为空 RUSTFS_REGION = "us-east-1" # RustFS 区域（可自定义） + +# ================================================= # +# *************** EIP 对接（出站 / 配置） *************** # +# ================================================= # +# PIS 主动调用 EIP 时使用（sync 插件 clients/eip_client） +EIP_BASE_URL = os.getenv("EIP_BASE_URL", "").rstrip("/") +EIP_API_KEY = os.getenv("EIP_API_KEY", "") +EIP_TIMEOUT = int(os.getenv("EIP_TIMEOUT", "30")) +# 同步重试（预留） +SYNC_RETRY_MAX = int(os.getenv("SYNC_RETRY_MAX", "3")) +SYNC_RETRY_DELAY = int(os.getenv("SYNC_RETRY_DELAY", "60")) \ No newline at end of file diff --git a/backend/plugins/eip_sync/sync/__init__.py b/backend/plugins/eip_sync/sync/__init__.py new file mode 100644 index 0000000..5cdd1d5 --- /dev/null +++ b/backend/plugins/eip_sync/sync/__init__.py @@ -0,0 +1 @@ +# EIP / external sync plugin package diff --git a/backend/plugins/eip_sync/sync/adapters/__init__.py b/backend/plugins/eip_sync/sync/adapters/__init__.py new file mode 100644 index 0000000..6ee423f --- /dev/null +++ b/backend/plugins/eip_sync/sync/adapters/__init__.py @@ -0,0 +1 @@ +# Adapters register themselves with SyncFactory on import. diff --git a/backend/plugins/eip_sync/sync/adapters/misc_material.py b/backend/plugins/eip_sync/sync/adapters/misc_material.py new file mode 100644 index 0000000..0210fa6 --- /dev/null +++ b/backend/plugins/eip_sync/sync/adapters/misc_material.py @@ -0,0 +1,133 @@ +from __future__ import annotations + +import json +import re +from typing import Any, Dict, Optional + +from apps.pisadmin.miscprocurement.models import MiscProcMaterial + +from sync.base import BaseSyncAdapter, SyncDirection +from sync.factory import SyncFactory + + +def _first_str(data: Dict[str, Any], *keys: str) -> str: + for k in keys: + v = data.get(k) + if v is None: + continue + s = str(v).strip() + if s: + return s + return "" + + +def _parse_category_id(raw: Any) -> Optional[int]: + if raw is None or raw == "": + return None + if isinstance(raw, int): + return raw + s = str(raw).strip() + if re.fullmatch(r"-?\d+", s): + return int(s) + return None + + +def _build_extra_description(data: Dict[str, Any], category_used_as_id: bool) -> str: + extra: Dict[str, Any] = {} + ac = data.get("accountCode") or data.get("account_code") + if ac is not None and str(ac).strip() != "": + extra["accountCode"] = str(ac).strip() + cat = data.get("Category") or data.get("category") + if cat is not None and str(cat).strip() != "" and not category_used_as_id: + extra["category"] = str(cat).strip() + for label, keys in ( + ("materialNameEn", ("materialNameEn", "material_name_en")), + ("materialNameVi", ("materialNameVi", "material_name_vi")), + ): + v = _first_str(data, *keys) + if v: + extra[label] = v + if not extra: + return "" + try: + blob = json.dumps(extra, ensure_ascii=False) + except (TypeError, ValueError): + return "" + if len(blob) <= 255: + return blob + return blob[:252] + "..." + + +class MiscMaterialSyncAdapter(BaseSyncAdapter): + adapter_name = "misc_material" + direction = SyncDirection.EIP_TO_PIS + + def validate(self, data: Dict[str, Any]) -> bool: + company = _first_str(data, "companyCode", "company_code") + code = _first_str(data, "materialCode", "material_code") + return bool(company and code) + + def transform_to_local(self, data: Dict[str, Any]) -> Dict[str, Any]: + company_code = _first_str(data, "companyCode", "company_code") + partid = _first_str(data, "materialCode", "material_code") + partid_name = _first_str( + data, + "materialName", + "materialNameZh", + "material_name", + "material_name_zh", + ) + specification = _first_str(data, "specification") + unit = _first_str(data, "unit") + + cat_raw = data.get("Category") if "Category" in data else data.get("category") + partid_category_id = _parse_category_id(cat_raw) + category_used_as_id = partid_category_id is not None + + description = _build_extra_description(data, category_used_as_id) + if not description: + desc_existing = _first_str(data, "description") + if desc_existing: + description = desc_existing[:255] + + external_id = f"{company_code}_{partid}" + return { + "company_code": company_code, + "partid": partid, + "partid_name": partid_name, + "specification": specification, + "unit": unit, + "partid_category_id": partid_category_id, + "description": description, + "external_id": external_id, + "status": 1, + } + + def save_to_local(self, transformed_data: Dict[str, Any]) -> bool: + external_id = transformed_data.pop("external_id", "") + _ = external_id + company_code = transformed_data.get("company_code") + partid = transformed_data.get("partid") + defaults = { + "partid_name": transformed_data.get("partid_name") or "", + "specification": transformed_data.get("specification") or "", + "unit": transformed_data.get("unit") or "", + "partid_category_id": transformed_data.get("partid_category_id"), + "status": transformed_data.get("status", 1), + } + desc = transformed_data.get("description") + if desc: + defaults["description"] = desc[:255] + + MiscProcMaterial.objects.update_or_create( + company_code=company_code, + partid=partid, + defaults=defaults, + ) + return True + + def transform_to_remote(self, data: Dict[str, Any]) -> Dict[str, Any]: + return {} + + +SyncFactory.register("misc_material", MiscMaterialSyncAdapter) diff --git a/backend/plugins/eip_sync/sync/apps.py b/backend/plugins/eip_sync/sync/apps.py new file mode 100644 index 0000000..302ad81 --- /dev/null +++ b/backend/plugins/eip_sync/sync/apps.py @@ -0,0 +1,11 @@ +from django.apps import AppConfig + + +class SyncConfig(AppConfig): + default_auto_field = "django.db.models.BigAutoField" + name = "sync" + verbose_name = "EIP 数据同步" + + def ready(self): + # Register adapters with SyncFactory + from sync.adapters import misc_material # noqa: F401 diff --git a/backend/plugins/eip_sync/sync/base.py b/backend/plugins/eip_sync/sync/base.py new file mode 100644 index 0000000..28ba837 --- /dev/null +++ b/backend/plugins/eip_sync/sync/base.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +from abc import ABC, abstractmethod +from dataclasses import dataclass +from datetime import datetime +from enum import Enum +from typing import Any, Dict, Optional + + +class SyncDirection(str, Enum): + EIP_TO_PIS = "eip_to_pis" + PIS_TO_EIP = "pis_to_eip" + + +class SyncStatus(str, Enum): + PENDING = "pending" + SUCCESS = "success" + FAILED = "failed" + RETRYING = "retrying" + + +@dataclass +class SyncOperationResult: + """In-memory result of a single sync operation (not the ORM SyncRecord).""" + + adapter_name: str = "" + direction: SyncDirection = SyncDirection.EIP_TO_PIS + external_id: str = "" + payload: Optional[Dict[str, Any]] = None + response: Optional[Dict[str, Any]] = None + status: SyncStatus = SyncStatus.PENDING + error_message: str = "" + retry_count: int = 0 + created_at: Optional[datetime] = None + + +class BaseSyncAdapter(ABC): + adapter_name: str = "" + direction: SyncDirection = SyncDirection.EIP_TO_PIS + + def __init__(self, eip_client=None, logger=None): + self.eip_client = eip_client + self.logger = logger + + @abstractmethod + def validate(self, data: Dict[str, Any]) -> bool: + pass + + @abstractmethod + def transform_to_local(self, data: Dict[str, Any]) -> Dict[str, Any]: + pass + + def transform_to_remote(self, data: Dict[str, Any]) -> Dict[str, Any]: + pass + + @abstractmethod + def save_to_local(self, transformed_data: Dict[str, Any]) -> bool: + pass + + def push_to_local(self, data: Dict[str, Any]) -> SyncOperationResult: + created_at = datetime.now() + record = SyncOperationResult( + adapter_name=self.adapter_name, + direction=self.direction, + payload=data, + created_at=created_at, + ) + + if not self.validate(data): + record.status = SyncStatus.FAILED + record.error_message = "数据校验失败" + return record + + try: + transformed = self.transform_to_local(data) + except Exception as e: + record.status = SyncStatus.FAILED + record.error_message = f"数据转换失败: {e}" + return record + + try: + success = self.save_to_local(transformed) + record.status = SyncStatus.SUCCESS if success else SyncStatus.FAILED + record.external_id = transformed.get("external_id", "") + if not success: + record.error_message = record.error_message or "保存失败" + except Exception as e: + record.status = SyncStatus.FAILED + record.error_message = f"保存失败: {e}" + + return record diff --git a/backend/plugins/eip_sync/sync/clients/__init__.py b/backend/plugins/eip_sync/sync/clients/__init__.py new file mode 100644 index 0000000..ed4de40 --- /dev/null +++ b/backend/plugins/eip_sync/sync/clients/__init__.py @@ -0,0 +1,3 @@ +from sync.clients.eip_client import EipClient + +__all__ = ["EipClient"] diff --git a/backend/plugins/eip_sync/sync/clients/eip_client.py b/backend/plugins/eip_sync/sync/clients/eip_client.py new file mode 100644 index 0000000..ff6ebcf --- /dev/null +++ b/backend/plugins/eip_sync/sync/clients/eip_client.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +from typing import Any, Dict, Optional + +import requests +from django.conf import settings + + +class EipClient: + """HTTP client for PIS -> EIP outbound calls (placeholder for future use).""" + + def __init__( + self, + base_url: Optional[str] = None, + api_key: Optional[str] = None, + timeout: int = 30, + ): + self.base_url = (base_url or getattr(settings, "EIP_BASE_URL", "") or "").rstrip("/") + self.api_key = api_key or getattr(settings, "EIP_API_KEY", "") or "" + self.timeout = timeout + self.session = requests.Session() + if self.api_key: + self.session.headers.update( + { + "Content-Type": "application/json", + "Authorization": f"Bearer {self.api_key}", + } + ) + + def post(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]: + url = f"{self.base_url}{endpoint}" + try: + response = self.session.post(url, json=data, timeout=self.timeout) + response.raise_for_status() + return response.json() + except requests.RequestException as e: + return {"Status": "fail", "Message": str(e)} + + def get(self, endpoint: str, params: Dict[str, Any] | None = None) -> Dict[str, Any]: + url = f"{self.base_url}{endpoint}" + try: + response = self.session.get(url, params=params or {}, timeout=self.timeout) + response.raise_for_status() + return response.json() + except requests.RequestException as e: + return {"Status": "fail", "Message": str(e)} diff --git a/backend/plugins/eip_sync/sync/factory.py b/backend/plugins/eip_sync/sync/factory.py new file mode 100644 index 0000000..47f2e67 --- /dev/null +++ b/backend/plugins/eip_sync/sync/factory.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +from typing import Any, Dict, Type + +from sync.base import BaseSyncAdapter + + +class SyncFactory: + """Creates sync adapters by registered name.""" + + _adapters: Dict[str, Type[BaseSyncAdapter]] = {} + + @classmethod + def register(cls, name: str, adapter_class: Type[BaseSyncAdapter]) -> None: + cls._adapters[name] = adapter_class + + @classmethod + def create(cls, name: str, **kwargs: Any) -> BaseSyncAdapter: + if name not in cls._adapters: + raise ValueError(f"未注册的同步适配器: {name}") + return cls._adapters[name](**kwargs) + + @classmethod + def list_adapters(cls) -> list: + return list(cls._adapters.keys()) diff --git a/backend/plugins/eip_sync/sync/logger.py b/backend/plugins/eip_sync/sync/logger.py new file mode 100644 index 0000000..d52250e --- /dev/null +++ b/backend/plugins/eip_sync/sync/logger.py @@ -0,0 +1,23 @@ +import logging +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from sync.base import SyncOperationResult + + +class SyncLogger: + def __init__(self, name: str = "pis.sync"): + self.logger = logging.getLogger(name) + + def log_sync(self, record: "SyncOperationResult", level: str = "INFO") -> None: + message = ( + f"[{record.adapter_name}] direction={record.direction.value} " + f"status={record.status.value} external_id={record.external_id} " + f"error={record.error_message or 'none'}" + ) + if level == "ERROR": + self.logger.error(message) + elif level == "WARNING": + self.logger.warning(message) + else: + self.logger.info(message) diff --git a/backend/plugins/eip_sync/sync/manager.py b/backend/plugins/eip_sync/sync/manager.py new file mode 100644 index 0000000..07bb3df --- /dev/null +++ b/backend/plugins/eip_sync/sync/manager.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +from django.utils import timezone + +from sync.base import SyncOperationResult, SyncStatus +from sync.factory import SyncFactory +from sync.logger import SyncLogger +from sync.models import SyncRecord + + +class SyncManager: + """Routes inbound EIP webhooks to registered adapters.""" + + SUCCESS_MESSAGE_MISC = "物料信息已成功抛转至PIS" + + def __init__(self): + self.logger = SyncLogger() + + def process_eip_webhook(self, adapter_name: str, data: Dict[str, Any]) -> Dict[str, Any]: + if adapter_name not in SyncFactory.list_adapters(): + return {"Status": "fail", "Message": f"未知的适配器: {adapter_name}"} + + adapter = SyncFactory.create(adapter_name) + op = adapter.push_to_local(data) + self._persist_record(op) + self.logger.log_sync(op) + + if op.status == SyncStatus.SUCCESS: + message = ( + self.SUCCESS_MESSAGE_MISC if adapter_name == "misc_material" else "同步成功" + ) + return {"Status": "success", "Message": message} + + return {"Status": "fail", "Message": op.error_message or "同步失败"} + + def _persist_record(self, op: SyncOperationResult) -> None: + completed_at = timezone.now() + SyncRecord.objects.create( + adapter_name=op.adapter_name, + direction=op.direction.value, + external_id=op.external_id or "", + payload=op.payload, + response=op.response, + status=op.status.value, + error_message=op.error_message or "", + retry_count=op.retry_count, + completed_at=completed_at, + ) + + def get_sync_history( + self, + adapter_name: Optional[str] = None, + direction: Optional[str] = None, + status: Optional[str] = None, + limit: int = 100, + ) -> List[SyncRecord]: + qs = SyncRecord.objects.all() + if adapter_name: + qs = qs.filter(adapter_name=adapter_name) + if direction: + qs = qs.filter(direction=direction) + if status: + qs = qs.filter(status=status) + return list(qs[:limit]) diff --git a/backend/plugins/eip_sync/sync/models.py b/backend/plugins/eip_sync/sync/models.py new file mode 100644 index 0000000..bf7bb39 --- /dev/null +++ b/backend/plugins/eip_sync/sync/models.py @@ -0,0 +1,50 @@ +from django.db import models + +from dvadmin.utils.models import CoreModel, table_prefix + + +class SyncRecord(CoreModel): + """Inbound/outbound sync audit row.""" + + SYNC_DIRECTION_CHOICES = ( + ("eip_to_pis", "EIP to PIS"), + ("pis_to_eip", "PIS to EIP"), + ) + + SYNC_STATUS_CHOICES = ( + ("pending", "待处理"), + ("success", "成功"), + ("failed", "失败"), + ("retrying", "重试中"), + ) + + adapter_name = models.CharField(max_length=50, verbose_name="适配器名称", db_index=True) + direction = models.CharField(max_length=20, choices=SYNC_DIRECTION_CHOICES, verbose_name="同步方向") + external_id = models.CharField(max_length=100, blank=True, verbose_name="外部单号", db_index=True) + payload = models.JSONField(null=True, blank=True, verbose_name="请求数据") + response = models.JSONField(null=True, blank=True, verbose_name="响应数据") + status = models.CharField( + max_length=20, + choices=SYNC_STATUS_CHOICES, + default="pending", + verbose_name="状态", + db_index=True, + ) + error_message = models.TextField(blank=True, verbose_name="错误信息") + retry_count = models.IntegerField(default=0, verbose_name="重试次数") + retry_max = models.IntegerField(default=3, verbose_name="最大重试次数") + completed_at = models.DateTimeField(null=True, blank=True, verbose_name="完成时间") + + class Meta: + db_table = table_prefix + "sync_record" + verbose_name = "数据同步记录" + verbose_name_plural = verbose_name + ordering = ("-create_datetime",) + indexes = [ + models.Index(fields=["adapter_name", "status"]), + models.Index(fields=["external_id"]), + models.Index(fields=["create_datetime"]), + ] + + def __str__(self) -> str: + return f"{self.adapter_name} - {self.direction} - {self.status}" diff --git a/backend/plugins/eip_sync/sync/urls.py b/backend/plugins/eip_sync/sync/urls.py new file mode 100644 index 0000000..f40e14d --- /dev/null +++ b/backend/plugins/eip_sync/sync/urls.py @@ -0,0 +1,7 @@ +from django.urls import path + +from sync.views import MiscMaterialSyncView + +urlpatterns = [ + path("misc-material/", MiscMaterialSyncView.as_view(), name="sync-misc-material"), +] diff --git a/backend/plugins/eip_sync/sync/views.py b/backend/plugins/eip_sync/sync/views.py new file mode 100644 index 0000000..4e26f56 --- /dev/null +++ b/backend/plugins/eip_sync/sync/views.py @@ -0,0 +1,23 @@ +from rest_framework.permissions import IsAuthenticated +from rest_framework.response import Response +from rest_framework.views import APIView + +from dvadmin_ak_sk.libs.authentication import AkSkAuthentication + +from sync.manager import SyncManager + + +class MiscMaterialSyncView(APIView): + """ + EIP -> PIS: miscellaneous procurement material master (杂采料号). + Authenticated via dvadmin-ak-sk (X-NSF-* signature headers). + """ + + authentication_classes = [AkSkAuthentication] + permission_classes = [IsAuthenticated] + + def post(self, request, *args, **kwargs): + manager = SyncManager() + result = manager.process_eip_webhook("misc_material", request.data) + status_code = 200 if result.get("Status") == "success" else 400 + return Response(result, status=status_code) diff --git a/backend/plugins/test/__init__.py b/backend/plugins/test/__init__.py new file mode 100644 index 0000000..df93c36 --- /dev/null +++ b/backend/plugins/test/__init__.py @@ -0,0 +1 @@ +# Plugin tests (EIP sync, etc.) diff --git a/backend/plugins/test/test_sync_misc_material.py b/backend/plugins/test/test_sync_misc_material.py new file mode 100644 index 0000000..cd42275 --- /dev/null +++ b/backend/plugins/test/test_sync_misc_material.py @@ -0,0 +1,184 @@ +""" +杂采料号 EIP 同步：适配器、SyncManager、HTTP 接口。 + +使用 pytest + django_db，客户端与 fixtures 与项目 conftest 一致（api_client / authenticate）。 +""" +import pytest +from rest_framework import status + +from apps.pisadmin.miscprocurement.models import MiscProcMaterial +from sync.adapters.misc_material import MiscMaterialSyncAdapter +from sync.base import SyncStatus +from sync.factory import SyncFactory +from sync.manager import SyncManager +from sync.models import SyncRecord + + +@pytest.mark.django_db +class TestMiscMaterialSyncAdapter: + """MiscMaterialSyncAdapter：校验、转换、落库。""" + + def test_validate_requires_company_and_material(self): + adapter = MiscMaterialSyncAdapter() + assert adapter.validate({}) is False + assert adapter.validate({"companyCode": "A01"}) is False + assert adapter.validate({"materialCode": "P1"}) is False + assert adapter.validate({"companyCode": "A01", "materialCode": "P1"}) is True + assert adapter.validate({"company_code": "A01", "material_code": "P1"}) is True + + def test_transform_maps_eip_fields(self): + adapter = MiscMaterialSyncAdapter() + data = { + "companyCode": "VC01", + "materialCode": "M-001", + "materialName": "说明A", + "specification": "SPEC-1", + "unit": "PCS", + "Category": "2", + "accountCode": "6001", + } + t = adapter.transform_to_local(data) + assert t["company_code"] == "VC01" + assert t["partid"] == "M-001" + assert t["partid_name"] == "说明A" + assert t["specification"] == "SPEC-1" + assert t["unit"] == "PCS" + assert t["partid_category_id"] == 2 + assert t["external_id"] == "VC01_M-001" + + def test_transform_non_numeric_category_goes_to_description(self): + adapter = MiscMaterialSyncAdapter() + data = { + "companyCode": "VC01", + "materialCode": "M-002", + "Category": "模治具", + "accountCode": "ACC1", + } + t = adapter.transform_to_local(data) + assert t["partid_category_id"] is None + assert "模治具" in (t.get("description") or "") + assert "ACC1" in (t.get("description") or "") + + def test_push_to_local_creates_misc_proc_material(self): + adapter = MiscMaterialSyncAdapter() + payload = { + "companyCode": "VC01", + "materialCode": "NEW-PART", + "materialNameZh": "新料号", + "specification": "S", + "unit": "EA", + } + op = adapter.push_to_local(payload) + assert op.status == SyncStatus.SUCCESS + obj = MiscProcMaterial.objects.get(company_code="VC01", partid="NEW-PART") + assert obj.partid_name == "新料号" + assert obj.specification == "S" + assert obj.unit == "EA" + + def test_push_to_local_update_existing(self): + adapter = MiscMaterialSyncAdapter() + MiscProcMaterial.objects.create( + company_code="VC01", + partid="UP-1", + partid_name="旧名", + specification="old", + unit="U", + ) + op = adapter.push_to_local( + { + "companyCode": "VC01", + "materialCode": "UP-1", + "materialName": "新名", + "specification": "new", + "unit": "U", + } + ) + assert op.status == SyncStatus.SUCCESS + obj = MiscProcMaterial.objects.get(company_code="VC01", partid="UP-1") + assert obj.partid_name == "新名" + assert obj.specification == "new" + + +@pytest.mark.django_db +class TestSyncManager: + """SyncManager.process_eip_webhook""" + + def test_unknown_adapter(self): + mgr = SyncManager() + out = mgr.process_eip_webhook("no_such_adapter", {}) + assert out["Status"] == "fail" + assert "未知" in out["Message"] + + def test_success_writes_sync_record(self): + mgr = SyncManager() + payload = { + "companyCode": "VC01", + "materialCode": "SR-1", + "materialName": "记录测试", + "specification": "", + "unit": "", + } + out = mgr.process_eip_webhook("misc_material", payload) + assert out["Status"] == "success" + assert out["Message"] == mgr.SUCCESS_MESSAGE_MISC + assert SyncRecord.objects.filter(adapter_name="misc_material", status="success").exists() + + def test_validation_failure_writes_failed_record(self): + mgr = SyncManager() + out = mgr.process_eip_webhook("misc_material", {"companyCode": "X"}) + assert out["Status"] == "fail" + assert SyncRecord.objects.filter(adapter_name="misc_material", status="failed").exists() + + +@pytest.mark.django_db +class TestMiscMaterialSyncAPI: + """POST /api/pisadmin/sync/misc-material/（DRF force_authenticate 绕过 AK/SK 验签，仅测业务链）。""" + + URL = "/api/pisadmin/sync/misc-material/" + + def test_post_success(self, api_client, admin_user): + api_client.force_authenticate(user=admin_user) + payload = { + "companyCode": "VC01", + "materialCode": "API-001", + "materialName": "接口测", + "specification": "SP", + "unit": "KG", + } + response = api_client.post(self.URL, data=payload, format="json") + assert response.status_code == status.HTTP_200_OK + assert response.data["Status"] == "success" + assert "物料信息已成功抛转至PIS" in response.data["Message"] + assert MiscProcMaterial.objects.filter(partid="API-001").exists() + + def test_post_validation_error(self, api_client, admin_user): + api_client.force_authenticate(user=admin_user) + response = api_client.post( + self.URL, data={"companyCode": ""}, format="json" + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.data["Status"] == "fail" + + def test_post_unauthenticated(self, api_client): + response = api_client.post( + self.URL, + data={ + "companyCode": "VC01", + "materialCode": "X", + "materialName": "n", + "specification": "", + "unit": "", + }, + format="json", + ) + assert response.status_code in ( + status.HTTP_401_UNAUTHORIZED, + status.HTTP_403_FORBIDDEN, + ) + + +@pytest.mark.django_db +def test_sync_factory_registers_misc_material(): + assert "misc_material" in SyncFactory.list_adapters() + adapter = SyncFactory.create("misc_material") + assert isinstance(adapter, MiscMaterialSyncAdapter) diff --git a/backend/pytest.ini b/backend/pytest.ini index 84b63b2..1c5bece 100644 --- a/backend/pytest.ini +++ b/backend/pytest.ini @@ -4,4 +4,4 @@ python_files = test_*.py python_classes = Test* python_functions = test_* addopts = -v --tb=short -testpaths = apps tests +testpaths = apps tests plugins diff --git a/backend/requirements.txt b/backend/requirements.txt index cd5b9a0..e6e318d 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -32,6 +32,7 @@ dvadmin3-celery==3.1.6 boto3>=1.34.0 pytest-django>=4.5.0 factory_boy>=3.3.0 +dvadmin-ak-sk==1.0.0 # Git hooks pre-commit>=3.0.0