GDPR Testing & Privacy Validation: Automate Compliance Checks

# tests/gdpr/test_right_to_erasure.py

def test_user_deletion_removes_personal_data():
    """Exercising the right to erasure must remove all personal data."""
    # Create a user with data across multiple systems
    user = create_test_user(
        name="Jane Smith",
        email="jane@example.com",
        address="123 Main St, Berlin",
        payment_method=PaymentMethod(card_number="4111111111111111")
    )
    
    # Generate some activity
    make_purchase(user, amount=99.99)
    create_support_ticket(user, "Test ticket")
    add_to_marketing_list(user)
    
    user_id = user.id
    
    # Exercise right to erasure
    response = api_client.delete(f"/api/users/{user_id}/data")
    assert response.status_code == 200
    
    # Verify personal data is removed
    assert not db.users.exists(id=user_id), "User record not deleted"
    
    # Orders should be anonymized, not deleted (legal retention requirement)
    orders = db.orders.find(user_id=user_id)
    for order in orders:
        assert order.user_id is None or order.user_id == "DELETED"
        assert order.customer_name is None or order.customer_name == "DELETED"
        assert order.customer_email is None

def test_deletion_removes_from_all_systems():
    """User deletion must propagate to all data stores."""
    user = create_test_user(email="delete-test@example.com")
    user_id = user.id
    email = user.email
    
    exercise_right_to_erasure(user_id)
    
    # Check primary database
    assert not primary_db.users.exists(id=user_id), "Not deleted from primary DB"
    
    # Check marketing database
    assert not marketing_db.contacts.exists(email=email), \
        "Not deleted from marketing system"
    
    # Check analytics (should be anonymized)
    analytics_events = analytics_db.events.find(user_id=user_id)
    for event in analytics_events:
        assert event.user_id is None, "Analytics events not anonymized"
    
    # Check email service
    subscribers = email_service.find_subscribers(email=email)
    assert len(subscribers) == 0, "Not unsubscribed from email service"

def test_deletion_does_not_remove_legally_required_data():
    """Financial records must be retained for legal compliance (7 years)."""
    user = create_test_user()
    make_purchase(user, amount=500.00)
    user_id = user.id
    
    exercise_right_to_erasure(user_id)
    
    # Financial transaction records must be retained (but anonymized)
    transactions = db.transactions.find(original_user_id=user_id)
    assert len(transactions) > 0, "Financial records deleted (required for tax compliance)"
    
    for txn in transactions:
        # Must be anonymized but retained
        assert txn.customer_name is None
        assert txn.customer_email is None
        assert txn.amount is not None  # Amount must be retained
        assert txn.transaction_date is not None  # Date must be retained

def test_data_export_contains_all_personal_data():
    """Data export must include all personal data in machine-readable format."""
    user = create_test_user(
        name="John Doe",
        email="john@example.com",
        phone="+491234567890"
    )
    make_purchase(user)
    create_profile(user, bio="Test bio", location="Berlin")
    
    response = api_client.post(f"/api/users/{user.id}/data-export")
    assert response.status_code == 200
    
    export_url = response.json()["download_url"]
    export_data = download_and_parse(export_url)
    
    # Verify format is machine-readable
    assert export_data.format in ["json", "csv", "xml"]
    
    # Verify completeness
    assert export_data.user["name"] == "John Doe"
    assert export_data.user["email"] == "john@example.com"
    assert export_data.user["phone"] == "+491234567890"
    assert len(export_data.orders) > 0
    assert export_data.profile["bio"] == "Test bio"

def test_data_export_delivered_within_30_days():
    """Data export requests must be fulfilled within 30 days per GDPR."""
    request = submit_data_export_request(user_id="test_user")
    
    # In practice you'd check the SLA, but validate the request is queued
    assert request.status == "queued"
    assert request.deadline <= request.submitted_at + timedelta(days=30)

def test_processing_requires_valid_consent():
    """Marketing emails must not be sent without consent."""
    user = create_test_user(marketing_consent=False)
    
    # Attempt to add to marketing campaign
    result = marketing_service.add_to_campaign(user.email, campaign_id="newsletter")
    
    assert result.success == False
    assert result.reason == "no_consent"
    
    # Verify not added
    subscribers = marketing_service.get_subscribers("newsletter")
    user_emails = [s.email for s in subscribers]
    assert user.email not in user_emails

def test_consent_withdrawal_stops_processing():
    """Withdrawing consent must immediately stop related processing."""
    user = create_test_user(marketing_consent=True)
    marketing_service.add_to_campaign(user.email, "newsletter")
    
    # Withdraw consent
    api_client.delete(f"/api/users/{user.id}/consent/marketing")
    
    # Verify immediately unsubscribed
    subscribers = marketing_service.get_subscribers("newsletter")
    user_emails = [s.email for s in subscribers]
    assert user.email not in user_emails

def test_consent_record_maintained():
    """Consent grants and withdrawals must be logged for accountability."""
    user = create_test_user()
    
    # Grant consent
    api_client.post(f"/api/users/{user.id}/consent", json={"type": "marketing"})
    
    # Withdraw consent
    api_client.delete(f"/api/users/{user.id}/consent/marketing")
    
    # Consent audit log should show both events
    consent_log = db.consent_log.find(user_id=user.id)
    assert len(consent_log) == 2
    assert consent_log[0].action == "granted"
    assert consent_log[1].action == "withdrawn"
    assert consent_log[0].timestamp is not None
    assert consent_log[1].timestamp is not None

def test_registration_only_collects_required_fields():
    """Registration should not collect more data than necessary."""
    # Test that unnecessary fields are rejected
    response = api_client.post("/api/register", json={
        "email": "test@example.com",
        "password": "SecurePass123!",
        # These should not be accepted at registration
        "date_of_birth": "1990-01-01",
        "national_id": "DE123456789",
        "salary": 100000
    })
    
    user = response.json()
    assert "date_of_birth" not in user, "DOB collected but not required"
    assert "national_id" not in user, "National ID collected but not required"
    assert "salary" not in user, "Salary collected but not required"

def test_analytics_events_are_anonymized():
    """Analytics must not contain personally identifiable information."""
    user = create_test_user(email="analytics-test@example.com")
    
    # Perform actions that generate analytics events
    visit_page(user, "/products")
    make_purchase(user, amount=49.99)
    
    analytics_events = analytics_db.get_events(user_id=user.id)
    
    for event in analytics_events:
        # Should use anonymous session ID, not user email
        assert "analytics-test@example.com" not in str(event)
        assert user.name not in str(event)
        # Device fingerprinting data should be hashed
        assert len(event.get("device_id", "")) <= 64  # Should be hashed, not raw

def test_eu_data_stays_in_eu():
    """Personal data of EU residents must be processed in EU data centers."""
    eu_user = create_test_user(country="DE", email="eu-user@example.com")
    
    # Verify data is stored in EU region
    user_data_location = get_data_storage_location(eu_user.id)
    
    EU_REGIONS = ["eu-west-1", "eu-central-1", "eu-north-1", "europe-west1"]
    assert user_data_location.region in EU_REGIONS, \
        f"EU user data stored in non-EU region: {user_data_location.region}"

def test_third_party_data_sharing_has_dpa():
    """Third-party processors must have Data Processing Agreements."""
    third_party_processors = [
        "stripe",      # Payment processor
        "sendgrid",    # Email service
        "intercom",    # Support chat
        "amplitude",   # Analytics
    ]
    
    dpa_registry = get_dpa_registry()
    
    for processor in third_party_processors:
        assert processor in dpa_registry, \
            f"No DPA found for {processor}"
        assert dpa_registry[processor].status == "active", \
            f"DPA for {processor} is not active"
        assert dpa_registry[processor].expiry > date.today(), \
            f"DPA for {processor} has expired"

def run_gdpr_compliance_checks():
    """Daily automated GDPR compliance validation."""
    checks = [
        ("Data Retention - Auto-delete old data", test_data_retention_policy),
        ("Consent - Marketing list compliance", test_no_unconsented_marketing),
        ("Erasure - Pending requests processed", test_erasure_requests_fulfilled),
        ("Encryption - PII fields encrypted", test_pii_fields_encrypted),
        ("Logging - No PII in logs", test_no_pii_in_logs),
    ]
    
    for check_name, check_fn in checks:
        try:
            check_fn()
            log_compliance_result(check_name, "PASS")
        except AssertionError as e:
            log_compliance_result(check_name, "FAIL", str(e))
            alert_dpo(check_name, str(e))