70 lines
2.2 KiB
Python
70 lines
2.2 KiB
Python
"""
|
|
DAT10: 전체 유닛 및 통합 테스트 스위트 (pytest 호환)
|
|
"""
|
|
|
|
import pytest
|
|
import os
|
|
import sys
|
|
import pandas as pd
|
|
|
|
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
|
from src.preprocess.dat10_preprocessor import TacticalDataPreprocessor
|
|
from src.models.dat10_baseline_model import P2PInventoryBaselineSync
|
|
from src.models.dat10_advanced_detector import AdvancedCRDTSyncEngine, AdvancedSupplyChainDisasterDetector
|
|
from src.verification.dat10_confidence_engine import TacticalConfidenceEngine, MerkleTreeVerifier
|
|
from src.hitl.dat10_approval_workflow import HITLApprovalWorkflow
|
|
|
|
def test_preprocessor_validation():
|
|
prep = TacticalDataPreprocessor(":memory:")
|
|
raw = [{
|
|
"transaction_id": "TX-TEST-001",
|
|
"terminal_id": "TERM-A101",
|
|
"unit_id": "BN-01",
|
|
"nsn_code": "1005-01-584-1234",
|
|
"quantity": -50, # 음수 -> 보정되어야 함
|
|
"tx_type": "ISSUE"
|
|
}]
|
|
res = prep.process_batch(raw)
|
|
assert res["total_processed"] == 1
|
|
assert res["records"][0]["quantity"] == 50
|
|
|
|
def test_crdt_pn_counter():
|
|
engine_a = AdvancedCRDTSyncEngine("NODE_A")
|
|
engine_b = AdvancedCRDTSyncEngine("NODE_B")
|
|
|
|
engine_a.increment("ITEM_1", 100)
|
|
engine_a.decrement("ITEM_1", 20)
|
|
|
|
engine_b.increment("ITEM_1", 50)
|
|
engine_b.decrement("ITEM_1", 10)
|
|
|
|
engine_a.merge(engine_b.pn_counters)
|
|
assert engine_a.value("ITEM_1") == 120
|
|
|
|
def test_confidence_engine():
|
|
conf = TacticalConfidenceEngine()
|
|
tx = {
|
|
"transaction_id": "TX-01",
|
|
"is_valid": 1,
|
|
"lamport_timestamp": 5,
|
|
"grid_coords": "52SDJ12345678",
|
|
"stock_after": 500
|
|
}
|
|
eval_res = conf.evaluate_confidence(tx)
|
|
assert eval_res["confidence_score"] >= 0.85
|
|
assert eval_res["confidence_grade"] == "HIGH"
|
|
|
|
def test_merkle_tree():
|
|
hashes = ["a"*64, "b"*64, "c"*64, "d"*64]
|
|
merkle = MerkleTreeVerifier(hashes)
|
|
assert len(merkle.root) == 64
|
|
|
|
def test_hitl_workflow():
|
|
hitl = HITLApprovalWorkflow(":memory:")
|
|
res = hitl.resolve_conflict("CONF-01", "1005-01", "APPROVE_LOCAL", "Officer", "Verified", 500)
|
|
assert res["status"] == "RESOLVED"
|
|
assert len(hitl.get_audit_trail()) == 1
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main(["-v", __file__])
|