Test Data Management in DevOps Pipelines: Synchronization, Masking, and Versioning Strategies

# data-classification-schema.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: data-classification
data:
  classification-rules.json: |
    {
      "classifications": {
        "PII": {
          "level": "sensitive",
          "patterns": [
            "\\b[A-Z]{1}[a-z]+\\s[A-Z]{1}[a-z]+\\b",
            "\\b\\d{3}-\\d{2}-\\d{4}\\b",
            "\\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Z|a-z]{2,}\\b"
          ],
          "fields": ["name", "email", "ssn", "phone", "address"],
          "handling": {
            "storage": "encrypted",
            "transit": "tls-required",
            "retention": "30-days",
            "masking": "required"
          }
        },
        "PHI": {
          "level": "highly-sensitive",
          "fields": ["diagnosis", "medical_record", "prescription"],
          "compliance": ["HIPAA"],
          "handling": {
            "storage": "encrypted-at-rest",
            "access": "audit-logged",
            "masking": "tokenization"
          }
        },
        "Financial": {
          "level": "sensitive",
          "patterns": [
            "\\b\\d{4}[\\s-]?\\d{4}[\\s-]?\\d{4}[\\s-]?\\d{4}\\b",
            "\\b\\d{3,4}\\b"
          ],
          "fields": ["credit_card", "bank_account", "routing_number"],
          "compliance": ["PCI-DSS"],
          "handling": {
            "storage": "tokenized",
            "masking": "format-preserving-encryption"
          }
        }
      }
    }

# test-data-provisioning/provisioner.py
import json
import hashlib
import random
from datetime import datetime, timedelta
from typing import Dict, List, Any
import boto3
import psycopg2
from faker import Faker
from dataclasses import dataclass

@dataclass
class TestDataRequest:
    environment: str
    dataset_size: str  # small, medium, large
    data_freshness: str  # real-time, daily, weekly
    compliance_requirements: List[str]
    seed_value: int

class TestDataProvisioner:
    def __init__(self, config_path: str):
        with open(config_path, 'r') as f:
            self.config = json.load(f)
        self.faker = Faker()
        self.s3 = boto3.client('s3')

    def provision_dataset(self, request: TestDataRequest) -> Dict[str, Any]:
        """Main provisioning workflow"""
        dataset = {
            'metadata': self._generate_metadata(request),
            'data': {}
        }

        # Determine data sources based on requirements
        if 'real-time' in request.data_freshness:
            dataset['data'] = self._extract_production_subset(request)
        else:
            dataset['data'] = self._generate_synthetic_data(request)

        # Apply compliance transformations
        if request.compliance_requirements:
            dataset['data'] = self._apply_compliance_masking(
                dataset['data'],
                request.compliance_requirements
            )

        # Version and store the dataset
        dataset_id = self._version_dataset(dataset)

        # Provision to target environment
        self._deploy_to_environment(dataset_id, request.environment)

        return {
            'dataset_id': dataset_id,
            'environment': request.environment,
            'status': 'provisioned',
            'timestamp': datetime.utcnow().isoformat()
        }

    def _extract_production_subset(self, request: TestDataRequest) -> Dict:
        """Extract and subset production data"""
        conn = psycopg2.connect(
            host=self.config['production_db']['host'],
            port=self.config['production_db']['port'],
            database=self.config['production_db']['database'],
            user=self.config['production_db']['user'],
            password=self.config['production_db']['password']
        )

        subset_config = self._get_subset_config(request.dataset_size)

        query = f"""
        WITH sampled_users AS (
            SELECT * FROM users
            TABLESAMPLE BERNOULLI ({subset_config['sampling_percentage']})
            WHERE created_at > NOW() - INTERVAL '{subset_config['time_window']}'
            LIMIT {subset_config['max_records']}
        ),
        related_orders AS (
            SELECT o.* FROM orders o
            INNER JOIN sampled_users u ON o.user_id = u.id
        ),
        related_transactions AS (
            SELECT t.* FROM transactions t
            INNER JOIN related_orders o ON t.order_id = o.id
        )
        SELECT
            json_build_object(
                'users', (SELECT json_agg(u) FROM sampled_users u),
                'orders', (SELECT json_agg(o) FROM related_orders o),
                'transactions', (SELECT json_agg(t) FROM related_transactions t)
            ) as dataset
        """

        cursor = conn.cursor()
        cursor.execute(query)
        result = cursor.fetchone()[0]

        conn.close()
        return result

    def _generate_synthetic_data(self, request: TestDataRequest) -> Dict:
        """Generate synthetic test data"""
        Faker.seed(request.seed_value)
        random.seed(request.seed_value)

        dataset_config = self._get_dataset_config(request.dataset_size)

        users = []
        for _ in range(dataset_config['user_count']):
            user = {
                'id': self.faker.uuid4(),
                'name': self.faker.name(),
                'email': self.faker.email(),
                'phone': self.faker.phone_number(),
                'address': {
                    'street': self.faker.street_address(),
                    'city': self.faker.city(),
                    'state': self.faker.state(),
                    'zip': self.faker.postcode()
                },
                'created_at': self.faker.date_time_between(
                    start_date='-1y',
                    end_date='now'
                ).isoformat()
            }
            users.append(user)

        orders = []
        for user in users[:int(len(users) * 0.7)]:  # 70% of users have orders
            order_count = random.randint(1, 5)
            for _ in range(order_count):
                order = {
                    'id': self.faker.uuid4(),
                    'user_id': user['id'],
                    'total': round(random.uniform(10, 500), 2),
                    'status': random.choice(['pending', 'completed', 'cancelled']),
                    'created_at': self.faker.date_time_between(
                        start_date=user['created_at'],
                        end_date='now'
                    ).isoformat()
                }
                orders.append(order)

        return {
            'users': users,
            'orders': orders,
            'generated_at': datetime.utcnow().isoformat(),
            'seed': request.seed_value
        }

# masking-engine/masker.py
import re
import hashlib
import secrets
from typing import Any, Dict, List
from cryptography.fernet import Fernet
from datetime import datetime

class DataMaskingEngine:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.fernet = Fernet(config['encryption_key'].encode())
        self.token_vault = {}

    def mask_dataset(self, data: Dict, rules: List[Dict]) -> Dict:
        """Apply masking rules to dataset"""
        masked_data = {}

        for table_name, records in data.items():
            masked_records = []
            for record in records:
                masked_record = self._mask_record(record, rules)
                masked_records.append(masked_record)
            masked_data[table_name] = masked_records

        return masked_data

    def _mask_record(self, record: Dict, rules: List[Dict]) -> Dict:
        """Apply masking rules to individual record"""
        masked = record.copy()

        for field, value in record.items():
            for rule in rules:
                if self._field_matches_rule(field, value, rule):
                    masked[field] = self._apply_masking_technique(
                        value,
                        rule['technique'],
                        rule.get('params', {})
                    )
                    break

        return masked

    def _apply_masking_technique(self, value: Any, technique: str, params: Dict) -> Any:
        """Apply specific masking technique"""
        if technique == 'hash':
            salt = params.get('salt', 'default-salt')
            return hashlib.sha256(f"{value}{salt}".encode()).hexdigest()[:len(str(value))]

        elif technique == 'tokenize':
            if value not in self.token_vault:
                token = secrets.token_urlsafe(32)
                self.token_vault[value] = token
                self._persist_token_mapping(value, token)
            return self.token_vault[value]

        elif technique == 'format_preserving':
            return self._format_preserving_encryption(value, params)

        elif technique == 'partial':
            mask_char = params.get('mask_char', '*')
            visible_chars = params.get('visible_chars', 4)
            if len(str(value)) > visible_chars:
                return str(value)[:visible_chars] + mask_char * (len(str(value)) - visible_chars)
            return value

        elif technique == 'shuffle':
            import random
            chars = list(str(value))
            random.shuffle(chars)
            return ''.join(chars)

        elif technique == 'date_shift':
            shift_days = params.get('shift_days', 30)
            if isinstance(value, str):
                dt = datetime.fromisoformat(value)
                shifted = dt.timestamp() + (shift_days * 86400)
                return datetime.fromtimestamp(shifted).isoformat()
            return value

        elif technique == 'redact':
            return params.get('replacement', '[REDACTED]')

        return value

    def _format_preserving_encryption(self, value: str, params: Dict) -> str:
        """Implement format-preserving encryption"""
        # Preserve format while encrypting
        if re.match(r'\d{4}-\d{4}-\d{4}-\d{4}', value):  # Credit card
            encrypted = self.fernet.encrypt(value.encode()).decode()
            # Generate format-preserving output
            hash_val = hashlib.md5(encrypted.encode()).hexdigest()
            return f"{hash_val[:4]}-{hash_val[4:8]}-{hash_val[8:12]}-{hash_val[12:16]}"

        elif re.match(r'\d{3}-\d{2}-\d{4}', value):  # SSN
            encrypted = self.fernet.encrypt(value.encode()).decode()
            hash_val = hashlib.md5(encrypted.encode()).hexdigest()
            nums = ''.join(filter(str.isdigit, hash_val))[:9]
            return f"{nums[:3]}-{nums[3:5]}-{nums[5:9]}"

        return value

# gdpr-compliance/pipeline.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: gdpr-compliance-rules
data:
  compliance-config.yaml: |
    gdpr:
      data_categories:
        personal_data:
          - name
          - email
          - phone
          - address
          - ip_address
        special_categories:
          - race_ethnicity
          - political_opinions
          - religious_beliefs
          - health_data
          - biometric_data

      processing_rules:
        right_to_erasure:
          retention_period: 30
          deletion_method: "secure_wipe"

        data_minimization:
          fields_to_exclude:
            - unnecessary_metadata
            - internal_ids
            - system_timestamps

        pseudonymization:
          technique: "tokenization"
          reversible: true
          key_storage: "hsm"

      test_data_requirements:
        consent_simulation: true
        audit_logging: mandatory
        encryption_at_rest: required
        cross_border_transfer: prohibited

# synthetic-generator/generator.py
import numpy as np
import pandas as pd
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
from scipy import stats
import tensorflow as tf
from typing import Tuple, Dict, Any

class SyntheticDataGenerator:
    def __init__(self, source_data_path: str, config: Dict[str, Any]):
        self.source_data = pd.read_csv(source_data_path)
        self.config = config
        self.statistical_properties = {}

    def analyze_source_distribution(self) -> Dict[str, Any]:
        """Analyze statistical properties of source data"""
        properties = {}

        for column in self.source_data.columns:
            col_data = self.source_data[column]

            if pd.api.types.is_numeric_dtype(col_data):
                properties[column] = {
                    'type': 'numeric',
                    'mean': col_data.mean(),
                    'std': col_data.std(),
                    'min': col_data.min(),
                    'max': col_data.max(),
                    'distribution': self._fit_distribution(col_data),
                    'correlations': self._calculate_correlations(column)
                }
            elif pd.api.types.is_categorical_dtype(col_data) or col_data.dtype == 'object':
                properties[column] = {
                    'type': 'categorical',
                    'categories': col_data.value_counts().to_dict(),
                    'probabilities': col_data.value_counts(normalize=True).to_dict()
                }
            elif pd.api.types.is_datetime64_any_dtype(col_data):
                properties[column] = {
                    'type': 'datetime',
                    'min': col_data.min(),
                    'max': col_data.max(),
                    'frequency': pd.infer_freq(col_data)
                }

        self.statistical_properties = properties
        return properties

    def generate_synthetic_dataset(self, num_records: int) -> pd.DataFrame:
        """Generate synthetic dataset maintaining statistical properties"""
        if not self.statistical_properties:
            self.analyze_source_distribution()

        synthetic_data = {}

        # Generate base columns
        for column, props in self.statistical_properties.items():
            if props['type'] == 'numeric':
                synthetic_data[column] = self._generate_numeric_column(
                    props, num_records
                )
            elif props['type'] == 'categorical':
                synthetic_data[column] = self._generate_categorical_column(
                    props, num_records
                )
            elif props['type'] == 'datetime':
                synthetic_data[column] = self._generate_datetime_column(
                    props, num_records
                )

        df = pd.DataFrame(synthetic_data)

        # Apply correlations
        df = self._apply_correlations(df)

        # Validate statistical similarity
        validation_results = self._validate_synthetic_data(df)

        return df

    def _generate_numeric_column(self, props: Dict, num_records: int) -> np.ndarray:
        """Generate numeric column based on distribution"""
        dist_name = props['distribution']['name']
        dist_params = props['distribution']['params']

        if dist_name == 'normal':
            data = np.random.normal(
                dist_params['loc'],
                dist_params['scale'],
                num_records
            )
        elif dist_name == 'exponential':
            data = np.random.exponential(
                dist_params['scale'],
                num_records
            )
        elif dist_name == 'uniform':
            data = np.random.uniform(
                props['min'],
                props['max'],
                num_records
            )
        else:
            # Fallback to normal distribution
            data = np.random.normal(
                props['mean'],
                props['std'],
                num_records
            )

        # Clip to original bounds
        data = np.clip(data, props['min'], props['max'])

        return data

    def _fit_distribution(self, data: pd.Series) -> Dict[str, Any]:
        """Fit statistical distribution to data"""
        distributions = ['normal', 'exponential', 'uniform', 'gamma', 'beta']
        best_dist = None
        best_params = None
        best_ks_stat = float('inf')

        for dist_name in distributions:
            try:
                dist = getattr(stats, dist_name)
                params = dist.fit(data.dropna())
                ks_stat, _ = stats.kstest(data.dropna(), lambda x: dist.cdf(x, *params))

                if ks_stat < best_ks_stat:
                    best_ks_stat = ks_stat
                    best_dist = dist_name
                    best_params = params
            except:
                continue

        return {
            'name': best_dist,
            'params': dict(zip(['loc', 'scale'], best_params[:2])),
            'ks_statistic': best_ks_stat
        }

# gan-generator/tabular_gan.py
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
import numpy as np

class TabularGAN:
    def __init__(self, input_dim: int, latent_dim: int = 100):
        self.input_dim = input_dim
        self.latent_dim = latent_dim
        self.generator = self._build_generator()
        self.discriminator = self._build_discriminator()
        self.gan = self._build_gan()

    def _build_generator(self) -> keras.Model:
        """Build generator network"""
        model = keras.Sequential([
            layers.Dense(256, activation='relu', input_dim=self.latent_dim),
            layers.BatchNormalization(),
            layers.Dropout(0.3),
            layers.Dense(512, activation='relu'),
            layers.BatchNormalization(),
            layers.Dropout(0.3),
            layers.Dense(1024, activation='relu'),
            layers.BatchNormalization(),
            layers.Dropout(0.3),
            layers.Dense(self.input_dim, activation='tanh')
        ])
        return model

    def _build_discriminator(self) -> keras.Model:
        """Build discriminator network"""
        model = keras.Sequential([
            layers.Dense(1024, activation='relu', input_dim=self.input_dim),
            layers.Dropout(0.3),
            layers.Dense(512, activation='relu'),
            layers.Dropout(0.3),
            layers.Dense(256, activation='relu'),
            layers.Dropout(0.3),
            layers.Dense(1, activation='sigmoid')
        ])

        model.compile(
            optimizer=keras.optimizers.Adam(learning_rate=0.0002, beta_1=0.5),
            loss='binary_crossentropy',
            metrics=['accuracy']
        )
        return model

    def _build_gan(self) -> keras.Model:
        """Combine generator and discriminator"""
        self.discriminator.trainable = False

        gan_input = keras.Input(shape=(self.latent_dim,))
        generated = self.generator(gan_input)
        gan_output = self.discriminator(generated)

        model = keras.Model(gan_input, gan_output)
        model.compile(
            optimizer=keras.optimizers.Adam(learning_rate=0.0002, beta_1=0.5),
            loss='binary_crossentropy'
        )
        return model

    def train(self, real_data: np.ndarray, epochs: int = 1000, batch_size: int = 32):
        """Train the GAN"""
        for epoch in range(epochs):
            # Train discriminator
            idx = np.random.randint(0, real_data.shape[0], batch_size)
            real_batch = real_data[idx]

            noise = np.random.normal(0, 1, (batch_size, self.latent_dim))
            generated_batch = self.generator.predict(noise, verbose=0)

            d_loss_real = self.discriminator.train_on_batch(
                real_batch,
                np.ones((batch_size, 1))
            )
            d_loss_fake = self.discriminator.train_on_batch(
                generated_batch,
                np.zeros((batch_size, 1))
            )
            d_loss = 0.5 * np.add(d_loss_real, d_loss_fake)

            # Train generator
            noise = np.random.normal(0, 1, (batch_size, self.latent_dim))
            g_loss = self.gan.train_on_batch(
                noise,
                np.ones((batch_size, 1))
            )

            if epoch % 100 == 0:
                print(f"Epoch {epoch}, D Loss: {d_loss[0]:.4f}, G Loss: {g_loss:.4f}")

    def generate_synthetic_data(self, num_samples: int) -> np.ndarray:
        """Generate synthetic samples"""
        noise = np.random.normal(0, 1, (num_samples, self.latent_dim))
        return self.generator.predict(noise)

// Jenkinsfile
@Library('test-data-lib') _

pipeline {
    agent any

    parameters {
        choice(name: 'ENVIRONMENT',
               choices: ['dev', 'qa', 'staging', 'performance'],
               description: 'Target environment')
        choice(name: 'DATA_SOURCE',
               choices: ['production-subset', 'synthetic', 'hybrid'],
               description: 'Data source strategy')
        choice(name: 'DATASET_SIZE',
               choices: ['small', 'medium', 'large', 'xlarge'],
               description: 'Dataset size')
        multiChoice(name: 'COMPLIANCE_REQUIREMENTS',
                   choices: ['GDPR', 'CCPA', 'HIPAA', 'PCI-DSS'],
                   description: 'Compliance requirements')
    }

    environment {
        VAULT_ADDR = 'https://vault.example.com'
        DATA_LAKE_BUCKET = 's3://test-data-lake'
    }

    stages {
        stage('Initialize Test Data Pipeline') {
            steps {
                script {
                    // Initialize configuration
                    testDataConfig = [
                        environment: params.ENVIRONMENT,
                        dataSource: params.DATA_SOURCE,
                        datasetSize: params.DATASET_SIZE,
                        compliance: params.COMPLIANCE_REQUIREMENTS,
                        timestamp: new Date().format('yyyy-MM-dd-HH-mm-ss')
                    ]

                    // Generate unique dataset ID
                    testDataConfig.datasetId = generateDatasetId(testDataConfig)
                }
            }
        }

        stage('Fetch Source Data') {
            when {
                expression { params.DATA_SOURCE != 'synthetic' }
            }
            steps {
                script {
                    withCredentials([
                        usernamePassword(
                            credentialsId: 'production-db-readonly',
                            usernameVariable: 'DB_USER',
                            passwordVariable: 'DB_PASS'
                        )
                    ]) {
                        sh """
                            python3 scripts/extract_production_subset.py \
                                --host ${PROD_DB_HOST} \
                                --user ${DB_USER} \
                                --password ${DB_PASS} \
                                --size ${params.DATASET_SIZE} \
                                --output /tmp/raw-data.json
                        """
                    }
                }
            }
        }

        stage('Generate Synthetic Data') {
            when {
                expression { params.DATA_SOURCE in ['synthetic', 'hybrid'] }
            }
            steps {
                script {
                    sh """
                        python3 scripts/generate_synthetic_data.py \
                            --config configs/synthetic-${params.ENVIRONMENT}.yaml \
                            --size ${params.DATASET_SIZE} \
                            --seed ${BUILD_NUMBER} \
                            --output /tmp/synthetic-data.json
                    """
                }
            }
        }

        stage('Apply Data Masking') {
            steps {
                script {
                    def maskingRules = loadMaskingRules(params.COMPLIANCE_REQUIREMENTS)

                    sh """
                        python3 scripts/apply_data_masking.py \
                            --input /tmp/raw-data.json \
                            --rules ${maskingRules} \
                            --compliance ${params.COMPLIANCE_REQUIREMENTS.join(',')} \
                            --output /tmp/masked-data.json
                    """
                }
            }
        }

        stage('Validate Data Quality') {
            steps {
                script {
                    sh """
                        python3 scripts/validate_test_data.py \
                            --data /tmp/masked-data.json \
                            --schema schemas/${params.ENVIRONMENT}-schema.json \
                            --rules validation-rules.yaml \
                            --report /tmp/validation-report.html
                    """

                    publishHTML(target: [
                        allowMissing: false,
                        alwaysLinkToLastBuild: true,
                        keepAll: true,
                        reportDir: '/tmp',
                        reportFiles: 'validation-report.html',
                        reportName: 'Data Validation Report'
                    ])
                }
            }
        }

        stage('Version and Store Dataset') {
            steps {
                script {
                    sh """
                        # Compress and encrypt dataset
                        tar -czf /tmp/dataset-${testDataConfig.datasetId}.tar.gz \
                            /tmp/masked-data.json

                        # Upload to S3 with versioning
                        aws s3 cp /tmp/dataset-${testDataConfig.datasetId}.tar.gz \
                            ${DATA_LAKE_BUCKET}/${params.ENVIRONMENT}/ \
                            --server-side-encryption aws:kms \
                            --metadata "build=${BUILD_NUMBER},environment=${params.ENVIRONMENT}"

                        # Register in data catalog
                        python3 scripts/register_dataset.py \
                            --dataset-id ${testDataConfig.datasetId} \
                            --location ${DATA_LAKE_BUCKET}/${params.ENVIRONMENT}/ \
                            --metadata '${groovy.json.JsonOutput.toJson(testDataConfig)}'
                    """
                }
            }
        }

        stage('Deploy to Test Environment') {
            steps {
                script {
                    parallel(
                        'Database': {
                            sh """
                                python3 scripts/load_to_database.py \
                                    --dataset /tmp/masked-data.json \
                                    --target ${params.ENVIRONMENT} \
                                    --connection-string \${${params.ENVIRONMENT.toUpperCase()}_DB_URL}
                            """
                        },
                        'Cache': {
                            sh """
                                python3 scripts/load_to_cache.py \
                                    --dataset /tmp/masked-data.json \
                                    --redis-host \${${params.ENVIRONMENT.toUpperCase()}_REDIS_HOST} \
                                    --ttl 3600
                            """
                        },
                        'File System': {
                            sh """
                                kubectl cp /tmp/masked-data.json \
                                    ${params.ENVIRONMENT}/test-data-pod:/data/test-data.json
                            """
                        }
                    )
                }
            }
        }

        stage('Run Data Verification Tests') {
            steps {
                script {
                    sh """
                        pytest tests/data_verification/ \
                            --environment ${params.ENVIRONMENT} \
                            --dataset-id ${testDataConfig.datasetId} \
                            --junitxml=test-results.xml
                    """
                }
            }
        }
    }

    post {
        always {
            junit 'test-results.xml'
            cleanWs()
        }
        success {
            emailext(
                subject: "Test Data Provisioned: ${testDataConfig.datasetId}",
                body: """
                    Test data successfully provisioned for ${params.ENVIRONMENT}

                    Dataset ID: ${testDataConfig.datasetId}
                    Size: ${params.DATASET_SIZE}
                    Source: ${params.DATA_SOURCE}
                    Compliance: ${params.COMPLIANCE_REQUIREMENTS}

                    Access the dataset:
                    ${env.BUILD_URL}
                """,
                to: 'qa-team@example.com'
            )
        }
        failure {
            sh """
                # Rollback any partial deployments
                python3 scripts/rollback_test_data.py \
                    --environment ${params.ENVIRONMENT} \
                    --dataset-id ${testDataConfig.datasetId}
            """
        }
    }
}

# .gitlab-ci.yml
stages:
  - prepare
  - extract
  - transform
  - load
  - validate

variables:
  DATA_PIPELINE_VERSION: "2.1.0"
  PYTHON_IMAGE: "python:3.9-slim"

.test-data-template:
  image: ${PYTHON_IMAGE}
  before_script:
    - pip install -r requirements.txt
    - export DATASET_ID=$(date +%Y%m%d-%H%M%S)-${CI_PIPELINE_ID}

prepare:environment:
  stage: prepare
  script:
    - echo "Preparing test data environment for ${CI_ENVIRONMENT_NAME}"
    - |
      python3 scripts/prepare_environment.py \
        --environment ${CI_ENVIRONMENT_NAME} \
        --clean-previous ${CLEAN_PREVIOUS_DATA}

extract:production-subset:
  stage: extract
  only:
    variables:
      - $DATA_SOURCE == "production"
  script:
    - |
      python3 scripts/extract_subset.py \
        --source ${PRODUCTION_DB_URL} \
        --size ${DATASET_SIZE} \
        --filters config/extraction-filters.yaml \
        --output artifacts/raw-data.json
  artifacts:
    paths:
      - artifacts/raw-data.json
    expire_in: 1 hour

generate:synthetic-data:
  stage: extract
  only:
    variables:
      - $DATA_SOURCE == "synthetic"
  script:
    - |
      python3 scripts/synthetic_generator.py \
        --model models/data-generation-model.pkl \
        --size ${DATASET_SIZE} \
        --config config/synthetic-config.yaml \
        --output artifacts/synthetic-data.json
  artifacts:
    paths:
      - artifacts/synthetic-data.json
    expire_in: 1 hour

transform:apply-masking:
  stage: transform
  dependencies:
    - extract:production-subset
    - generate:synthetic-data
  script:
    - |
      python3 scripts/data_masker.py \
        --input artifacts/*.json \
        --rules config/masking-rules-${COMPLIANCE_LEVEL}.yaml \
        --output artifacts/masked-data.json \
        --audit-log artifacts/masking-audit.log
  artifacts:
    paths:
      - artifacts/masked-data.json
      - artifacts/masking-audit.log
    expire_in: 1 day

load:to-database:
  stage: load
  dependencies:
    - transform:apply-masking
  script:
    - |
      python3 scripts/database_loader.py \
        --data artifacts/masked-data.json \
        --target ${CI_ENVIRONMENT_NAME} \
        --connection ${TEST_DB_CONNECTION_STRING} \
        --batch-size 1000

validate:data-quality:
  stage: validate
  dependencies:
    - load:to-database
  script:
    - |
      python3 scripts/data_validator.py \
        --environment ${CI_ENVIRONMENT_NAME} \
        --checks config/validation-checks.yaml \
        --report artifacts/validation-report.html
  artifacts:
    reports:
      junit: artifacts/validation-results.xml
    paths:
      - artifacts/validation-report.html
    expire_in: 30 days

# security/data-security-framework.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: test-data-security
data:
  security-controls.yaml: |
    encryption:
      at_rest:
        algorithm: "AES-256-GCM"
        key_rotation: "30-days"
        key_storage: "HSM"

      in_transit:
        protocol: "TLS 1.3"
        certificate_validation: "required"
        mutual_tls: "enabled"

    access_control:
      authentication:
        method: "oauth2"
        mfa_required: true
        session_timeout: "30-minutes"

      authorization:
        model: "RBAC"
        roles:
          - name: "test-data-admin"
            permissions: ["create", "read", "update", "delete", "mask"]
          - name: "test-data-user"
            permissions: ["read"]
          - name: "compliance-auditor"
            permissions: ["read", "audit"]

      audit_logging:
        enabled: true
        retention: "7-years"
        immutable_storage: true
        events:
          - data_access
          - data_modification
          - masking_operations
          - compliance_violations

    data_lifecycle:
      retention:
        test_data: "30-days"
        masked_data: "90-days"
        synthetic_data: "unlimited"

      disposal:
        method: "cryptographic-erasure"
        verification: "required"
        certificate_generation: true

# compliance-monitoring/dashboard.py
from flask import Flask, jsonify, render_template
from datetime import datetime, timedelta
import psycopg2
import redis

app = Flask(__name__)

class ComplianceMonitor:
    def __init__(self):
        self.db = psycopg2.connect(
            host="localhost",
            database="compliance_db",
            user="monitor",
            password="secure_password"
        )
        self.cache = redis.Redis(host='localhost', port=6379)

    def get_compliance_metrics(self) -> dict:
        """Collect compliance metrics"""
        metrics = {
            'gdpr': self._check_gdpr_compliance(),
            'pci_dss': self._check_pci_compliance(),
            'hipaa': self._check_hipaa_compliance(),
            'data_quality': self._check_data_quality(),
            'last_updated': datetime.utcnow().isoformat()
        }
        return metrics

    def _check_gdpr_compliance(self) -> dict:
        cursor = self.db.cursor()
        cursor.execute("""
            SELECT
                COUNT(*) FILTER (WHERE is_masked = true) as masked_records,
                COUNT(*) as total_records,
                COUNT(DISTINCT dataset_id) as datasets,
                MAX(created_at) as last_processing
            FROM test_data_records
            WHERE contains_pii = true
        """)
        result = cursor.fetchone()

        return {
            'compliant': result[0] == result[1],
            'masked_percentage': (result[0] / result[1] * 100) if result[1] > 0 else 100,
            'total_datasets': result[2],
            'last_processing': result[3].isoformat() if result[3] else None
        }

@app.route('/api/compliance/status')
def compliance_status():
    monitor = ComplianceMonitor()
    return jsonify(monitor.get_compliance_metrics())

@app.route('/api/compliance/audit-trail/<dataset_id>')
def audit_trail(dataset_id):
    monitor = ComplianceMonitor()
    trail = monitor.get_audit_trail(dataset_id)
    return jsonify(trail)

if __name__ == '__main__':
    app.run(debug=False, port=5000)