测试环境治理：环境配置管理与版本控制_千锋教育

import hashlib
import random

def desensitize_phone(phone):
    """手机号脱敏：保留前3后4"""
    return phone[:3] + "****" + phone[-4:]

def desensitize_id_card(id_card):
    """身份证脱敏：保留前6后4"""
    return id_card[:6] + "********" + id_card[-4:]

def desensitize_name(name):
    """姓名脱敏：保留姓，名替换为*"""
    return name[0] + "*" * (len(name) - 1)

# 批量脱敏示例
test_data = [
    {"name": "张三", "phone": "13800138000", "id_card": "110101199001011234"},
    {"name": "李四", "phone": "13900139000", "id_card": "310101199002022345"}
]

for record in test_data:
    record["name"] = desensitize_name(record["name"])
    record["phone"] = desensitize_phone(record["phone"])
    record["id_card"] = desensitize_id_card(record["id_card"])
    print(record)

# .gitlab-ci.yml 示例
stages:
  - validate
  - deploy

validate_config:
  stage: validate
  script:
    - python validate_config.py --env $CI_ENVIRONMENT_NAME
  only:
    - main
    - develop

deploy_config:
  stage: deploy
  script:
    - ansible-playbook deploy_config.yml -e env=$CI_ENVIRONMENT_NAME
  environment:
    name: $CI_ENVIRONMENT_NAME
  only:
    - main

# ansible/vars/secrets.yml 加密版本
db_host: "{{ vault_db_host }}"
db_password: "{{ vault_db_password }}"
api_key: "{{ vault_api_key }}"

# ansible-playbook命令
ansible-playbook deploy.yml --vault-password-file vault_pass.txt

# terraform/consul_config.tf
resource "consul_keys" "test_env" {
  key {
    path = "test/database/host"
    value = "test-db.cluster-xxx.rds.amazonaws.com"
  }
  key {
    path = "test/database/port"
    value = "3306"
  }
  key {
    path = "test/redis/host"
    value = "test-redis.xxx.cache.amazonaws.com"
  }
}

# env_governance.py
import json
import yaml
import git
from datetime import datetime

class EnvironmentGovernance:
    def __init__(self, env_name):
        self.env_name = env_name
        self.config_path = f"configs/{env_name}.yaml"
        self.repo = git.Repo(".")
    
    def validate_config(self):
        """验证配置完整性"""
        with open(self.config_path, 'r') as f:
            config = yaml.safe_load(f)
        
        required_fields = ['database', 'redis', 'api_keys']
        missing_fields = [f for f in required_fields if f not in config]
        
        if missing_fields:
            raise ValueError(f"Missing config fields: {missing_fields}")
        
        return True
    
    def check_data_compliance(self):
        """检查数据合规性"""
        compliance_rules = {
            'phone_mask': True,
            'id_card_mask': True,
            'email_domain': True
        }
        
        # 模拟检查过程
        for rule, expected in compliance_rules.items():
            actual = self._check_rule(rule)
            if actual != expected:
                self._create_alert(f"Compliance violation: {rule}")
        
        return all(compliance_rules.values())
    
    def _check_rule(self, rule):
        """内部规则检查方法"""
        # 实际实现会连接数据库验证脱敏状态
        return True
    
    def create_snapshot(self):
        """创建环境快照"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        snapshot_name = f"{self.env_name}_snapshot_{timestamp}"
        
        # 保存配置版本
        self.repo.index.add([self.config_path])
        self.repo.index.commit(f"Snapshot: {snapshot_name}")
        
        return snapshot_name

# 使用示例
governance = EnvironmentGovernance("test")
governance.validate_config()
governance.check_data_compliance()
snapshot = governance.create_snapshot()
print(f"Snapshot created: {snapshot}")