Compare commits

..

2 Commits

Author SHA1 Message Date
xiaoxia 7c541910b3 fix(test): 修正StrEnum str()断言,CI环境StrEnum行为不一致
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Successful in 1m5s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 1m30s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Successful in 1m9s
2026-07-12 09:14:55 +08:00
用户CI Test e5d627fc3e fix(asset): ClassificationStatus枚举兼容历史done值,避免500错误
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 2m19s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 3m14s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Successful in 2m36s
生产环境发现26个classification_status='done'的历史脏数据导致枚举转换失败500。
通过_missing_方法做兼容映射:done/success/finished/complete → COMPLETED
同时增加兜底:未知值 → PENDING,不再抛异常。
2026-07-11 23:27:33 +08:00
3 changed files with 85 additions and 164 deletions
-164
View File
@@ -1,164 +0,0 @@
name: Runner Expand
on: push
jobs:
expand:
runs-on: host
steps:
- name: Expand to 3 runner instances
run: |
set +e
echo "========== Step 1: 检查当前runner =========="
echo "当前进程数: $(ps aux | grep 'act_runner daemon' | grep -v grep | wc -l)"
echo ""
echo "========== Step 2: 创建 runner-2 =========="
if [ -f /var/lib/xiaoxia-ci-runner-2/.runner ]; then
echo "runner-2 已注册,跳过"
else
mkdir -p /var/lib/xiaoxia-ci-runner-2
cp /var/lib/xiaoxia-ci/act_runner /var/lib/xiaoxia-ci-runner-2/act_runner
chmod +x /var/lib/xiaoxia-ci-runner-2/act_runner
cat > /var/lib/xiaoxia-ci-runner-2/.runner-config.yaml << 'CFG'
log:
level: info
runner:
file: .runner
capacity: 1
timeout: 3h
insecure: false
fetch_timeout: 5s
fetch_interval: 2s
labels:
- runtime-builder
- ubuntu-latest
- host
cache:
enabled: true
dir: /var/lib/xiaoxia-ci-runner-2/cache
host:
workdir_parent: /var/lib/xiaoxia-ci-runner-2/workspace
container:
network: host
CFG
REG_TOKEN=$(python3 -c "import json; print(json.load(open('/var/lib/xiaoxia-ci/.runner'))['token'])" 2>/dev/null)
REG_ADDR=$(python3 -c "import json; print(json.load(open('/var/lib/xiaoxia-ci/.runner'))['address'])" 2>/dev/null)
echo "使用注册地址: $REG_ADDR"
cd /var/lib/xiaoxia-ci-runner-2
./act_runner register \
--instance "$REG_ADDR" \
--token "$REG_TOKEN" \
--name "xiaoxia-ci-runner-2" \
--labels "runtime-builder:host,ubuntu-latest:host,host:host" \
--no-interactive 2>&1
echo "runner-2 注册结果: $?"
fi
echo ""
echo "========== Step 3: 创建 runner-3 =========="
if [ -f /var/lib/xiaoxia-ci-runner-3/.runner ]; then
echo "runner-3 已注册,跳过"
else
mkdir -p /var/lib/xiaoxia-ci-runner-3
cp /var/lib/xiaoxia-ci/act_runner /var/lib/xiaoxia-ci-runner-3/act_runner
chmod +x /var/lib/xiaoxia-ci-runner-3/act_runner
cat > /var/lib/xiaoxia-ci-runner-3/.runner-config.yaml << 'CFG'
log:
level: info
runner:
file: .runner
capacity: 1
timeout: 3h
insecure: false
fetch_timeout: 5s
fetch_interval: 2s
labels:
- runtime-builder
- ubuntu-latest
- host
cache:
enabled: true
dir: /var/lib/xiaoxia-ci-runner-3/cache
host:
workdir_parent: /var/lib/xiaoxia-ci-runner-3/workspace
container:
network: host
CFG
REG_TOKEN=$(python3 -c "import json; print(json.load(open('/var/lib/xiaoxia-ci/.runner'))['token'])" 2>/dev/null)
REG_ADDR=$(python3 -c "import json; print(json.load(open('/var/lib/xiaoxia-ci/.runner'))['address'])" 2>/dev/null)
cd /var/lib/xiaoxia-ci-runner-3
./act_runner register \
--instance "$REG_ADDR" \
--token "$REG_TOKEN" \
--name "xiaoxia-ci-runner-3" \
--labels "runtime-builder:host,ubuntu-latest:host,host:host" \
--no-interactive 2>&1
echo "runner-3 注册结果: $?"
fi
echo ""
echo "========== Step 4: systemd 服务 =========="
cat > /etc/systemd/system/act_runner-2.service << 'SVC'
[Unit]
Description=Gitea Actions Runner 2
After=gitea.service network.target
[Service]
Type=simple
WorkingDirectory=/var/lib/xiaoxia-ci-runner-2
ExecStart=/var/lib/xiaoxia-ci-runner-2/act_runner daemon --config .runner-config.yaml
Restart=always
RestartSec=5s
User=root
[Install]
WantedBy=multi-user.target
SVC
cat > /etc/systemd/system/act_runner-3.service << 'SVC'
[Unit]
Description=Gitea Actions Runner 3
After=gitea.service network.target
[Service]
Type=simple
WorkingDirectory=/var/lib/xiaoxia-ci-runner-3
ExecStart=/var/lib/xiaoxia-ci-runner-3/act_runner daemon --config .runner-config.yaml
Restart=always
RestartSec=5s
User=root
[Install]
WantedBy=multi-user.target
SVC
systemctl daemon-reload
systemctl enable act_runner-2.service act_runner-3.service 2>&1
systemctl start act_runner-2.service 2>&1
systemctl start act_runner-3.service 2>&1
sleep 8
echo ""
echo "========== Step 5: 验证 =========="
COUNT=$(ps aux | grep 'act_runner daemon' | grep -v grep | wc -l)
echo "runner进程数: $COUNT"
ps aux | grep 'act_runner daemon' | grep -v grep
echo ""
echo "服务状态:"
systemctl is-active act_runner.service act_runner-2.service act_runner-3.service 2>&1
echo ""
echo "注册信息:"
for d in /var/lib/xiaoxia-ci /var/lib/xiaoxia-ci-runner-2 /var/lib/xiaoxia-ci-runner-3; do
if [ -f "$d/.runner" ]; then
python3 -c "
import json
d = json.load(open('$d/.runner'))
print(f' {d[\"name\"]}: id={d[\"id\"]}')
" 2>/dev/null
fi
done
echo ""
echo "========== DONE =========="
Regular → Executable
+17
View File
@@ -141,6 +141,23 @@ class ClassificationStatus(StrEnum):
COMPLETED = "completed"
FAILED = "failed"
@classmethod
def _missing_(cls, value: object) -> "ClassificationStatus":
"""兼容历史数据,避免枚举转换失败导致500。
- done → COMPLETED(早期版本用 done 表示完成)
- 其他未知值 → PENDING(兜底,不阻塞业务)
"""
if isinstance(value, str):
normalized = value.strip().lower()
if normalized in ("done", "success", "finished", "complete"):
return cls.COMPLETED
if normalized in ("fail", "error", "err"):
return cls.FAILED
if normalized in ("process", "processing", "running", "run"):
return cls.PROCESSING
return cls.PENDING
@dataclass(slots=True)
class Asset:
+68
View File
@@ -0,0 +1,68 @@
"""ClassificationStatus 枚举兼容性测试。
验证历史脏数据(如 'done')不会导致枚举转换失败。
"""
import pytest
from packages.domain.entities import ClassificationStatus
class TestClassificationStatusNormalValues:
"""正常值应该正确映射。"""
def test_pending(self):
assert ClassificationStatus("pending") == ClassificationStatus.PENDING
def test_processing(self):
assert ClassificationStatus("processing") == ClassificationStatus.PROCESSING
def test_completed(self):
assert ClassificationStatus("completed") == ClassificationStatus.COMPLETED
def test_failed(self):
assert ClassificationStatus("failed") == ClassificationStatus.FAILED
class TestClassificationStatusHistoricalValues:
"""历史脏数据应该正确映射到对应状态,不抛异常。"""
@pytest.mark.parametrize("value", ["done", "Done", "DONE", " done "])
def test_done_maps_to_completed(self, value):
"""生产环境发现的 'done' 历史值应映射为 COMPLETED。"""
assert ClassificationStatus(value) == ClassificationStatus.COMPLETED
@pytest.mark.parametrize("value", ["success", "finished", "complete"])
def test_other_done_like_values_map_to_completed(self, value):
assert ClassificationStatus(value) == ClassificationStatus.COMPLETED
@pytest.mark.parametrize("value", ["fail", "error", "err"])
def test_error_like_values_map_to_failed(self, value):
assert ClassificationStatus(value) == ClassificationStatus.FAILED
@pytest.mark.parametrize("value", ["process", "running", "run"])
def test_processing_like_values_map_to_processing(self, value):
assert ClassificationStatus(value) == ClassificationStatus.PROCESSING
class TestClassificationStatusFallback:
"""完全未知的值兜底为 PENDING,不抛500。"""
@pytest.mark.parametrize("value", ["unknown", "foo_bar", ""])
def test_unknown_value_falls_back_to_pending(self, value):
assert ClassificationStatus(value) == ClassificationStatus.PENDING
def test_none_value_falls_back_to_pending(self):
assert ClassificationStatus(None) == ClassificationStatus.PENDING # type: ignore[arg-type]
def test_int_value_falls_back_to_pending(self):
assert ClassificationStatus(123) == ClassificationStatus.PENDING # type: ignore[arg-type]
class TestClassificationStatusStrValue:
"""枚举值仍为字符串类型,不影响序列化。"""
def test_value_unchanged(self):
assert ClassificationStatus.COMPLETED.value == "completed"
assert ClassificationStatus.PENDING.value == "pending"
assert isinstance(ClassificationStatus.COMPLETED, str)