11 · 测试与 CI：125 个用例怎么保护一个平台

class TraceContextTests(TestCase):
    def test_parse_valid_traceparent(self):
        ctx = tracing.parse_traceparent('00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01')
        self.assertEqual(ctx['trace_id'], '4bf92f3577b34da6a3ce929d0e0e4736')
        self.assertEqual(ctx['span_id'], '00f067aa0ba902b7')
        self.assertEqual(ctx['flags'], '01')

    def test_parse_rejects_invalid(self):
        bad = [
            '',
            'not-a-traceparent',
            '00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7',  # 缺 flags
            '00-00000000000000000000000000000000-00f067aa0ba902b7-01',  # 全零 trace-id
            '00-4bf92f3577b34da6a3ce929d0e0e4736-0000000000000000-01',  # 全零 span-id
            'ff-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01',  # version ff
            '00-4BF92F3577B34DA6A3CE929D0E0E4736-00f067aa0ba902b7-01',  # 大写非法
            '00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-0g',  # flags 非 hex
        ]
        for value in bad:
            self.assertIsNone(tracing.parse_traceparent(value), value)

class ForDurationTests(EngineBaseTests):
    def _policy(self, **kw):
        defaults = dict(name='p', metric_key='test.metric', operator='>',
                        threshold=10, level='P2', for_minutes=2)
        defaults.update(kw)
        return AlertPolicy.objects.create(**defaults)

    def _metric(self, value):
        # 测试用注册表桩：把 metric_value 指到可控数据
        CustomMetric.objects.create(name='metric', value=value, created_at=timezone.now())

    def test_spike_does_not_fire(self):
        """单次毛刺：不触发"""
        policy = self._policy(for_minutes=2)
        self._metric(99)
        fired, _ = engine.evaluate_once()
        self.assertEqual((fired, bool(AlertEvent.objects.count())), (0, False))
        # 毛刺过去（观察期内条件解除）
        self._metric(1)
        engine.evaluate_once()
        policy.refresh_from_db()
        self.assertIsNone(policy.breach_since)   # 计时被重置
        self.assertEqual(AlertEvent.objects.count(), 0)

    def test_sustained_breach_fires(self):
        """持续越限满 for_minutes：触发"""
        policy = self._policy(for_minutes=2)
        self._metric(99)
        engine.evaluate_once()               # 第 1 轮：记时
        self.assertIsNone(AlertEvent.objects.first())
        AlertPolicy.objects.filter(pk=policy.pk).update(
            breach_since=timezone.now() - timedelta(minutes=3))   # 时间快进 3 分钟
        engine.evaluate_once()               # 第 2 轮：观察期满，触发
        fired = AlertEvent.objects.get()
        self.assertEqual(fired.status, 'firing')

class LeadershipTests(TransactionTestCase):
    def test_acquire_exclusive(self):
        ok1, _ = leadership.acquire('t1', holder_id='A')
        ok2, _ = leadership.acquire('t1', holder_id='B')
        self.assertTrue(ok1)
        self.assertFalse(ok2)                      # B 未过期抢不走

    def test_expired_taken_over_with_term_bump(self):
        leadership.acquire('t2', holder_id='A', ttl=1)
        TaskLease.objects.filter(name='t2').update(
            expires_at=timezone.now() - timedelta(seconds=1))    # 租约过期
        ok, term = leadership.acquire('t2', holder_id='B')
        self.assertTrue(ok)
        self.assertEqual(term, 2)                  # 接管，任期 +1

    def test_previous_holder_must_yield(self):
        ok, _ = leadership.acquire('t3', holder_id='A', ttl=1)
        self.assertTrue(ok)
        TaskLease.objects.filter(name='t3').update(
            expires_at=timezone.now() - timedelta(seconds=1))
        ok2, _ = leadership.acquire('t3', holder_id='B')         # B 接管
        self.assertTrue(ok2)
        ok3, _ = leadership.acquire('t3', holder_id='A')         # A 必须认输
        self.assertFalse(ok3)

class SsrfGuardTests(TestCase):
    def test_blocks_metadata_and_reserved(self):
        for url in [
            'http://169.254.169.254/latest/meta-data/',
            'http://0.0.0.0/',
            'http://224.0.0.1/',
        ]:
            err = urlsafe.validate_url(url)
            self.assertIsNotNone(err, url)

    def test_blocks_non_http_schemes(self):
        for url in ['file:///etc/passwd', 'ftp://example.com', 'gopher://x']:
            self.assertIsNotNone(urlsafe.validate_url(url))

    def test_allows_loopback_by_default_and_blocks_when_disabled(self):
        self.assertIsNone(urlsafe.validate_url('http://127.0.0.1:8014/'))
        with override_settings(OBS_ALLOW_LOOPBACK_URL=False):
            # urlsafe 模块级开关读环境变量——测试通过 reload/monkeypatch 验证
            ...

class CleanSystemTempTests(TestCase):
    def test_rejects_outside_whitelist_root(self):
        """白名单根之外的目录必须被拒绝——哪怕请求里长得像系统临时目录"""
        ...

    def test_junction_symlink_pruned(self):
        """临时目录里塞一个指向外部的 Junction/符号链接：遍历必须剪枝，不进入"""
        # Windows 上构造 Junction 指向 C:\Users（或其他真实目录），
        # 断言清理后该目标目录原封不动
        ...

    def test_unknown_item_rejected(self):
        """未知清理项一律拒绝（防止配置层新增"顺手"项绕过白名单）"""
        ...

# ruff：CI 与本地共用的静态检查配置（与现有代码风格对齐，宽松起步）
target-version = "py310"
line-length = 110

exclude = ["migrations", "static", "venv", ".venv", "展示图片"]

[lint]
# E/W 基础 pycodestyle + F 逻辑错误；B 常见陷阱；UP 过时写法
select = ["E", "F", "W", "B", "UP"]
ignore = [
    "E501",  # 行宽交给 line-length 之上的例外（中文注释常超）
]

[lint.per-file-ignores]
"**/tests.py" = ["B011"]  # 测试里 assert False 类模式不受限

[format]
quote-style = "single"

pip install ruff
ruff check .

# CI：静态检查 + 全量测试（SQLite 内存库，无需外部服务）
name: ci

on:
  push:
    branches: [main, master]
  pull_request:

jobs:
  lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install ruff
      - run: ruff check .

  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: pip
      - run: pip install -r requirements.txt
      - name: Django system check
        run: python manage.py check
      - name: Migration completeness
        run: python manage.py makemigrations --check --dry-run
      - name: Tests
        run: python manage.py test --verbosity 2
        env:
          OBS_DISABLE_WORKERS: "1"

# monitor/tests.py 新手起手式
from django.test import TestCase

from monitor import tracing
from monitor.middleware import _sanitize_sql


class TracingSmokeTest(TestCase):
    def test_w3c_shape(self):
        tid = tracing.new_trace_id()
        self.assertEqual(len(tid), 32)
        self.assertTrue(set(tid) <= set('0123456789abcdef'))

    def test_rejects_garbage(self):
        self.assertIsNone(tracing.parse_traceparent(''))
        self.assertIsNone(tracing.parse_traceparent('xx' * 28))


class SanitizeSmokeTest(TestCase):
    def test_secret_masked(self):
        # 测试值是无意义占位，不是真实凭据
        sql = "SELECT * FROM t WHERE api_key = 'demo-123'"
        self.assertIn('***', _sanitize_sql(sql))

    def test_normal_column_untouched(self):
        sql = "SELECT * FROM t WHERE author_id = 7"
        self.assertEqual(_sanitize_sql(sql), sql)

python manage.py test            # 全部
python manage.py test monitor    # 单 app
python manage.py test monitor.tests.SanitizeSmokeTest   # 单类